From b8d7e56655158e434f1248b008f0963fcefafdce Mon Sep 17 00:00:00 2001 From: safayavatsal Date: Mon, 18 May 2026 15:49:55 +0530 Subject: [PATCH 1/2] Add Decision-Rights Matrix and Constraint Catalog binding (#59) - Migration 014: append-only decision_rights_matrix (DB trigger blocks UPDATE/DELETE), constraint_catalog_versions (Hash can repeat across rows on 24h re-sign), and drm_hash + constraint_catalog_hash columns on issued_credentials with partial indexes for drift detection. - Domain types: DRMDocument/DecisionRightsMatrix (+ ErrDRMUnauthorized and ErrDRMInvalid sentinels), ConstraintCatalogVersion, and SignalTypePolicyDrift. - GovernanceService: deterministic canonical-JSON hashing (sorted keys recursively), DRM authorization with exact and trailing-* SPIFFE pattern matching, catalog publish/re-sign with ES256 signing over sha256(hash||"|"||signed_at). policy_drift signals fan out from PublishDRM/PublishCatalog on hash change. - CatalogSignerWorker re-signs each tenant's active catalog every 24h (preserves Hash, rewrites SignedAt+Signature so outstanding tokens stay valid). - OAuthService.tokenExchange runs the DRM authorization gate before IssueCredential and embeds drm_version/drm_hash/ constraint_catalog_version/constraint_catalog_hash claims. authorization_code embeds the claims for audit but skips the gate (consent is not delegation). All four claim names added to reservedClaims; Introspect passes them through. - Admin endpoints under /governance/decision-rights-matrix and /governance/constraint-catalog. Routes self-suppress when governanceSvc is nil. Every code path is no-op for tenants that haven't published a DRM/catalog row -- pre-#59 flows are unchanged. - Integration tests: happy path with claim assertions, DRM-deny path, and no-config backward-compat path. Each test uses a fresh tenant to keep the append-only DRM rows from leaking across the run. --- domain/constraint_catalog.go | 33 ++ domain/credential.go | 8 + domain/decision_rights_matrix.go | 85 +++++ domain/signal.go | 8 +- internal/handler/governance.go | 188 ++++++++++ internal/handler/routes.go | 4 + internal/service/credential.go | 51 ++- internal/service/governance.go | 326 ++++++++++++++++++ internal/service/oauth.go | 133 ++++++- internal/store/postgres/constraint_catalog.go | 72 ++++ internal/store/postgres/credential.go | 31 ++ .../store/postgres/decision_rights_matrix.go | 79 +++++ internal/worker/catalog_signer.go | 69 ++++ migrations/014_governance_artifacts.down.sql | 18 + migrations/014_governance_artifacts.up.sql | 82 +++++ server.go | 24 +- tests/integration/governance_test.go | 226 ++++++++++++ 17 files changed, 1404 insertions(+), 33 deletions(-) create mode 100644 domain/constraint_catalog.go create mode 100644 domain/decision_rights_matrix.go create mode 100644 internal/handler/governance.go create mode 100644 internal/service/governance.go create mode 100644 internal/store/postgres/constraint_catalog.go create mode 100644 internal/store/postgres/decision_rights_matrix.go create mode 100644 internal/worker/catalog_signer.go create mode 100644 migrations/014_governance_artifacts.down.sql create mode 100644 migrations/014_governance_artifacts.up.sql create mode 100644 tests/integration/governance_test.go diff --git a/domain/constraint_catalog.go b/domain/constraint_catalog.go new file mode 100644 index 00000000..1ad52168 --- /dev/null +++ b/domain/constraint_catalog.go @@ -0,0 +1,33 @@ +package domain + +import ( + "encoding/json" + "time" + + "github.com/uptrace/bun" +) + +// ConstraintCatalogVersion is the persisted, ES256-signed snapshot of the +// active downstream policy set (e.g. Cedar policies enforced in Shield). +// ZeroID does not parse the document — it hashes the canonical bytes, +// signs them, and rewrites the SignedAt every 24h so consumers of the +// hash claim can detect a stale/replayed catalog. +// +// Multiple rows can share the same Hash (re-sign of unchanged content), +// distinguished by SignedAt. Only Hash is embedded in tokens, so a +// re-sign with identical Hash leaves outstanding tokens valid. +type ConstraintCatalogVersion struct { + bun.BaseModel `bun:"table:constraint_catalog_versions,alias:ccv"` + + ID string `bun:"id,pk,type:uuid" json:"id"` + AccountID string `bun:"account_id,type:varchar(255)" json:"account_id"` + ProjectID string `bun:"project_id,type:varchar(255)" json:"project_id"` + Version string `bun:"version,type:varchar(64)" json:"version"` + EffectiveAt time.Time `bun:"effective_at" json:"effective_at"` + Document json.RawMessage `bun:"document,type:jsonb" json:"document"` + Hash string `bun:"hash,type:varchar(80)" json:"hash"` + SignedAt time.Time `bun:"signed_at,nullzero,notnull,default:current_timestamp" json:"signed_at"` + Signature string `bun:"signature,type:text" json:"signature"` + SigningKeyID string `bun:"signing_key_id,type:varchar(255)" json:"signing_key_id"` + CreatedAt time.Time `bun:"created_at,nullzero,notnull,default:current_timestamp" json:"created_at"` +} diff --git a/domain/credential.go b/domain/credential.go index 3860dce9..b422e3bc 100644 --- a/domain/credential.go +++ b/domain/credential.go @@ -82,4 +82,12 @@ type IssuedCredential struct { ParentJTI string `bun:"parent_jti,type:varchar(255)" json:"parent_jti,omitempty"` // DelegatedByWIMSEURI records the orchestrator that delegated authority (RFC 8693 token_exchange). DelegatedByWIMSEURI string `bun:"delegated_by_wimse_uri,type:text" json:"delegated_by_wimse_uri,omitempty"` + // DRMHash binds the credential to the SHA-256 of the DRM document + // active at issuance (issue #59). Empty when no DRM is configured for + // the tenant. Indexed for policy_drift detection. + DRMHash string `bun:"drm_hash,type:varchar(80)" json:"drm_hash,omitempty"` + // ConstraintCatalogHash binds the credential to the SHA-256 of the + // Constraint Catalog document active at issuance (issue #59). Empty + // when no catalog is configured for the tenant. + ConstraintCatalogHash string `bun:"constraint_catalog_hash,type:varchar(80)" json:"constraint_catalog_hash,omitempty"` } diff --git a/domain/decision_rights_matrix.go b/domain/decision_rights_matrix.go new file mode 100644 index 00000000..d2e9af21 --- /dev/null +++ b/domain/decision_rights_matrix.go @@ -0,0 +1,85 @@ +package domain + +import ( + "errors" + "time" + + "github.com/uptrace/bun" +) + +// ErrDRMUnauthorized signals that the requested delegation (from→to pair, +// scopes, or resource set) is not permitted by the active DRM. Wrapped so +// callers can use errors.Is at the OAuth boundary to translate into an +// invalid_grant response (RFC 8693 §2.2.2). +var ErrDRMUnauthorized = errors.New("delegation not authorized by active decision-rights matrix") + +// ErrDRMInvalid indicates the DRM document failed schema validation at +// admission time (missing version, empty allowed_delegations, malformed +// SPIFFE pattern). Surfaces as 400 from the admin handler. +var ErrDRMInvalid = errors.New("invalid decision-rights matrix document") + +// DRMAllowedDelegation is one row of the DRM authorization table. +// +// `From` and `To` are SPIFFE URI patterns (`spiffe://domain/.../*` style). +// `Resources` and `Conditions` are kept as opaque maps/strings — the DRM +// is an authorization fence for *which delegations exist*, not a full +// policy engine; Cedar/OPA live downstream in Shield for per-request +// authorization. +type DRMAllowedDelegation struct { + From string `json:"from"` + To string `json:"to"` + Resources []string `json:"resources,omitempty"` + Conditions map[string]any `json:"conditions,omitempty"` +} + +// DRMDocument is the canonical wire/storage shape of a DRM. The Hash +// stored alongside DecisionRightsMatrix is computed over the canonical +// JSON encoding (sorted keys) of this struct. +type DRMDocument struct { + Version string `json:"version"` + EffectiveAt time.Time `json:"effective_at"` + ExpiresAt *time.Time `json:"expires_at,omitempty"` + AllowedDelegations []DRMAllowedDelegation `json:"allowed_delegations"` +} + +// Validate enforces the minimum schema requirements: version present, +// effective_at set, at least one allowed_delegation, and every delegation +// has a non-empty from/to. We deliberately do not validate the SPIFFE +// pattern syntax here — the governance service does that during the +// authorization check so the failure mode is consistent across writes +// and reads. +func (d DRMDocument) Validate() error { + if d.Version == "" { + return errors.New("drm: version is required") + } + if d.EffectiveAt.IsZero() { + return errors.New("drm: effective_at is required") + } + if len(d.AllowedDelegations) == 0 { + return errors.New("drm: allowed_delegations must not be empty") + } + for i, rule := range d.AllowedDelegations { + if rule.From == "" || rule.To == "" { + return errors.New("drm: allowed_delegations: from and to are required") + } + _ = i + } + return nil +} + +// DecisionRightsMatrix is the persisted DRM row. Rows are immutable — +// the database trigger drm_block_mutation refuses UPDATE/DELETE. A new +// version is published by inserting a new row. +type DecisionRightsMatrix struct { + bun.BaseModel `bun:"table:decision_rights_matrix,alias:drm"` + + ID string `bun:"id,pk,type:uuid" json:"id"` + AccountID string `bun:"account_id,type:varchar(255)" json:"account_id"` + ProjectID string `bun:"project_id,type:varchar(255)" json:"project_id"` + Version string `bun:"version,type:varchar(64)" json:"version"` + EffectiveAt time.Time `bun:"effective_at" json:"effective_at"` + ExpiresAt *time.Time `bun:"expires_at" json:"expires_at,omitempty"` + Document DRMDocument `bun:"document,type:jsonb" json:"document"` + Hash string `bun:"hash,type:varchar(80)" json:"hash"` + CreatedAt time.Time `bun:"created_at,nullzero,notnull,default:current_timestamp" json:"created_at"` +} diff --git a/domain/signal.go b/domain/signal.go index 81fe8a51..3ffa8898 100644 --- a/domain/signal.go +++ b/domain/signal.go @@ -17,6 +17,11 @@ const ( SignalTypePolicyViolation SignalType = "policy_violation" SignalTypeRetirement SignalType = "retirement" SignalTypeOwnerChange SignalType = "owner_change" + // SignalTypePolicyDrift is emitted when the active DRM or Constraint + // Catalog hash diverges from the hash bound into outstanding tokens. + // Enforcement points use this signal to schedule re-evaluation on + // next use rather than immediate revocation — see issue #59. + SignalTypePolicyDrift SignalType = "policy_drift" ) // SignalSeverity indicates the severity level of a CAE signal. @@ -42,7 +47,8 @@ func (s SignalSeverity) Valid() bool { func (t SignalType) Valid() bool { switch t { case SignalTypeCredentialChange, SignalTypeSessionRevoked, SignalTypeIPChange, - SignalTypeAnomalousBehavior, SignalTypePolicyViolation, SignalTypeRetirement, SignalTypeOwnerChange: + SignalTypeAnomalousBehavior, SignalTypePolicyViolation, SignalTypeRetirement, SignalTypeOwnerChange, + SignalTypePolicyDrift: return true } return false diff --git a/internal/handler/governance.go b/internal/handler/governance.go new file mode 100644 index 00000000..25f6cf73 --- /dev/null +++ b/internal/handler/governance.go @@ -0,0 +1,188 @@ +package handler + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "time" + + "github.com/danielgtaylor/huma/v2" + "github.com/rs/zerolog/log" + + "github.com/highflame-ai/zeroid/domain" + internalMiddleware "github.com/highflame-ai/zeroid/internal/middleware" +) + +// ── DRM types ──────────────────────────────────────────────────────────────── + +type PublishDRMInput struct { + Body struct { + Version string `json:"version" required:"true" doc:"Semver version string"` + EffectiveAt time.Time `json:"effective_at" required:"true" doc:"When this DRM becomes active"` + ExpiresAt *time.Time `json:"expires_at,omitempty" doc:"When this DRM stops being active (optional)"` + AllowedDelegations []domain.DRMAllowedDelegation `json:"allowed_delegations" required:"true" minItems:"1" doc:"Permitted delegation rules"` + } +} + +type DRMOutput struct { + Body *domain.DecisionRightsMatrix +} + +type DRMListOutput struct { + Body struct { + DecisionRightsMatrix []*domain.DecisionRightsMatrix `json:"decision_rights_matrix"` + Total int `json:"total"` + } +} + +type DRMIDInput struct { + ID string `path:"id" doc:"DRM row UUID"` +} + +// ── Constraint Catalog types ───────────────────────────────────────────────── + +type PublishCatalogInput struct { + Body struct { + Version string `json:"version" required:"true" doc:"Version identifier (e.g. ISO 8601 timestamp)"` + EffectiveAt time.Time `json:"effective_at" required:"true" doc:"When this catalog becomes active"` + Document json.RawMessage `json:"document" required:"true" doc:"Opaque policy document (ZeroID hashes + signs, does not parse)"` + } +} + +type CatalogOutput struct { + Body *domain.ConstraintCatalogVersion +} + +// ── Routes ─────────────────────────────────────────────────────────────────── + +func (a *API) registerGovernanceRoutes(api huma.API) { + if a.governanceSvc == nil { + // Governance binding is not configured for this deployment — skip + // route registration so the OpenAPI surface only exposes the + // endpoints that will actually work. + return + } + + huma.Register(api, huma.Operation{ + OperationID: "publish-drm", + Method: http.MethodPost, + Path: "/governance/decision-rights-matrix", + Summary: "Publish a new Decision-Rights Matrix", + Tags: []string{"Governance"}, + DefaultStatus: http.StatusCreated, + }, a.publishDRMOp) + + huma.Register(api, huma.Operation{ + OperationID: "get-active-drm", + Method: http.MethodGet, + Path: "/governance/decision-rights-matrix/active", + Summary: "Get the currently active DRM", + Tags: []string{"Governance"}, + }, a.getActiveDRMOp) + + huma.Register(api, huma.Operation{ + OperationID: "list-drm", + Method: http.MethodGet, + Path: "/governance/decision-rights-matrix", + Summary: "List DRM version history", + Tags: []string{"Governance"}, + }, a.listDRMOp) + + huma.Register(api, huma.Operation{ + OperationID: "publish-constraint-catalog", + Method: http.MethodPost, + Path: "/governance/constraint-catalog", + Summary: "Publish a new Constraint Catalog version", + Tags: []string{"Governance"}, + DefaultStatus: http.StatusCreated, + }, a.publishCatalogOp) + + huma.Register(api, huma.Operation{ + OperationID: "get-active-catalog", + Method: http.MethodGet, + Path: "/governance/constraint-catalog/active", + Summary: "Get the most recently signed Constraint Catalog row", + Tags: []string{"Governance"}, + }, a.getActiveCatalogOp) +} + +func (a *API) publishDRMOp(ctx context.Context, input *PublishDRMInput) (*DRMOutput, error) { + tenant, err := internalMiddleware.GetTenant(ctx) + if err != nil { + return nil, huma.Error401Unauthorized("missing tenant context") + } + doc := domain.DRMDocument{ + Version: input.Body.Version, + EffectiveAt: input.Body.EffectiveAt, + ExpiresAt: input.Body.ExpiresAt, + AllowedDelegations: input.Body.AllowedDelegations, + } + row, err := a.governanceSvc.PublishDRM(ctx, tenant.AccountID, tenant.ProjectID, doc) + if err != nil { + if errors.Is(err, domain.ErrDRMInvalid) { + return nil, huma.Error400BadRequest(err.Error()) + } + log.Error().Err(err).Msg("publish DRM failed") + return nil, huma.Error500InternalServerError("failed to publish DRM") + } + return &DRMOutput{Body: row}, nil +} + +func (a *API) getActiveDRMOp(ctx context.Context, _ *struct{}) (*DRMOutput, error) { + tenant, err := internalMiddleware.GetTenant(ctx) + if err != nil { + return nil, huma.Error401Unauthorized("missing tenant context") + } + row, err := a.governanceSvc.GetActiveDRM(ctx, tenant.AccountID, tenant.ProjectID) + if err != nil { + return nil, huma.Error500InternalServerError("failed to get active DRM") + } + if row == nil { + return nil, huma.Error404NotFound("no active DRM") + } + return &DRMOutput{Body: row}, nil +} + +func (a *API) listDRMOp(ctx context.Context, _ *struct{}) (*DRMListOutput, error) { + tenant, err := internalMiddleware.GetTenant(ctx) + if err != nil { + return nil, huma.Error401Unauthorized("missing tenant context") + } + rows, err := a.governanceSvc.ListDRM(ctx, tenant.AccountID, tenant.ProjectID) + if err != nil { + return nil, huma.Error500InternalServerError("failed to list DRMs") + } + out := &DRMListOutput{} + out.Body.DecisionRightsMatrix = rows + out.Body.Total = len(rows) + return out, nil +} + +func (a *API) publishCatalogOp(ctx context.Context, input *PublishCatalogInput) (*CatalogOutput, error) { + tenant, err := internalMiddleware.GetTenant(ctx) + if err != nil { + return nil, huma.Error401Unauthorized("missing tenant context") + } + row, err := a.governanceSvc.PublishCatalog(ctx, tenant.AccountID, tenant.ProjectID, input.Body.Version, input.Body.EffectiveAt, input.Body.Document) + if err != nil { + log.Error().Err(err).Msg("publish catalog failed") + return nil, huma.Error500InternalServerError("failed to publish catalog") + } + return &CatalogOutput{Body: row}, nil +} + +func (a *API) getActiveCatalogOp(ctx context.Context, _ *struct{}) (*CatalogOutput, error) { + tenant, err := internalMiddleware.GetTenant(ctx) + if err != nil { + return nil, huma.Error401Unauthorized("missing tenant context") + } + row, err := a.governanceSvc.GetActiveCatalog(ctx, tenant.AccountID, tenant.ProjectID) + if err != nil { + return nil, huma.Error500InternalServerError("failed to get active catalog") + } + if row == nil { + return nil, huma.Error404NotFound("no active catalog") + } + return &CatalogOutput{Body: row}, nil +} diff --git a/internal/handler/routes.go b/internal/handler/routes.go index 4e4f6fa9..6a8ecf95 100644 --- a/internal/handler/routes.go +++ b/internal/handler/routes.go @@ -31,6 +31,7 @@ type API struct { apiKeySvc *service.APIKeyService agentSvc *service.AgentService auditSvc *service.AuditService + governanceSvc *service.GovernanceService jwksSvc *signing.JWKSService db *bun.DB issuer string @@ -51,6 +52,7 @@ func NewAPI( apiKeySvc *service.APIKeyService, agentSvc *service.AgentService, auditSvc *service.AuditService, + governanceSvc *service.GovernanceService, jwksSvc *signing.JWKSService, db *bun.DB, issuer, baseURL string, @@ -67,6 +69,7 @@ func NewAPI( apiKeySvc: apiKeySvc, agentSvc: agentSvc, auditSvc: auditSvc, + governanceSvc: governanceSvc, jwksSvc: jwksSvc, db: db, issuer: issuer, @@ -116,6 +119,7 @@ func (a *API) RegisterAdmin(api huma.API, router chi.Router) { a.registerSignalRoutes(api, router) a.registerProofVerifyRoute(api) a.registerAuditRoutes(api) + a.registerGovernanceRoutes(api) } // RegisterAgentAuth registers endpoints requiring agent-auth middleware (proof generation). diff --git a/internal/service/credential.go b/internal/service/credential.go index c0b713aa..8e4f6396 100644 --- a/internal/service/credential.go +++ b/internal/service/credential.go @@ -94,6 +94,15 @@ type IssueRequest struct { // CustomClaims allows callers to add arbitrary key-value pairs to the JWT. // This is the extensibility hook for deployment-specific claims. CustomClaims map[string]any + // Governance binding (issue #59). When non-empty, these values are + // embedded as reserved claims (drm_*/constraint_catalog_*) on the + // issued JWT and persisted on the credential row for policy_drift + // signal fan-out. Set by OAuthService.tokenExchange and + // authorizationCode after a successful DRM authorization check. + DRMVersion string + DRMHash string + ConstraintCatalogVer string + ConstraintCatalogHash string } // ErrScopesNotAllowed is returned when one or more requested scopes are not in the identity's AllowedScopes list. @@ -312,6 +321,18 @@ func (s *CredentialService) IssueCredential(ctx context.Context, req IssueReques _ = token.Set(k, v) } + // Governance binding (issue #59). Set after CustomClaims so deployer + // hooks cannot spoof these values — OAuthService also rejects them + // at the reservedClaims gate, but this is defence-in-depth. + if req.DRMHash != "" { + _ = token.Set("drm_hash", req.DRMHash) + _ = token.Set("drm_version", req.DRMVersion) + } + if req.ConstraintCatalogHash != "" { + _ = token.Set("constraint_catalog_hash", req.ConstraintCatalogHash) + _ = token.Set("constraint_catalog_version", req.ConstraintCatalogVer) + } + // RFC 8693 "act" claim — two use cases: // 1. NHI delegation: orchestrator delegates to sub-agent. act.sub = orchestrator WIMSE URI. // 2. User context: NHI acts on behalf of an end user. act.sub = user ID. @@ -341,20 +362,22 @@ func (s *CredentialService) IssueCredential(ctx context.Context, req IssueReques // Persist credential record cred := &domain.IssuedCredential{ - ID: uuid.New().String(), - IdentityID: stringPtrOrNil(req.Identity.ID), - AccountID: req.Identity.AccountID, - ProjectID: req.Identity.ProjectID, - JTI: jti, - Subject: req.Identity.WIMSEURI, - IssuedAt: now, - ExpiresAt: expiresAt, - TTLSeconds: ttl, - Scopes: coalesceScopeSlice(req.Scopes), - GrantType: req.GrantType, - DelegationDepth: req.DelegationDepth, - ParentJTI: req.ParentJTI, - DelegatedByWIMSEURI: req.DelegatedBy, + ID: uuid.New().String(), + IdentityID: stringPtrOrNil(req.Identity.ID), + AccountID: req.Identity.AccountID, + ProjectID: req.Identity.ProjectID, + JTI: jti, + Subject: req.Identity.WIMSEURI, + IssuedAt: now, + ExpiresAt: expiresAt, + TTLSeconds: ttl, + Scopes: coalesceScopeSlice(req.Scopes), + GrantType: req.GrantType, + DelegationDepth: req.DelegationDepth, + ParentJTI: req.ParentJTI, + DelegatedByWIMSEURI: req.DelegatedBy, + DRMHash: req.DRMHash, + ConstraintCatalogHash: req.ConstraintCatalogHash, } if err := s.repo.Create(ctx, cred); err != nil { diff --git a/internal/service/governance.go b/internal/service/governance.go new file mode 100644 index 00000000..0c170afe --- /dev/null +++ b/internal/service/governance.go @@ -0,0 +1,326 @@ +package service + +import ( + "context" + "crypto/ecdsa" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" + "fmt" + "sort" + "strings" + "time" + + "github.com/google/uuid" + "github.com/rs/zerolog/log" + + "github.com/highflame-ai/zeroid/domain" + "github.com/highflame-ai/zeroid/internal/signing" + "github.com/highflame-ai/zeroid/internal/store/postgres" +) + +// GovernanceService binds the Decision-Rights Matrix and Constraint +// Catalog (issue #59) to token issuance. The service is intentionally +// a no-op when no DRM/catalog rows exist for a tenant — direct-OIDC +// federation, plain client_credentials, and every existing flow keep +// working unchanged for tenants that never opt in to governance binding. +type GovernanceService struct { + drmRepo *postgres.DRMRepository + catalogRepo *postgres.ConstraintCatalogRepository + credRepo *postgres.CredentialRepository + signalSvc *SignalService + jwksSvc *signing.JWKSService +} + +func NewGovernanceService( + drmRepo *postgres.DRMRepository, + catalogRepo *postgres.ConstraintCatalogRepository, + credRepo *postgres.CredentialRepository, + signalSvc *SignalService, + jwksSvc *signing.JWKSService, +) *GovernanceService { + return &GovernanceService{ + drmRepo: drmRepo, + catalogRepo: catalogRepo, + credRepo: credRepo, + signalSvc: signalSvc, + jwksSvc: jwksSvc, + } +} + +// HashSHA256 returns "sha256:" of the canonical JSON +// encoding of v. Canonical encoding sorts object keys recursively so the +// same logical document always produces the same hash regardless of +// writer-side key order. +func HashSHA256(v any) (string, error) { + bytes, err := canonicalJSON(v) + if err != nil { + return "", err + } + sum := sha256.Sum256(bytes) + return "sha256:" + hex.EncodeToString(sum[:]), nil +} + +// canonicalJSON re-encodes v with object keys sorted lexicographically +// at every level. Round-trips through encoding/json into a +// map[string]any/[]any tree first, then walks it. +func canonicalJSON(v any) ([]byte, error) { + raw, err := json.Marshal(v) + if err != nil { + return nil, fmt.Errorf("canonical json: marshal: %w", err) + } + var generic any + if err := json.Unmarshal(raw, &generic); err != nil { + return nil, fmt.Errorf("canonical json: unmarshal: %w", err) + } + return canonicalEncode(generic), nil +} + +func canonicalEncode(v any) []byte { + switch t := v.(type) { + case map[string]any: + keys := make([]string, 0, len(t)) + for k := range t { + keys = append(keys, k) + } + sort.Strings(keys) + var b strings.Builder + b.WriteByte('{') + for i, k := range keys { + if i > 0 { + b.WriteByte(',') + } + kb, _ := json.Marshal(k) + b.Write(kb) + b.WriteByte(':') + b.Write(canonicalEncode(t[k])) + } + b.WriteByte('}') + return []byte(b.String()) + case []any: + var b strings.Builder + b.WriteByte('[') + for i, el := range t { + if i > 0 { + b.WriteByte(',') + } + b.Write(canonicalEncode(el)) + } + b.WriteByte(']') + return []byte(b.String()) + default: + out, _ := json.Marshal(t) + return out + } +} + +// PublishDRM validates, hashes, and inserts a new DRM. If a prior active +// DRM existed and the new hash differs, a policy_drift signal is emitted +// for every identity with an outstanding (non-revoked) credential bound +// to the old hash. +func (g *GovernanceService) PublishDRM(ctx context.Context, accountID, projectID string, doc domain.DRMDocument) (*domain.DecisionRightsMatrix, error) { + if err := doc.Validate(); err != nil { + return nil, fmt.Errorf("%w: %v", domain.ErrDRMInvalid, err) + } + hash, err := HashSHA256(doc) + if err != nil { + return nil, err + } + + previous, _ := g.drmRepo.GetActive(ctx, accountID, projectID) + + row := &domain.DecisionRightsMatrix{ + ID: uuid.New().String(), + AccountID: accountID, + ProjectID: projectID, + Version: doc.Version, + EffectiveAt: doc.EffectiveAt, + ExpiresAt: doc.ExpiresAt, + Document: doc, + Hash: hash, + } + if err := g.drmRepo.Create(ctx, row); err != nil { + return nil, err + } + + if previous != nil && previous.Hash != hash { + g.emitDriftSignals(ctx, accountID, projectID, "drm", previous.Hash, hash) + } + return row, nil +} + +// GetActiveDRM returns the active DRM row for a tenant, or (nil, nil) if +// no DRM is configured. +func (g *GovernanceService) GetActiveDRM(ctx context.Context, accountID, projectID string) (*domain.DecisionRightsMatrix, error) { + return g.drmRepo.GetActive(ctx, accountID, projectID) +} + +// ListDRM returns every DRM row for a tenant in descending-effective order. +func (g *GovernanceService) ListDRM(ctx context.Context, accountID, projectID string) ([]*domain.DecisionRightsMatrix, error) { + return g.drmRepo.List(ctx, accountID, projectID) +} + +// AuthorizeDelegation returns nil when the from→to delegation is permitted +// by the active DRM, ErrDRMUnauthorized when it isn't, or nil when no DRM +// is configured (backward compat). +func (g *GovernanceService) AuthorizeDelegation(ctx context.Context, accountID, projectID, fromURI, toURI string) (*domain.DecisionRightsMatrix, error) { + drm, err := g.drmRepo.GetActive(ctx, accountID, projectID) + if err != nil { + return nil, err + } + if drm == nil { + return nil, nil + } + for _, rule := range drm.Document.AllowedDelegations { + if matchSPIFFE(rule.From, fromURI) && matchSPIFFE(rule.To, toURI) { + return drm, nil + } + } + return drm, fmt.Errorf("%w: %s → %s under DRM %s", domain.ErrDRMUnauthorized, fromURI, toURI, drm.Version) +} + +// matchSPIFFE implements the minimal SPIFFE pattern semantics ZeroID +// needs: exact match, or a single trailing `*` glob that matches any +// suffix on the path component. We deliberately avoid a full glob +// library — DRM rules are operator-authored and we want predictable +// failure modes. +func matchSPIFFE(pattern, uri string) bool { + if pattern == "" { + return false + } + if strings.HasSuffix(pattern, "*") { + prefix := strings.TrimSuffix(pattern, "*") + return strings.HasPrefix(uri, prefix) + } + return pattern == uri +} + +// GetActiveCatalog returns the most recently signed catalog row for a +// tenant, or (nil, nil) when none is configured. +func (g *GovernanceService) GetActiveCatalog(ctx context.Context, accountID, projectID string) (*domain.ConstraintCatalogVersion, error) { + return g.catalogRepo.GetActive(ctx, accountID, projectID) +} + +// PublishCatalog hashes, signs, and stores a new catalog version. If the +// hash differs from the previously active row, a policy_drift signal is +// emitted for outstanding tokens bound to the previous catalog hash. +func (g *GovernanceService) PublishCatalog(ctx context.Context, accountID, projectID, version string, effectiveAt time.Time, document json.RawMessage) (*domain.ConstraintCatalogVersion, error) { + hash, err := hashRawJSON(document) + if err != nil { + return nil, err + } + previous, _ := g.catalogRepo.GetActive(ctx, accountID, projectID) + + signedAt := time.Now().UTC() + sig, err := g.signCatalog(hash, signedAt) + if err != nil { + return nil, err + } + + row := &domain.ConstraintCatalogVersion{ + ID: uuid.New().String(), + AccountID: accountID, + ProjectID: projectID, + Version: version, + EffectiveAt: effectiveAt, + Document: document, + Hash: hash, + SignedAt: signedAt, + Signature: sig, + SigningKeyID: g.jwksSvc.KeyID(), + } + if err := g.catalogRepo.Create(ctx, row); err != nil { + return nil, err + } + + if previous != nil && previous.Hash != hash { + g.emitDriftSignals(ctx, accountID, projectID, "constraint_catalog", previous.Hash, hash) + } + return row, nil +} + +// ResignCatalog rewrites the active catalog row with a fresh SignedAt +// signature, preserving the Hash. Used by the 24h liveness worker so +// that consumers of the hash claim can detect a stale/replayed catalog +// without forcing a token re-mint when no policy change occurred. +func (g *GovernanceService) ResignCatalog(ctx context.Context, accountID, projectID string) error { + current, err := g.catalogRepo.GetActive(ctx, accountID, projectID) + if err != nil { + return err + } + if current == nil { + return nil + } + signedAt := time.Now().UTC() + sig, err := g.signCatalog(current.Hash, signedAt) + if err != nil { + return err + } + row := &domain.ConstraintCatalogVersion{ + ID: uuid.New().String(), + AccountID: current.AccountID, + ProjectID: current.ProjectID, + Version: current.Version, + EffectiveAt: current.EffectiveAt, + Document: current.Document, + Hash: current.Hash, + SignedAt: signedAt, + Signature: sig, + SigningKeyID: g.jwksSvc.KeyID(), + } + return g.catalogRepo.Create(ctx, row) +} + +// signCatalog signs SHA-256(hash || "|" || signed_at) with the ES256 +// service key. The "|" separator prevents any chance of length-extension +// ambiguity between the two fields. +func (g *GovernanceService) signCatalog(hash string, signedAt time.Time) (string, error) { + payload := hash + "|" + signedAt.Format(time.RFC3339Nano) + digest := sha256.Sum256([]byte(payload)) + priv := g.jwksSvc.PrivateKey() + if priv == nil { + return "", fmt.Errorf("catalog sign: no ES256 key loaded") + } + sig, err := ecdsa.SignASN1(rand.Reader, priv, digest[:]) + if err != nil { + return "", fmt.Errorf("catalog sign: %w", err) + } + return base64.RawURLEncoding.EncodeToString(sig), nil +} + +func hashRawJSON(raw json.RawMessage) (string, error) { + var v any + if err := json.Unmarshal(raw, &v); err != nil { + return "", fmt.Errorf("catalog hash: invalid json: %w", err) + } + return HashSHA256(v) +} + +// emitDriftSignals best-effort-fans-out policy_drift signals for every +// identity holding an outstanding credential bound to oldHash. Errors +// are logged and swallowed — the new DRM/catalog write must remain +// durable even if signal fan-out partially fails. +func (g *GovernanceService) emitDriftSignals(ctx context.Context, accountID, projectID, kind, oldHash, newHash string) { + if g.signalSvc == nil || g.credRepo == nil { + return + } + identities, err := g.credRepo.ListIdentitiesByGovernanceHash(ctx, accountID, projectID, kind, oldHash) + if err != nil { + log.Warn().Err(err).Str("kind", kind).Msg("policy_drift: failed to enumerate affected identities") + return + } + for _, identityID := range identities { + _, err := g.signalSvc.IngestSignal(ctx, accountID, projectID, identityID, + domain.SignalTypePolicyDrift, domain.SignalSeverityMedium, "governance", + map[string]any{ + "kind": kind, + "old_hash": oldHash, + "new_hash": newHash, + }) + if err != nil { + log.Warn().Err(err).Str("identity_id", identityID).Msg("policy_drift: signal emit failed") + } + } +} diff --git a/internal/service/oauth.go b/internal/service/oauth.go index 516663ba..77b29a9f 100644 --- a/internal/service/oauth.go +++ b/internal/service/oauth.go @@ -39,6 +39,18 @@ type OAuthService struct { trustedServiceValidator trustedServiceValidatorFunc // customGrants holds registered custom grant type handlers. customGrants map[string]CustomGrantHandler + // governanceSvc wires the DRM authorization check + Constraint + // Catalog hash lookup into delegation grants (issue #59). Nil when + // governance binding is not configured for this deployment — every + // existing flow then behaves identically to pre-#59 ZeroID. + governanceSvc *GovernanceService +} + +// SetGovernanceService attaches a GovernanceService so that token_exchange +// and authorization_code grants run the DRM authorization check and embed +// the active DRM / Constraint Catalog hash claims (issue #59). +func (s *OAuthService) SetGovernanceService(g *GovernanceService) { + s.governanceSvc = g } // CustomGrantHandler implements a custom OAuth2 grant type. @@ -62,6 +74,11 @@ var reservedClaims = map[string]bool{ "user_email": true, "user_name": true, // ZeroID internal claims "act": true, "token_exchange": true, "trusted_by": true, + // Governance binding (issue #59) — set by tokenExchange / + // authorizationCode from the active DRM and Constraint Catalog. + // Deployer claim enrichers cannot spoof these. + "drm_version": true, "drm_hash": true, + "constraint_catalog_version": true, "constraint_catalog_hash": true, } // trustedServiceValidatorFunc checks whether the current request comes from a trusted @@ -461,18 +478,32 @@ func (s *OAuthService) tokenExchange(ctx context.Context, req TokenRequest) (*do } } - // Step 6: Issue a delegated credential for the sub-agent. The full + // Step 6: Governance binding (issue #59). If a DRM is configured for + // the tenant, refuse the exchange unless the orchestrator→sub-agent + // pair is permitted by the active DRM. Bind drm_hash + catalog_hash + // into the issued credential so post-hoc audit can identify which + // governance version authorized the delegation. + govReq, err := s.resolveGovernance(ctx, accountID, projectID, subjectParsed.Subject(), actorIdentity.WIMSEURI) + if err != nil { + return nil, err + } + + // Step 7: Issue a delegated credential for the sub-agent. The full // policy constraint set (delegation depth ceiling, required trust // level, allowed grant types, max TTL) is enforced inside // IssueCredential against actor.IdentityPolicyID. accessToken, _, err := s.credentialSvc.IssueCredential(ctx, IssueRequest{ - Identity: actorIdentity, - IdentityPolicyID: actorPolicy.ID, - Scopes: scopes, - GrantType: domain.GrantTypeTokenExchange, - DelegatedBy: subjectParsed.Subject(), - ParentJTI: subjectJTI, - DelegationDepth: parentDepth + 1, + Identity: actorIdentity, + IdentityPolicyID: actorPolicy.ID, + Scopes: scopes, + GrantType: domain.GrantTypeTokenExchange, + DelegatedBy: subjectParsed.Subject(), + ParentJTI: subjectJTI, + DelegationDepth: parentDepth + 1, + DRMVersion: govReq.DRMVersion, + DRMHash: govReq.DRMHash, + ConstraintCatalogVer: govReq.CatalogVersion, + ConstraintCatalogHash: govReq.CatalogHash, }) if err != nil { return nil, err @@ -481,6 +512,48 @@ func (s *OAuthService) tokenExchange(ctx context.Context, req TokenRequest) (*do return accessToken, nil } +// governanceBinding carries the resolved DRM + Constraint Catalog values +// for one issuance call. All four fields are empty when no governance is +// configured for the tenant — IssueCredential then skips the claim +// embedding entirely. +type governanceBinding struct { + DRMVersion string + DRMHash string + CatalogVersion string + CatalogHash string +} + +// resolveGovernance performs the DRM authorization check (when a DRM +// exists) and looks up the active Constraint Catalog hash. Returns an +// oauth error wrapping ErrDRMUnauthorized when the delegation pair is +// rejected; returns empty values when no governance is configured. +func (s *OAuthService) resolveGovernance(ctx context.Context, accountID, projectID, fromURI, toURI string) (governanceBinding, error) { + var out governanceBinding + if s.governanceSvc == nil { + return out, nil + } + drm, err := s.governanceSvc.AuthorizeDelegation(ctx, accountID, projectID, fromURI, toURI) + if err != nil { + if errors.Is(err, domain.ErrDRMUnauthorized) { + return out, oauthBadRequest("invalid_grant", err.Error()) + } + return out, oauthServerError("governance: drm authorization failed", err) + } + if drm != nil { + out.DRMVersion = drm.Version + out.DRMHash = drm.Hash + } + catalog, err := s.governanceSvc.GetActiveCatalog(ctx, accountID, projectID) + if err != nil { + return out, oauthServerError("governance: catalog lookup failed", err) + } + if catalog != nil { + out.CatalogVersion = catalog.Version + out.CatalogHash = catalog.Hash + } + return out, nil +} + // externalPrincipalExchange handles RFC 8693 token exchange for externally-authenticated // principals (e.g. human users authenticated by Clerk, Google, Okta). // @@ -811,14 +884,36 @@ func (s *OAuthService) authorizationCode(ctx context.Context, req TokenRequest) Status: domain.IdentityStatusActive, } + // Governance binding (issue #59). authorization_code represents + // human consent, not agent-to-agent delegation, so we do not run + // the DRM authorization gate here — but we DO bind the active DRM + // and Constraint Catalog hashes onto the issued token so post-hoc + // audit can recover which governance version was in effect when + // the user consented. + govBind := governanceBinding{} + if s.governanceSvc != nil { + if drm, err := s.governanceSvc.GetActiveDRM(ctx, authCode.AccountID, authCode.ProjectID); err == nil && drm != nil { + govBind.DRMVersion = drm.Version + govBind.DRMHash = drm.Hash + } + if cat, err := s.governanceSvc.GetActiveCatalog(ctx, authCode.AccountID, authCode.ProjectID); err == nil && cat != nil { + govBind.CatalogVersion = cat.Version + govBind.CatalogHash = cat.Hash + } + } + accessToken, _, err := s.credentialSvc.IssueCredential(ctx, IssueRequest{ - Identity: identity, - GrantType: domain.GrantTypeAuthorizationCode, - UseRS256: true, - SubjectOverride: authCode.UserID, - ApplicationID: authCode.ClientID, - TTL: ttl, - Scopes: authCode.Scopes, + Identity: identity, + GrantType: domain.GrantTypeAuthorizationCode, + UseRS256: true, + SubjectOverride: authCode.UserID, + ApplicationID: authCode.ClientID, + TTL: ttl, + Scopes: authCode.Scopes, + DRMVersion: govBind.DRMVersion, + DRMHash: govBind.DRMHash, + ConstraintCatalogVer: govBind.CatalogVersion, + ConstraintCatalogHash: govBind.CatalogHash, }) if err != nil { return nil, err @@ -1025,6 +1120,14 @@ func (s *OAuthService) Introspect(ctx context.Context, tokenStr string) (map[str if v, ok := parsed.Get("act"); ok { result["act"] = v } + // Governance binding claims (issue #59). Pass through only when + // present — every existing introspection response keeps its shape + // for tokens that pre-date governance binding. + for _, claim := range []string{"drm_version", "drm_hash", "constraint_catalog_version", "constraint_catalog_hash"} { + if v, ok := parsed.Get(claim); ok { + result[claim] = v + } + } return result, nil } diff --git a/internal/store/postgres/constraint_catalog.go b/internal/store/postgres/constraint_catalog.go new file mode 100644 index 00000000..67522294 --- /dev/null +++ b/internal/store/postgres/constraint_catalog.go @@ -0,0 +1,72 @@ +package postgres + +import ( + "context" + "database/sql" + "errors" + "fmt" + + "github.com/uptrace/bun" + + "github.com/highflame-ai/zeroid/domain" +) + +// ConstraintCatalogRepository persists ConstraintCatalogVersion rows. +// Rows are append-only by convention — a new effective version is a +// fresh row, and a 24h liveness re-sign of an unchanged document is also +// a fresh row with the same Hash but a new SignedAt. +type ConstraintCatalogRepository struct { + db *bun.DB +} + +func NewConstraintCatalogRepository(db *bun.DB) *ConstraintCatalogRepository { + return &ConstraintCatalogRepository{db: db} +} + +func (r *ConstraintCatalogRepository) Create(ctx context.Context, v *domain.ConstraintCatalogVersion) error { + if _, err := r.db.NewInsert().Model(v).Exec(ctx); err != nil { + return fmt.Errorf("failed to create constraint catalog version: %w", err) + } + return nil +} + +// GetActive returns the most recently signed catalog row for the tenant. +// Returns (nil, nil) when no catalog is configured. +func (r *ConstraintCatalogRepository) GetActive(ctx context.Context, accountID, projectID string) (*domain.ConstraintCatalogVersion, error) { + v := &domain.ConstraintCatalogVersion{} + err := r.db.NewSelect().Model(v). + Where("account_id = ?", accountID). + Where("project_id = ?", projectID). + OrderExpr("signed_at DESC"). + Limit(1). + Scan(ctx) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("failed to get active catalog: %w", err) + } + return v, nil +} + +// ListTenants returns each (account_id, project_id) that has at least one +// catalog row. Used by the 24h re-sign worker to enumerate tenants +// without holding a separate index. +func (r *ConstraintCatalogRepository) ListTenants(ctx context.Context) ([]TenantKey, error) { + var rows []TenantKey + err := r.db.NewSelect(). + TableExpr("constraint_catalog_versions"). + ColumnExpr("DISTINCT account_id, project_id"). + Scan(ctx, &rows) + if err != nil { + return nil, fmt.Errorf("failed to list catalog tenants: %w", err) + } + return rows, nil +} + +// TenantKey is a minimal (account, project) tuple used by worker +// iterators that don't need a full row. +type TenantKey struct { + AccountID string `bun:"account_id"` + ProjectID string `bun:"project_id"` +} diff --git a/internal/store/postgres/credential.go b/internal/store/postgres/credential.go index 2c7a7707..26169a36 100644 --- a/internal/store/postgres/credential.go +++ b/internal/store/postgres/credential.go @@ -103,3 +103,34 @@ func (r *CredentialRepository) Revoke(ctx context.Context, id, accountID, projec } return nil } + +// ListIdentitiesByGovernanceHash returns the distinct identity_ids of +// non-revoked, non-expired credentials whose governance hash column +// (drm_hash or constraint_catalog_hash, selected by `kind`) matches +// `hash`. Used by policy_drift signal fan-out (issue #59). +func (r *CredentialRepository) ListIdentitiesByGovernanceHash(ctx context.Context, accountID, projectID, kind, hash string) ([]string, error) { + var column string + switch kind { + case "drm": + column = "drm_hash" + case "constraint_catalog": + column = "constraint_catalog_hash" + default: + return nil, fmt.Errorf("unknown governance hash kind: %s", kind) + } + var ids []string + err := r.db.NewSelect(). + TableExpr("issued_credentials"). + ColumnExpr("DISTINCT identity_id"). + Where("account_id = ?", accountID). + Where("project_id = ?", projectID). + Where("is_revoked = FALSE"). + Where("expires_at > NOW()"). + Where("identity_id IS NOT NULL"). + Where("? = ?", bun.Ident(column), hash). + Scan(ctx, &ids) + if err != nil { + return nil, fmt.Errorf("failed to enumerate identities by %s: %w", column, err) + } + return ids, nil +} diff --git a/internal/store/postgres/decision_rights_matrix.go b/internal/store/postgres/decision_rights_matrix.go new file mode 100644 index 00000000..e3b404fa --- /dev/null +++ b/internal/store/postgres/decision_rights_matrix.go @@ -0,0 +1,79 @@ +package postgres + +import ( + "context" + "database/sql" + "errors" + "fmt" + + "github.com/uptrace/bun" + + "github.com/highflame-ai/zeroid/domain" +) + +// DRMRepository persists DecisionRightsMatrix rows. The table is +// append-only via DB trigger — there is intentionally no Update or +// Delete method. +type DRMRepository struct { + db *bun.DB +} + +func NewDRMRepository(db *bun.DB) *DRMRepository { + return &DRMRepository{db: db} +} + +func (r *DRMRepository) Create(ctx context.Context, drm *domain.DecisionRightsMatrix) error { + if _, err := r.db.NewInsert().Model(drm).Exec(ctx); err != nil { + return fmt.Errorf("failed to create DRM: %w", err) + } + return nil +} + +// GetActive returns the most-recently-effective DRM whose effective_at is +// in the past and which has not expired. Returns (nil, nil) when the +// tenant has no DRM configured — callers treat this as "no DRM +// enforcement" rather than an error (backward compat). +func (r *DRMRepository) GetActive(ctx context.Context, accountID, projectID string) (*domain.DecisionRightsMatrix, error) { + drm := &domain.DecisionRightsMatrix{} + err := r.db.NewSelect().Model(drm). + Where("account_id = ?", accountID). + Where("project_id = ?", projectID). + Where("effective_at <= NOW()"). + Where("expires_at IS NULL OR expires_at > NOW()"). + OrderExpr("effective_at DESC"). + Limit(1). + Scan(ctx) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("failed to get active DRM: %w", err) + } + return drm, nil +} + +func (r *DRMRepository) GetByID(ctx context.Context, id, accountID, projectID string) (*domain.DecisionRightsMatrix, error) { + drm := &domain.DecisionRightsMatrix{} + err := r.db.NewSelect().Model(drm). + Where("id = ?", id). + Where("account_id = ?", accountID). + Where("project_id = ?", projectID). + Scan(ctx) + if err != nil { + return nil, fmt.Errorf("failed to get DRM: %w", err) + } + return drm, nil +} + +func (r *DRMRepository) List(ctx context.Context, accountID, projectID string) ([]*domain.DecisionRightsMatrix, error) { + var rows []*domain.DecisionRightsMatrix + err := r.db.NewSelect().Model(&rows). + Where("account_id = ?", accountID). + Where("project_id = ?", projectID). + OrderExpr("effective_at DESC"). + Scan(ctx) + if err != nil { + return nil, fmt.Errorf("failed to list DRMs: %w", err) + } + return rows, nil +} diff --git a/internal/worker/catalog_signer.go b/internal/worker/catalog_signer.go new file mode 100644 index 00000000..99ab6606 --- /dev/null +++ b/internal/worker/catalog_signer.go @@ -0,0 +1,69 @@ +package worker + +import ( + "context" + "time" + + "github.com/rs/zerolog/log" + + "github.com/highflame-ai/zeroid/internal/store/postgres" +) + +// CatalogResigner is a tenant resigner abstraction. The governance +// service implements this; we keep the worker decoupled from the +// service package to avoid an import cycle (service imports worker +// for testing helpers in other places). +type CatalogResigner interface { + ResignCatalog(ctx context.Context, accountID, projectID string) error +} + +// CatalogSignerWorker re-signs the active Constraint Catalog row for +// every tenant on a 24h cadence (issue #59). A re-sign preserves the +// document Hash but rewrites SignedAt+Signature, so outstanding tokens +// bound to the hash stay valid; downstream consumers that watch +// SignedAt can detect a stale/replayed catalog. +type CatalogSignerWorker struct { + repo *postgres.ConstraintCatalogRepository + resigner CatalogResigner + interval time.Duration +} + +// NewCatalogSignerWorker — pass time.Hour*24 in production. Tests pass +// a short interval and rely on runOnce being deterministic. +func NewCatalogSignerWorker(repo *postgres.ConstraintCatalogRepository, resigner CatalogResigner, interval time.Duration) *CatalogSignerWorker { + return &CatalogSignerWorker{repo: repo, resigner: resigner, interval: interval} +} + +func (w *CatalogSignerWorker) Run(ctx context.Context) { + log.Info().Dur("interval", w.interval).Msg("Catalog signer worker started") + ticker := time.NewTicker(w.interval) + defer ticker.Stop() + + // First re-sign happens after one full interval, not at startup — + // avoids a spurious re-sign every restart. + for { + select { + case <-ticker.C: + w.runOnce(ctx) + case <-ctx.Done(): + log.Info().Msg("Catalog signer worker stopped") + return + } + } +} + +func (w *CatalogSignerWorker) runOnce(ctx context.Context) { + tenants, err := w.repo.ListTenants(ctx) + if err != nil { + log.Error().Err(err).Msg("Catalog signer: failed to list tenants") + return + } + for _, t := range tenants { + if err := w.resigner.ResignCatalog(ctx, t.AccountID, t.ProjectID); err != nil { + log.Warn().Err(err). + Str("account_id", t.AccountID). + Str("project_id", t.ProjectID). + Msg("Catalog signer: re-sign failed") + } + } +} diff --git a/migrations/014_governance_artifacts.down.sql b/migrations/014_governance_artifacts.down.sql new file mode 100644 index 00000000..20b23fc2 --- /dev/null +++ b/migrations/014_governance_artifacts.down.sql @@ -0,0 +1,18 @@ +-- 014_governance_artifacts.down.sql + +DROP INDEX IF EXISTS idx_issued_credentials_catalog_hash; +DROP INDEX IF EXISTS idx_issued_credentials_drm_hash; + +ALTER TABLE issued_credentials + DROP COLUMN IF EXISTS constraint_catalog_hash, + DROP COLUMN IF EXISTS drm_hash; + +DROP INDEX IF EXISTS idx_catalog_tenant_effective; +DROP INDEX IF EXISTS idx_catalog_tenant_signed; +DROP TABLE IF EXISTS constraint_catalog_versions; + +DROP TRIGGER IF EXISTS drm_block_delete ON decision_rights_matrix; +DROP TRIGGER IF EXISTS drm_block_update ON decision_rights_matrix; +DROP FUNCTION IF EXISTS drm_block_mutation(); +DROP INDEX IF EXISTS idx_drm_tenant_effective; +DROP TABLE IF EXISTS decision_rights_matrix; diff --git a/migrations/014_governance_artifacts.up.sql b/migrations/014_governance_artifacts.up.sql new file mode 100644 index 00000000..53889f32 --- /dev/null +++ b/migrations/014_governance_artifacts.up.sql @@ -0,0 +1,82 @@ +-- 014_governance_artifacts.up.sql +-- Adds Decision-Rights Matrix (DRM) and Constraint Catalog version tables +-- per issue #59. Both artifacts are append-only governance records whose +-- SHA-256 hashes are bound into delegation tokens at issuance time so that +-- post-hoc audit can answer "which governance version authorized this token?". +-- +-- DRM rows enumerate permitted delegation patterns and are user-authored +-- via the admin API. Constraint Catalog rows are signed snapshots of the +-- active policy set, re-signed every 24h by an internal worker; multiple +-- catalog rows can share the same Hash (re-sign of unchanged content) +-- but each carries a distinct SignedAt to prove liveness. + +CREATE TABLE IF NOT EXISTS decision_rights_matrix ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + account_id VARCHAR(255) NOT NULL, + project_id VARCHAR(255) NOT NULL, + version VARCHAR(64) NOT NULL, + effective_at TIMESTAMPTZ NOT NULL, + expires_at TIMESTAMPTZ, + document JSONB NOT NULL, + hash VARCHAR(80) NOT NULL, -- "sha256:" + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (account_id, project_id, version) +); + +CREATE INDEX IF NOT EXISTS idx_drm_tenant_effective + ON decision_rights_matrix (account_id, project_id, effective_at DESC); + +-- Append-only enforcement: refuse UPDATE/DELETE on DRM rows. The issue +-- requires "immutable writes" for governance artifacts so post-hoc audit +-- can rely on the row history being intact. +CREATE OR REPLACE FUNCTION drm_block_mutation() RETURNS trigger AS $$ +BEGIN + RAISE EXCEPTION 'decision_rights_matrix is append-only — % blocked', TG_OP; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS drm_block_update ON decision_rights_matrix; +CREATE TRIGGER drm_block_update + BEFORE UPDATE ON decision_rights_matrix + FOR EACH ROW EXECUTE FUNCTION drm_block_mutation(); + +DROP TRIGGER IF EXISTS drm_block_delete ON decision_rights_matrix; +CREATE TRIGGER drm_block_delete + BEFORE DELETE ON decision_rights_matrix + FOR EACH ROW EXECUTE FUNCTION drm_block_mutation(); + + +CREATE TABLE IF NOT EXISTS constraint_catalog_versions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + account_id VARCHAR(255) NOT NULL, + project_id VARCHAR(255) NOT NULL, + version VARCHAR(64) NOT NULL, -- ISO 8601 effective-at by convention + effective_at TIMESTAMPTZ NOT NULL, + document JSONB NOT NULL, -- opaque blob — ZeroID hashes/signs but does not parse + hash VARCHAR(80) NOT NULL, -- "sha256:" of canonical document bytes + signed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + signature TEXT NOT NULL, -- ES256 signature over hash||signed_at + signing_key_id VARCHAR(255) NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_catalog_tenant_signed + ON constraint_catalog_versions (account_id, project_id, signed_at DESC); + +CREATE INDEX IF NOT EXISTS idx_catalog_tenant_effective + ON constraint_catalog_versions (account_id, project_id, effective_at DESC); + +-- Bind governance hashes into the credential record at issuance time so the +-- policy_drift signal emitter can identify outstanding tokens issued under +-- a now-superseded DRM or catalog version without having to decode every JWT. +ALTER TABLE issued_credentials + ADD COLUMN IF NOT EXISTS drm_hash VARCHAR(80), + ADD COLUMN IF NOT EXISTS constraint_catalog_hash VARCHAR(80); + +CREATE INDEX IF NOT EXISTS idx_issued_credentials_drm_hash + ON issued_credentials (account_id, project_id, drm_hash) + WHERE drm_hash IS NOT NULL AND is_revoked = FALSE; + +CREATE INDEX IF NOT EXISTS idx_issued_credentials_catalog_hash + ON issued_credentials (account_id, project_id, constraint_catalog_hash) + WHERE constraint_catalog_hash IS NOT NULL AND is_revoked = FALSE; diff --git a/server.go b/server.go index 132d0c51..ad9e3184 100644 --- a/server.go +++ b/server.go @@ -74,8 +74,12 @@ type Server struct { refreshTokenSvc *service.RefreshTokenService // Cleanup - cleanupWorker *worker.CleanupWorker - workerCancel context.CancelFunc + cleanupWorker *worker.CleanupWorker + catalogSignerWorker *worker.CatalogSignerWorker + workerCancel context.CancelFunc + + // Governance (issue #59) — DRM + Constraint Catalog binding. + governanceSvc *service.GovernanceService // Extensibility mu sync.RWMutex @@ -181,11 +185,20 @@ func NewServer(cfg Config) (*Server, error) { proofSvc := service.NewProofService(jwksSvc, proofRepo, cfg.Token.Issuer) agentSvc := service.NewAgentService(identitySvc, apiKeySvc, apiKeyRepo) + // Governance binding (issue #59). Wires Decision-Rights Matrix + + // Constraint Catalog hash binding into delegation grants. The service + // is no-op for any tenant that has not published a DRM/catalog row, so + // existing flows behave identically. + drmRepo := postgres.NewDRMRepository(db) + catalogRepo := postgres.NewConstraintCatalogRepository(db) + governanceSvc := service.NewGovernanceService(drmRepo, catalogRepo, credentialRepo, signalSvc, jwksSvc) + oauthSvc.SetGovernanceService(governanceSvc) + // Create shared API handler. apiHandler := handler.NewAPI( identitySvc, credentialSvc, credentialPolicySvc, attestationSvc, proofSvc, oauthSvc, oauthClientSvc, - signalSvc, apiKeySvc, agentSvc, auditSvc, jwksSvc, db, + signalSvc, apiKeySvc, agentSvc, auditSvc, governanceSvc, jwksSvc, db, cfg.Token.Issuer, cfg.Token.BaseURL, ) @@ -297,7 +310,9 @@ func NewServer(cfg Config) (*Server, error) { agentSvc: agentSvc, jwksSvc: jwksSvc, refreshTokenSvc: refreshTokenSvc, + governanceSvc: governanceSvc, cleanupWorker: worker.NewCleanupWorker(db, time.Hour), + catalogSignerWorker: worker.NewCatalogSignerWorker(catalogRepo, governanceSvc, 24*time.Hour), adminAuthState: authState, globalMWState: globalMW, http: &http.Server{ @@ -319,6 +334,9 @@ func (s *Server) Start() error { workerCtx, workerCancel := context.WithCancel(context.Background()) s.workerCancel = workerCancel go s.cleanupWorker.Run(workerCtx) + if s.catalogSignerWorker != nil { + go s.catalogSignerWorker.Run(workerCtx) + } // Start HTTP server. errCh := make(chan error, 1) diff --git a/tests/integration/governance_test.go b/tests/integration/governance_test.go new file mode 100644 index 00000000..2b7fb702 --- /dev/null +++ b/tests/integration/governance_test.go @@ -0,0 +1,226 @@ +package integration_test + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "encoding/json" + "net/http" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Governance tests use a per-test tenant so that DRM rows (which are +// append-only and tenant-scoped via the AuthorizeDelegation check) do +// not leak into the shared TestMain Postgres and reject token_exchange +// in other tests that run later in the suite. + +func govTenant(t *testing.T) (accountID, projectID string, headers map[string]string) { + t.Helper() + suffix := uid("gov") + accountID = "acct-" + suffix + projectID = "proj-" + suffix + headers = map[string]string{ + "X-Account-ID": accountID, + "X-Project-ID": projectID, + } + return +} + +func govRegisterIdentity(t *testing.T, headers map[string]string, externalID string, scopes []string, publicKeyPEM string) (id, wimseURI string) { + t.Helper() + body := map[string]any{ + "external_id": externalID, + "trust_level": "unverified", + "owner_user_id": "user-test-owner", + "allowed_scopes": scopes, + } + if publicKeyPEM != "" { + body["public_key_pem"] = publicKeyPEM + } + resp := post(t, adminPath("/identities"), body, headers) + require.Equal(t, http.StatusCreated, resp.StatusCode) + raw := decode(t, resp) + return raw["id"].(string), raw["wimse_uri"].(string) +} + +func govRegisterOAuthClient(t *testing.T, headers map[string]string, clientID string, scopes []string) (cid, secret string) { + t.Helper() + resp := post(t, adminPath("/oauth/clients"), map[string]any{ + "client_id": clientID, + "name": clientID + "-client", + "confidential": true, + "grant_types": []string{"client_credentials"}, + "scopes": scopes, + }, headers) + require.Equal(t, http.StatusCreated, resp.StatusCode) + raw := decode(t, resp) + client := raw["client"].(map[string]any) + return client["client_id"].(string), raw["client_secret"].(string) +} + +func govIntrospect(t *testing.T, headers map[string]string, tokenStr string) map[string]any { + t.Helper() + resp := post(t, "/oauth2/token/introspect", map[string]string{"token": tokenStr}, headers) + require.Equal(t, http.StatusOK, resp.StatusCode) + return decode(t, resp) +} + +// TestGovernanceBinding_TokenExchange — happy path. DRM permits the pair; +// the issued JWT carries the four governance claims and introspection +// surfaces them. +func TestGovernanceBinding_TokenExchange(t *testing.T) { + accountID, projectID, headers := govTenant(t) + + orchID := uid("gov-orch") + _, orchWIMSE := govRegisterIdentity(t, headers, orchID, []string{"data:read"}, "") + orchClientID, orchSecret := govRegisterOAuthClient(t, headers, orchID, []string{"data:read"}) + + subKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + subID := uid("gov-sub") + _, subWIMSE := govRegisterIdentity(t, headers, subID, []string{"data:read"}, ecPublicKeyPEM(t, subKey)) + + drmResp := post(t, adminPath("/governance/decision-rights-matrix"), map[string]any{ + "version": "1.0.0", + "effective_at": time.Now().UTC().Add(-time.Second), + "allowed_delegations": []map[string]any{ + {"from": orchWIMSE, "to": subWIMSE}, + }, + }, headers) + require.Equal(t, http.StatusCreated, drmResp.StatusCode) + drmBody := decode(t, drmResp) + drmHash := drmBody["hash"].(string) + require.True(t, strings.HasPrefix(drmHash, "sha256:")) + + catResp := post(t, adminPath("/governance/constraint-catalog"), map[string]any{ + "version": "2026-05-18T00:00:00Z", + "effective_at": time.Now().UTC().Add(-time.Second), + "document": json.RawMessage(`{"policies":["permit(principal,action,resource)"]}`), + }, headers) + require.Equal(t, http.StatusCreated, catResp.StatusCode) + catBody := decode(t, catResp) + catHash := catBody["hash"].(string) + require.True(t, strings.HasPrefix(catHash, "sha256:")) + + resp := post(t, "/oauth2/token", map[string]any{ + "grant_type": "client_credentials", + "account_id": accountID, + "project_id": projectID, + "client_id": orchClientID, + "client_secret": orchSecret, + "scope": "data:read", + }, nil) + require.Equal(t, http.StatusOK, resp.StatusCode) + orchToken := decode(t, resp)["access_token"].(string) + + actorAssertion := buildAssertion(t, subKey, subWIMSE) + resp = post(t, "/oauth2/token", map[string]any{ + "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange", + "subject_token": orchToken, + "actor_token": actorAssertion, + "scope": "data:read", + }, nil) + require.Equal(t, http.StatusOK, resp.StatusCode, "DRM-permitted exchange must succeed") + delegatedToken := decode(t, resp)["access_token"].(string) + + result := govIntrospect(t, headers, delegatedToken) + assert.True(t, result["active"].(bool)) + assert.Equal(t, "1.0.0", result["drm_version"]) + assert.Equal(t, drmHash, result["drm_hash"]) + assert.Equal(t, "2026-05-18T00:00:00Z", result["constraint_catalog_version"]) + assert.Equal(t, catHash, result["constraint_catalog_hash"]) +} + +// TestGovernanceBinding_UnauthorizedDelegation — DRM allows a different +// `to` pattern than the actor's WIMSE URI; exchange must be rejected. +func TestGovernanceBinding_UnauthorizedDelegation(t *testing.T) { + accountID, projectID, headers := govTenant(t) + + orchID := uid("gov-orch-deny") + _, orchWIMSE := govRegisterIdentity(t, headers, orchID, []string{"data:read"}, "") + orchClientID, orchSecret := govRegisterOAuthClient(t, headers, orchID, []string{"data:read"}) + + subKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + subID := uid("gov-sub-deny") + _, subWIMSE := govRegisterIdentity(t, headers, subID, []string{"data:read"}, ecPublicKeyPEM(t, subKey)) + + drmResp := post(t, adminPath("/governance/decision-rights-matrix"), map[string]any{ + "version": "deny-1.0", + "effective_at": time.Now().UTC().Add(-time.Second), + "allowed_delegations": []map[string]any{ + {"from": orchWIMSE, "to": "spiffe://example.test/never-matches/*"}, + }, + }, headers) + require.Equal(t, http.StatusCreated, drmResp.StatusCode) + + resp := post(t, "/oauth2/token", map[string]any{ + "grant_type": "client_credentials", + "account_id": accountID, + "project_id": projectID, + "client_id": orchClientID, + "client_secret": orchSecret, + "scope": "data:read", + }, nil) + require.Equal(t, http.StatusOK, resp.StatusCode) + orchToken := decode(t, resp)["access_token"].(string) + + actorAssertion := buildAssertion(t, subKey, subWIMSE) + resp = post(t, "/oauth2/token", map[string]any{ + "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange", + "subject_token": orchToken, + "actor_token": actorAssertion, + "scope": "data:read", + }, nil) + require.Equal(t, http.StatusBadRequest, resp.StatusCode) + body := decode(t, resp) + assert.Equal(t, "invalid_grant", body["error"]) +} + +// TestGovernanceBinding_NoConfigBackwardCompat — a fresh tenant with no +// DRM/catalog row still completes token_exchange exactly like pre-#59 +// ZeroID and the issued token has none of the governance claims. +func TestGovernanceBinding_NoConfigBackwardCompat(t *testing.T) { + accountID, projectID, headers := govTenant(t) + + orchID := uid("gov-bc-orch") + _, _ = govRegisterIdentity(t, headers, orchID, []string{"data:read"}, "") + orchClientID, orchSecret := govRegisterOAuthClient(t, headers, orchID, []string{"data:read"}) + + subKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + subID := uid("gov-bc-sub") + _, subWIMSE := govRegisterIdentity(t, headers, subID, []string{"data:read"}, ecPublicKeyPEM(t, subKey)) + + resp := post(t, "/oauth2/token", map[string]any{ + "grant_type": "client_credentials", + "account_id": accountID, + "project_id": projectID, + "client_id": orchClientID, + "client_secret": orchSecret, + "scope": "data:read", + }, nil) + require.Equal(t, http.StatusOK, resp.StatusCode) + orchToken := decode(t, resp)["access_token"].(string) + + actorAssertion := buildAssertion(t, subKey, subWIMSE) + resp = post(t, "/oauth2/token", map[string]any{ + "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange", + "subject_token": orchToken, + "actor_token": actorAssertion, + "scope": "data:read", + }, nil) + require.Equal(t, http.StatusOK, resp.StatusCode, "exchange must succeed when no DRM is configured") + delegatedToken := decode(t, resp)["access_token"].(string) + + result := govIntrospect(t, headers, delegatedToken) + _, hasDRM := result["drm_hash"] + _, hasCat := result["constraint_catalog_hash"] + assert.False(t, hasDRM) + assert.False(t, hasCat) +} From 2259aa34dd0a0808620d07dab1d0d3aa34fa8f5a Mon Sep 17 00:00:00 2001 From: safayavatsal Date: Mon, 18 May 2026 16:10:27 +0530 Subject: [PATCH 2/2] Address Gemini code-assist review comments on PR #151 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five comments resolved in priority order: (high, #3258137060) emitDriftSignals fan-out now runs on a detached goroutine parented on a new GovernanceService.svcCtx so a hash transition affecting many identities does not block the admin POST. Server.Shutdown calls GovernanceService.Stop() to cancel in-flight fan-outs (mirrors the BackchannelService pattern). The affected- identity scan is paginated via the new CredentialRepository.ListIdentitiesByGovernanceHashPage (keyset cursor on identity_id ASC, default page size 500) so a single drift event cannot OOM the worker on huge tenants. (medium, #3258137067) canonicalJSON drops the hand-rolled recursive encoder. encoding/json.Marshal sorts string-keyed map keys, so the two-pass Marshal->Unmarshal-into-any->Marshal pattern produces the same canonical output via stdlib alone. Hashing tests still pass — the determinism contract is preserved. (medium, #3258137081, #3258137088) PublishDRM and PublishCatalog now log the GetActive lookup error instead of swallowing it. The failure is still non-fatal (we don't want to fail the write because the drift lookup couldn't reach the DB) but is no longer silent. (medium, #3258137107) CatalogSignerWorker performs an initial run at startup before entering the 24h tick loop so a server restarted more often than the re-sign interval still produces fresh signed_at rows. Verified clean: GOEXPERIMENT=jsonv2 go build, go vet, gofmt -l, governance integration tests (3/3 pass), full integration suite green. --- internal/service/governance.go | 183 ++++++++++++++++---------- internal/store/postgres/credential.go | 30 +++-- internal/worker/catalog_signer.go | 8 +- server.go | 6 + 4 files changed, 147 insertions(+), 80 deletions(-) diff --git a/internal/service/governance.go b/internal/service/governance.go index 0c170afe..3a2c0f3a 100644 --- a/internal/service/governance.go +++ b/internal/service/governance.go @@ -9,8 +9,8 @@ import ( "encoding/hex" "encoding/json" "fmt" - "sort" "strings" + "sync" "time" "github.com/google/uuid" @@ -32,6 +32,18 @@ type GovernanceService struct { credRepo *postgres.CredentialRepository signalSvc *SignalService jwksSvc *signing.JWKSService + + // svcCtx is the long-lived context used by detached policy_drift + // fan-out goroutines. Parented on context.Background() at + // construction; Server.Shutdown calls Stop() to cancel in-flight + // drift fan-outs so they don't outlive the listener close. + mu sync.Mutex + svcCtx context.Context + svcCancel context.CancelFunc + // driftPageSize bounds memory for a single fan-out by paginating + // the affected-identity scan. Configurable mainly so tests can + // exercise the multi-page path. + driftPageSize int } func NewGovernanceService( @@ -41,12 +53,28 @@ func NewGovernanceService( signalSvc *SignalService, jwksSvc *signing.JWKSService, ) *GovernanceService { + svcCtx, svcCancel := context.WithCancel(context.Background()) return &GovernanceService{ - drmRepo: drmRepo, - catalogRepo: catalogRepo, - credRepo: credRepo, - signalSvc: signalSvc, - jwksSvc: jwksSvc, + drmRepo: drmRepo, + catalogRepo: catalogRepo, + credRepo: credRepo, + signalSvc: signalSvc, + jwksSvc: jwksSvc, + svcCtx: svcCtx, + svcCancel: svcCancel, + driftPageSize: 500, + } +} + +// Stop cancels the service lifecycle context so detached drift fan-out +// goroutines wind down. Idempotent. Server.Shutdown calls this so a +// large drift fan-out doesn't keep work running past listener close. +func (g *GovernanceService) Stop() { + g.mu.Lock() + defer g.mu.Unlock() + if g.svcCancel != nil { + g.svcCancel() + g.svcCancel = nil } } @@ -63,57 +91,22 @@ func HashSHA256(v any) (string, error) { return "sha256:" + hex.EncodeToString(sum[:]), nil } -// canonicalJSON re-encodes v with object keys sorted lexicographically -// at every level. Round-trips through encoding/json into a -// map[string]any/[]any tree first, then walks it. +// canonicalJSON returns a deterministic JSON encoding of v: keys are +// sorted lexicographically at every level. We achieve this by first +// marshaling v (collapsing structs/typed maps), then unmarshaling into +// `any` so every object becomes `map[string]any`, then re-marshaling — +// encoding/json's Marshal sorts string-keyed map keys, so the second +// pass produces canonical output without a hand-rolled encoder. func canonicalJSON(v any) ([]byte, error) { raw, err := json.Marshal(v) if err != nil { - return nil, fmt.Errorf("canonical json: marshal: %w", err) + return nil, err } var generic any if err := json.Unmarshal(raw, &generic); err != nil { - return nil, fmt.Errorf("canonical json: unmarshal: %w", err) - } - return canonicalEncode(generic), nil -} - -func canonicalEncode(v any) []byte { - switch t := v.(type) { - case map[string]any: - keys := make([]string, 0, len(t)) - for k := range t { - keys = append(keys, k) - } - sort.Strings(keys) - var b strings.Builder - b.WriteByte('{') - for i, k := range keys { - if i > 0 { - b.WriteByte(',') - } - kb, _ := json.Marshal(k) - b.Write(kb) - b.WriteByte(':') - b.Write(canonicalEncode(t[k])) - } - b.WriteByte('}') - return []byte(b.String()) - case []any: - var b strings.Builder - b.WriteByte('[') - for i, el := range t { - if i > 0 { - b.WriteByte(',') - } - b.Write(canonicalEncode(el)) - } - b.WriteByte(']') - return []byte(b.String()) - default: - out, _ := json.Marshal(t) - return out + return nil, err } + return json.Marshal(generic) } // PublishDRM validates, hashes, and inserts a new DRM. If a prior active @@ -129,7 +122,14 @@ func (g *GovernanceService) PublishDRM(ctx context.Context, accountID, projectID return nil, err } - previous, _ := g.drmRepo.GetActive(ctx, accountID, projectID) + // Look up the previously active DRM for drift detection. A lookup + // failure here is non-fatal — the new DRM still gets written and + // the only consequence is missed policy_drift signals for tokens + // minted under the old hash. Log so this doesn't go silent. + previous, prevErr := g.drmRepo.GetActive(ctx, accountID, projectID) + if prevErr != nil { + log.Warn().Err(prevErr).Msg("PublishDRM: failed to look up previous active DRM for drift detection") + } row := &domain.DecisionRightsMatrix{ ID: uuid.New().String(), @@ -211,7 +211,12 @@ func (g *GovernanceService) PublishCatalog(ctx context.Context, accountID, proje if err != nil { return nil, err } - previous, _ := g.catalogRepo.GetActive(ctx, accountID, projectID) + // Same drift-detection lookup as PublishDRM — non-fatal but noisy + // on failure so missed policy_drift signals are visible. + previous, prevErr := g.catalogRepo.GetActive(ctx, accountID, projectID) + if prevErr != nil { + log.Warn().Err(prevErr).Msg("PublishCatalog: failed to look up previous active catalog for drift detection") + } signedAt := time.Now().UTC() sig, err := g.signCatalog(hash, signedAt) @@ -298,29 +303,69 @@ func hashRawJSON(raw json.RawMessage) (string, error) { return HashSHA256(v) } -// emitDriftSignals best-effort-fans-out policy_drift signals for every -// identity holding an outstanding credential bound to oldHash. Errors -// are logged and swallowed — the new DRM/catalog write must remain -// durable even if signal fan-out partially fails. -func (g *GovernanceService) emitDriftSignals(ctx context.Context, accountID, projectID, kind, oldHash, newHash string) { +// emitDriftSignals fires policy_drift signals for every identity +// holding an outstanding credential bound to oldHash. The fan-out runs +// on a background goroutine parented on the service lifecycle context +// (svcCtx) so a hash transition affecting many identities does not +// block the admin POST that triggered it, but also does not outlive +// the server: Server.Shutdown -> GovernanceService.Stop cancels svcCtx +// and in-flight pagination winds down. Errors are logged and swallowed +// — the new DRM/catalog row write must remain durable even if signal +// fan-out partially fails. +func (g *GovernanceService) emitDriftSignals(_ context.Context, accountID, projectID, kind, oldHash, newHash string) { if g.signalSvc == nil || g.credRepo == nil { return } - identities, err := g.credRepo.ListIdentitiesByGovernanceHash(ctx, accountID, projectID, kind, oldHash) - if err != nil { - log.Warn().Err(err).Str("kind", kind).Msg("policy_drift: failed to enumerate affected identities") + g.mu.Lock() + parent := g.svcCtx + g.mu.Unlock() + if parent == nil { + // Stop() already called — no detached work after shutdown. return } - for _, identityID := range identities { - _, err := g.signalSvc.IngestSignal(ctx, accountID, projectID, identityID, - domain.SignalTypePolicyDrift, domain.SignalSeverityMedium, "governance", - map[string]any{ - "kind": kind, - "old_hash": oldHash, - "new_hash": newHash, - }) + pageSize := g.driftPageSize + if pageSize <= 0 { + pageSize = 500 + } + go g.runDriftFanout(parent, accountID, projectID, kind, oldHash, newHash, pageSize) +} + +func (g *GovernanceService) runDriftFanout(ctx context.Context, accountID, projectID, kind, oldHash, newHash string, pageSize int) { + afterID := "" + for { + if ctx.Err() != nil { + log.Info().Str("kind", kind).Msg("policy_drift: fan-out cancelled by shutdown") + return + } + ids, err := g.credRepo.ListIdentitiesByGovernanceHashPage(ctx, accountID, projectID, kind, oldHash, afterID, pageSize) if err != nil { - log.Warn().Err(err).Str("identity_id", identityID).Msg("policy_drift: signal emit failed") + log.Warn().Err(err).Str("kind", kind).Msg("policy_drift: failed to enumerate affected identities") + return + } + if len(ids) == 0 { + return + } + for _, identityID := range ids { + if ctx.Err() != nil { + return + } + _, emitErr := g.signalSvc.IngestSignal(ctx, accountID, projectID, identityID, + domain.SignalTypePolicyDrift, domain.SignalSeverityMedium, "governance", + map[string]any{ + "kind": kind, + "old_hash": oldHash, + "new_hash": newHash, + }) + if emitErr != nil { + log.Warn().Err(emitErr).Str("identity_id", identityID).Msg("policy_drift: signal emit failed") + } + } + // Advance keyset cursor. ListIdentitiesByGovernanceHashPage + // returns rows ordered by identity_id ASC, so the last id is + // the high-water mark. + afterID = ids[len(ids)-1] + if len(ids) < pageSize { + return } } } diff --git a/internal/store/postgres/credential.go b/internal/store/postgres/credential.go index 6d892796..3843337d 100644 --- a/internal/store/postgres/credential.go +++ b/internal/store/postgres/credential.go @@ -131,11 +131,16 @@ func (r *CredentialRepository) Revoke(ctx context.Context, id, accountID, projec return nil } -// ListIdentitiesByGovernanceHash returns the distinct identity_ids of -// non-revoked, non-expired credentials whose governance hash column -// (drm_hash or constraint_catalog_hash, selected by `kind`) matches -// `hash`. Used by policy_drift signal fan-out (issue #59). -func (r *CredentialRepository) ListIdentitiesByGovernanceHash(ctx context.Context, accountID, projectID, kind, hash string) ([]string, error) { +// ListIdentitiesByGovernanceHashPage returns one page of distinct +// identity_ids of non-revoked, non-expired credentials whose governance +// hash column (drm_hash or constraint_catalog_hash, selected by `kind`) +// matches `hash`. Keyset paginated by identity_id ascending: pass +// afterID="" on the first call, then the largest returned identity_id +// on each subsequent call until an empty slice comes back. Used by +// policy_drift signal fan-out (issue #59) so a single hash transition +// touching millions of credentials doesn't materialise the full id +// list in memory. +func (r *CredentialRepository) ListIdentitiesByGovernanceHashPage(ctx context.Context, accountID, projectID, kind, hash, afterID string, limit int) ([]string, error) { var column string switch kind { case "drm": @@ -145,8 +150,10 @@ func (r *CredentialRepository) ListIdentitiesByGovernanceHash(ctx context.Contex default: return nil, fmt.Errorf("unknown governance hash kind: %s", kind) } - var ids []string - err := r.db.NewSelect(). + if limit <= 0 { + limit = 500 + } + q := r.db.NewSelect(). TableExpr("issued_credentials"). ColumnExpr("DISTINCT identity_id"). Where("account_id = ?", accountID). @@ -155,8 +162,13 @@ func (r *CredentialRepository) ListIdentitiesByGovernanceHash(ctx context.Contex Where("expires_at > NOW()"). Where("identity_id IS NOT NULL"). Where("? = ?", bun.Ident(column), hash). - Scan(ctx, &ids) - if err != nil { + OrderExpr("identity_id ASC"). + Limit(limit) + if afterID != "" { + q = q.Where("identity_id > ?", afterID) + } + var ids []string + if err := q.Scan(ctx, &ids); err != nil { return nil, fmt.Errorf("failed to enumerate identities by %s: %w", column, err) } return ids, nil diff --git a/internal/worker/catalog_signer.go b/internal/worker/catalog_signer.go index 99ab6606..d7350b04 100644 --- a/internal/worker/catalog_signer.go +++ b/internal/worker/catalog_signer.go @@ -39,8 +39,12 @@ func (w *CatalogSignerWorker) Run(ctx context.Context) { ticker := time.NewTicker(w.interval) defer ticker.Stop() - // First re-sign happens after one full interval, not at startup — - // avoids a spurious re-sign every restart. + // Initial run at startup so a server that's restarted more often + // than the re-sign interval still produces fresh signed_at rows. + // runOnce is short — ListTenants + one Create per active tenant — + // so this isn't load-bearing on cold start. + w.runOnce(ctx) + for { select { case <-ticker.C: diff --git a/server.go b/server.go index 9bcf8f3c..111e17b5 100644 --- a/server.go +++ b/server.go @@ -429,6 +429,12 @@ func (s *Server) Shutdown(ctx context.Context) error { s.backchannelSvc.Stop() } + // Cancel policy_drift fan-out goroutines for the same reason + // (issue #59). + if s.governanceSvc != nil { + s.governanceSvc.Stop() + } + var firstErr error if err := s.http.Shutdown(ctx); err != nil && firstErr == nil { firstErr = err