From 80ff72a0fef1a68092130370eee024acf4587c93 Mon Sep 17 00:00:00 2001 From: Efremenko Arhip Date: Sun, 26 Jul 2026 19:18:50 +0300 Subject: [PATCH] =?UTF-8?q?feat(coordinator):=20worker=20trust=20tiers=20(?= =?UTF-8?q?C1)=20=E2=80=94=20enroll=20volunteers,=20quarantine=20untrusted?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - migration 0012: workers.owner_id + trust_level (trusted/untrusted) - verifier/authctx read the JWT verified claim; IsTrusted() = admin||verified - /workers/register resolves trust from auth: service token or verified/admin JWT -> trusted; plain user JWT -> untrusted, tagged with owner_id - claim quarantines untrusted workers (no tasks) until quorum (C2) lands - unit tests for trust resolution, quarantine, and the verified claim Additive and backward compatible: shared-token workers stay trusted, so the existing worker flow and team tests are unchanged. Quorum verification (C2) is deferred. --- coordinator/internal/authctx/authctx.go | 10 ++- coordinator/internal/domain/worker.go | 28 ++++++-- .../internal/storage/postgres/worker_repo.go | 8 ++- coordinator/internal/token/verifier.go | 10 +-- coordinator/internal/token/verifier_test.go | 21 +++++- .../internal/transport/http/handlers.go | 19 +++++- .../internal/transport/http/middleware.go | 5 +- coordinator/internal/usecase/dto.go | 7 ++ coordinator/internal/usecase/task.go | 7 ++ coordinator/internal/usecase/usecase_test.go | 65 +++++++++++++++++++ coordinator/internal/usecase/worker.go | 7 ++ .../migrations/0012_worker_trust.down.sql | 8 +++ .../migrations/0012_worker_trust.up.sql | 18 +++++ 13 files changed, 195 insertions(+), 18 deletions(-) create mode 100644 coordinator/migrations/0012_worker_trust.down.sql create mode 100644 coordinator/migrations/0012_worker_trust.up.sql diff --git a/coordinator/internal/authctx/authctx.go b/coordinator/internal/authctx/authctx.go index 74bf86e..398cc38 100644 --- a/coordinator/internal/authctx/authctx.go +++ b/coordinator/internal/authctx/authctx.go @@ -15,13 +15,19 @@ import ( // Requester at all (From returns ok=false), which is how worker traffic and // legacy unauthenticated-user traffic stay owner-less. type Requester struct { - UserID uuid.UUID - Role string + UserID uuid.UUID + Role string + Verified bool } // IsAdmin reports whether the requester may act on any user's jobs. func (r Requester) IsAdmin() bool { return r.Role == "admin" } +// IsTrusted reports whether workers this requester registers produce results +// the coordinator accepts without quorum. Admins and verified contributors are +// trusted; a plain unverified user is not. +func (r Requester) IsTrusted() bool { return r.IsAdmin() || r.Verified } + type ctxKey struct{} // With returns a copy of ctx carrying r. diff --git a/coordinator/internal/domain/worker.go b/coordinator/internal/domain/worker.go index e7e0bf9..75ee27b 100644 --- a/coordinator/internal/domain/worker.go +++ b/coordinator/internal/domain/worker.go @@ -14,14 +14,30 @@ const ( WorkerOffline WorkerStatus = "offline" ) +// WorkerTrust says whether a worker's results are accepted directly or must +// clear quorum cross-checking. +type WorkerTrust string + +const ( + // WorkerTrusted — lab machine (shared token) or a verified/admin contributor. + WorkerTrusted WorkerTrust = "trusted" + // WorkerUntrusted — a plain enthusiast; results are quarantined until quorum. + WorkerUntrusted WorkerTrust = "untrusted" +) + // Worker is a registered process/machine allowed to claim tasks. Its // capabilities are the allowlisted workload names it can run; the coordinator // never hands it a task outside that set. type Worker struct { - ID uuid.UUID - Name string - Capabilities []string - Status WorkerStatus + ID uuid.UUID + Name string + Capabilities []string + Status WorkerStatus + // OwnerID is the userservice user who registered this worker; nil for a + // worker registered with the shared service token. + OwnerID *uuid.UUID + // TrustLevel decides whether this worker's results need quorum. + TrustLevel WorkerTrust LastHeartbeatAt time.Time CreatedAt time.Time UpdatedAt time.Time @@ -29,6 +45,9 @@ type Worker struct { // NewWorker registers a worker. A worker with no capabilities could never be // handed a task, so an empty set is rejected rather than silently stored. +// +// Trust defaults to WorkerTrusted (the shared-token lab worker); the caller +// overrides it for a volunteer registered through the userservice. func NewWorker(name string, capabilities []string, now time.Time) (*Worker, error) { if len(capabilities) == 0 { return nil, ErrInvalidInput @@ -38,6 +57,7 @@ func NewWorker(name string, capabilities []string, now time.Time) (*Worker, erro Name: name, Capabilities: capabilities, Status: WorkerOnline, + TrustLevel: WorkerTrusted, LastHeartbeatAt: now, CreatedAt: now, UpdatedAt: now, diff --git a/coordinator/internal/storage/postgres/worker_repo.go b/coordinator/internal/storage/postgres/worker_repo.go index 1df545e..5f75183 100644 --- a/coordinator/internal/storage/postgres/worker_repo.go +++ b/coordinator/internal/storage/postgres/worker_repo.go @@ -23,13 +23,13 @@ func NewWorkerRepo(pool *pgxpool.Pool) *WorkerRepo { return &WorkerRepo{pool: pool} } -var workerColumns = []string{"id", "name", "capabilities", "status", "last_heartbeat_at", "created_at", "updated_at"} +var workerColumns = []string{"id", "name", "capabilities", "status", "owner_id", "trust_level", "last_heartbeat_at", "created_at", "updated_at"} func (r *WorkerRepo) Insert(ctx context.Context, w *domain.Worker) error { sql, args, err := psql.Insert("workers"). Columns(workerColumns...). // capabilities is a jsonb column; pgx marshals the []string to a JSON array. - Values(w.ID, w.Name, w.Capabilities, string(w.Status), + Values(w.ID, w.Name, w.Capabilities, string(w.Status), w.OwnerID, string(w.TrustLevel), w.LastHeartbeatAt, w.CreatedAt, w.UpdatedAt). ToSql() if err != nil { @@ -95,11 +95,13 @@ func scanWorker(row pgx.Row) (*domain.Worker, error) { var ( w domain.Worker status string + trust string ) - if err := row.Scan(&w.ID, &w.Name, &w.Capabilities, &status, + if err := row.Scan(&w.ID, &w.Name, &w.Capabilities, &status, &w.OwnerID, &trust, &w.LastHeartbeatAt, &w.CreatedAt, &w.UpdatedAt); err != nil { return nil, err } w.Status = domain.WorkerStatus(status) + w.TrustLevel = domain.WorkerTrust(trust) return &w, nil } diff --git a/coordinator/internal/token/verifier.go b/coordinator/internal/token/verifier.go index a366dcb..93b78ee 100644 --- a/coordinator/internal/token/verifier.go +++ b/coordinator/internal/token/verifier.go @@ -13,8 +13,9 @@ import ( // Claims is the subset of a userservice token the coordinator cares about. type Claims struct { - UserID uuid.UUID - Role string + UserID uuid.UUID + Role string + Verified bool } // Verifier checks tokens against the shared HS256 secret. @@ -32,7 +33,8 @@ func NewVerifier(secret string) *Verifier { } type claims struct { - Role string `json:"role"` + Role string `json:"role"` + Verified bool `json:"verified"` jwt.RegisteredClaims } @@ -54,5 +56,5 @@ func (v *Verifier) Verify(raw string) (Claims, error) { if err != nil { return Claims{}, fmt.Errorf("token subject is not a uuid: %w", err) } - return Claims{UserID: id, Role: c.Role}, nil + return Claims{UserID: id, Role: c.Role, Verified: c.Verified}, nil } diff --git a/coordinator/internal/token/verifier_test.go b/coordinator/internal/token/verifier_test.go index 238d150..9c3b0f4 100644 --- a/coordinator/internal/token/verifier_test.go +++ b/coordinator/internal/token/verifier_test.go @@ -11,9 +11,15 @@ import ( const secret = "coordinator-verify-secret-32-bytes!!" func sign(t *testing.T, method jwt.SigningMethod, key any, sub, role string, exp time.Time) string { + t.Helper() + return signVerified(t, method, key, sub, role, false, exp) +} + +func signVerified(t *testing.T, method jwt.SigningMethod, key any, sub, role string, verified bool, exp time.Time) string { t.Helper() tok := jwt.NewWithClaims(method, claims{ - Role: role, + Role: role, + Verified: verified, RegisteredClaims: jwt.RegisteredClaims{ Subject: sub, ExpiresAt: jwt.NewNumericDate(exp), @@ -26,6 +32,19 @@ func sign(t *testing.T, method jwt.SigningMethod, key any, sub, role string, exp return raw } +func TestVerifyCarriesVerifiedClaim(t *testing.T) { + v := NewVerifier(secret) + raw := signVerified(t, jwt.SigningMethodHS256, []byte(secret), uuid.New().String(), "user", true, time.Now().Add(time.Hour)) + + claims, err := v.Verify(raw) + if err != nil { + t.Fatalf("verify: %v", err) + } + if !claims.Verified { + t.Error("verified claim not read from token") + } +} + func TestNewVerifierNilWhenNoSecret(t *testing.T) { if NewVerifier("") != nil { t.Error("empty secret must yield a nil verifier (auth disabled)") diff --git a/coordinator/internal/transport/http/handlers.go b/coordinator/internal/transport/http/handlers.go index e8c827b..b002cd6 100644 --- a/coordinator/internal/transport/http/handlers.go +++ b/coordinator/internal/transport/http/handlers.go @@ -12,6 +12,7 @@ import ( "github.com/google/uuid" + "github.com/emil28092005/SciMesh/coordinator/internal/authctx" "github.com/emil28092005/SciMesh/coordinator/internal/domain" "github.com/emil28092005/SciMesh/coordinator/internal/usecase" ) @@ -56,10 +57,24 @@ func (s *Server) handleRegister(w http.ResponseWriter, r *http.Request) { return } - worker, err := s.uc.RegisterWorker.Execute(ctx, usecase.RegisterWorkerInput{ + // Resolve the worker's trust tier from how the caller authenticated: + // - shared service token (no requester) -> trusted lab worker + // - verified/admin user JWT -> trusted volunteer + // - plain user JWT -> untrusted (quarantined) + in := usecase.RegisterWorkerInput{ Name: req.Name, Capabilities: req.Capabilities, - }) + TrustLevel: domain.WorkerTrusted, + } + if requester, ok := authctx.From(ctx); ok { + id := requester.UserID + in.OwnerID = &id + if !requester.IsTrusted() { + in.TrustLevel = domain.WorkerUntrusted + } + } + + worker, err := s.uc.RegisterWorker.Execute(ctx, in) if err != nil { s.writeError(w, r, err) return diff --git a/coordinator/internal/transport/http/middleware.go b/coordinator/internal/transport/http/middleware.go index bfa0dea..c304827 100644 --- a/coordinator/internal/transport/http/middleware.go +++ b/coordinator/internal/transport/http/middleware.go @@ -70,8 +70,9 @@ func withAuth(token string, verifier *tokenpkg.Verifier) func(http.Handler) http if verifier != nil && presented != "" { if claims, err := verifier.Verify(presented); err == nil { ctx := authctx.With(r.Context(), authctx.Requester{ - UserID: claims.UserID, - Role: claims.Role, + UserID: claims.UserID, + Role: claims.Role, + Verified: claims.Verified, }) next.ServeHTTP(w, r.WithContext(ctx)) return diff --git a/coordinator/internal/usecase/dto.go b/coordinator/internal/usecase/dto.go index 17e0498..2ab4fe0 100644 --- a/coordinator/internal/usecase/dto.go +++ b/coordinator/internal/usecase/dto.go @@ -4,6 +4,8 @@ import ( "io" "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" ) // Use-case boundary types. Adapters map their wire formats onto these, so the @@ -28,6 +30,11 @@ type ChunkInput struct { type RegisterWorkerInput struct { Name string Capabilities []string + // OwnerID is the userservice user registering this worker; nil for a + // shared-token registration. TrustLevel is resolved by the transport layer + // from how the caller authenticated. + OwnerID *uuid.UUID + TrustLevel domain.WorkerTrust } type ClaimTaskInput struct { diff --git a/coordinator/internal/usecase/task.go b/coordinator/internal/usecase/task.go index 1af5bf8..f724c07 100644 --- a/coordinator/internal/usecase/task.go +++ b/coordinator/internal/usecase/task.go @@ -51,6 +51,13 @@ func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.Cl if err != nil { return nil, err } + // C1 quarantine: an untrusted volunteer worker may register but receives + // no tasks, because there is not yet (until quorum, C2) any way to verify + // its results. Report an empty queue rather than an error, so its poller + // simply idles. + if worker.TrustLevel == domain.WorkerUntrusted { + return nil, nil + } // Never trust caller-supplied capabilities: registration is the durable // worker identity and its allowlist. workloads = worker.Capabilities diff --git a/coordinator/internal/usecase/usecase_test.go b/coordinator/internal/usecase/usecase_test.go index 2272c51..b1ed39d 100644 --- a/coordinator/internal/usecase/usecase_test.go +++ b/coordinator/internal/usecase/usecase_test.go @@ -265,6 +265,71 @@ func TestClaimEmptyQueueReturnsNil(t *testing.T) { } } +func TestRegisterWorkerDefaultsToTrusted(t *testing.T) { + h := newHarness() + // A shared-token registration carries no owner and no explicit trust. + w, err := h.register.Execute(ctx, usecase.RegisterWorkerInput{ + Name: "lab", Capabilities: []string{"w"}, + }) + if err != nil { + t.Fatal(err) + } + if w.TrustLevel != domain.WorkerTrusted { + t.Errorf("trust = %q, want trusted", w.TrustLevel) + } + if w.OwnerID != nil { + t.Errorf("owner = %v, want nil for a shared-token worker", w.OwnerID) + } +} + +func TestRegisterWorkerRecordsOwnerAndUntrusted(t *testing.T) { + h := newHarness() + owner := uuid.New() + w, err := h.register.Execute(ctx, usecase.RegisterWorkerInput{ + Name: "volunteer", Capabilities: []string{"w"}, + OwnerID: &owner, TrustLevel: domain.WorkerUntrusted, + }) + if err != nil { + t.Fatal(err) + } + if w.TrustLevel != domain.WorkerUntrusted { + t.Errorf("trust = %q, want untrusted", w.TrustLevel) + } + if w.OwnerID == nil || *w.OwnerID != owner { + t.Errorf("owner = %v, want %v", w.OwnerID, owner) + } +} + +func TestUntrustedWorkerIsQuarantinedFromClaims(t *testing.T) { + h := newHarness() + h.seedJob(t, "w", 1) // a task is waiting + owner := uuid.New() + worker, err := h.register.Execute(ctx, usecase.RegisterWorkerInput{ + Name: "volunteer", Capabilities: []string{"w"}, + OwnerID: &owner, TrustLevel: domain.WorkerUntrusted, + }) + if err != nil { + t.Fatal(err) + } + + // Even with a matching task available, an untrusted worker gets nothing: + // its results cannot be verified until quorum (C2) exists. + claimed, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{WorkerID: worker.ID.String()}) + if err != nil { + t.Fatalf("claim: %v", err) + } + if claimed != nil { + t.Error("untrusted worker must receive no task (quarantine)") + } + + // A trusted worker still drains the same queue. + trusted, _ := h.register.Execute(ctx, usecase.RegisterWorkerInput{Name: "lab", Capabilities: []string{"w"}}) + got, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{WorkerID: trusted.ID.String()}) + if err != nil || got == nil { + t.Fatalf("trusted claim = (%v, %v), want a task", got, err) + } +} + func TestClaimRequiresWorkerID(t *testing.T) { h := newHarness() if _, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{}); !errors.Is(err, domain.ErrInvalidInput) { diff --git a/coordinator/internal/usecase/worker.go b/coordinator/internal/usecase/worker.go index c8ccab2..572f8a2 100644 --- a/coordinator/internal/usecase/worker.go +++ b/coordinator/internal/usecase/worker.go @@ -22,6 +22,13 @@ func (uc *RegisterWorker) Execute(ctx context.Context, in RegisterWorkerInput) ( if err != nil { return nil, err } + w.OwnerID = in.OwnerID + // The transport layer resolves trust from the caller's credentials; fall + // back to the domain default (trusted) only when it was left unset, so a + // zero-value input never silently downgrades a shared-token worker. + if in.TrustLevel != "" { + w.TrustLevel = in.TrustLevel + } if err := uc.workers.Insert(ctx, w); err != nil { return nil, err } diff --git a/coordinator/migrations/0012_worker_trust.down.sql b/coordinator/migrations/0012_worker_trust.down.sql new file mode 100644 index 0000000..103c168 --- /dev/null +++ b/coordinator/migrations/0012_worker_trust.down.sql @@ -0,0 +1,8 @@ +BEGIN; + +DROP INDEX IF EXISTS ix_workers_owner; +ALTER TABLE workers DROP COLUMN IF EXISTS trust_level; +ALTER TABLE workers DROP COLUMN IF EXISTS owner_id; +DROP TYPE IF EXISTS worker_trust; + +COMMIT; diff --git a/coordinator/migrations/0012_worker_trust.up.sql b/coordinator/migrations/0012_worker_trust.up.sql new file mode 100644 index 0000000..36114fb --- /dev/null +++ b/coordinator/migrations/0012_worker_trust.up.sql @@ -0,0 +1,18 @@ +BEGIN; + +-- Whether a worker's results are accepted directly or must clear quorum. +-- 'trusted' — lab machine (shared token) or a verified/admin contributor. +-- 'untrusted' — a plain enthusiast; results are quarantined until quorum (C2). +CREATE TYPE worker_trust AS ENUM ('trusted', 'untrusted'); + +-- Who registered this worker (userservice user id, from the JWT sub). NULL for +-- workers registered with the shared service token. Not a foreign key: users +-- live in a separate service/database. +ALTER TABLE workers ADD COLUMN owner_id uuid; + +-- Existing rows were all shared-token lab workers, hence 'trusted'. +ALTER TABLE workers ADD COLUMN trust_level worker_trust NOT NULL DEFAULT 'trusted'; + +CREATE INDEX ix_workers_owner ON workers (owner_id); + +COMMIT;