From 6ca6657f7a12e718098199f0a83151d84e3b4165 Mon Sep 17 00:00:00 2001 From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com> Date: Mon, 7 Sep 2026 10:06:33 +0800 Subject: [PATCH] feat(edge): add modern permission decision receipts (#2353) Co-authored-by: Codex --- edge-server/internal/api/handlers.go | 5 + .../internal/api/handlers_approvals.go | 143 +++++-- .../internal/api/handlers_run_callback.go | 5 +- .../internal/api/permission_receipts.go | 183 ++++++++ .../internal/api/permission_receipts_test.go | 390 ++++++++++++++++++ 5 files changed, 690 insertions(+), 36 deletions(-) create mode 100644 edge-server/internal/api/permission_receipts.go create mode 100644 edge-server/internal/api/permission_receipts_test.go diff --git a/edge-server/internal/api/handlers.go b/edge-server/internal/api/handlers.go index 3d296cd01..b8edbcc55 100644 --- a/edge-server/internal/api/handlers.go +++ b/edge-server/internal/api/handlers.go @@ -64,6 +64,11 @@ type Handler struct { PermissionRegistry *permission.PermissionRegistry PermissionBroker *adapters.PermissionDecisionBroker + // permissionReceipts provides bounded warm-replay receipts for modern + // POST /v1/permissions/decide controls. Lazy and nil-safe in production; + // it is process-local and never authoritative. + permissionReceipts *permissionReceiptCache + // PlanApprovalBroker manages pending orchestrator plans and connects // them to user approval/rejection decisions (P0 #3: plan confirmation gate). PlanApprovalBroker *orchestrator.PlanApprovalBroker diff --git a/edge-server/internal/api/handlers_approvals.go b/edge-server/internal/api/handlers_approvals.go index 5cefe2cdc..6ccb7c18a 100644 --- a/edge-server/internal/api/handlers_approvals.go +++ b/edge-server/internal/api/handlers_approvals.go @@ -10,19 +10,26 @@ import ( "github.com/agenthub/edge-server/internal/permission" ) -// Handler holds dependencies for HTTP and WebSocket handlers. +// permissionDecideRequest is the POST /v1/permissions/decide payload. controlId +// and hubTaskId are optional together: when either is present both are required, +// and the request is modern and idempotent; when both are absent the existing +// one-shot receiver semantics are unchanged. +type permissionDecideRequest struct { + ControlID string `json:"controlId"` + HubTaskID string `json:"hubTaskId"` + RunID string `json:"runId"` + RequestID string `json:"requestId"` + Decision string `json:"decision"` + Reason string `json:"reason,omitempty"` +} + func (h *Handler) PostPermissionDecide(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { errcode.Write(w, errcode.ErrMethodNotAllowed) return } - var req struct { - RunID string `json:"runId"` - RequestID string `json:"requestId"` - Decision string `json:"decision"` - Reason string `json:"reason,omitempty"` - } + var req permissionDecideRequest if err := decodeOptionalJSON(r, &req); err != nil { errcode.Write(w, errcode.ErrInvalidJSON) return @@ -30,6 +37,8 @@ func (h *Handler) PostPermissionDecide(w http.ResponseWriter, r *http.Request) { req.RunID = strings.TrimSpace(req.RunID) req.RequestID = strings.TrimSpace(req.RequestID) req.Decision = strings.TrimSpace(req.Decision) + req.ControlID = strings.TrimSpace(req.ControlID) + req.HubTaskID = strings.TrimSpace(req.HubTaskID) if req.RunID == "" { errcode.Write(w, errcode.ErrRunIDRequired) return @@ -42,51 +51,117 @@ func (h *Handler) PostPermissionDecide(w http.ResponseWriter, r *http.Request) { errcode.Write(w, errcode.ErrInvalidDecision) return } + if (req.ControlID == "") != (req.HubTaskID == "") { + errcode.Write(w, errcode.ErrBadRequest.WithMessage("controlId and hubTaskId must be provided together")) + return + } - // Ownership gate, resolved before any state change (broker decide / registry - // consume / event publish): without it any caller past the Edge's coarse auth - // could allow or deny a live tool call of somebody else's run, i.e. make the - // victim's agent perform the call. The 404 body is byte-identical to the - // "no such request" path below (same errcode, no distinguishing message) so a - // foreign runId and a nonexistent runId stay indistinguishable — this endpoint - // must not become a runId existence oracle. Local single-tenant mode resolves - // to the documented bypass sentinel and is unaffected; an empty principal under - // Hub JWT fails closed (AH-SR-045). - if !isRunOwnedBy(ensureStore(h), req.RunID, h.ownerUserID(r)) { + // Ownership and task binding are resolved before any state change: + // broker decide, registry consume, event publish, and receipt storage all + // happen only for the real run owner and the exact stored Hub task. The + // 404 body is byte-identical for a foreign runId, a missing runId, a + // wrong hubTaskId, and a missing pending request so this endpoint is not an + // existence oracle. Local single-tenant mode resolves to the documented + // bypass sentinel; empty principal under Hub JWT fails closed (AH-SR-045). + repo := ensureStore(h) + owner := h.ownerUserID(r) + if !isRunOwnedBy(repo, req.RunID, owner) { errcode.Write(w, errcode.ErrPermissionRequestNotFound) return } + if req.ControlID != "" { + run, ok := repo.GetRun(req.RunID) + if !ok || run.HubTaskID != req.HubTaskID { + errcode.Write(w, errcode.ErrPermissionRequestNotFound) + return + } + } registry := h.ensurePermissionRegistry() - permission, ok := pendingPermissionFromBroker(h.ensurePermissionBroker(), req.RunID, req.RequestID, req.Decision, req.Reason) - if ok { - _, _ = registry.Consume(req.RunID, req.RequestID) - } else { - permission, ok = registry.Consume(req.RunID, req.RequestID) + if req.ControlID == "" { + h.decideLegacyPermission(w, registry, req) + return + } + h.decideModernPermission(w, registry, req) +} + +func (h *Handler) decideLegacyPermission(w http.ResponseWriter, registry *permission.PermissionRegistry, req permissionDecideRequest) { + pending, ok := h.resolvePendingPermission(registry, req) + if !ok { + errcode.Write(w, errcode.ErrPermissionRequestNotFound) + return + } + h.publishPermissionDecision(pending, req) + slog.Info("permission decided by Desktop", "requestId", req.RequestID, "decision", req.Decision) + writeSuccess(w, http.StatusOK, map[string]any{"status": "ok"}) +} + +func (h *Handler) decideModernPermission(w http.ResponseWriter, registry *permission.PermissionRegistry, req permissionDecideRequest) { + receipt := permissionReceipt{ + ControlID: req.ControlID, + HubTaskID: req.HubTaskID, + RunID: req.RunID, + RequestID: req.RequestID, + Decision: req.Decision, + Reason: req.Reason, + } + result := h.ensurePermissionReceipts().apply(receipt, func() bool { + pending, ok := h.resolvePendingPermission(registry, req) if !ok { - errcode.Write(w, errcode.ErrPermissionRequestNotFound) - return + return false } + h.publishPermissionDecision(pending, req) + return true + }) + switch { + case result.Conflict: + errcode.Write(w, errcode.ErrConflict.WithMessage("permission decision conflict")) + return + case result.Full: + errcode.Write(w, errcode.ErrTooManyRequests.WithMessage("permission decision receipt cache is full")) + return + case !result.Applied: + errcode.Write(w, errcode.ErrPermissionRequestNotFound) + return } + slog.Info("permission decision applied by Desktop", "requestId", req.RequestID, "controlId", req.ControlID, "decision", req.Decision) + writeSuccess(w, http.StatusOK, map[string]any{ + "status": "ok", + "controlId": result.Receipt.ControlID, + "hubTaskId": result.Receipt.HubTaskID, + "runId": result.Receipt.RunID, + "requestId": result.Receipt.RequestID, + "decision": result.Receipt.Decision, + "applied": true, + "deduplicated": result.Deduplicated, + }) +} + +func (h *Handler) resolvePendingPermission(registry *permission.PermissionRegistry, req permissionDecideRequest) (permission.PendingPermission, bool) { + pending, ok := pendingPermissionFromBroker(h.ensurePermissionBroker(), req.RunID, req.RequestID, req.Decision, req.Reason) + if ok { + _, _ = registry.Consume(req.RunID, req.RequestID) + return pending, true + } + return registry.Consume(req.RunID, req.RequestID) +} - scope := map[string]any{"runId": permission.RunID} - if permission.ProjectID != "" { - scope["projectId"] = permission.ProjectID +func (h *Handler) publishPermissionDecision(pending permission.PendingPermission, req permissionDecideRequest) { + scope := map[string]any{"runId": pending.RunID} + if pending.ProjectID != "" { + scope["projectId"] = pending.ProjectID } - if permission.ThreadID != "" { - scope["threadId"] = permission.ThreadID + if pending.ThreadID != "" { + scope["threadId"] = pending.ThreadID } ensureBus(h).Publish(adapters.BusEventPermissionDecided, scope, map[string]any{ "runId": req.RunID, "requestId": req.RequestID, - "toolName": permission.ToolName, - "toolUseId": permission.ToolUseID, + "toolName": pending.ToolName, + "toolUseId": pending.ToolUseID, "decision": req.Decision, "reason": req.Reason, }) - - slog.Info("permission decided by Desktop", "requestId", req.RequestID, "decision", req.Decision) - writeSuccess(w, http.StatusOK, map[string]any{"status": "ok"}) } func pendingPermissionFromBroker(broker *adapters.PermissionDecisionBroker, runID, requestID, decision, reason string) (permission.PendingPermission, bool) { diff --git a/edge-server/internal/api/handlers_run_callback.go b/edge-server/internal/api/handlers_run_callback.go index 49dcb0ab5..f28e944aa 100644 --- a/edge-server/internal/api/handlers_run_callback.go +++ b/edge-server/internal/api/handlers_run_callback.go @@ -50,7 +50,8 @@ func validateReplayCallbackOwner(req runRequest, run store.Run) *errcode.Error { func (h *Handler) runCallbackCapabilities() map[string]bool { return map[string]bool{ - "runCallbackOwnership": true, - "directHubCallbacks": h.directHubCallbacksConfigured(), + "runCallbackOwnership": true, + "directHubCallbacks": h.directHubCallbacksConfigured(), + "permissionDecisionReceipts": true, } } diff --git a/edge-server/internal/api/permission_receipts.go b/edge-server/internal/api/permission_receipts.go new file mode 100644 index 000000000..dc8b1fbd4 --- /dev/null +++ b/edge-server/internal/api/permission_receipts.go @@ -0,0 +1,183 @@ +package api + +import ( + "strings" + "sync" + "time" + + "github.com/agenthub/edge-server/internal/deliverydedup" +) + +const ( + permissionReceiptDefaultCapacity = deliverydedup.DefaultCapacity + permissionReceiptDefaultTTL = deliverydedup.DefaultTTL +) + +// permissionRunRequestKey is the business identity that may accept only one +// modern control. A different control ID for the same run/request is a +// conflict, never a second application. +type permissionRunRequestKey struct { + runID string + requestID string +} + +// permissionReceipt is an applied modern permission decision. It is a warm, +// bounded replay receipt only; it is never authority and never a pending +// request. +type permissionReceipt struct { + ControlID string + HubTaskID string + RunID string + RequestID string + Decision string + Reason string + expiresAt time.Time +} + +func (r permissionReceipt) sameAs(other permissionReceipt) bool { + return r.ControlID == other.ControlID && + r.HubTaskID == other.HubTaskID && + r.RunID == other.RunID && + r.RequestID == other.RequestID && + r.Decision == other.Decision && + r.Reason == other.Reason +} + +func (r permissionReceipt) runRequestKey() permissionRunRequestKey { + return permissionRunRequestKey{runID: r.RunID, requestID: r.RequestID} +} + +// permissionReceiptApplyResult reports the outcome of an atomic modern +// decision application. Full and Conflict are returned before the callback +// runs, so no broker/consume/event effect happens after those failures. +type permissionReceiptApplyResult struct { + Receipt permissionReceipt + Applied bool + Deduplicated bool + Conflict bool + Full bool +} + +// permissionReceiptCache is a process-local, bounded TTL cache of applied +// modern permission receipts. It deliberately has no durable state and does +// not retain pending requests: a cold miss must fall through to the existing +// broker/registry and return not-found instead of inventing success. Capacity +// pressure rejects a new decision before effects; it does not evict an applied +// receipt to admit another control. +type permissionReceiptCache struct { + mu sync.Mutex + capacity int + ttl time.Duration + now func() time.Time + byControl map[string]permissionReceipt + byRunReq map[permissionRunRequestKey]struct{} +} + +func newPermissionReceiptCache(capacity int, ttl time.Duration) *permissionReceiptCache { + if capacity <= 0 { + panic("api: permission receipt cache capacity must be > 0") + } + if ttl <= 0 { + panic("api: permission receipt cache ttl must be > 0") + } + return &permissionReceiptCache{ + capacity: capacity, + ttl: ttl, + now: time.Now, + byControl: make(map[string]permissionReceipt, capacity), + byRunReq: make(map[permissionRunRequestKey]struct{}, capacity), + } +} + +func (c *permissionReceiptCache) withClock(clock func() time.Time) *permissionReceiptCache { + c.mu.Lock() + defer c.mu.Unlock() + c.now = clock + return c +} + +// apply atomically decides whether a modern request is a warm replay, a +// conflict, a capacity rejection, or a newly pending application. The callback +// runs while the cache lock is held so one winner performs every effect and +// concurrent retries observe a stable receipt afterwards. +func (c *permissionReceiptCache) apply(request permissionReceipt, applyFn func() bool) permissionReceiptApplyResult { + c.mu.Lock() + defer c.mu.Unlock() + request.ControlID = strings.TrimSpace(request.ControlID) + request.HubTaskID = strings.TrimSpace(request.HubTaskID) + request.RunID = strings.TrimSpace(request.RunID) + request.RequestID = strings.TrimSpace(request.RequestID) + request.Decision = strings.TrimSpace(request.Decision) + if request.ControlID == "" || request.HubTaskID == "" || request.RunID == "" || request.RequestID == "" { + return permissionReceiptApplyResult{} + } + c.purgeExpiredLocked(c.now()) + + if stored, ok := c.byControl[request.ControlID]; ok { + if stored.sameAs(request) { + return permissionReceiptApplyResult{ + Receipt: stored, + Applied: true, + Deduplicated: true, + } + } + return permissionReceiptApplyResult{Conflict: true} + } + if _, ok := c.byRunReq[request.runRequestKey()]; ok { + return permissionReceiptApplyResult{Conflict: true} + } + if len(c.byControl) >= c.capacity { + return permissionReceiptApplyResult{Full: true} + } + if !applyFn() { + return permissionReceiptApplyResult{} + } + request.expiresAt = c.now().Add(c.ttl) + c.storeLocked(request) + return permissionReceiptApplyResult{ + Receipt: request, + Applied: true, + } +} + +func (c *permissionReceiptCache) storeLocked(receipt permissionReceipt) { + receipt.ControlID = strings.TrimSpace(receipt.ControlID) + receipt.HubTaskID = strings.TrimSpace(receipt.HubTaskID) + receipt.RunID = strings.TrimSpace(receipt.RunID) + receipt.RequestID = strings.TrimSpace(receipt.RequestID) + receipt.Decision = strings.TrimSpace(receipt.Decision) + controlID := receipt.ControlID + if controlID == "" { + return + } + c.byControl[controlID] = receipt + c.byRunReq[receipt.runRequestKey()] = struct{}{} +} + +func (c *permissionReceiptCache) purgeExpiredLocked(now time.Time) { + for id, receipt := range c.byControl { + if now.After(receipt.expiresAt) { + delete(c.byControl, id) + delete(c.byRunReq, receipt.runRequestKey()) + } + } +} + +func (c *permissionReceiptCache) len() int { + c.mu.Lock() + defer c.mu.Unlock() + c.purgeExpiredLocked(c.now()) + return len(c.byControl) +} + +func (h *Handler) ensurePermissionReceipts() *permissionReceiptCache { + h.permissionRegistryMu.Lock() + defer h.permissionRegistryMu.Unlock() + if h.permissionReceipts == nil { + h.permissionReceipts = newPermissionReceiptCache( + permissionReceiptDefaultCapacity, + permissionReceiptDefaultTTL, + ) + } + return h.permissionReceipts +} diff --git a/edge-server/internal/api/permission_receipts_test.go b/edge-server/internal/api/permission_receipts_test.go new file mode 100644 index 000000000..dd1356ee1 --- /dev/null +++ b/edge-server/internal/api/permission_receipts_test.go @@ -0,0 +1,390 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/agenthub/edge-server/internal/adapters" + "github.com/agenthub/edge-server/internal/errcode" + "github.com/agenthub/edge-server/internal/permission" + "github.com/agenthub/edge-server/internal/store" +) + +func seedModernApprovalRun(t *testing.T, repo store.Repository, key, runID, ownerID, hubTaskID string) { + t.Helper() + projectID := "proj-receipt-" + key + threadID := "thread-receipt-" + key + if _, err := repo.CreateProject(projectID, "Receipt "+key, ownerID); err != nil { + t.Fatalf("CreateProject(%s): %v", projectID, err) + } + if _, err := repo.CreateThread(threadID, projectID, "Receipt "+key, "", "", ""); err != nil { + t.Fatalf("CreateThread(%s): %v", threadID, err) + } + if _, err := repo.CreateRun(runID, projectID, threadID); err != nil { + t.Fatalf("CreateRun(%s): %v", runID, err) + } + if _, ok := repo.SetRunHubTaskID(runID, hubTaskID); !ok { + t.Fatalf("SetRunHubTaskID(%s) failed", runID) + } +} + +func registerModernPendingRegistry(t *testing.T, h *Handler, runID, requestID, projectID, threadID string) { + t.Helper() + registry := h.ensurePermissionRegistry() + if !registry.Register(permission.PendingPermission{ + ProjectID: projectID, + ThreadID: threadID, + RunID: runID, + RequestID: requestID, + ToolName: "Bash", + ToolUseID: "tool-" + requestID, + }) { + t.Fatalf("Register(%s/%s) failed", runID, requestID) + } +} + +func registerModernBrokerPending(t *testing.T, h *Handler, runID, requestID, projectID, threadID string) func(context.Context) adapters.PermissionDecision { + t.Helper() + wait, ok := h.ensurePermissionBroker().Begin(adapters.PermissionScope{ + ProjectID: projectID, + ThreadID: threadID, + RunID: runID, + }, adapters.PermissionRequest{ + RequestID: requestID, + ToolName: "Bash", + ToolUseID: "tool-" + requestID, + }) + if !ok { + t.Fatalf("broker.Begin(%s/%s) failed", runID, requestID) + } + return wait +} + +func modernPermissionDecisionBody(controlID, hubTaskID, runID, requestID, decision, reason string) string { + return fmt.Sprintf( + `{"controlId":%q,"hubTaskId":%q,"runId":%q,"requestId":%q,"decision":%q,"reason":%q}`, + controlID, hubTaskID, runID, requestID, decision, reason, + ) +} + +func modernReceiptData(t *testing.T, rec *httptest.ResponseRecorder) map[string]any { + t.Helper() + var body map[string]any + if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { + t.Fatalf("decode %q: %v", rec.Body.String(), err) + } + return unwrapSuccess(body) +} + +func assertModernReceipt(t *testing.T, rec *httptest.ResponseRecorder, controlID, hubTaskID, runID, requestID, decision string, deduplicated bool) { + t.Helper() + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec.Code, rec.Body.String()) + } + data := modernReceiptData(t, rec) + for key, want := range map[string]any{ + "status": "ok", + "controlId": controlID, + "hubTaskId": hubTaskID, + "runId": runID, + "requestId": requestID, + "decision": decision, + "applied": true, + "deduplicated": deduplicated, + } { + if got := data[key]; got != want { + t.Fatalf("data[%q] = %#v, want %#v; body=%s", key, got, want, rec.Body.String()) + } + } +} + +func TestPostPermissionDecideModernReceiptReplayIsSingleApplication(t *testing.T) { + h := newTestHandler() + h.HubJWTSecret = "test-secret" + defer h.Bus.Close() + seedModernApprovalRun(t, h.Store, "replay", "run-replay", "user-a", "task-replay") + wait := registerModernBrokerPending(t, h, "run-replay", "req_1", "proj-receipt-replay", "thread-receipt-replay") + registerModernPendingRegistry(t, h, "run-replay", "req_1", "proj-receipt-replay", "thread-receipt-replay") + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + resultCh := make(chan adapters.PermissionDecision, 1) + go func() { + resultCh <- wait(ctx) + }() + + body := modernPermissionDecisionBody("control-replay", "task-replay", "run-replay", "req_1", "allow", "approved") + first := doPermissionDecideAsUser(h, "user-a", body) + assertModernReceipt(t, first, "control-replay", "task-replay", "run-replay", "req_1", "allow", false) + + select { + case got := <-resultCh: + if got.Behavior != "allow" || got.Message != "approved" { + t.Fatalf("broker decision = %#v, want allow/approved", got) + } + case <-time.After(time.Second): + t.Fatal("broker waiter was not woken") + } + if h.ensurePermissionBroker().PendingPermission("run-replay", "req_1") { + t.Fatal("broker pending entry remained after first modern decision") + } + if _, ok := h.PermissionRegistry.Consume("run-replay", "req_1"); ok { + t.Fatal("registry pending entry remained after first modern decision") + } + if got := h.Bus.HistoryLen(); got != 1 { + t.Fatalf("event history after first decision = %d, want 1", got) + } + + replay := doPermissionDecideAsUser(h, "user-a", body) + assertModernReceipt(t, replay, "control-replay", "task-replay", "run-replay", "req_1", "allow", true) + if got := h.Bus.HistoryLen(); got != 1 { + t.Fatalf("event history after replay = %d, want 1", got) + } + if got := h.permissionReceipts.len(); got != 1 { + t.Fatalf("receipt cache length = %d, want 1", got) + } +} + +func TestPostPermissionDecideModernConflictIsFailClosed(t *testing.T) { + h := newTestHandler() + h.HubJWTSecret = "test-secret" + defer h.Bus.Close() + seedModernApprovalRun(t, h.Store, "conflict", "run-conflict", "user-a", "task-conflict") + registerModernPendingRegistry(t, h, "run-conflict", "req_1", "proj-receipt-conflict", "thread-receipt-conflict") + + body := modernPermissionDecisionBody("control-ok", "task-conflict", "run-conflict", "req_1", "allow", "yes") + first := doPermissionDecideAsUser(h, "user-a", body) + assertModernReceipt(t, first, "control-ok", "task-conflict", "run-conflict", "req_1", "allow", false) + if got := h.Bus.HistoryLen(); got != 1 { + t.Fatalf("event history after first decision = %d, want 1", got) + } + + altered := doPermissionDecideAsUser(h, "user-a", modernPermissionDecisionBody("control-ok", "task-conflict", "run-conflict", "req_1", "deny", "changed")) + if altered.Code != http.StatusConflict { + t.Fatalf("altered replay status = %d, want 409; body=%s", altered.Code, altered.Body.String()) + } + assertErrorCode(t, altered.Body.String(), errcode.ErrConflict.Code) + if got := h.Bus.HistoryLen(); got != 1 { + t.Fatalf("altered replay published %d events, want 1", got) + } + + different := doPermissionDecideAsUser(h, "user-a", modernPermissionDecisionBody("control-other", "task-conflict", "run-conflict", "req_1", "allow", "yes")) + if different.Code != http.StatusConflict { + t.Fatalf("different control status = %d, want 409; body=%s", different.Code, different.Body.String()) + } + assertErrorCode(t, different.Body.String(), errcode.ErrConflict.Code) + if got := h.Bus.HistoryLen(); got != 1 { + t.Fatalf("different control published %d events, want 1", got) + } +} + +func TestPostPermissionDecideModernOwnershipAndBindingFailClosed(t *testing.T) { + h := newTestHandler() + h.HubJWTSecret = "test-secret" + defer h.Bus.Close() + seedModernApprovalRun(t, h.Store, "auth", "run-auth", "user-a", "task-auth") + seedModernApprovalRun(t, h.Store, "auth-other", "run-auth-other", "user-a", "task-auth") + _ = registerModernBrokerPending(t, h, "run-auth", "req_1", "proj-receipt-auth", "thread-receipt-auth") + _ = registerModernBrokerPending(t, h, "run-auth-other", "req_2", "proj-receipt-auth-other", "thread-receipt-auth-other") + registerModernPendingRegistry(t, h, "run-auth", "req_1", "proj-receipt-auth", "thread-receipt-auth") + registerModernPendingRegistry(t, h, "run-auth-other", "req_2", "proj-receipt-auth-other", "thread-receipt-auth-other") + + nonOwner := doPermissionDecideAsUser(h, "user-b", modernPermissionDecisionBody("control-user", "task-auth", "run-auth", "req_1", "allow", "x")) + if nonOwner.Code != http.StatusNotFound { + t.Fatalf("non-owner status = %d, want 404; body=%s", nonOwner.Code, nonOwner.Body.String()) + } + if got := h.Bus.HistoryLen(); got != 0 { + t.Fatalf("non-owner published %d events, want 0", got) + } + if !h.ensurePermissionBroker().PendingPermission("run-auth", "req_1") { + t.Fatal("non-owner consumed the pending request") + } + + wrongTask := doPermissionDecideAsUser(h, "user-a", modernPermissionDecisionBody("control-task", "wrong-task", "run-auth", "req_1", "allow", "x")) + if wrongTask.Code != http.StatusNotFound { + t.Fatalf("wrong task status = %d, want 404; body=%s", wrongTask.Code, wrongTask.Body.String()) + } + if got := h.Bus.HistoryLen(); got != 0 { + t.Fatalf("wrong task published %d events, want 0", got) + } + if !h.ensurePermissionBroker().PendingPermission("run-auth", "req_1") { + t.Fatal("wrong task consumed the pending request") + } + + owner := doPermissionDecideAsUser(h, "user-a", modernPermissionDecisionBody("control-ok", "task-auth", "run-auth", "req_1", "allow", "x")) + assertModernReceipt(t, owner, "control-ok", "task-auth", "run-auth", "req_1", "allow", false) + if got := h.Bus.HistoryLen(); got != 1 { + t.Fatalf("owner published %d events, want 1", got) + } + + crossRun := doPermissionDecideAsUser(h, "user-a", modernPermissionDecisionBody("control-ok", "task-auth", "run-auth-other", "req_2", "allow", "x")) + if crossRun.Code != http.StatusConflict { + t.Fatalf("cross-run status = %d, want 409; body=%s", crossRun.Code, crossRun.Body.String()) + } + if got := h.Bus.HistoryLen(); got != 1 { + t.Fatalf("cross-run published %d events, want 1", got) + } + if !h.ensurePermissionBroker().PendingPermission("run-auth-other", "req_2") { + t.Fatal("cross-run consumed the other run's pending request") + } +} + +func TestPostPermissionDecideModernCapacityFullRejectsBeforeEffect(t *testing.T) { + h := newTestHandler() + h.HubJWTSecret = "test-secret" + h.permissionReceipts = newPermissionReceiptCache(1, time.Minute) + defer h.Bus.Close() + seedModernApprovalRun(t, h.Store, "cap-a", "run-cap-a", "user-a", "task-cap-a") + seedModernApprovalRun(t, h.Store, "cap-b", "run-cap-b", "user-a", "task-cap-b") + _ = registerModernBrokerPending(t, h, "run-cap-a", "req_a", "proj-receipt-cap-a", "thread-receipt-cap-a") + _ = registerModernBrokerPending(t, h, "run-cap-b", "req_b", "proj-receipt-cap-b", "thread-receipt-cap-b") + registerModernPendingRegistry(t, h, "run-cap-a", "req_a", "proj-receipt-cap-a", "thread-receipt-cap-a") + registerModernPendingRegistry(t, h, "run-cap-b", "req_b", "proj-receipt-cap-b", "thread-receipt-cap-b") + + first := doPermissionDecideAsUser(h, "user-a", modernPermissionDecisionBody("control-a", "task-cap-a", "run-cap-a", "req_a", "allow", "x")) + assertModernReceipt(t, first, "control-a", "task-cap-a", "run-cap-a", "req_a", "allow", false) + + full := doPermissionDecideAsUser(h, "user-a", modernPermissionDecisionBody("control-b", "task-cap-b", "run-cap-b", "req_b", "allow", "x")) + if full.Code != http.StatusTooManyRequests { + t.Fatalf("capacity full status = %d, want 429; body=%s", full.Code, full.Body.String()) + } + assertErrorCode(t, full.Body.String(), errcode.ErrTooManyRequests.Code) + if got := h.Bus.HistoryLen(); got != 1 { + t.Fatalf("capacity full published %d events, want 1", got) + } + if !h.ensurePermissionBroker().PendingPermission("run-cap-b", "req_b") { + t.Fatal("capacity full consumed the pending request") + } + if got := h.permissionReceipts.len(); got != 1 { + t.Fatalf("receipt cache length = %d, want 1", got) + } +} + +func TestPostPermissionDecideModernConcurrentSameKeyOneWinner(t *testing.T) { + h := newTestHandler() + h.HubJWTSecret = "test-secret" + h.permissionReceipts = newPermissionReceiptCache(8, time.Minute) + defer h.Bus.Close() + seedModernApprovalRun(t, h.Store, "concurrent", "run-concurrent", "user-a", "task-concurrent") + wait := registerModernBrokerPending(t, h, "run-concurrent", "req_1", "proj-receipt-concurrent", "thread-receipt-concurrent") + registerModernPendingRegistry(t, h, "run-concurrent", "req_1", "proj-receipt-concurrent", "thread-receipt-concurrent") + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + resultCh := make(chan adapters.PermissionDecision, 1) + go func() { + resultCh <- wait(ctx) + }() + + const n = 8 + body := modernPermissionDecisionBody("control-concurrent", "task-concurrent", "run-concurrent", "req_1", "allow", "yes") + start := make(chan struct{}) + results := make(chan *httptest.ResponseRecorder, n) + for i := 0; i < n; i++ { + go func() { + <-start + results <- doPermissionDecideAsUser(h, "user-a", body) + }() + } + close(start) + + applied := 0 + deduplicated := 0 + for i := 0; i < n; i++ { + rec := <-results + if rec.Code != http.StatusOK { + t.Fatalf("concurrent status = %d, want 200; body=%s", rec.Code, rec.Body.String()) + } + data := modernReceiptData(t, rec) + if got, _ := data["applied"].(bool); !got { + t.Fatalf("concurrent receipt not applied: %#v", data) + } + if got, _ := data["deduplicated"].(bool); got { + deduplicated++ + } else { + applied++ + } + } + if applied != 1 || deduplicated != n-1 { + t.Fatalf("winner/applied=%d replay/deduplicated=%d, want 1/%d", applied, deduplicated, n-1) + } + if h.ensurePermissionBroker().PendingPermission("run-concurrent", "req_1") { + t.Fatal("concurrent broker pending remained") + } + if _, ok := h.PermissionRegistry.Consume("run-concurrent", "req_1"); ok { + t.Fatal("concurrent registry pending remained") + } + if got := h.Bus.HistoryLen(); got != 1 { + t.Fatalf("concurrent event history = %d, want 1", got) + } + select { + case got := <-resultCh: + if got.Behavior != "allow" || got.Message != "yes" { + t.Fatalf("concurrent broker decision = %#v", got) + } + case <-time.After(time.Second): + t.Fatal("concurrent broker waiter was not woken") + } +} + +func TestPermissionDecisionReceiptsCapability(t *testing.T) { + h := newTestHandler() + defer h.Bus.Close() + rec := httptest.NewRecorder() + h.GetHealth(rec, httptest.NewRequest(http.MethodGet, "/v1/health", nil)) + var health struct { + Capabilities map[string]bool `json:"capabilities"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &health); err != nil { + t.Fatalf("decode health: %v", err) + } + if !health.Capabilities["permissionDecisionReceipts"] { + t.Fatalf("health capabilities missing permissionDecisionReceipts: %#v", health.Capabilities) + } +} +func TestPermissionReceiptCacheExpiresAndNeverGrowsPastCapacity(t *testing.T) { + now := time.Unix(1000, 0) + cache := newPermissionReceiptCache(2, time.Minute).withClock(func() time.Time { return now }) + entry := permissionReceipt{ + ControlID: "control-ttl", + HubTaskID: "task-ttl", + RunID: "run-ttl", + RequestID: "req-ttl", + Decision: "allow", + } + if result := cache.apply(entry, func() bool { return true }); !result.Applied { + t.Fatalf("first apply = %#v", result) + } + now = now.Add(30 * time.Second) + if result := cache.apply(entry, func() bool { return false }); !result.Deduplicated { + t.Fatalf("warm replay = %#v", result) + } + now = now.Add(31 * time.Second) + if got := cache.len(); got != 0 { + t.Fatalf("expired receipt remaining in cache = %d, want 0", got) + } + + capacity := newPermissionReceiptCache(1, time.Minute) + calls := 0 + if result := capacity.apply(entry, func() bool { calls++; return true }); !result.Applied { + t.Fatalf("capacity first apply = %#v", result) + } + other := entry + other.ControlID = "control-other" + other.RunID = "run-other" + other.RequestID = "req-other" + if result := capacity.apply(other, func() bool { calls++; return true }); !result.Full { + t.Fatalf("capacity full apply = %#v", result) + } + if calls != 1 { + t.Fatalf("capacity full callback ran %d times, want 1", calls) + } + if got := capacity.len(); got != 1 { + t.Fatalf("capacity full cache length = %d, want 1", got) + } +}