feat(coordinator): worker trust tiers (C1) — enroll volunteers, quarantine untrusted

- 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.
This commit is contained in:
Efremenko Arhip
2026-07-26 19:18:50 +03:00
parent c6a66747eb
commit 80ff72a0fe
13 changed files with 195 additions and 18 deletions
+8 -2
View File
@@ -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.
+24 -4
View File
@@ -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,
@@ -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
}
+6 -4
View File
@@ -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
}
+20 -1
View File
@@ -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)")
@@ -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
@@ -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
+7
View File
@@ -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 {
+7
View File
@@ -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
@@ -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) {
+7
View File
@@ -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
}
@@ -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;
@@ -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;