diff --git a/docs/mcp/configuration.md b/docs/mcp/configuration.md index 36aa126b..01f5d07d 100644 --- a/docs/mcp/configuration.md +++ b/docs/mcp/configuration.md @@ -104,8 +104,9 @@ mcp: identity from the authenticated request and POSTs `{server: , subject}`, so each user gets **their own** token *and its own connection*. It is **inherently lazy**: `required: true` is rejected (there is no user at - startup), and until the platform has a grant for that user the call fails - auth-required. + startup), and until the platform has a grant for that user the call + **pauses** on the [auth-required gate](#delegated-consent--the-auth-required-gate-330) + (rather than failing) until they consent. - **`tools.schemas`** is **required** for `type: user` — the server has no connection at startup (no user), so it can't run `tools/list`. The platform **materializes** the tool schemas from the registry entry into config, and @@ -175,6 +176,51 @@ mcp: > platform re-materializes the registry entry (a redeploy) — the same > snapshot semantics as the `allow: ["*"]` discovery filter. +#### Delegated consent — the auth-required gate (#330) + +When a `type: user` call has **no grant yet** for the requesting user, the +tool call does not fail — it **pauses** on the auth-required gate, the user is +prompted to consent, and the call **resumes** with their token once a grant +exists. This is the delegated analog of the [DEFER](../security/defer-decisions.md) +park/resume: DEFER waits for a human to *approve an action*; the auth gate +waits for a user to *complete OAuth consent*. + +``` +type: user call, no grant → PARK (task status → auth-required) + → prompt the user to consent + → user consents → platform holds a grant + → RESUME → the call re-resolves and proceeds +``` + +- **One prompt per user, not per call.** The gate is keyed by + `{subject, server}`, so a user's concurrent calls (in any task) share **one** + gate and **one** consent; a single grant resumes them all. +- **The gate never sees a token.** It only unblocks the executor to re-resolve + through the normal delegated path — the token is fetched by the resolver, not + handed through the agent (AARM R10; `design-tool-registry.md` §18.5). +- **Bounded.** A call parks for at most the gate timeout (default 10m), then + fails auth-required. Audit events: `mcp_auth_required` → `mcp_auth_resolved` + / `mcp_auth_timeout`. + +**Resuming a parked call — two modes (§18.4):** + +| Mode | Consent delivery + callback | Resume signal to Forge | +|------|-----------------------------|------------------------| +| **Managed** | The platform prompts (Slack DM / console), hosts the OAuth callback, and holds the token. | `POST /mcp/consent` with `{subject, server}` (optionally `granted:false` to refuse). Carries **no token** — a pure "a grant now exists, re-resolve" signal. | +| **Standalone** | Forge hosts its own loopback callback `GET /mcp/oauth/callback`. | The callback validates the OAuth `state` (single-use, expiring, **session-bound** — cross-session/replayed/expired callbacks are rejected), exchanges the code for a token, then resumes. | + +> **Never resume before the grant exists.** In standalone mode the callback +> resumes the gate **only after** the code→token exchange succeeds — resuming +> with no stored token would just re-park the call. Delegation follows +> authorization. + +> **Swappable token store.** The per-user access-token cache is the +> `SubjectTokenStore` interface (in-process by default). A managed broker can +> substitute a shared/durable implementation so grants survive restarts and are +> shared across replicas — the resolver and agent are unchanged. Only +> short-lived **access** tokens live here; refresh tokens never leave the +> broker. + #### OAuth discovery & dynamic client registration (#316) For `type: oauth`, `client_id` / `authorize_url` / `token_url` are all diff --git a/forge-cli/runtime/mcp_authgate.go b/forge-cli/runtime/mcp_authgate.go new file mode 100644 index 00000000..7c419f80 --- /dev/null +++ b/forge-cli/runtime/mcp_authgate.go @@ -0,0 +1,237 @@ +package runtime + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" + + "github.com/initializ/forge/forge-cli/server" + "github.com/initializ/forge/forge-core/a2a" + "github.com/initializ/forge/forge-core/auth" + "github.com/initializ/forge/forge-core/mcp" + coreruntime "github.com/initializ/forge/forge-core/runtime" + "github.com/initializ/forge/forge-core/security/authgate" +) + +// ConsentDeliverer delivers an MCP auth-required consent prompt to the +// requesting user (e.g. a Slack DM with a "Connect Atlassian" link, or an +// A2A `auth-required` artifact). It is the auth-gate analog of +// DeferralNotifier. +// +// Mode split (design-tool-registry.md §18.4): in MANAGED mode the platform +// owns delivery + the consent callback + token custody, so Forge is handed +// a deliverer that hands off to the platform. In STANDALONE mode the default +// is nil (no delivery yet) until the loopback resolver lands (#330 inc 4); +// the gate still parks and the resume endpoint still works, so an operator +// or the platform can drive consent out-of-band. +// +// Best-effort: a delivery error is logged, never fatal — the parked call +// still resumes when a grant arrives via the resume endpoint, and blocking +// on a channel outage would be strictly worse. +type ConsentDeliverer func(ctx context.Context, subject, server, taskID string, deadline time.Time) error + +// mcpAuthGate is the runtime implementation of adapters.AuthGate (#330). It +// bridges the pure authgate.Engine to the runtime concerns the engine +// deliberately doesn't know about: identity extraction, task-status flips, +// consent delivery, and audit. +type mcpAuthGate struct { + engine *authgate.Engine + store TaskStatusStore // flips task status → auth-required while parked + audit *coreruntime.AuditLogger + deliverer ConsentDeliverer // optional; nil ⇒ no prompt delivery (park + resume still work) + logger coreruntime.Logger +} + +// Await parks the executor because `server` has no grant for the requesting +// user in ctx, resuming once consent lands (→ nil, the caller re-resolves) +// or the wait ends without it (→ error wrapping mcp.ErrNoToken, so the tool +// adapter classifies it as `no_token` — the same reason code as the failure +// it replaced). +func (g *mcpAuthGate) Await(ctx context.Context, server string) error { + subject := delegatedSubject(ctx) + if subject == "" { + // No requesting user ⇒ a gate can't be addressed to anyone. Fail + // through as auth-required rather than parking a nameless call. + return fmt.Errorf("%w: no requesting user in context to consent for %q", mcp.ErrNoToken, server) + } + taskID := coreruntime.TaskIDFromContext(ctx) + correlationID := coreruntime.CorrelationIDFromContext(ctx) + + handle, first, err := g.engine.Await(subject, server, authgate.Spec{TaskID: taskID}) + if err != nil { + return fmt.Errorf("%w: %v", mcp.ErrNoToken, err) + } + + // Flip THIS call's task → auth-required so parallel GET /tasks/{id} + // readers see it's blocked on consent, not hung. Every parked caller + // flips its own task; only the gate creator (first) delivers a prompt, + // so one user gets one prompt no matter how many calls pile on. + originalStatus := g.setAuthRequired(taskID, subject, server) + + if first { + g.emit(ctx, coreruntime.EventMCPAuthRequired, correlationID, taskID, map[string]any{ + "server": server, + "subject": subject, + "deadline": handle.Deadline().UTC().Format(time.RFC3339), + "timeout_ms": time.Until(handle.Deadline()).Milliseconds(), + }) + g.deliver(ctx, subject, server, taskID, handle.Deadline()) + } + + start := time.Now() + res, waitErr := handle.WaitCtx(ctx) + if waitErr != nil { + // ctx cancelled — caller abandoned the request. Restore status; the + // engine tears the gate down when the last waiter leaves. + g.restore(taskID, originalStatus) + return waitErr + } + waitMs := time.Since(start).Milliseconds() + + switch res.Decision { + case authgate.DecisionGranted: + g.restore(taskID, originalStatus) + g.emit(ctx, coreruntime.EventMCPAuthResolved, correlationID, taskID, map[string]any{ + "server": server, "subject": subject, "wait_ms": waitMs, + }) + return nil + default: // DecisionTimeout / DecisionCanceled + g.emit(ctx, coreruntime.EventMCPAuthTimeout, correlationID, taskID, map[string]any{ + "server": server, "subject": subject, "wait_ms": waitMs, "decision": string(res.Decision), + }) + return fmt.Errorf("%w: consent for %q not granted (%s)", mcp.ErrNoToken, server, res.Decision) + } +} + +// setAuthRequired flips the task to auth-required, returning the prior status +// so Await can restore it on resume. nil store (unit tests) ⇒ no-op. +func (g *mcpAuthGate) setAuthRequired(taskID, subject, server string) a2a.TaskStatus { + if g.store == nil || taskID == "" { + return a2a.TaskStatus{} + } + return g.store.SetStatus(taskID, a2a.TaskStatus{ + State: a2a.TaskStateAuthRequired, + Message: &a2a.Message{ + Role: a2a.MessageRoleAgent, + Parts: []a2a.Part{ + a2a.NewTextPart(fmt.Sprintf( + "Authorization required: connect %s to continue (as %s)", server, subject)), + }, + }, + }) +} + +func (g *mcpAuthGate) restore(taskID string, prev a2a.TaskStatus) { + if g.store == nil || taskID == "" { + return + } + g.store.SetStatus(taskID, prev) +} + +func (g *mcpAuthGate) deliver(ctx context.Context, subject, server, taskID string, deadline time.Time) { + if g.deliverer == nil { + return + } + if err := g.deliverer(ctx, subject, server, taskID, deadline); err != nil && g.logger != nil { + g.logger.Warn("mcp consent delivery failed", map[string]any{ + "server": server, "subject": subject, "error": err.Error(), + }) + } +} + +func (g *mcpAuthGate) emit(ctx context.Context, event, correlationID, taskID string, fields map[string]any) { + if g.audit == nil { + return + } + g.audit.EmitFromContext(ctx, coreruntime.AuditEvent{ + Event: event, + CorrelationID: correlationID, + TaskID: taskID, + Fields: fields, + }) +} + +// delegatedSubject extracts the consenting user from ctx: email preferred +// (stable, human-meaningful, matches the delegated token subject), else the +// opaque UserID. Mirrors mcp.delegatedSubject, duplicated here to avoid +// exporting it across the module boundary. +func delegatedSubject(ctx context.Context) string { + id := auth.IdentityFromContext(ctx) + if id == nil { + return "" + } + if id.Email != "" { + return id.Email + } + return id.UserID +} + +// registerMCPConsentEndpoint wires POST /mcp/consent so the platform (managed +// mode) or an operator (standalone) can signal that a user's consent +// completed and a grant now exists — resuming every call parked on +// {subject, server}. Mirrors POST /tasks/{id}/decisions: 404 when no call is +// parked, 400 on a malformed body, 200 on resume. +// +// The endpoint carries NO token — Forge never receives the credential. It is +// a pure "a grant now exists, re-resolve" signal; the token is fetched +// through the normal delegated resolver on resume. +func (r *Runner) registerMCPConsentEndpoint(srv *server.Server, auditLogger *coreruntime.AuditLogger) { + if r.authGateEngine == nil { + return + } + srv.RegisterHTTPHandler("POST /mcp/consent", makeMCPConsentHandler(r.authGateEngine, auditLogger)) +} + +// makeMCPConsentHandler is extracted so tests can exercise the status-code +// paths without a full server. +func makeMCPConsentHandler(engine *authgate.Engine, auditLogger *coreruntime.AuditLogger) http.HandlerFunc { + return func(w http.ResponseWriter, req *http.Request) { + var body struct { + Subject string `json:"subject"` + Server string `json:"server"` + // Granted defaults true (the signal is "consent completed"). + // false lets the platform report a refusal so the parked call + // fails fast instead of idling to its timeout. + Granted *bool `json:"granted"` + } + if err := json.NewDecoder(req.Body).Decode(&body); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid body: " + err.Error()}) + return + } + subject := strings.TrimSpace(body.Subject) + serverName := strings.TrimSpace(body.Server) + if subject == "" || serverName == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "subject and server are required"}) + return + } + if _, ok := engine.Peek(subject, serverName); !ok { + writeJSON(w, http.StatusNotFound, map[string]string{"error": "no parked call for this subject/server"}) + return + } + decision := authgate.DecisionGranted + if body.Granted != nil && !*body.Granted { + decision = authgate.DecisionTimeout // explicit refusal → fail fast + } + if err := engine.Resolve(subject, serverName, decision); err != nil { + // Race: the gate resolved (timeout / another signal) between Peek + // and Resolve. 409 is the honest signal. + writeJSON(w, http.StatusConflict, map[string]string{"error": err.Error()}) + return + } + if auditLogger != nil { + auditLogger.Emit(coreruntime.AuditEvent{ + Event: coreruntime.EventMCPAuthResolved, + Fields: map[string]any{ + "server": serverName, "subject": subject, + "decision": string(decision), "via": "consent_endpoint", + }, + }) + } + writeJSON(w, http.StatusOK, map[string]any{ + "subject": subject, "server": serverName, "decision": string(decision), + }) + } +} diff --git a/forge-cli/runtime/mcp_authgate_test.go b/forge-cli/runtime/mcp_authgate_test.go new file mode 100644 index 00000000..caadc01d --- /dev/null +++ b/forge-cli/runtime/mcp_authgate_test.go @@ -0,0 +1,198 @@ +package runtime + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/initializ/forge/forge-core/a2a" + "github.com/initializ/forge/forge-core/auth" + "github.com/initializ/forge/forge-core/mcp" + coreruntime "github.com/initializ/forge/forge-core/runtime" + "github.com/initializ/forge/forge-core/security/authgate" +) + +// userCtx carries a requesting user + a task id so the gate can address the +// consent and flip that task's status. Reuses fakeStatusStore from +// defer_test.go. +func userCtx(subject, taskID string) context.Context { + ctx := auth.WithIdentity(context.Background(), &auth.Identity{Email: subject}) + return coreruntime.WithTaskID(ctx, taskID) +} + +// A parked call resumes (Await → nil) once consent is signalled, and the +// task is flipped auth-required then restored. +func TestMCPAuthGate_ParksThenResumesOnConsent(t *testing.T) { + store := newFakeStatusStore() + store.SetStatus("task-1", a2a.TaskStatus{State: a2a.TaskStateWorking}) // seed prior status + var mu sync.Mutex + var delivered int + gate := &mcpAuthGate{ + engine: authgate.New(), + store: store, + deliverer: func(context.Context, string, string, string, time.Time) error { + mu.Lock() + delivered++ + mu.Unlock() + return nil + }, + } + + done := make(chan error, 1) + go func() { done <- gate.Await(userCtx("alice@corp.com", "task-1"), "atl") }() + + // Wait for the call to park, then signal consent. + waitFor(t, func() bool { _, ok := gate.engine.Peek("alice@corp.com", "atl"); return ok }) + if err := gate.engine.Resolve("alice@corp.com", "atl", authgate.DecisionGranted); err != nil { + t.Fatalf("Resolve: %v", err) + } + + select { + case err := <-done: + if err != nil { + t.Fatalf("Await after consent must return nil, got %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Await did not resume after consent") + } + mu.Lock() + d := delivered + mu.Unlock() + if d != 1 { + t.Fatalf("consent delivered %d times, want 1", d) + } + // working(seed) → auth-required → working(restored) + got := store.statuses() + if len(got) < 3 || got[1] != a2a.TaskStateAuthRequired || got[len(got)-1] != a2a.TaskStateWorking { + t.Fatalf("status trail = %v, want working→auth-required→working", got) + } +} + +// A gate that never gets consent fails the call with an ErrNoToken-wrapping +// error (so the tool adapter classifies it `no_token`). +func TestMCPAuthGate_TimeoutFailsAsNoToken(t *testing.T) { + gate := &mcpAuthGate{engine: authgate.New(), store: newFakeStatusStore()} + // Spec timeout is set inside Await via engine default; force the outcome + // by resolving as timeout ourselves after it parks. + done := make(chan error, 1) + go func() { done <- gate.Await(userCtx("alice@corp.com", "task-1"), "atl") }() + waitFor(t, func() bool { _, ok := gate.engine.Peek("alice@corp.com", "atl"); return ok }) + _ = gate.engine.Resolve("alice@corp.com", "atl", authgate.DecisionTimeout) + + select { + case err := <-done: + if !errors.Is(err, mcp.ErrNoToken) { + t.Fatalf("timeout must wrap ErrNoToken, got %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Await did not return on timeout") + } +} + +// No requesting user in ctx ⇒ the gate can't be addressed; Await fails +// immediately (no park) with ErrNoToken. +func TestMCPAuthGate_NoSubject_FailsFast(t *testing.T) { + gate := &mcpAuthGate{engine: authgate.New(), store: newFakeStatusStore()} + err := gate.Await(context.Background(), "atl") + if !errors.Is(err, mcp.ErrNoToken) { + t.Fatalf("no-subject Await must fail as ErrNoToken, got %v", err) + } + if gate.engine.Pending() != 0 { + t.Fatalf("no gate should be parked for a nameless call; Pending=%d", gate.engine.Pending()) + } +} + +func TestMCPConsentHandler_StatusCodes(t *testing.T) { + engine := authgate.New() + h := makeMCPConsentHandler(engine, nil) + + // 400 — malformed body. + if code := doConsent(h, "atl", []byte("{not json")); code != http.StatusBadRequest { + t.Errorf("malformed body → %d, want 400", code) + } + // 400 — missing fields. + if code := doConsent(h, "", mustJSON(map[string]any{"server": "atl"})); code != http.StatusBadRequest { + t.Errorf("missing subject → %d, want 400", code) + } + // 404 — no parked call. + if code := doConsent(h, "atl", mustJSON(map[string]any{"subject": "nobody@corp.com", "server": "atl"})); code != http.StatusNotFound { + t.Errorf("no parked call → %d, want 404", code) + } + + // 200 — a real parked call resolves. + done := make(chan authgate.Resolution, 1) + handle, _, err := engine.Await("alice@corp.com", "atl", authgate.Spec{Timeout: time.Hour}) + if err != nil { + t.Fatal(err) + } + go func() { r, _ := handle.WaitCtx(context.Background()); done <- r }() + if code := doConsent(h, "atl", mustJSON(map[string]any{"subject": "alice@corp.com", "server": "atl"})); code != http.StatusOK { + t.Fatalf("valid consent → %d, want 200", code) + } + select { + case r := <-done: + if !r.Granted() { + t.Fatalf("parked call resolved %q, want granted", r.Decision) + } + case <-time.After(2 * time.Second): + t.Fatal("consent endpoint did not resume the parked call") + } +} + +// granted:false is an explicit refusal → the parked call fails fast (timeout +// decision) rather than idling to its window. +func TestMCPConsentHandler_ExplicitRefusal(t *testing.T) { + engine := authgate.New() + h := makeMCPConsentHandler(engine, nil) + handle, _, err := engine.Await("alice@corp.com", "atl", authgate.Spec{Timeout: time.Hour}) + if err != nil { + t.Fatal(err) + } + done := make(chan authgate.Resolution, 1) + go func() { r, _ := handle.WaitCtx(context.Background()); done <- r }() + + refuse := mustJSON(map[string]any{"subject": "alice@corp.com", "server": "atl", "granted": false}) + if code := doConsent(h, "atl", refuse); code != http.StatusOK { + t.Fatalf("refusal → %d, want 200", code) + } + select { + case r := <-done: + if r.Granted() { + t.Fatal("explicit refusal must NOT resume as granted") + } + case <-time.After(2 * time.Second): + t.Fatal("refusal did not resolve the parked call") + } +} + +// --- helpers --- + +func doConsent(h http.HandlerFunc, _ string, body []byte) int { + req := httptest.NewRequest("POST", "/mcp/consent", bytes.NewReader(body)) + rec := httptest.NewRecorder() + h(rec, req) + return rec.Code +} + +func mustJSON(v any) []byte { + b, _ := json.Marshal(v) + return b +} + +func waitFor(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(2 * time.Millisecond) + } + t.Fatal("condition not met within 2s") +} diff --git a/forge-cli/runtime/mcp_consent_callback.go b/forge-cli/runtime/mcp_consent_callback.go new file mode 100644 index 00000000..e1161689 --- /dev/null +++ b/forge-cli/runtime/mcp_consent_callback.go @@ -0,0 +1,205 @@ +package runtime + +import ( + "net/http" + "sync" + "time" + + "github.com/initializ/forge/forge-cli/server" + "github.com/initializ/forge/forge-core/llm/oauth" + coreruntime "github.com/initializ/forge/forge-core/runtime" + "github.com/initializ/forge/forge-core/security/authgate" +) + +// This file is the STANDALONE-mode consent loop (design-tool-registry.md +// §18.4). In managed mode the platform hosts the callback + token custody +// and only signals Forge via POST /mcp/consent (mcp_authgate.go). In +// standalone mode Forge hosts its own loopback callback here: it validates +// the OAuth `state` it issued, exchanges the code for a token via the +// injected completer, and resumes the parked call. +// +// The security-critical piece is the state binding: a callback is honored +// only if its `state` was issued by us, hasn't been used, hasn't expired, +// and arrives in the same session that started the flow — rejecting +// cross-session, replayed, and stale callbacks (the #317/#330 acceptance). + +// stateBinding is what an issued OAuth `state` is bound to. The callback +// must match all of it (modulo an empty session) to be honored. +type stateBinding struct { + subject string + server string + session string // the initiating session; callback must match (cross-session guard) + deadline time.Time +} + +// stateBinder issues single-use, expiring OAuth `state` values bound to the +// {subject, server, session} that initiated a consent flow, and validates +// them on the callback. In-process (like the gate itself); a restart drops +// pending flows, which is safe — the user just re-initiates. +type stateBinder struct { + mu sync.Mutex + m map[string]stateBinding + ttl time.Duration + now func() time.Time +} + +// defaultStateTTL bounds how long an issued state is honored — long enough +// to click through an IdP consent screen, short enough that a leaked state +// is quickly useless. +const defaultStateTTL = 10 * time.Minute + +func newStateBinder(ttl time.Duration) *stateBinder { + if ttl <= 0 { + ttl = defaultStateTTL + } + return &stateBinder{m: make(map[string]stateBinding), ttl: ttl, now: time.Now} +} + +// Issue mints a cryptographically-random state bound to {subject, server, +// session} and returns it for embedding in the authorize URL. +func (s *stateBinder) Issue(subject, server, session string) (string, error) { + state, err := oauth.GenerateState() + if err != nil { + return "", err + } + s.mu.Lock() + defer s.mu.Unlock() + s.sweepLocked() // opportunistic GC so abandoned flows don't accumulate + s.m[state] = stateBinding{subject: subject, server: server, session: session, deadline: s.now().Add(s.ttl)} + return state, nil +} + +// Consume looks up and REMOVES the binding for state. It returns ok=false +// for an unknown state (forged), an already-used state (replay — the entry +// is gone after the first Consume), or an expired one. Single-use + expiry +// are enforced here; the caller enforces the session match. +func (s *stateBinder) Consume(state string) (stateBinding, bool) { + if state == "" { + return stateBinding{}, false + } + s.mu.Lock() + defer s.mu.Unlock() + b, ok := s.m[state] + if !ok { + return stateBinding{}, false // unknown or already consumed (replay) + } + delete(s.m, state) // single-use: a replay of this state now finds nothing + if s.now().After(b.deadline) { + return stateBinding{}, false // expired + } + return b, true +} + +// sweepLocked drops expired bindings. Caller holds s.mu. +func (s *stateBinder) sweepLocked() { + now := s.now() + for k, b := range s.m { + if now.After(b.deadline) { + delete(s.m, k) + } + } +} + +// CallbackCompleter exchanges an OAuth authorization code for a token and +// stores it for {subject, server}, so the resumed call finds a grant. The +// standalone interactive resolver provides one; when nil the loopback +// callback is NOT registered (managed mode hosts its own callback and never +// hands Forge a code). +type CallbackCompleter func(subject, server, code string) error + +// sessionFromRequest extracts the caller's session id for the cross-session +// check. Prefers an explicit forge session header; falls back to a session +// cookie (the header isn't set on a browser redirect, so the cookie is the +// real path for the IdP callback). An empty result fails the mandatory +// session match in the handler — the network-exposed callback never +// downgrades to single-use+expiry alone. +func sessionFromRequest(r *http.Request) string { + if h := r.Header.Get("X-Forge-Session"); h != "" { + return h + } + if c, err := r.Cookie("forge_session"); err == nil { + return c.Value + } + return "" +} + +// registerMCPCallbackEndpoint wires the standalone loopback callback. It is +// registered ONLY when a CallbackCompleter is set (standalone interactive +// mode); managed deployments never register it. +func (r *Runner) registerMCPCallbackEndpoint(srv *server.Server, auditLogger *coreruntime.AuditLogger) { + if r.authGateEngine == nil || r.callbackCompleter == nil { + return + } + if r.stateBinder == nil { + r.stateBinder = newStateBinder(defaultStateTTL) + } + srv.RegisterHTTPHandler("GET /mcp/oauth/callback", + makeMCPCallbackHandler(r.stateBinder, r.authGateEngine, r.callbackCompleter, sessionFromRequest, auditLogger)) +} + +// makeMCPCallbackHandler validates the state, enforces the session match, +// exchanges the code for a token, and resumes the parked call. Extracted so +// tests can drive every rejection path without a real IdP. +func makeMCPCallbackHandler( + binder *stateBinder, + engine *authgate.Engine, + complete CallbackCompleter, + sessionOf func(*http.Request) string, + auditLogger *coreruntime.AuditLogger, +) http.HandlerFunc { + return func(w http.ResponseWriter, req *http.Request) { + q := req.URL.Query() + state, code := q.Get("state"), q.Get("code") + if state == "" || code == "" { + http.Error(w, "missing state or code", http.StatusBadRequest) + return + } + // Single-use + expiry are enforced in Consume; unknown/replayed/ + // expired all collapse to !ok — a forged or stale callback can't + // resume anyone. + b, ok := binder.Consume(state) + if !ok { + http.Error(w, "invalid, expired, or already-used state", http.StatusBadRequest) + return + } + // Cross-session guard — MANDATORY. This endpoint is unauthenticated + // (a browser redirect carries no bearer token) and is registered on + // the network-exposed main server (cfg.Host, possibly 0.0.0.0), so + // the state binding is its ENTIRE security. Single-use + expiry alone + // would let anyone who obtains a state within its TTL complete the + // flow, so we additionally require the callback to land in the SAME + // session that started it. An empty bound session is a config/Issue- + // side bug (a network-exposed flow must bind a session) and is + // rejected fail-closed rather than silently downgrading to + // single-use+expiry. A future loopback-bound variant that wanted to + // relax this would bind the listener to localhost. + if b.session == "" || sessionOf(req) != b.session { + http.Error(w, "state/session mismatch", http.StatusBadRequest) + return + } + // Exchange the code for a token and store it for {subject, server}. + // Only AFTER this succeeds do we resume — resolving the gate with no + // grant would just re-park the call (delegation follows + // authorization: never resume before the grant exists). + if err := complete(b.subject, b.server, code); err != nil { + http.Error(w, "authorization exchange failed", http.StatusBadGateway) + return + } + // Grant exists now → wake every call parked on {subject, server}. + // A missing gate (call already timed out / was canceled) is benign: + // the token is stored, so a fresh call will just succeed. + _ = engine.Resolve(b.subject, b.server, authgate.DecisionGranted) + if auditLogger != nil { + auditLogger.Emit(coreruntime.AuditEvent{ + Event: coreruntime.EventMCPAuthResolved, + Fields: map[string]any{ + "server": b.server, "subject": b.subject, "via": "loopback_callback", + }, + }) + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`Connected` + + `

Authorization complete. You can return to your conversation.

`)) + } +} diff --git a/forge-cli/runtime/mcp_consent_callback_test.go b/forge-cli/runtime/mcp_consent_callback_test.go new file mode 100644 index 00000000..e9a85d34 --- /dev/null +++ b/forge-cli/runtime/mcp_consent_callback_test.go @@ -0,0 +1,275 @@ +package runtime + +import ( + "context" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/initializ/forge/forge-core/security/authgate" +) + +// A freshly-issued state consumes once and carries its binding. +func TestStateBinder_IssueConsume(t *testing.T) { + b := newStateBinder(time.Hour) + state, err := b.Issue("alice@corp.com", "atl", "sess-1") + if err != nil { + t.Fatal(err) + } + got, ok := b.Consume(state) + if !ok { + t.Fatal("freshly issued state must consume") + } + if got.subject != "alice@corp.com" || got.server != "atl" || got.session != "sess-1" { + t.Fatalf("binding = %+v, want alice/atl/sess-1", got) + } +} + +// Replay: a state is single-use — the second Consume finds nothing. +func TestStateBinder_ReplayRejected(t *testing.T) { + b := newStateBinder(time.Hour) + state, _ := b.Issue("alice@corp.com", "atl", "sess-1") + if _, ok := b.Consume(state); !ok { + t.Fatal("first consume must succeed") + } + if _, ok := b.Consume(state); ok { + t.Fatal("replayed state must be rejected (single-use)") + } +} + +// Unknown/forged state is rejected. +func TestStateBinder_UnknownRejected(t *testing.T) { + b := newStateBinder(time.Hour) + if _, ok := b.Consume("never-issued"); ok { + t.Fatal("forged state must be rejected") + } + if _, ok := b.Consume(""); ok { + t.Fatal("empty state must be rejected") + } +} + +// Expired state is rejected even though it was validly issued. +func TestStateBinder_ExpiredRejected(t *testing.T) { + b := newStateBinder(time.Hour) + now := time.Now() + b.now = func() time.Time { return now } + state, _ := b.Issue("alice@corp.com", "atl", "sess-1") + // Advance past the TTL. + b.now = func() time.Time { return now.Add(2 * time.Hour) } + if _, ok := b.Consume(state); ok { + t.Fatal("expired state must be rejected") + } +} + +// Issue opportunistically sweeps expired bindings so abandoned flows don't +// accumulate. +func TestStateBinder_SweepsExpired(t *testing.T) { + b := newStateBinder(time.Minute) + now := time.Now() + b.now = func() time.Time { return now } + _, _ = b.Issue("alice@corp.com", "atl", "s1") + _, _ = b.Issue("bob@corp.com", "atl", "s2") + // Jump past TTL and issue a third — the sweep should drop the first two. + b.now = func() time.Time { return now.Add(2 * time.Minute) } + _, _ = b.Issue("carol@corp.com", "atl", "s3") + b.mu.Lock() + n := len(b.m) + b.mu.Unlock() + if n != 1 { + t.Fatalf("after sweep, %d bindings remain, want 1 (only carol's)", n) + } +} + +// fakeCompleter records the code exchange and "stores" a grant. +type fakeCompleter struct { + mu sync.Mutex + calls int + lastCode string + err error +} + +func (f *fakeCompleter) complete(_, _, code string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.calls++ + f.lastCode = code + return f.err +} + +func newCallback(t *testing.T, binder *stateBinder, engine *authgate.Engine, comp *fakeCompleter) http.HandlerFunc { + t.Helper() + return makeMCPCallbackHandler(binder, engine, comp.complete, sessionFromRequest, nil) +} + +// The happy path: valid state + code → exchange runs → the parked call +// resumes granted. +func TestMCPCallback_HappyPath_ResumesGate(t *testing.T) { + binder := newStateBinder(time.Hour) + engine := authgate.New() + comp := &fakeCompleter{} + + // Park a call for alice. + handle, _, err := engine.Await("alice@corp.com", "atl", authgate.Spec{Timeout: time.Hour}) + if err != nil { + t.Fatal(err) + } + resumed := make(chan authgate.Resolution, 1) + go func() { r, _ := handle.WaitCtx(context.Background()); resumed <- r }() + + state, _ := binder.Issue("alice@corp.com", "atl", "sess-1") + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/mcp/oauth/callback?state="+state+"&code=authcode123", nil) + req.Header.Set("X-Forge-Session", "sess-1") + newCallback(t, binder, engine, comp)(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("callback → %d, want 200", rec.Code) + } + if comp.calls != 1 || comp.lastCode != "authcode123" { + t.Fatalf("completer calls=%d code=%q, want 1/authcode123", comp.calls, comp.lastCode) + } + select { + case r := <-resumed: + if !r.Granted() { + t.Fatalf("parked call resolved %q, want granted", r.Decision) + } + case <-time.After(2 * time.Second): + t.Fatal("callback did not resume the parked call") + } +} + +// Cross-session: a valid state used from a DIFFERENT session is rejected, +// the code is NOT exchanged, and no gate resumes. +func TestMCPCallback_CrossSessionRejected(t *testing.T) { + binder := newStateBinder(time.Hour) + engine := authgate.New() + comp := &fakeCompleter{} + state, _ := binder.Issue("alice@corp.com", "atl", "sess-INITIATED") + + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/mcp/oauth/callback?state="+state+"&code=x", nil) + req.Header.Set("X-Forge-Session", "sess-ATTACKER") + newCallback(t, binder, engine, comp)(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("cross-session callback → %d, want 400", rec.Code) + } + if comp.calls != 0 { + t.Fatal("a cross-session callback must NOT exchange the code") + } +} + +// An empty bound session is rejected fail-closed — the network-exposed +// callback must never downgrade to single-use+expiry alone (finding 2). +func TestMCPCallback_EmptySessionRejected(t *testing.T) { + binder := newStateBinder(time.Hour) + engine := authgate.New() + comp := &fakeCompleter{} + // Issue with an empty session (a config/Issue-side bug on a network- + // exposed flow) — even a request that also presents no session must fail. + state, _ := binder.Issue("alice@corp.com", "atl", "") + + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/mcp/oauth/callback?state="+state+"&code=x", nil) + // No X-Forge-Session / cookie either. + newCallback(t, binder, engine, comp)(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("empty-session binding → %d, want 400 (fail-closed)", rec.Code) + } + if comp.calls != 0 { + t.Fatal("empty-session callback must NOT exchange the code") + } +} + +// A request that presents no session against a session-bound state is +// rejected (the mandatory cross-session guard, missing-side). +func TestMCPCallback_MissingRequestSessionRejected(t *testing.T) { + binder := newStateBinder(time.Hour) + engine := authgate.New() + comp := &fakeCompleter{} + state, _ := binder.Issue("alice@corp.com", "atl", "sess-1") + + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/mcp/oauth/callback?state="+state+"&code=x", nil) + // Deliberately no session header/cookie on the request. + newCallback(t, binder, engine, comp)(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("missing request session → %d, want 400", rec.Code) + } + if comp.calls != 0 { + t.Fatal("a sessionless request must NOT exchange the code") + } +} + +// A replayed callback (state already consumed) is rejected without a second +// exchange. +func TestMCPCallback_ReplayRejected(t *testing.T) { + binder := newStateBinder(time.Hour) + engine := authgate.New() + comp := &fakeCompleter{} + state, _ := binder.Issue("alice@corp.com", "atl", "sess-1") + + mk := func() *httptest.ResponseRecorder { + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/mcp/oauth/callback?state="+state+"&code=x", nil) + req.Header.Set("X-Forge-Session", "sess-1") + newCallback(t, binder, engine, comp)(rec, req) + return rec + } + if code := mk().Code; code != http.StatusOK { + t.Fatalf("first callback → %d, want 200", code) + } + if code := mk().Code; code != http.StatusBadRequest { + t.Fatalf("replayed callback → %d, want 400", code) + } + if comp.calls != 1 { + t.Fatalf("code exchanged %d times, want exactly 1 (replay must not re-exchange)", comp.calls) + } +} + +// Missing params are a 400. +func TestMCPCallback_MissingParams(t *testing.T) { + binder := newStateBinder(time.Hour) + engine := authgate.New() + comp := &fakeCompleter{} + h := newCallback(t, binder, engine, comp) + for _, q := range []string{"?state=x", "?code=y", ""} { + rec := httptest.NewRecorder() + h(rec, httptest.NewRequest("GET", "/mcp/oauth/callback"+q, nil)) + if rec.Code != http.StatusBadRequest { + t.Errorf("query %q → %d, want 400", q, rec.Code) + } + } +} + +// A completer failure (bad code / IdP down) surfaces as 502 and does NOT +// resume the gate — never resume before the grant exists. +func TestMCPCallback_ExchangeFailure_DoesNotResume(t *testing.T) { + binder := newStateBinder(time.Hour) + engine := authgate.New() + comp := &fakeCompleter{err: context.DeadlineExceeded} + + handle, _, _ := engine.Await("alice@corp.com", "atl", authgate.Spec{Timeout: time.Hour}) + resumed := make(chan struct{}, 1) + go func() { _, _ = handle.WaitCtx(context.Background()); resumed <- struct{}{} }() + + state, _ := binder.Issue("alice@corp.com", "atl", "sess-1") + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/mcp/oauth/callback?state="+state+"&code=x", nil) + req.Header.Set("X-Forge-Session", "sess-1") + newCallback(t, binder, engine, comp)(rec, req) + + if rec.Code != http.StatusBadGateway { + t.Fatalf("exchange failure → %d, want 502", rec.Code) + } + select { + case <-resumed: + t.Fatal("gate must NOT resume when the token exchange failed") + case <-time.After(150 * time.Millisecond): + // still parked — correct. + } +} diff --git a/forge-cli/runtime/runner.go b/forge-cli/runtime/runner.go index 33ad70af..41fc6b91 100644 --- a/forge-cli/runtime/runner.go +++ b/forge-cli/runtime/runner.go @@ -43,6 +43,7 @@ import ( "github.com/initializ/forge/forge-core/scheduler" "github.com/initializ/forge/forge-core/secrets" "github.com/initializ/forge/forge-core/security" + "github.com/initializ/forge/forge-core/security/authgate" deferengine "github.com/initializ/forge/forge-core/security/deferpolicy" "github.com/initializ/forge/forge-core/security/intent" "github.com/initializ/forge/forge-core/security/stepup" @@ -168,6 +169,10 @@ type Runner struct { intentEngine *intent.Engine // R3 (#208) intent-alignment engine; nil when disabled stepUpEngine *stepup.Engine // R4b (#210) step-up authorization engine; nil when disabled deferEngine *deferengine.Engine // R4c (#211) deferred-authorization engine; nil when disabled + authGateEngine *authgate.Engine // R10 (#330) MCP auth-required gate; nil until an MCP manager with a type=user server is wired + consentDeliverer ConsentDeliverer // optional: delivers MCP consent prompts to channels (#330); nil in standalone (no delivery yet) + callbackCompleter CallbackCompleter // optional: standalone loopback code→token exchange (#330); nil ⇒ no loopback callback (managed hosts its own) + stateBinder *stateBinder // standalone OAuth state binding (single-use/expiring/session-bound); lazily built with the callback endpoint taskStore *a2a.TaskStore // shared task store, populated once srv is built; read by defer hook when it fires platformCommandGuard *coreruntime.PlatformCommandGuard // #238 (ASI02) operator-authored command deny, applied to every tool call; empty when no layer declares denied_command_patterns } @@ -206,6 +211,23 @@ func (r *Runner) SetDeferralNotifier(fn DeferralNotifier) { r.deferralNotifier = fn } +// SetConsentDeliverer sets the callback used to deliver MCP auth-required +// consent prompts (#330) — the managed platform injects one that hands off +// to its consent flow; standalone leaves it nil until the loopback resolver +// lands. Must be called before Run(). +func (r *Runner) SetConsentDeliverer(fn ConsentDeliverer) { + r.consentDeliverer = fn +} + +// SetCallbackCompleter enables the STANDALONE loopback consent callback +// (#330): the injected func exchanges an OAuth code for a token and stores +// it for {subject, server}. When set, GET /mcp/oauth/callback is registered; +// when nil (managed mode), it is not — the platform hosts its own callback. +// Must be called before Run(). +func (r *Runner) SetCallbackCompleter(fn CallbackCompleter) { + r.callbackCompleter = fn +} + // ResolveAuth resolves the auth token early (before Run). This is needed so // channel adapters can be configured with the token before Run() blocks. // Safe to call multiple times — subsequent calls are no-ops. @@ -1016,12 +1038,27 @@ func (r *Runner) Run(ctx context.Context) error { return fmt.Errorf("starting mcp manager: %w", err) } else if mcpMgr != nil { defer func() { _ = mcpMgr.Stop() }() + // Auth-required gate (#330): parks a delegated (type=user) call + // that has no grant for the requesting user until they consent, + // instead of failing. Harmless for non-delegated servers — only + // an ErrNoToken from the per-user resolver ever trips it. + if r.authGateEngine == nil { + r.authGateEngine = authgate.New() + } + mcpGate := &mcpAuthGate{ + engine: r.authGateEngine, + store: r, // Runner.SetStatus flips task status while parked + audit: auditLogger, + deliverer: r.consentDeliverer, + logger: r.logger, + } for _, h := range mcpMgr.Tools() { mcpTool, ctorErr := adapters.NewMCPTool(adapters.MCPToolOpts{ Server: h.Server, Descriptor: h.Descriptor, Client: h.Client, Resolver: h.Resolver, // per-call client resolution (#317) + AuthGate: mcpGate, // park-on-no-grant, resume-on-consent (#330) Audit: auditLogger, }) if ctorErr != nil { @@ -2291,6 +2328,16 @@ func (r *Runner) registerRESTHandlers(srv *server.Server, executor coreruntime.A // resolve pending deferrals. No-op wire (nothing registered) // when defer is disabled. r.registerDecisionsEndpoint(srv, auditLogger) + + // R10 (#330) — consent-resume endpoint the platform/operator calls + // when a delegated MCP grant lands, unblocking calls parked on the + // auth-required gate. No-op wire when no type=user MCP server is active. + r.registerMCPConsentEndpoint(srv, auditLogger) + + // R10 (#330) — standalone loopback OAuth callback. Registered only when + // a CallbackCompleter is set (standalone interactive mode); managed + // deployments host their own callback and skip this. + r.registerMCPCallbackEndpoint(srv, auditLogger) } // serveJWKS is the handler for /.well-known/forge-audit-keys. Split diff --git a/forge-core/auth/middleware.go b/forge-core/auth/middleware.go index ec63eedd..143376e8 100644 --- a/forge-core/auth/middleware.go +++ b/forge-core/auth/middleware.go @@ -28,16 +28,28 @@ type errorResponse struct { // emitted by the handler); removable after one release cycle func DefaultSkipPaths() map[string]bool { return map[string]bool{ - "GET /": true, - "GET /.well-known/agent-card.json": true, - "GET /.well-known/agent.json": true, - "GET /healthz": true, - "GET /health": true, + "GET /": true, + "GET /.well-known/agent-card.json": true, + "GET /.well-known/agent.json": true, + "GET /healthz": true, + "GET /health": true, + // The standalone MCP OAuth callback (#330) is a browser redirect + // FROM the IdP — it carries no bearer token, so it must bypass auth + // or the middleware 401s it before the handler runs. Its authenticity + // is the single-use, session-bound OAuth `state` it validates, not a + // bearer token. Exempt unconditionally: it is registered only in + // standalone mode (a CallbackCompleter is set); in managed mode the + // path is unregistered and a request just 404s after the skip. NOTE + // the contrast with POST /mcp/consent, which is deliberately NOT + // exempt — the managed platform authenticates when it posts the + // resume signal. + "GET /mcp/oauth/callback": true, "OPTIONS /": true, "OPTIONS /.well-known/agent-card.json": true, "OPTIONS /.well-known/agent.json": true, "OPTIONS /healthz": true, "OPTIONS /health": true, + "OPTIONS /mcp/oauth/callback": true, } } diff --git a/forge-core/auth/middleware_test.go b/forge-core/auth/middleware_test.go index 238d0b00..942b60ca 100644 --- a/forge-core/auth/middleware_test.go +++ b/forge-core/auth/middleware_test.go @@ -125,6 +125,26 @@ func TestMiddleware(t *testing.T) { path: "/tasks/send", wantStatus: http.StatusUnauthorized, }, + { + // #330: the standalone OAuth callback is a tokenless browser + // redirect — it MUST bypass auth (its authenticity is the state + // binding). Pins the exemption so a future refactor can't 401 it. + name: "GET /mcp/oauth/callback is public (tokenless IdP redirect)", + opts: MiddlewareOptions{Chain: chain, SkipPaths: DefaultSkipPaths()}, + method: "GET", + path: "/mcp/oauth/callback", + wantStatus: http.StatusOK, + }, + { + // #330: the consent RESUME signal is the opposite — the managed + // platform authenticates when it posts it, so it must NOT be + // exempt. Guards against accidentally skip-listing both together. + name: "POST /mcp/consent requires auth", + opts: MiddlewareOptions{Chain: chain, SkipPaths: DefaultSkipPaths()}, + method: "POST", + path: "/mcp/consent", + wantStatus: http.StatusUnauthorized, + }, { name: "case insensitive Bearer prefix", opts: MiddlewareOptions{Chain: chain, SkipPaths: DefaultSkipPaths()}, diff --git a/forge-core/mcp/platform_token.go b/forge-core/mcp/platform_token.go index 612ef88a..90a123a4 100644 --- a/forge-core/mcp/platform_token.go +++ b/forge-core/mcp/platform_token.go @@ -58,6 +58,11 @@ type PlatformSourceConfig struct { AgentIdentity string Ref string HTTPClient *http.Client + + // SubjectStore swaps the per-user delegated-token cache (§18.8 #2). + // Used only by the delegated (type=user) resolver. nil ⇒ the in-process + // default; a managed broker substitutes a shared/durable store. + SubjectStore SubjectTokenStore } // Token returns a valid access token, fetching from the platform when the @@ -180,8 +185,10 @@ type delegatedTokenSource struct { ref string client *http.Client - mu sync.Mutex - cache map[string]cachedToken // key: subject + // store caches access tokens per subject. Swappable (§18.8 #2): the + // default is in-process; a managed broker can substitute a shared/durable + // impl so grants survive restarts and are shared across replicas. + store SubjectTokenStore } type cachedToken struct { @@ -190,34 +197,31 @@ type cachedToken struct { } func newDelegatedTokenSource(cfg PlatformSourceConfig) *delegatedTokenSource { + store := cfg.SubjectStore + if store == nil { + store = newMemSubjectTokenStore(0) + } return &delegatedTokenSource{ endpoint: cfg.TokenEndpoint, identity: cfg.AgentIdentity, ref: cfg.Ref, client: cfg.HTTPClient, + store: store, } } // TokenForSubject returns a valid access token for the requesting user, -// fetching from the platform when the per-subject cache is empty/expiring. +// fetching from the platform when the per-subject store is empty/expiring. func (d *delegatedTokenSource) TokenForSubject(ctx context.Context, subject string) (string, error) { if subject == "" { return "", fmt.Errorf("%w: delegated token requires a requesting-user subject", ErrNoToken) } - // Fast path: a cached, unexpired token for this subject. The lock is - // NOT held across the network fetch, so a slow fetch for user A never - // blocks user B — the multi-user path is the whole point of #317. - d.mu.Lock() - if c, ok := d.cache[subject]; ok { - if c.token != "" && time.Now().Before(c.expiresAt.Add(-platformTokenSkew)) { - d.mu.Unlock() - return c.token, nil - } - // Evict the stale entry — don't hold a sensitive token past use - // (#327 review finding 2). - delete(d.cache, subject) + // Fast path: a cached, unexpired token for this subject. The store is not + // held across the network fetch, so a slow fetch for user A never blocks + // user B — the multi-user path is the whole point of #317. + if tok, ok := d.store.Get(subject); ok { + return tok, nil } - d.mu.Unlock() tok, ttl, status, err := doPlatformTokenRequest(ctx, d.client, d.endpoint, d.identity, d.ref, subject) if err != nil { @@ -230,20 +234,7 @@ func (d *delegatedTokenSource) TokenForSubject(ctx context.Context, subject stri return "", fmt.Errorf("%w: platform token endpoint returned %d for server %q (subject %q)", ErrProtocolError, status, d.ref, subject) } - d.mu.Lock() - if d.cache == nil { - d.cache = map[string]cachedToken{} - } - // Opportunistic sweep so the map can't grow unbounded with one-shot - // users' stale tokens — a background-free TTL bound (#327 review). - now := time.Now() - for s, c := range d.cache { - if !now.Before(c.expiresAt) { - delete(d.cache, s) - } - } - d.cache[subject] = cachedToken{token: tok, expiresAt: now.Add(ttl)} - d.mu.Unlock() + d.store.Put(subject, tok, ttl) return tok, nil } diff --git a/forge-core/mcp/subject_token_store.go b/forge-core/mcp/subject_token_store.go new file mode 100644 index 00000000..3bc8c6e9 --- /dev/null +++ b/forge-core/mcp/subject_token_store.go @@ -0,0 +1,87 @@ +package mcp + +import ( + "sync" + "time" +) + +// SubjectTokenStore caches delegated access tokens per requesting user +// (#317/#330). It is the seam §18.8 #2 calls out: the default is +// process-memory (memSubjectTokenStore), but a managed broker can substitute +// a shared/durable implementation so per-user grants survive restarts and +// are shared across replicas, without the resolver or the agent changing. +// +// Contract: +// - Get must NOT return an expired token — freshness (including any early- +// refresh skew) is the store's policy, so the resolver just asks and +// trusts the answer. +// - The store holds only ACCESS tokens (short-lived). Refresh tokens never +// reach it — they never leave the broker (invariant 8). +// - Implementations must be safe for concurrent use by many goroutines +// (distinct users resolve in parallel — the whole point of #317). +type SubjectTokenStore interface { + // Get returns a cached, unexpired access token for subject, or ok=false + // when there is none (or it is too close to expiry to reuse). + Get(subject string) (token string, ok bool) + // Put stores token for subject, valid for ttl from now. + Put(subject, token string, ttl time.Duration) + // Evict drops any cached token for subject (e.g. on a 401 from the + // downstream server — a stale token must not linger). + Evict(subject string) +} + +// memSubjectTokenStore is the default in-process SubjectTokenStore: a +// per-subject map with early-refresh skew and an opportunistic sweep so it +// can't grow unbounded with one-shot users, without a background goroutine. +type memSubjectTokenStore struct { + mu sync.Mutex + m map[string]cachedToken + skew time.Duration + now func() time.Time +} + +// newMemSubjectTokenStore builds the default store. skew<=0 uses +// platformTokenSkew (re-fetch slightly before nominal expiry so a token +// never expires mid-flight downstream). +func newMemSubjectTokenStore(skew time.Duration) *memSubjectTokenStore { + if skew <= 0 { + skew = platformTokenSkew + } + return &memSubjectTokenStore{m: make(map[string]cachedToken), skew: skew, now: time.Now} +} + +func (s *memSubjectTokenStore) Get(subject string) (string, bool) { + s.mu.Lock() + defer s.mu.Unlock() + c, ok := s.m[subject] + if !ok { + return "", false + } + if c.token != "" && s.now().Before(c.expiresAt.Add(-s.skew)) { + return c.token, true + } + // Stale — evict eagerly so a sensitive token isn't held past use + // (#327 review finding 2). + delete(s.m, subject) + return "", false +} + +func (s *memSubjectTokenStore) Put(subject, token string, ttl time.Duration) { + s.mu.Lock() + defer s.mu.Unlock() + now := s.now() + // Opportunistic sweep so the map can't grow unbounded with one-shot + // users' stale tokens — a background-free TTL bound (#327 review). + for sub, c := range s.m { + if !now.Before(c.expiresAt) { + delete(s.m, sub) + } + } + s.m[subject] = cachedToken{token: token, expiresAt: now.Add(ttl)} +} + +func (s *memSubjectTokenStore) Evict(subject string) { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.m, subject) +} diff --git a/forge-core/mcp/subject_token_store_test.go b/forge-core/mcp/subject_token_store_test.go new file mode 100644 index 00000000..553270bd --- /dev/null +++ b/forge-core/mcp/subject_token_store_test.go @@ -0,0 +1,85 @@ +package mcp + +import ( + "testing" + "time" +) + +func TestMemSubjectTokenStore_GetPutEvict(t *testing.T) { + s := newMemSubjectTokenStore(0) + if _, ok := s.Get("alice@corp.com"); ok { + t.Fatal("empty store must miss") + } + s.Put("alice@corp.com", "tok-a", time.Hour) + s.Put("bob@corp.com", "tok-b", time.Hour) + + if tok, ok := s.Get("alice@corp.com"); !ok || tok != "tok-a" { + t.Fatalf("alice = (%q, %v), want tok-a/true", tok, ok) + } + // Distinct users never share a token. + if tok, _ := s.Get("bob@corp.com"); tok != "tok-b" { + t.Fatalf("bob = %q, want tok-b (per-subject isolation)", tok) + } + + s.Evict("alice@corp.com") + if _, ok := s.Get("alice@corp.com"); ok { + t.Fatal("evicted token must be gone") + } + if _, ok := s.Get("bob@corp.com"); !ok { + t.Fatal("evicting alice must not touch bob") + } +} + +// A token within the skew window of expiry is treated as already stale so it +// never expires mid-flight downstream — and is evicted on that read. +func TestMemSubjectTokenStore_SkewAndEvictOnStale(t *testing.T) { + now := time.Unix(1_000_000, 0) + s := newMemSubjectTokenStore(30 * time.Second) + s.now = func() time.Time { return now } + + s.Put("alice@corp.com", "tok-a", 20*time.Second) // expires in 20s, < 30s skew + if _, ok := s.Get("alice@corp.com"); ok { + t.Fatal("a token inside the skew window must be treated as stale") + } + // The stale read must have evicted it (no sensitive token held past use). + s.mu.Lock() + _, present := s.m["alice@corp.com"] + s.mu.Unlock() + if present { + t.Fatal("stale token must be evicted on read") + } +} + +func TestMemSubjectTokenStore_ExpiredMiss(t *testing.T) { + now := time.Unix(1_000_000, 0) + s := newMemSubjectTokenStore(0) + s.now = func() time.Time { return now } + s.Put("alice@corp.com", "tok-a", time.Minute) + s.now = func() time.Time { return now.Add(2 * time.Minute) } + if _, ok := s.Get("alice@corp.com"); ok { + t.Fatal("expired token must miss") + } +} + +// Put opportunistically sweeps expired entries so one-shot users don't leak. +func TestMemSubjectTokenStore_PutSweepsExpired(t *testing.T) { + now := time.Unix(1_000_000, 0) + s := newMemSubjectTokenStore(0) + s.now = func() time.Time { return now } + s.Put("alice@corp.com", "tok-a", time.Minute) + s.Put("bob@corp.com", "tok-b", time.Minute) + + // Advance past expiry, then Put a third — the sweep drops alice & bob. + s.now = func() time.Time { return now.Add(2 * time.Minute) } + s.Put("carol@corp.com", "tok-c", time.Minute) + + s.mu.Lock() + n := len(s.m) + s.mu.Unlock() + if n != 1 { + t.Fatalf("after sweep %d entries remain, want 1 (only carol)", n) + } +} + +// The default store satisfies the interface (compile-time guard). +var _ SubjectTokenStore = (*memSubjectTokenStore)(nil) diff --git a/forge-core/runtime/audit.go b/forge-core/runtime/audit.go index 4dc6b6f2..e3c0427a 100644 --- a/forge-core/runtime/audit.go +++ b/forge-core/runtime/audit.go @@ -66,6 +66,22 @@ const ( EventMCPToolConflict = "mcp_tool_conflict" EventMCPTokenRefresh = "mcp_token_refresh" + // Auth-required gate events (R10 / #330). A delegated (type=user) MCP + // call with no grant for the requesting user PARKS on the authgate + // engine instead of failing; these trace park → resume/timeout. The + // gate never sees a token — it only unblocks the executor to re-resolve + // once the platform holds a grant. + // + // EventMCPAuthRequired — a call parked awaiting user consent; the + // task flipped to `auth-required` and (first + // waiter only) a consent prompt was delivered. + // EventMCPAuthResolved — consent arrived (or the wait was canceled); + // parked call(s) resumed and re-resolved. + // EventMCPAuthTimeout — no consent within the window; the call fails. + EventMCPAuthRequired = "mcp_auth_required" + EventMCPAuthResolved = "mcp_auth_resolved" + EventMCPAuthTimeout = "mcp_auth_timeout" + // Agent Card events. Emitted once at agent startup with the // finalized A2A Agent Card content for traceability. Carries the // card's name, version, URL, protocolVersion, skill count, and a diff --git a/forge-core/security/authgate/authgate.go b/forge-core/security/authgate/authgate.go new file mode 100644 index 00000000..84f049d8 --- /dev/null +++ b/forge-core/security/authgate/authgate.go @@ -0,0 +1,292 @@ +// Package authgate implements the AARM R10 auth-required gate — the +// pause-and-resume primitive behind delegated MCP consent (#330). +// +// It is the sibling of deferpolicy (R4c): where DEFER parks the executor +// until a human *approves an action*, authgate parks it until a user +// *completes an OAuth consent* and the platform holds a grant. Both share +// the same shape — a goroutine blocks on a handle, an out-of-band signal +// resolves it, a timeout auto-fails — but the semantics differ enough to +// justify a distinct engine: +// +// - Keyed by {subject, server}, NOT taskID. A grant is per user per +// server; once alice consents to "atl", EVERY parked call of hers to +// "atl" resumes — across tasks/sessions. So one gate fans out to many +// waiters, and one Resolve wakes them all. (deferpolicy is one waiter +// per taskID.) +// - The resolution is binary — granted or timed-out. There is no +// approver/reject: the "approver" is the OAuth flow itself, and the +// signal is "a grant now exists," delivered by the platform's consent +// callback or a bounded re-poll of the delegated resolver. +// +// The gate never mints a token and never sees one: it only unblocks the +// executor so it can re-resolve through the normal delegated path, which +// now succeeds because the grant exists (delegation follows authorization, +// design-tool-registry.md §18.5). Token custody stays with the resolver. +// +// In-process only, like deferpolicy — a restart abandons pending gates +// (each blocked goroutine's stack disappears). The Engine is the seam for +// a future persistent implementation. +package authgate + +import ( + "context" + "errors" + "fmt" + "sync" + "time" +) + +// defaultTimeout bounds how long a call parks awaiting consent before it +// auto-fails. Consent is a human-in-the-loop step (open a link, approve in +// an IdP), so the window is generous relative to deferpolicy's action +// approvals — but still finite so a never-answered prompt can't wedge a +// tool call forever. +const defaultTimeout = 10 * time.Minute + +// Decision is how a parked gate resolved. +type Decision string + +const ( + // DecisionGranted: the user consented and the platform now holds a + // grant — the executor should re-resolve and proceed. + DecisionGranted Decision = "granted" + // DecisionTimeout: no consent arrived within the window — the call + // fails, same as the pre-gate ErrNoToken behavior. + DecisionTimeout Decision = "timeout" + // DecisionCanceled: every waiter abandoned the gate (all contexts + // cancelled) before consent — the gate is torn down with nothing left + // to resume. + DecisionCanceled Decision = "canceled" +) + +// Resolution is the outcome delivered to every waiter on a gate. +type Resolution struct { + Decision Decision + At time.Time +} + +// Granted reports whether the resolution means "proceed." +func (r Resolution) Granted() bool { return r.Decision == DecisionGranted } + +// Spec carries the delivery/audit context for a gate. None of these fields +// are part of the dedup key — the key is {subject, server} — they travel +// with the FIRST waiter so the consent prompt can be addressed and audited. +type Spec struct { + // Timeout overrides defaultTimeout when > 0. + Timeout time.Duration + // TaskID / Session identify the request that first tripped the gate, + // for audit and for routing the consent prompt back to the right + // conversation. Later joiners' task/session are intentionally dropped — + // one consent serves them all. + TaskID string + Session string +} + +// Handle is a pending gate. Executors obtain one from Engine.Await and +// block on WaitCtx; the resume signal (or the timeout) resolves it, +// broadcasting the Resolution to every waiter at once via a closed channel. +// +// A Handle is shared: concurrent Await calls for the same {subject, server} +// return the SAME *Handle. It is opaque — the engine owns its lifecycle. +type Handle struct { + eng *Engine + key string + subject string + server string + spec Spec + deadline time.Time + + // done is closed exactly once, by whichever of Resolve / timeout / the + // last-waiter-leaves path wins the resolved Once. Closing (not sending) + // is the fan-out: every WaitCtx selecting on it wakes together. + done chan struct{} + res Resolution // written before done is closed; read after + resolved sync.Once + + // waiters counts live WaitCtx callers, guarded by Engine.mu. When it + // drops to zero before consent, the gate is torn down (DecisionCanceled) + // so an abandoned prompt doesn't idle to its full timeout. + waiters int +} + +// Subject returns the consenting user the gate is keyed to. +func (h *Handle) Subject() string { return h.subject } + +// Server returns the MCP server name the grant is for. +func (h *Handle) Server() string { return h.server } + +// Spec returns the delivery/audit context captured at first Await. +func (h *Handle) Spec() Spec { return h.spec } + +// Deadline returns the absolute time the gate auto-times-out. +func (h *Handle) Deadline() time.Time { return h.deadline } + +// WaitCtx blocks until the gate resolves (consent granted, timeout, or +// cancellation) or ctx is cancelled. On ctx cancellation it detaches this +// waiter — and if it was the last one holding the gate open, tears the gate +// down (DecisionCanceled) so an abandoned prompt doesn't linger to its +// timeout. Returns (Resolution{}, ctx.Err()) on cancellation. +func (h *Handle) WaitCtx(ctx context.Context) (Resolution, error) { + select { + case <-h.done: + return h.res, nil + case <-ctx.Done(): + h.eng.leave(h.key) + return Resolution{}, ctx.Err() + } +} + +// Engine coordinates pending auth-required gates across the runtime. +// Thread-safe. Every gate is single-flighted per {subject, server}: the +// first Await creates it and is told to deliver the consent prompt; later +// Awaits attach silently. +type Engine struct { + mu sync.Mutex + pending map[string]*Handle + timers map[string]*time.Timer + now func() time.Time +} + +// New constructs an empty Engine. +func New() *Engine { + return &Engine{ + pending: make(map[string]*Handle), + timers: make(map[string]*time.Timer), + now: time.Now, + } +} + +// gateKey composes the dedup key. The NUL separator can't appear in an +// email or a server name, so distinct (subject, server) pairs can't collide. +func gateKey(subject, server string) string { return subject + "\x00" + server } + +// Await returns the pending gate for {subject, server}, creating it if none +// exists. The bool is true only for the caller that CREATED the gate — that +// caller (and only that caller) should deliver the consent prompt; joiners +// get false and must not re-deliver, or one user would get N prompts for N +// concurrent calls. +// +// subject and server are required. A blank subject means no requesting user +// is in context, which is a caller bug (a gate can't be addressed to nobody). +func (e *Engine) Await(subject, server string, spec Spec) (*Handle, bool, error) { + if subject == "" { + return nil, false, errors.New("authgate: subject is required") + } + if server == "" { + return nil, false, errors.New("authgate: server is required") + } + if spec.Timeout <= 0 { + spec.Timeout = defaultTimeout + } + key := gateKey(subject, server) + + e.mu.Lock() + defer e.mu.Unlock() + if h, ok := e.pending[key]; ok { + h.waiters++ + return h, false, nil // join the in-flight gate; don't re-prompt + } + h := &Handle{ + eng: e, + key: key, + subject: subject, + server: server, + spec: spec, + deadline: e.now().Add(spec.Timeout), + done: make(chan struct{}), + waiters: 1, + } + e.pending[key] = h + e.timers[key] = time.AfterFunc(spec.Timeout, func() { e.timeout(key) }) + return h, true, nil +} + +// Resolve wakes every waiter on {subject, server} with the given decision +// (normally DecisionGranted, from a consent callback). Idempotent — a second +// Resolve, or a Resolve racing the timeout, is a no-op guarded by the gate's +// sync.Once. Returns an error only when no gate is pending (404-like). +func (e *Engine) Resolve(subject, server string, d Decision) error { + key := gateKey(subject, server) + e.mu.Lock() + h, ok := e.pending[key] + if !ok { + e.mu.Unlock() + return fmt.Errorf("authgate: no pending gate for subject %q server %q", subject, server) + } + e.clear(key) + e.mu.Unlock() + + h.finish(d, e.now()) + return nil +} + +// Peek reports whether a gate is pending for {subject, server}. Used by the +// resume endpoint to 404 signals that target no parked call. +func (e *Engine) Peek(subject, server string) (*Handle, bool) { + e.mu.Lock() + defer e.mu.Unlock() + h, ok := e.pending[gateKey(subject, server)] + return h, ok +} + +// Pending returns the count of currently-parked gates. For health checks +// and startup logs. +func (e *Engine) Pending() int { + e.mu.Lock() + defer e.mu.Unlock() + return len(e.pending) +} + +// timeout fires when a gate's window closes with no consent. +func (e *Engine) timeout(key string) { + e.mu.Lock() + h, ok := e.pending[key] + if !ok { + e.mu.Unlock() // Resolve or leave got here first. + return + } + e.clear(key) + e.mu.Unlock() + h.finish(DecisionTimeout, e.now()) +} + +// leave is called by a waiter whose ctx was cancelled. It decrements the +// gate's live-waiter count; when the last waiter departs before consent, the +// gate is cleared and resolved DecisionCanceled so an abandoned prompt is not +// left idling to its full timeout. If the gate was already resolved (the +// common race: Resolve closed done, then a ctx also fired) it's gone from the +// map and this is a no-op. +func (e *Engine) leave(key string) { + e.mu.Lock() + h, ok := e.pending[key] + if !ok { + e.mu.Unlock() // already resolved/cleared — nothing to release. + return + } + h.waiters-- + if h.waiters > 0 { + e.mu.Unlock() // siblings still parked — keep the gate alive. + return + } + e.clear(key) + e.mu.Unlock() + h.finish(DecisionCanceled, e.now()) +} + +// clear removes a gate + its timer under e.mu. Caller holds the lock. +func (e *Engine) clear(key string) { + if t, ok := e.timers[key]; ok { + t.Stop() + delete(e.timers, key) + } + delete(e.pending, key) +} + +// finish resolves the gate exactly once, publishing res before closing done +// so every WaitCtx observes a fully-written Resolution. +func (h *Handle) finish(d Decision, at time.Time) { + h.resolved.Do(func() { + h.res = Resolution{Decision: d, At: at} + close(h.done) + }) +} diff --git a/forge-core/security/authgate/authgate_test.go b/forge-core/security/authgate/authgate_test.go new file mode 100644 index 00000000..f82b168f --- /dev/null +++ b/forge-core/security/authgate/authgate_test.go @@ -0,0 +1,299 @@ +package authgate + +import ( + "context" + "sync" + "testing" + "time" +) + +// waitResult captures a WaitCtx return for assertions off the waiter goroutine. +type waitResult struct { + res Resolution + err error +} + +// awaitInBackground parks a waiter and returns a channel that yields its +// WaitCtx result. It blocks until the gate exists so callers can assert +// ordering deterministically. +func awaitInBackground(t *testing.T, e *Engine, subject, server string, spec Spec) (<-chan waitResult, *Handle, bool) { + t.Helper() + h, first, err := e.Await(subject, server, spec) + if err != nil { + t.Fatalf("Await(%q,%q): %v", subject, server, err) + } + ch := make(chan waitResult, 1) + go func() { + r, werr := h.WaitCtx(context.Background()) + ch <- waitResult{r, werr} + }() + return ch, h, first +} + +func TestAwait_RequiresSubjectAndServer(t *testing.T) { + e := New() + if _, _, err := e.Await("", "atl", Spec{}); err == nil { + t.Error("blank subject must error") + } + if _, _, err := e.Await("a@x", "", Spec{}); err == nil { + t.Error("blank server must error") + } +} + +// A granted signal wakes the parked waiter with DecisionGranted. +func TestResolve_Granted_WakesWaiter(t *testing.T) { + e := New() + ch, _, first := awaitInBackground(t, e, "alice@corp.com", "atl", Spec{}) + if !first { + t.Fatal("the creating Await must report first=true") + } + if e.Pending() != 1 { + t.Fatalf("Pending = %d, want 1", e.Pending()) + } + + if err := e.Resolve("alice@corp.com", "atl", DecisionGranted); err != nil { + t.Fatalf("Resolve: %v", err) + } + select { + case got := <-ch: + if got.err != nil { + t.Fatalf("WaitCtx err = %v", got.err) + } + if !got.res.Granted() { + t.Fatalf("decision = %q, want granted", got.res.Decision) + } + case <-time.After(2 * time.Second): + t.Fatal("waiter did not wake after Resolve") + } + if e.Pending() != 0 { + t.Fatalf("Pending after Resolve = %d, want 0", e.Pending()) + } +} + +// The decisive fan-out property: N concurrent calls for the same +// {subject, server} share ONE gate — only the first is told to deliver the +// prompt — and ONE Resolve wakes them all. +func TestAwait_FanOut_OnePromptManyWaiters(t *testing.T) { + e := New() + const n = 8 + var firstCount int + chans := make([]<-chan waitResult, n) + for i := range n { + ch, _, first := awaitInBackground(t, e, "alice@corp.com", "atl", Spec{}) + if first { + firstCount++ + } + chans[i] = ch + } + if firstCount != 1 { + t.Fatalf("first=true returned %d times, want exactly 1 (one prompt per user)", firstCount) + } + if e.Pending() != 1 { + t.Fatalf("Pending = %d, want 1 (all share one gate)", e.Pending()) + } + + if err := e.Resolve("alice@corp.com", "atl", DecisionGranted); err != nil { + t.Fatalf("Resolve: %v", err) + } + for i, ch := range chans { + select { + case got := <-ch: + if !got.res.Granted() { + t.Fatalf("waiter %d decision = %q, want granted", i, got.res.Decision) + } + case <-time.After(2 * time.Second): + t.Fatalf("waiter %d did not wake — Resolve must fan out to all", i) + } + } +} + +// Different subjects (or different servers) are independent gates: resolving +// one must not wake the other. +func TestAwait_DistinctKeysAreIndependent(t *testing.T) { + e := New() + chAlice, _, _ := awaitInBackground(t, e, "alice@corp.com", "atl", Spec{}) + chBob, _, bobFirst := awaitInBackground(t, e, "bob@corp.com", "atl", Spec{}) + if !bobFirst { + t.Fatal("bob is a distinct subject → must be first for his own gate") + } + if e.Pending() != 2 { + t.Fatalf("Pending = %d, want 2 distinct gates", e.Pending()) + } + + if err := e.Resolve("alice@corp.com", "atl", DecisionGranted); err != nil { + t.Fatalf("Resolve(alice): %v", err) + } + select { + case <-chAlice: // good + case <-time.After(2 * time.Second): + t.Fatal("alice did not wake") + } + select { + case got := <-chBob: + t.Fatalf("bob woke on alice's grant — keys must be independent: %+v", got) + case <-time.After(150 * time.Millisecond): + // bob still parked — correct. + } +} + +// The timeout auto-fails a gate that never gets consent. +func TestTimeout_AutoFails(t *testing.T) { + e := New() + ch, h, _ := awaitInBackground(t, e, "alice@corp.com", "atl", Spec{Timeout: 40 * time.Millisecond}) + if h.Deadline().IsZero() { + t.Error("deadline must be set") + } + select { + case got := <-ch: + if got.res.Decision != DecisionTimeout { + t.Fatalf("decision = %q, want timeout", got.res.Decision) + } + if got.res.Granted() { + t.Fatal("a timed-out gate must not report Granted") + } + case <-time.After(2 * time.Second): + t.Fatal("gate did not auto-time-out") + } + if e.Pending() != 0 { + t.Fatalf("Pending after timeout = %d, want 0", e.Pending()) + } +} + +// Resolve on a subject/server with no parked gate is a 404-like error. +func TestResolve_NoGate_Errors(t *testing.T) { + e := New() + if err := e.Resolve("nobody@corp.com", "atl", DecisionGranted); err == nil { + t.Error("Resolve with no pending gate must error") + } +} + +// Resolve is idempotent and race-safe against the timeout: whichever wins, +// the waiter sees exactly one resolution and a second Resolve is a no-op +// error (gate already cleared). +func TestResolve_Idempotent(t *testing.T) { + e := New() + ch, _, _ := awaitInBackground(t, e, "alice@corp.com", "atl", Spec{}) + if err := e.Resolve("alice@corp.com", "atl", DecisionGranted); err != nil { + t.Fatalf("first Resolve: %v", err) + } + <-ch + if err := e.Resolve("alice@corp.com", "atl", DecisionGranted); err == nil { + t.Error("second Resolve must error — gate already cleared") + } +} + +// When the last waiter's ctx cancels before consent, the gate is torn down +// (DecisionCanceled) rather than idling to its timeout. +func TestWaitCtx_LastWaiterCancel_TearsDown(t *testing.T) { + e := New() + h, first, err := e.Await("alice@corp.com", "atl", Spec{Timeout: time.Hour}) + if err != nil || !first { + t.Fatalf("Await: first=%v err=%v", first, err) + } + ctx, cancel := context.WithCancel(context.Background()) + ch := make(chan waitResult, 1) + go func() { + r, werr := h.WaitCtx(ctx) + ch <- waitResult{r, werr} + }() + // Let the waiter park, then cancel. + time.Sleep(20 * time.Millisecond) + cancel() + select { + case got := <-ch: + if got.err == nil { + t.Fatalf("WaitCtx must return ctx.Err() on cancel, got res=%+v", got.res) + } + case <-time.After(2 * time.Second): + t.Fatal("WaitCtx did not return after cancel") + } + // The gate must be gone — not left holding a live timer to its hour. + if e.Pending() != 0 { + t.Fatalf("Pending after last-waiter cancel = %d, want 0 (gate torn down)", e.Pending()) + } +} + +// One cancelled waiter among several must NOT tear down the shared gate — +// its siblings stay parked and still resume on consent. +func TestWaitCtx_OneOfManyCancel_KeepsGate(t *testing.T) { + e := New() + // Sibling that stays parked on a background ctx. + survivor, _, err := e.Await("alice@corp.com", "atl", Spec{Timeout: time.Hour}) + if err != nil { + t.Fatal(err) + } + survCh := make(chan waitResult, 1) + go func() { + r, werr := survivor.WaitCtx(context.Background()) + survCh <- waitResult{r, werr} + }() + + // Second waiter on a cancellable ctx (joins the same gate). + joiner, first, err := e.Await("alice@corp.com", "atl", Spec{}) + if err != nil || first { + t.Fatalf("joiner must not be first: first=%v err=%v", first, err) + } + ctx, cancel := context.WithCancel(context.Background()) + joinCh := make(chan waitResult, 1) + go func() { + r, werr := joiner.WaitCtx(ctx) + joinCh <- waitResult{r, werr} + }() + time.Sleep(20 * time.Millisecond) + cancel() + <-joinCh // joiner returns cancelled + + // Gate must still be pending for the survivor. + if e.Pending() != 1 { + t.Fatalf("Pending = %d, want 1 (survivor still parked)", e.Pending()) + } + if err := e.Resolve("alice@corp.com", "atl", DecisionGranted); err != nil { + t.Fatalf("Resolve: %v", err) + } + select { + case got := <-survCh: + if !got.res.Granted() { + t.Fatalf("survivor decision = %q, want granted", got.res.Decision) + } + case <-time.After(2 * time.Second): + t.Fatal("survivor did not resume after a sibling cancelled") + } +} + +// Peek reflects pending state for the resume endpoint's 404 decision. +func TestPeek(t *testing.T) { + e := New() + if _, ok := e.Peek("alice@corp.com", "atl"); ok { + t.Error("Peek must be false before any Await") + } + ch, _, _ := awaitInBackground(t, e, "alice@corp.com", "atl", Spec{}) + if h, ok := e.Peek("alice@corp.com", "atl"); !ok || h.Subject() != "alice@corp.com" { + t.Errorf("Peek after Await = (%v, %v)", h, ok) + } + _ = e.Resolve("alice@corp.com", "atl", DecisionGranted) + <-ch + if _, ok := e.Peek("alice@corp.com", "atl"); ok { + t.Error("Peek must be false after resolve") + } +} + +// Race sanity: many concurrent Awaits + a Resolve under -race must not +// deadlock or double-close. +func TestConcurrentAwaitResolve_Race(t *testing.T) { + e := New() + var wg sync.WaitGroup + for range 32 { + wg.Go(func() { + h, _, err := e.Await("alice@corp.com", "atl", Spec{Timeout: time.Hour}) + if err != nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, _ = h.WaitCtx(ctx) + }) + } + time.Sleep(30 * time.Millisecond) + _ = e.Resolve("alice@corp.com", "atl", DecisionGranted) + wg.Wait() +} diff --git a/forge-core/tools/adapters/mcp_authgate_test.go b/forge-core/tools/adapters/mcp_authgate_test.go new file mode 100644 index 00000000..69e83686 --- /dev/null +++ b/forge-core/tools/adapters/mcp_authgate_test.go @@ -0,0 +1,141 @@ +package adapters + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "github.com/initializ/forge/forge-core/mcp" +) + +// grantOnResumeResolver returns ErrNoToken until the auth gate "grants", +// then hands back a client — modeling the real per-user pool, which 401s +// with no grant and establishes the connection once one exists. +type grantOnResumeResolver struct { + granted *bool + client mcp.Client +} + +func (r grantOnResumeResolver) ClientFor(context.Context) (mcp.Client, error) { + if !*r.granted { + return nil, mcp.ErrNoToken + } + return r.client, nil +} + +// fakeGate models the runtime gate: Await flips the grant (as a real +// consent + platform grant would) and returns its configured error. +type fakeGate struct { + granted *bool + err error + awaitCalls int + sawServer string +} + +func (g *fakeGate) Await(_ context.Context, server string) error { + g.awaitCalls++ + g.sawServer = server + if g.err == nil { + *g.granted = true + } + return g.err +} + +// A delegated call with no grant parks on the gate; when the gate grants, +// Execute re-resolves and the call proceeds (#330). +func TestMCPTool_Execute_AuthGate_ParksThenResumes(t *testing.T) { + t.Parallel() + granted := false + client := &mockClient{res: &mcp.CallToolResult{Content: []mcp.ToolContent{{Type: "text", Text: "ok-after-consent"}}}} + gate := &fakeGate{granted: &granted} + tool, err := NewMCPTool(MCPToolOpts{ + Server: "atl", + Descriptor: mcp.MCPToolDescriptor{Name: "create_issue", InputSchema: json.RawMessage(`{"type":"object"}`)}, + Resolver: grantOnResumeResolver{granted: &granted, client: client}, + AuthGate: gate, + }) + if err != nil { + t.Fatal(err) + } + + got, err := tool.Execute(context.Background(), json.RawMessage(`{}`)) + if err != nil { + t.Fatalf("Execute after consent must succeed, got %v", err) + } + if got != "ok-after-consent" { + t.Fatalf("result = %q, want ok-after-consent", got) + } + if gate.awaitCalls != 1 { + t.Fatalf("Await called %d times, want 1", gate.awaitCalls) + } + if gate.sawServer != "atl" { + t.Fatalf("gate saw server %q, want atl", gate.sawServer) + } +} + +// When the gate gives up (timeout / no consent), the call fails — the pause +// is bounded, not indefinite. +func TestMCPTool_Execute_AuthGate_TimeoutFails(t *testing.T) { + t.Parallel() + granted := false + gate := &fakeGate{granted: &granted, err: errors.New("authgate: consent timed out")} + tool, err := NewMCPTool(MCPToolOpts{ + Server: "atl", + Descriptor: mcp.MCPToolDescriptor{Name: "create_issue", InputSchema: json.RawMessage(`{"type":"object"}`)}, + Resolver: grantOnResumeResolver{granted: &granted}, + AuthGate: gate, + }) + if err != nil { + t.Fatal(err) + } + if _, err := tool.Execute(context.Background(), json.RawMessage(`{}`)); err == nil { + t.Fatal("a gate that never grants must fail the call, not hang or succeed") + } + if gate.awaitCalls != 1 { + t.Fatalf("Await called %d times, want 1", gate.awaitCalls) + } +} + +// With no gate wired (nil), ErrNoToken surfaces directly — the pre-#330 +// behavior for non-delegated servers and standalone/tests. +func TestMCPTool_Execute_NoGate_ErrNoTokenSurfaces(t *testing.T) { + t.Parallel() + granted := false + tool, err := NewMCPTool(MCPToolOpts{ + Server: "atl", + Descriptor: mcp.MCPToolDescriptor{Name: "create_issue", InputSchema: json.RawMessage(`{"type":"object"}`)}, + Resolver: grantOnResumeResolver{granted: &granted}, + // AuthGate deliberately nil. + }) + if err != nil { + t.Fatal(err) + } + _, err = tool.Execute(context.Background(), json.RawMessage(`{}`)) + if !errors.Is(err, mcp.ErrNoToken) { + t.Fatalf("without a gate, ErrNoToken must surface; got %v", err) + } +} + +// The gate is only consulted for ErrNoToken — a different resolver error +// (e.g. transport) is NOT parked; it fails immediately. +func TestMCPTool_Execute_AuthGate_OnlyForNoToken(t *testing.T) { + t.Parallel() + granted := false + gate := &fakeGate{granted: &granted} + tool, err := NewMCPTool(MCPToolOpts{ + Server: "atl", + Descriptor: mcp.MCPToolDescriptor{Name: "create_issue", InputSchema: json.RawMessage(`{"type":"object"}`)}, + Resolver: resolverStub{err: mcp.ErrTransportUnavailable}, + AuthGate: gate, + }) + if err != nil { + t.Fatal(err) + } + if _, err := tool.Execute(context.Background(), json.RawMessage(`{}`)); !errors.Is(err, mcp.ErrTransportUnavailable) { + t.Fatalf("transport error must surface unchanged; got %v", err) + } + if gate.awaitCalls != 0 { + t.Fatalf("gate must NOT be consulted for non-auth errors; Await called %d times", gate.awaitCalls) + } +} diff --git a/forge-core/tools/adapters/mcp_tool.go b/forge-core/tools/adapters/mcp_tool.go index 02d3c3f8..daba5713 100644 --- a/forge-core/tools/adapters/mcp_tool.go +++ b/forge-core/tools/adapters/mcp_tool.go @@ -56,10 +56,27 @@ type MCPTool struct { resolver mcp.ClientResolver client mcp.Client + // authGate parks a delegated call that has no grant for the requesting + // user, resuming it after consent (#330). nil ⇒ no gating: an ErrNoToken + // surfaces as a normal failure (the pre-#330 behavior, and the path for + // non-delegated servers / tests). + authGate AuthGate + maxResultChars int audit *runtime.AuditLogger } +// AuthGate turns "this user has no grant yet" from a hard failure into a +// pause-and-resume. When resolving the per-user connection returns +// mcp.ErrNoToken, Execute calls Await instead of failing; the runtime +// implementation parks the executor on the authgate engine, delivers a +// consent prompt, and returns once a grant exists (→ nil, Execute +// re-resolves and proceeds) or the wait ends without one (→ error, Execute +// gives up). Implemented in forge-cli/runtime; nil in core/tests. +type AuthGate interface { + Await(ctx context.Context, server string) error +} + // MCPToolOpts configures a new MCPTool. type MCPToolOpts struct { // Server is the MCP server name from forge.yaml (e.g. "linear"). @@ -77,6 +94,12 @@ type MCPToolOpts struct { // connection. Optional — nil falls back to Client. Resolver mcp.ClientResolver + // AuthGate parks a delegated call lacking a grant until the requesting + // user consents (#330). Optional — nil disables gating (ErrNoToken fails + // the call as before). The runtime passes its authgate-backed impl for + // type=user servers. + AuthGate AuthGate + // MaxResultChars truncates tool results above this size. 0 ⇒ default. MaxResultChars int @@ -105,6 +128,7 @@ func NewMCPTool(opts MCPToolOpts) (*MCPTool, error) { descriptor: opts.Descriptor, resolver: opts.Resolver, client: opts.Client, + authGate: opts.AuthGate, maxResultChars: maxChars, audit: opts.Audit, }, nil @@ -178,6 +202,19 @@ func (m *MCPTool) Execute(ctx context.Context, args json.RawMessage) (string, er // static resolver returns the shared client (unchanged); a per-subject // pool returns that user's own connection, establishing it lazily. client, err := m.resolveClient(ctx) + if err != nil && m.authGate != nil && errors.Is(err, mcp.ErrNoToken) { + // No grant yet for this user (#330). Rather than fail the call, park + // the executor and let the user consent; on a granted resume, + // re-resolve — the delegated path now finds the grant and the + // per-user connection establishes. A gate error (timeout / cancel / + // no requesting user) means give up; it flows to the emit+return + // below and classifies like the underlying ErrNoToken. + if gateErr := m.authGate.Await(ctx, m.server); gateErr != nil { + err = gateErr + } else { + client, err = m.resolveClient(ctx) + } + } if err != nil { durMs := time.Since(start).Milliseconds() m.emitResult(correlationID, durMs, 0, false, classifyToolErr(err))