From fa76133efc96daa9740a1030d273d338d22b3e72 Mon Sep 17 00:00:00 2001 From: Emil Date: Mon, 27 Jul 2026 22:23:08 +0300 Subject: [PATCH] Secure user worker operations --- PLAN.md | 4 +- coordinator/cmd/coordinator/main.go | 4 +- .../internal/transport/http/server_test.go | 4 +- coordinator/internal/usecase/artifact.go | 8 +++- coordinator/internal/usecase/task.go | 30 +++++++----- coordinator/internal/usecase/usecase_test.go | 36 ++++++++++++-- .../internal/usecase/worker_authorization.go | 33 +++++++++++++ docs/user-service-api-contract.md | 47 +++++++++++++++++++ users/README.md | 3 ++ 9 files changed, 147 insertions(+), 22 deletions(-) create mode 100644 coordinator/internal/usecase/worker_authorization.go create mode 100644 docs/user-service-api-contract.md diff --git a/PLAN.md b/PLAN.md index 91f5922..2e43ebc 100644 --- a/PLAN.md +++ b/PLAN.md @@ -872,7 +872,9 @@ workload logic into the service. **Acceptance criteria:** -- the service has a versioned, documented API and owns user identity data; +- the service has a versioned, documented API in + [`docs/user-service-api-contract.md`](docs/user-service-api-contract.md) and + owns user identity data; - credentials and authentication tokens are stored and handled securely; they are never logged or exposed to workers; - authenticated identity is propagated to coordinator requests through an diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index dd6e33d..61de190 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/coordinator/cmd/coordinator/main.go @@ -78,10 +78,10 @@ func run() error { RenewLease: usecase.NewRenewLease(taskRepo, workerRepo, tx, clk, cfg.LeaseDuration), CompleteTask: usecase.NewCompleteTask(taskRepo, jobRepo, artifactRepo, workerRepo, taskResultRepo, tx, clk, cfg.QuorumSize), ReduceJob: usecase.NewReduceJob(jobRepo, taskRepo, artifactRepo, blobStore, tx, clk), - FailTask: usecase.NewFailTask(taskRepo, jobRepo, tx, clk), + FailTask: usecase.NewFailTask(taskRepo, jobRepo, workerRepo, tx, clk), GetJobStatus: usecase.NewGetJobStatus(jobRepo, taskRepo), CancelJob: usecase.NewCancelJob(jobRepo, taskRepo, tx, clk), - UploadArtifact: usecase.NewUploadArtifact(taskRepo, artifactRepo, blobStore, tx, clk), + UploadArtifact: usecase.NewUploadArtifact(taskRepo, workerRepo, artifactRepo, blobStore, tx, clk), DownloadArtifact: usecase.NewDownloadArtifact(artifactRepo, blobStore), GetJobResult: usecase.NewGetJobResult(jobRepo, usecase.NewDownloadArtifact(artifactRepo, blobStore)), GetTaskInput: usecase.NewGetTaskInput(taskRepo, artifactRepo, blobStore), diff --git a/coordinator/internal/transport/http/server_test.go b/coordinator/internal/transport/http/server_test.go index 56cb049..3d9748f 100644 --- a/coordinator/internal/transport/http/server_test.go +++ b/coordinator/internal/transport/http/server_test.go @@ -52,11 +52,11 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur RenewLease: usecase.NewRenewLease(tasks, work, tx, clk, lease), CompleteTask: usecase.NewCompleteTask(tasks, jobs, arts, work, memstore.NewTaskResultRepo(), tx, clk, 2), ReduceJob: usecase.NewReduceJob(jobs, tasks, arts, blobs, tx, clk), - FailTask: usecase.NewFailTask(tasks, jobs, tx, clk), + FailTask: usecase.NewFailTask(tasks, jobs, work, tx, clk), GetJobStatus: usecase.NewGetJobStatus(jobs, tasks), GetJobResult: usecase.NewGetJobResult(jobs, downloadArtifact), CancelJob: usecase.NewCancelJob(jobs, tasks, tx, clk), - UploadArtifact: usecase.NewUploadArtifact(tasks, arts, blobs, tx, clk), + UploadArtifact: usecase.NewUploadArtifact(tasks, work, arts, blobs, tx, clk), DownloadArtifact: downloadArtifact, GetTaskInput: usecase.NewGetTaskInput(tasks, arts, blobs), Dashboard: usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, work, arts)), diff --git a/coordinator/internal/usecase/artifact.go b/coordinator/internal/usecase/artifact.go index 7a0187f..4c10657 100644 --- a/coordinator/internal/usecase/artifact.go +++ b/coordinator/internal/usecase/artifact.go @@ -12,18 +12,22 @@ import ( // UploadArtifact stores a worker's partial-result bytes and records the metadata. type UploadArtifact struct { tasks TaskRepository + workers WorkerRepository artifacts ArtifactRepository blobs BlobStore tx TxManager clk Clock } -func NewUploadArtifact(tasks TaskRepository, artifacts ArtifactRepository, +func NewUploadArtifact(tasks TaskRepository, workers WorkerRepository, artifacts ArtifactRepository, blobs BlobStore, tx TxManager, clk Clock) *UploadArtifact { - return &UploadArtifact{tasks: tasks, artifacts: artifacts, blobs: blobs, tx: tx, clk: clk} + return &UploadArtifact{tasks: tasks, workers: workers, artifacts: artifacts, blobs: blobs, tx: tx, clk: clk} } func (uc *UploadArtifact) Execute(ctx context.Context, in UploadArtifactInput) (*domain.Artifact, error) { + if err := authorizeWorkerOwner(ctx, uc.workers, in.WorkerID); err != nil { + return nil, err + } task, err := uc.tasks.Get(ctx, in.TaskID) if err != nil { return nil, err diff --git a/coordinator/internal/usecase/task.go b/coordinator/internal/usecase/task.go index ca4ca4b..9cb9c40 100644 --- a/coordinator/internal/usecase/task.go +++ b/coordinator/internal/usecase/task.go @@ -7,7 +7,6 @@ import ( "github.com/google/uuid" - "github.com/emil28092005/SciMesh/coordinator/internal/authctx" "github.com/emil28092005/SciMesh/coordinator/internal/domain" ) @@ -59,11 +58,8 @@ func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.Cl // tier would be read off a caller-supplied worker_id, letting anyone who // knows a trusted worker's id claim as it. A shared-token caller (no // requester) is a lab operator and may act as any worker. - if r, ok := authctx.From(ctx); ok { - if worker.OwnerID == nil || *worker.OwnerID != r.UserID { - // Don't disclose that another user's worker exists. - return nil, domain.ErrWorkerNotFound - } + if err := authorizeWorkerOwner(ctx, uc.workers, in.WorkerID); err != nil { + return nil, err } // An untrusted volunteer may claim, but never a chunk its owner has // already voted on — so quorum needs genuinely independent computations. @@ -126,6 +122,9 @@ func NewRenewLease(tasks TaskRepository, workers WorkerRepository, tx TxManager, // locked: two concurrent heartbeats must not interleave into a lost update. // Whether the caller may renew at all is decided by the entity, not here. func (uc *RenewLease) Execute(ctx context.Context, in RenewLeaseInput) (*domain.ClaimedTask, error) { + if err := authorizeWorkerOwner(ctx, uc.workers, in.WorkerID); err != nil { + return nil, err + } var claimed domain.ClaimedTask err := uc.tx.WithinTx(ctx, func(ctx context.Context) error { @@ -186,6 +185,9 @@ func NewCompleteTask(tasks TaskRepository, jobs JobRepository, artifacts Artifac // Lease ownership, staleness, and idempotent replays are all decided by // Task.CompleteWith; this use case only orchestrates. func (uc *CompleteTask) Execute(ctx context.Context, in CompleteTaskInput) (*domain.Task, error) { + if err := authorizeWorkerOwner(ctx, uc.workers, in.WorkerID); err != nil { + return nil, err + } var out *domain.Task err := uc.tx.WithinTx(ctx, func(ctx context.Context) error { @@ -317,19 +319,23 @@ func (uc *CompleteTask) verifyResultArtifact(ctx context.Context, taskID uuid.UU // --- FailTask ------------------------------------------------------------ type FailTask struct { - tasks TaskRepository - jobs JobRepository - tx TxManager - clock Clock + tasks TaskRepository + jobs JobRepository + workers WorkerRepository + tx TxManager + clock Clock } -func NewFailTask(tasks TaskRepository, jobs JobRepository, tx TxManager, clock Clock) *FailTask { - return &FailTask{tasks: tasks, jobs: jobs, tx: tx, clock: clock} +func NewFailTask(tasks TaskRepository, jobs JobRepository, workers WorkerRepository, tx TxManager, clock Clock) *FailTask { + return &FailTask{tasks: tasks, jobs: jobs, workers: workers, tx: tx, clock: clock} } // Execute delegates the requeue-or-terminate decision to Task.Fail, then keeps // the parent job's status consistent in the same transaction. func (uc *FailTask) Execute(ctx context.Context, in FailTaskInput) (*domain.Task, error) { + if err := authorizeWorkerOwner(ctx, uc.workers, in.WorkerID); err != nil { + return nil, err + } var out *domain.Task err := uc.tx.WithinTx(ctx, func(ctx context.Context) error { diff --git a/coordinator/internal/usecase/usecase_test.go b/coordinator/internal/usecase/usecase_test.go index 246c452..b64111f 100644 --- a/coordinator/internal/usecase/usecase_test.go +++ b/coordinator/internal/usecase/usecase_test.go @@ -77,11 +77,11 @@ func newHarness() *harness { h.claim = usecase.NewClaimTask(h.tasks, h.jobs, h.work, tx, h.clk, lease) h.renew = usecase.NewRenewLease(h.tasks, h.work, tx, h.clk, lease) h.complete = usecase.NewCompleteTask(h.tasks, h.jobs, h.arts, h.work, h.taskResults, tx, h.clk, 2) - h.fail = usecase.NewFailTask(h.tasks, h.jobs, tx, h.clk) + h.fail = usecase.NewFailTask(h.tasks, h.jobs, h.work, tx, h.clk) h.status = usecase.NewGetJobStatus(h.jobs, h.tasks) h.results = usecase.NewListResults(h.tasks) h.register = usecase.NewRegisterWorker(h.work, h.clk) - h.uploadArt = usecase.NewUploadArtifact(h.tasks, h.arts, h.blobs, tx, h.clk) + h.uploadArt = usecase.NewUploadArtifact(h.tasks, h.work, h.arts, h.blobs, tx, h.clk) h.downloadArt = usecase.NewDownloadArtifact(h.arts, h.blobs) h.getInput = usecase.NewGetTaskInput(h.tasks, h.arts, h.blobs) h.expire = usecase.NewExpireLeases(h.tasks, h.jobs, tx, h.clk) @@ -318,6 +318,36 @@ func TestJWTCallerCannotClaimAsAnotherUsersWorker(t *testing.T) { } } +func TestJWTCallerCannotMutateAnotherUsersWorkerLease(t *testing.T) { + h := newHarness() + h.seedJob(t, "w", 1) + victimOwner := uuid.New() + victim, err := h.register.Execute(ctx, usecase.RegisterWorkerInput{ + Name: "victim", Capabilities: []string{"w"}, OwnerID: &victimOwner, TrustLevel: domain.WorkerTrusted, + }) + if err != nil { + t.Fatal(err) + } + claimed, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{WorkerID: victim.ID.String()}) + if err != nil || claimed == nil { + t.Fatalf("claim = (%v, %v)", claimed, err) + } + attacker := authctx.With(ctx, authctx.Requester{UserID: uuid.New(), Role: "user"}) + + if _, err := h.renew.Execute(attacker, usecase.RenewLeaseInput{TaskID: claimed.TaskID, WorkerID: victim.ID.String(), Attempt: claimed.Attempt}); !errors.Is(err, domain.ErrWorkerNotFound) { + t.Errorf("foreign heartbeat err = %v, want ErrWorkerNotFound", err) + } + if _, err := h.fail.Execute(attacker, usecase.FailTaskInput{TaskID: claimed.TaskID, WorkerID: victim.ID.String(), Attempt: claimed.Attempt, ErrorCode: "x"}); !errors.Is(err, domain.ErrWorkerNotFound) { + t.Errorf("foreign failure err = %v, want ErrWorkerNotFound", err) + } + if _, err := h.uploadArt.Execute(attacker, usecase.UploadArtifactInput{TaskID: claimed.TaskID, WorkerID: victim.ID.String(), Attempt: claimed.Attempt, Filename: "x.csv", ContentType: "text/csv", Body: strings.NewReader("x")}); !errors.Is(err, domain.ErrWorkerNotFound) { + t.Errorf("foreign upload err = %v, want ErrWorkerNotFound", err) + } + if _, err := h.complete.Execute(attacker, usecase.CompleteTaskInput{TaskID: claimed.TaskID, WorkerID: victim.ID.String(), Attempt: claimed.Attempt, ResultArtifactID: uuid.New()}); !errors.Is(err, domain.ErrWorkerNotFound) { + t.Errorf("foreign result err = %v, want ErrWorkerNotFound", err) + } +} + func TestJWTCallerClaimsAsOwnTrustedWorker(t *testing.T) { h := newHarness() h.seedJob(t, "w", 1) @@ -551,7 +581,7 @@ func TestUploadRejectsLeaseThatExpiresDuringStreaming(t *testing.T) { h.seedJob(t, "w", 1) taskID, attempt := h.leaseOne(t, "w1", "w") h.uploadArt = usecase.NewUploadArtifact( - h.tasks, h.arts, expiringBlobStore{BlobStore: h.blobs, clock: h.clk}, memstore.Tx{}, h.clk, + h.tasks, h.work, h.arts, expiringBlobStore{BlobStore: h.blobs, clock: h.clk}, memstore.Tx{}, h.clk, ) _, err := h.uploadArt.Execute(ctx, usecase.UploadArtifactInput{ diff --git a/coordinator/internal/usecase/worker_authorization.go b/coordinator/internal/usecase/worker_authorization.go new file mode 100644 index 0000000..258e889 --- /dev/null +++ b/coordinator/internal/usecase/worker_authorization.go @@ -0,0 +1,33 @@ +package usecase + +import ( + "context" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/authctx" + "github.com/emil28092005/SciMesh/coordinator/internal/domain" +) + +// authorizeWorkerOwner binds a JWT-authenticated requester to a worker. The +// shared coordinator token intentionally has no requester and retains its +// existing operator privileges. +func authorizeWorkerOwner(ctx context.Context, workers WorkerRepository, workerID string) error { + requester, ok := authctx.From(ctx) + if !ok { + return nil + } + id, err := uuid.Parse(workerID) + if err != nil { + return domain.ErrWorkerNotFound + } + worker, err := workers.Get(ctx, id) + if err != nil { + return err + } + if worker.OwnerID == nil || *worker.OwnerID != requester.UserID { + // Mask ownership and existence from another user. + return domain.ErrWorkerNotFound + } + return nil +} diff --git a/docs/user-service-api-contract.md b/docs/user-service-api-contract.md new file mode 100644 index 0000000..91cd3c9 --- /dev/null +++ b/docs/user-service-api-contract.md @@ -0,0 +1,47 @@ +# SciMesh User Service API contract (v1) + +**Status:** `v1`. The User Service owns user accounts and issues access tokens. +The coordinator never receives user passwords and never accesses the User +Service database. + +## Authentication boundary + +- User Service signs access tokens; coordinator verifies them before accepting + user-scoped requests. +- Tokens contain a UUID `sub`, `role` (`user` or `admin`), `verified`, `iat`, + and `exp` claims. +- A user-authenticated caller may operate only workers whose `owner_id` equals + `sub`. This applies to claim, heartbeat, result, failure, and artifact upload. +- Worker traffic authenticated with the coordinator's shared worker token has + no user identity and remains an operator-only compatibility path. +- Role or verification changes take effect when the access token is renewed. + Deployments needing immediate revocation must use a short token lifetime or a + revocation mechanism before enabling volunteer-worker trust. + +## Endpoints + +All JSON request bodies reject unknown fields and are size-limited. Error +responses are JSON with a stable `error` value and request ID. + +| Method | Path | Auth | Success | +| --- | --- | --- | --- | +| `GET` | `/health` | none | `200 {"status":"ok"}` | +| `POST` | `/register` | none | `201` user object | +| `POST` | `/login` | none | `200` user object and access token | +| `GET` | `/me` | Bearer access token | `200` current user | +| `POST` | `/users/{id}/verify` | Bearer admin token | `204` | +| `POST` | `/users/{id}/unverify` | Bearer admin token | `204` | +| `POST` | `/users/{id}/promote` | Bearer admin token | `204` | +| `POST` | `/users/{id}/demote` | Bearer admin token | `204` | + +`POST /register` accepts `{ "email": string, "password": string }` and +always creates role `user` with `verified: false`. `POST /login` accepts the +same shape and returns `{ "token": string, "user": User }`. Password hashes, +JWT signing material, and raw tokens must never be logged. + +## Coordinator integration tests + +The coordinator must test that a JWT user cannot claim or mutate another +user's worker lease, including heartbeat, failure, result, and artifact upload. +Job and artifact access is restricted to the job owner unless the caller has +the admin role. diff --git a/users/README.md b/users/README.md index 4a7efa8..83f5d20 100644 --- a/users/README.md +++ b/users/README.md @@ -5,6 +5,9 @@ and issues the JWTs the coordinator trusts. It is a **separate bounded context** from the coordinator: its own database, its own binary. The only thing shared between the two services is the JWT signing secret. +The versioned external contract is +[`docs/user-service-api-contract.md`](../docs/user-service-api-contract.md). + Built as a modular monolith following Clean Architecture — one binary, four layers, dependencies pointing strictly inward: