From f4ccdc43aa69bcdebdc4407298001e3e28103bf0 Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Sun, 10 May 2026 20:53:28 -0400 Subject: [PATCH 1/4] feat(listeners): Record pipeline denials as SessionDenied events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Close a symmetry gap in session-event recording: inbound rejects in the extproc listener were already emitted as SessionDenied events, but outbound rejects in extproc, outbound rejects in forwardproxy, and inbound rejects in reverseproxy all returned an HTTP error without appending any session event. Consequences: a guardrail plugin blocks an outbound MCP/inference call → 403 lands on the agent, but /v1/sessions and abctl show nothing. The observability surface operators use to understand what was blocked and why is blind to exactly the events they care about most. Fix: add a reject-recording helper in each listener, called from the reject path before the HTTP error is rendered. Symmetric shape to the existing extproc recordInboundReject: Phase: SessionDenied Invocations: phase-filtered from pctx.Extensions.Invocations Host: pctx.Host StatusCode + Error: from action.Violation Bucketing: - extproc outbound: ActiveSession() → default - forwardproxy: ActiveSession() → default - reverseproxy: A2A.SessionID → ActiveSession() → default (mirrors the accept path's fallback chain) Skip rule: denials with no Invocations are not recorded — a content-free SessionDenied event would be noise without attribution, and non-plugin-originated denials (body-too-large, etc.) belong in /stats counters, not the session stream. Response-phase rejects (RunResponse denying after the upstream returned) are out of scope — no in-tree plugin rejects in that path today; wiring it adds listener complexity without a current consumer. Enables IBAC-style outbound guardrails to surface their blocks to operators in abctl and /v1/sessions alongside the allow events. Tests: 8 new tests across the three listeners covering the happy-path recording, bucketing fallbacks, and skip-without- invocations rule. All existing tests pass. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- .../cmd/authbridge/listener/extproc/server.go | 55 +++++++++ .../listener/extproc/server_test.go | 114 ++++++++++++++++++ .../listener/forwardproxy/server.go | 71 +++++++++++ .../listener/forwardproxy/server_test.go | 75 +++++++++++- .../listener/reverseproxy/server.go | 77 ++++++++++++ .../listener/reverseproxy/server_test.go | 92 +++++++++++++- 6 files changed, 480 insertions(+), 4 deletions(-) diff --git a/authbridge/cmd/authbridge/listener/extproc/server.go b/authbridge/cmd/authbridge/listener/extproc/server.go index 8ff8ffe59..d9c3f5a0a 100644 --- a/authbridge/cmd/authbridge/listener/extproc/server.go +++ b/authbridge/cmd/authbridge/listener/extproc/server.go @@ -251,6 +251,59 @@ func (s *Server) recordInboundReject(pctx *pipeline.Context, action pipeline.Act s.Sessions.Append(inboundSessionID(pctx), ev) } +// recordOutboundReject emits a SessionDenied event for outbound requests +// a pipeline plugin rejected. Symmetric to recordInboundReject on the +// inbound side. Called BEFORE rejectFromAction returns, so denied +// outbound calls appear in /v1/sessions and abctl rather than vanishing +// with only a 4xx/5xx on the agent side — the observability surface +// that guardrail plugins (rate-limit, policy, intent-based) depend on +// to show operators what they blocked and why. +// +// Uses the same ActiveSession bucketing as recordOutboundSession: an +// outbound call inherits the most-recently-updated session. When no +// active session exists the event lands in DefaultSessionID. Matches +// the correctness envelope of the accept path. +// +// Skips recording when no Invocations were appended — the deny came +// from a plugin that didn't contribute diagnostic context, and a +// content-free SessionDenied event would be noise without attribution. +func (s *Server) recordOutboundReject(pctx *pipeline.Context, action pipeline.Action) { + if s.Sessions == nil || pctx.Extensions.Invocations == nil { + return + } + sid := s.Sessions.ActiveSession() + if sid == "" { + sid = session.DefaultSessionID + } + var status int + var code, message string + if action.Violation != nil { + status = action.Violation.Status + if status == 0 { + status = pipeline.StatusFromCode(action.Violation.Code) + } + code = action.Violation.Code + message = action.Violation.Reason + } + ev := pipeline.SessionEvent{ + At: time.Now(), + Direction: pipeline.Outbound, + Phase: pipeline.SessionDenied, + Invocations: snapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), + Plugins: snapshotPlugins(pctx.Extensions.Custom), + Identity: snapshotIdentity(pctx), + Host: pctx.Host, + StatusCode: status, + Error: &pipeline.EventError{ + Kind: "policy", + Code: code, + Message: message, + }, + Duration: durationSince(pctx.StartedAt), + } + s.Sessions.Append(sid, ev) +} + // snapshotInvocations returns a shallow copy of the Invocations extension // filtered by phase. Plugins append to pctx.Extensions.Invocations as // both OnRequest and OnResponse fire; the full list lives there for @@ -540,6 +593,7 @@ func (s *Server) handleOutbound(stream extprocv3.ExternalProcessor_ProcessServer originalAuth := pctx.Headers.Get("Authorization") action := s.OutboundPipeline.Run(ctx, pctx) if action.Type == pipeline.Reject { + s.recordOutboundReject(pctx, action) return rejectFromAction(action), nil } @@ -575,6 +629,7 @@ func (s *Server) handleOutboundBody(stream extprocv3.ExternalProcessor_ProcessSe originalAuth := pctx.Headers.Get("Authorization") action := s.OutboundPipeline.Run(ctx, pctx) if action.Type == pipeline.Reject { + s.recordOutboundReject(pctx, action) return rejectFromAction(action), nil } diff --git a/authbridge/cmd/authbridge/listener/extproc/server_test.go b/authbridge/cmd/authbridge/listener/extproc/server_test.go index a4b015976..e70ec6f36 100644 --- a/authbridge/cmd/authbridge/listener/extproc/server_test.go +++ b/authbridge/cmd/authbridge/listener/extproc/server_test.go @@ -1319,6 +1319,120 @@ func TestRecordInboundReject_SkipsWithoutAuth(t *testing.T) { } } +// TestRecordOutboundReject_EmitsDeniedPhase verifies that a rejected +// outbound request produces a SessionDenied event with the outbound +// direction tag, the plugin's Invocation context, and the Violation +// mapped onto StatusCode + EventError. Gap: before this, outbound +// denies (future IBAC, rate-limit, guardrail plugins) would surface +// only as a 4xx on the agent and never appear in /v1/sessions or abctl. +func TestRecordOutboundReject_EmitsDeniedPhase(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + // Seed an active session so ActiveSession() returns it instead of + // empty — mirrors the real request flow (inbound recorded first). + store.Append("sess-active", pipeline.SessionEvent{ + At: time.Now().Add(-50 * time.Millisecond), + Direction: pipeline.Inbound, Phase: pipeline.SessionRequest, + A2A: &pipeline.A2AExtension{Method: "message/send", SessionID: "sess-active"}, + }) + s := &Server{Sessions: store} + + pctx := &pipeline.Context{ + Direction: pipeline.Outbound, + StartedAt: time.Now().Add(-10 * time.Millisecond), + Host: "external.example", + Extensions: pipeline.Extensions{ + Invocations: &pipeline.Invocations{ + Outbound: []pipeline.Invocation{{ + Plugin: "ibac", + Phase: pipeline.InvocationPhaseRequest, + Action: pipeline.ActionDeny, + Reason: "blocked", + Details: map[string]string{ + "intent_preview": "summarize my emails", + "action": "POST http://external.example", + "llm_reason": "action unrelated to user intent", + }, + }}, + }, + }, + } + action := pipeline.DenyStatus(403, "ibac.blocked", "action unrelated to user intent") + s.recordOutboundReject(pctx, action) + + v := store.View("sess-active") + if v == nil || len(v.Events) != 2 { + t.Fatalf("expected 2 events under sess-active (seed + denied), got %v", v) + } + ev := v.Events[1] + if ev.Direction != pipeline.Outbound { + t.Errorf("Direction = %v, want Outbound", ev.Direction) + } + if ev.Phase != pipeline.SessionDenied { + t.Errorf("Phase = %v, want SessionDenied", ev.Phase) + } + if ev.StatusCode != 403 { + t.Errorf("StatusCode = %d, want 403", ev.StatusCode) + } + if ev.Error == nil || ev.Error.Code != "ibac.blocked" { + t.Errorf("Error = %+v, want code=ibac.blocked", ev.Error) + } + if ev.Host != "external.example" { + t.Errorf("Host = %q, want external.example", ev.Host) + } + if ev.Invocations == nil || len(ev.Invocations.Outbound) != 1 || + ev.Invocations.Outbound[0].Action != pipeline.ActionDeny { + t.Errorf("Invocations context lost on denied event: %+v", ev.Invocations) + } + if ev.Duration <= 0 { + t.Errorf("Duration = %v, want > 0", ev.Duration) + } +} + +// TestRecordOutboundReject_NoActiveSession verifies the default-bucket +// fallback: when no session has been appended yet, the denial lands +// under DefaultSessionID so operators can still see it in /v1/sessions. +func TestRecordOutboundReject_NoActiveSession(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + s := &Server{Sessions: store} + + pctx := &pipeline.Context{ + Direction: pipeline.Outbound, + StartedAt: time.Now(), + Extensions: pipeline.Extensions{ + Invocations: &pipeline.Invocations{ + Outbound: []pipeline.Invocation{{ + Plugin: "ibac", Action: pipeline.ActionDeny, + Phase: pipeline.InvocationPhaseRequest, Reason: "blocked", + }}, + }, + }, + } + action := pipeline.DenyStatus(403, "ibac.blocked", "blocked") + s.recordOutboundReject(pctx, action) + + if v := store.View(session.DefaultSessionID); v == nil || len(v.Events) != 1 { + t.Fatalf("expected 1 event under default session, got %v", v) + } +} + +// TestRecordOutboundReject_SkipsWithoutInvocations mirrors the inbound +// skip rule: a denial with no Invocation context would produce a +// content-free event, noise for operators without attribution. +func TestRecordOutboundReject_SkipsWithoutInvocations(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + s := &Server{Sessions: store} + + action := pipeline.DenyStatus(403, "policy.forbidden", "forbidden") + s.recordOutboundReject(&pipeline.Context{Direction: pipeline.Outbound}, action) + + if v := store.View(session.DefaultSessionID); v != nil { + t.Errorf("expected no event recorded, got %+v", v) + } +} + // TestSnapshotPlugins_FiltersByEventSuffix verifies the plugin-public // observability convention: only Custom entries with keys ending in // pipeline.PluginEventSuffix are promoted to SessionEvent.Plugins. diff --git a/authbridge/cmd/authbridge/listener/forwardproxy/server.go b/authbridge/cmd/authbridge/listener/forwardproxy/server.go index d7cac81d7..c7f34acf0 100644 --- a/authbridge/cmd/authbridge/listener/forwardproxy/server.go +++ b/authbridge/cmd/authbridge/listener/forwardproxy/server.go @@ -84,6 +84,7 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { action := s.OutboundPipeline.Run(r.Context(), pctx) if action.Type == pipeline.Reject { + s.recordOutboundReject(pctx, action) writeRejection(w, action) return } @@ -207,6 +208,76 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { } } +// recordOutboundReject emits a SessionDenied event for outbound +// requests a pipeline plugin rejected. Symmetric to the accept path's +// session recording (above). Lets guardrail plugins (rate-limit, +// intent-based, content policy) show operators what was blocked and +// why via /v1/sessions and abctl, instead of the block appearing only +// as a 4xx/5xx on the agent side. +// +// Skips when no Invocations were appended — the deny came from a +// plugin that didn't contribute diagnostic context, and a content-free +// SessionDenied event would be noise without attribution. +func (s *Server) recordOutboundReject(pctx *pipeline.Context, action pipeline.Action) { + if s.Sessions == nil || pctx.Extensions.Invocations == nil { + return + } + sid := s.Sessions.ActiveSession() + if sid == "" { + sid = session.DefaultSessionID + } + var status int + var code, message string + if action.Violation != nil { + status = action.Violation.Status + if status == 0 { + status = pipeline.StatusFromCode(action.Violation.Code) + } + code = action.Violation.Code + message = action.Violation.Reason + } + ev := pipeline.SessionEvent{ + At: time.Now(), + Direction: pipeline.Outbound, + Phase: pipeline.SessionDenied, + Invocations: filterInvocationsByPhase(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), + Host: pctx.Host, + StatusCode: status, + Error: &pipeline.EventError{ + Kind: "policy", + Code: code, + Message: message, + }, + } + s.Sessions.Append(sid, ev) +} + +// filterInvocationsByPhase returns only the invocations that match the +// given phase. Used by reject-event recording to avoid emitting response- +// phase entries on a request-phase denial (the pipeline's Reject +// terminates before RunResponse, so phase filtering is redundant in +// practice, but stays consistent with the extproc listener's helper). +func filterInvocationsByPhase(ext *pipeline.Invocations, phase pipeline.InvocationPhase) *pipeline.Invocations { + if ext == nil { + return nil + } + out := &pipeline.Invocations{} + for _, inv := range ext.Inbound { + if inv.Phase == "" || inv.Phase == phase { + out.Inbound = append(out.Inbound, inv) + } + } + for _, inv := range ext.Outbound { + if inv.Phase == "" || inv.Phase == phase { + out.Outbound = append(out.Outbound, inv) + } + } + if out.Inbound == nil && out.Outbound == nil { + return nil + } + return out +} + func extractBearer(authHeader string) string { if len(authHeader) > 7 && strings.EqualFold(authHeader[:7], "bearer ") { return authHeader[7:] diff --git a/authbridge/cmd/authbridge/listener/forwardproxy/server_test.go b/authbridge/cmd/authbridge/listener/forwardproxy/server_test.go index 2fe6ec382..77e7d43df 100644 --- a/authbridge/cmd/authbridge/listener/forwardproxy/server_test.go +++ b/authbridge/cmd/authbridge/listener/forwardproxy/server_test.go @@ -9,14 +9,16 @@ import ( "net/url" "strings" "testing" + "time" "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" - "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/cache" - "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/exchange" "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/jwtvalidation/validation" "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/plugintesting" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/cache" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/exchange" "github.com/kagenti/kagenti-extensions/authbridge/authlib/routing" - "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/jwtvalidation/validation" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/session" ) type mockVerifier struct { @@ -335,3 +337,70 @@ func TestForwardProxy_NoBodyBuffering_WhenNotNeeded(t *testing.T) { t.Errorf("status = %d, want 200", resp.StatusCode) } } + +// TestRecordOutboundReject_EmitsDeniedPhase verifies the forward-proxy +// listener's reject-event recording: a rejected outbound request +// produces a SessionDenied event with the plugin Invocation context +// and the Violation mapped to StatusCode + EventError. +func TestRecordOutboundReject_EmitsDeniedPhase(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + store.Append("sess-active", pipeline.SessionEvent{ + At: time.Now().Add(-50 * time.Millisecond), + Direction: pipeline.Inbound, + Phase: pipeline.SessionRequest, + A2A: &pipeline.A2AExtension{Method: "message/send", SessionID: "sess-active"}, + }) + s := &Server{Sessions: store} + + pctx := &pipeline.Context{ + Direction: pipeline.Outbound, + Host: "external.example", + Extensions: pipeline.Extensions{ + Invocations: &pipeline.Invocations{ + Outbound: []pipeline.Invocation{{ + Plugin: "ibac", Action: pipeline.ActionDeny, + Phase: pipeline.InvocationPhaseRequest, Reason: "blocked", + Details: map[string]string{"llm_reason": "unrelated to user intent"}, + }}, + }, + }, + } + action := pipeline.DenyStatus(403, "ibac.blocked", "unrelated to user intent") + s.recordOutboundReject(pctx, action) + + v := store.View("sess-active") + if v == nil || len(v.Events) != 2 { + t.Fatalf("expected 2 events under sess-active, got %+v", v) + } + ev := v.Events[1] + if ev.Direction != pipeline.Outbound || ev.Phase != pipeline.SessionDenied { + t.Errorf("Direction/Phase = %v/%v, want Outbound/SessionDenied", ev.Direction, ev.Phase) + } + if ev.StatusCode != 403 || ev.Error == nil || ev.Error.Code != "ibac.blocked" { + t.Errorf("Status/Error = %d/%+v, want 403/ibac.blocked", ev.StatusCode, ev.Error) + } + if ev.Host != "external.example" { + t.Errorf("Host = %q, want external.example", ev.Host) + } + if ev.Invocations == nil || len(ev.Invocations.Outbound) != 1 { + t.Errorf("Invocations lost on denied event: %+v", ev.Invocations) + } +} + +// TestRecordOutboundReject_SkipsWithoutInvocations confirms the skip +// rule matches extproc's equivalent: denials with no diagnostic +// context are not recorded, so session stream attribution stays +// meaningful. +func TestRecordOutboundReject_SkipsWithoutInvocations(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + s := &Server{Sessions: store} + + action := pipeline.DenyStatus(403, "policy.forbidden", "forbidden") + s.recordOutboundReject(&pipeline.Context{Direction: pipeline.Outbound}, action) + + if v := store.View(session.DefaultSessionID); v != nil { + t.Errorf("expected no event, got %+v", v) + } +} \ No newline at end of file diff --git a/authbridge/cmd/authbridge/listener/reverseproxy/server.go b/authbridge/cmd/authbridge/listener/reverseproxy/server.go index ac40ed98a..6af92e66a 100644 --- a/authbridge/cmd/authbridge/listener/reverseproxy/server.go +++ b/authbridge/cmd/authbridge/listener/reverseproxy/server.go @@ -95,6 +95,7 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { action := s.InboundPipeline.Run(r.Context(), pctx) if action.Type == pipeline.Reject { + s.recordInboundReject(pctx, action) writeRejection(w, action) return } @@ -177,6 +178,82 @@ func (s *Server) errorHandler(w http.ResponseWriter, _ *http.Request, err error) http.Error(w, `{"error":"bad gateway"}`, http.StatusBadGateway) } +// recordInboundReject emits a SessionDenied event for inbound requests +// a pipeline plugin rejected. Lets gate plugins (jwt-validation and +// future inbound guardrails) show operators what was blocked and why +// via /v1/sessions and abctl, instead of the block appearing only as +// a 401/403 on the caller side. +// +// Skips when no Invocations were appended — the deny came from a +// plugin that didn't contribute diagnostic context, and a content-free +// SessionDenied event would be noise without attribution. +func (s *Server) recordInboundReject(pctx *pipeline.Context, action pipeline.Action) { + if s.Sessions == nil || pctx.Extensions.Invocations == nil { + return + } + // Inbound uses the A2A-stated contextId when available; otherwise + // falls through to the default bucket. Matches the accept path's + // bucketing rule (A2A request event at line 112-125). + sid := "" + if pctx.Extensions.A2A != nil { + sid = pctx.Extensions.A2A.SessionID + } + if sid == "" { + sid = s.Sessions.ActiveSession() + } + if sid == "" { + sid = session.DefaultSessionID + } + var status int + var code, message string + if action.Violation != nil { + status = action.Violation.Status + if status == 0 { + status = pipeline.StatusFromCode(action.Violation.Code) + } + code = action.Violation.Code + message = action.Violation.Reason + } + ev := pipeline.SessionEvent{ + At: time.Now(), + Direction: pipeline.Inbound, + Phase: pipeline.SessionDenied, + Invocations: filterInvocationsByPhase(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), + Host: pctx.Host, + StatusCode: status, + Error: &pipeline.EventError{ + Kind: "policy", + Code: code, + Message: message, + }, + } + s.Sessions.Append(sid, ev) +} + +// filterInvocationsByPhase returns only the invocations matching the +// phase. Mirrors the filter used by the extproc listener so reject +// events carry the request-phase entries that attributed the block. +func filterInvocationsByPhase(ext *pipeline.Invocations, phase pipeline.InvocationPhase) *pipeline.Invocations { + if ext == nil { + return nil + } + out := &pipeline.Invocations{} + for _, inv := range ext.Inbound { + if inv.Phase == "" || inv.Phase == phase { + out.Inbound = append(out.Inbound, inv) + } + } + for _, inv := range ext.Outbound { + if inv.Phase == "" || inv.Phase == phase { + out.Outbound = append(out.Outbound, inv) + } + } + if out.Inbound == nil && out.Outbound == nil { + return nil + } + return out +} + // writeRejection renders a pipeline Reject to the http.ResponseWriter, // preserving the plugin's status, headers, and body. func writeRejection(w http.ResponseWriter, action pipeline.Action) { diff --git a/authbridge/cmd/authbridge/listener/reverseproxy/server_test.go b/authbridge/cmd/authbridge/listener/reverseproxy/server_test.go index 9447312f3..f16a791ae 100644 --- a/authbridge/cmd/authbridge/listener/reverseproxy/server_test.go +++ b/authbridge/cmd/authbridge/listener/reverseproxy/server_test.go @@ -8,12 +8,14 @@ import ( "net/http/httptest" "strings" "testing" + "time" "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" "github.com/kagenti/kagenti-extensions/authbridge/authlib/bypass" "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" - "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/plugintesting" "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/jwtvalidation/validation" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/plugintesting" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/session" ) type mockVerifier struct { @@ -295,3 +297,91 @@ func TestReverseProxy_BodyNotBuffered_WhenNotNeeded(t *testing.T) { t.Errorf("status = %d, want 200", resp.StatusCode) } } + +// TestRecordInboundReject_EmitsDeniedPhase verifies the reverse-proxy +// listener's inbound reject-event recording. Gap: before this, an +// inbound request denied by jwt-validation or any other gate plugin +// produced a 401/403 on the wire but no SessionDenied event — abctl +// and /v1/sessions showed nothing, making misconfigurations invisible. +func TestRecordInboundReject_EmitsDeniedPhase(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + s := &Server{Sessions: store} + + pctx := &pipeline.Context{ + Direction: pipeline.Inbound, + Host: "agent.example", + Extensions: pipeline.Extensions{ + A2A: &pipeline.A2AExtension{SessionID: "sess-abc"}, + Invocations: &pipeline.Invocations{ + Inbound: []pipeline.Invocation{{ + Plugin: "jwt-validation", Action: pipeline.ActionDeny, + Phase: pipeline.InvocationPhaseRequest, Reason: "jwt_failed", + Details: map[string]string{ + "expected_issuer": "http://issuer.example", + }, + }}, + }, + }, + } + action := pipeline.DenyStatus(401, "auth.unauthorized", "token validation failed") + s.recordInboundReject(pctx, action) + + v := store.View("sess-abc") + if v == nil || len(v.Events) != 1 { + t.Fatalf("expected 1 event under sess-abc, got %+v", v) + } + ev := v.Events[0] + if ev.Direction != pipeline.Inbound || ev.Phase != pipeline.SessionDenied { + t.Errorf("Direction/Phase = %v/%v, want Inbound/SessionDenied", ev.Direction, ev.Phase) + } + if ev.StatusCode != 401 || ev.Error == nil || ev.Error.Code != "auth.unauthorized" { + t.Errorf("Status/Error = %d/%+v, want 401/auth.unauthorized", ev.StatusCode, ev.Error) + } + if ev.Invocations == nil || len(ev.Invocations.Inbound) != 1 { + t.Errorf("Invocations lost on denied event: %+v", ev.Invocations) + } +} + +// TestRecordInboundReject_FallbackBucketing confirms the bucket +// selection falls through A2A.SessionID → ActiveSession → default. +// A request with no A2A context (bypass path that denied in a later +// gate) lands in the default bucket so operators still see it. +func TestRecordInboundReject_FallbackBucketing(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + s := &Server{Sessions: store} + + pctx := &pipeline.Context{ + Direction: pipeline.Inbound, + Extensions: pipeline.Extensions{ + Invocations: &pipeline.Invocations{ + Inbound: []pipeline.Invocation{{ + Plugin: "jwt-validation", Action: pipeline.ActionDeny, + Phase: pipeline.InvocationPhaseRequest, Reason: "no_header", + }}, + }, + }, + } + action := pipeline.DenyStatus(401, "auth.unauthorized", "no bearer token") + s.recordInboundReject(pctx, action) + + if v := store.View(session.DefaultSessionID); v == nil || len(v.Events) != 1 { + t.Fatalf("expected 1 event in default bucket, got %+v", v) + } +} + +// TestRecordInboundReject_SkipsWithoutInvocations confirms the skip +// rule: denials with no diagnostic context are not recorded. +func TestRecordInboundReject_SkipsWithoutInvocations(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + s := &Server{Sessions: store} + + action := pipeline.DenyStatus(401, "auth.unauthorized", "denied") + s.recordInboundReject(&pipeline.Context{Direction: pipeline.Inbound}, action) + + if v := store.View(session.DefaultSessionID); v != nil { + t.Errorf("expected no event, got %+v", v) + } +} From 28bd660e97a533fac822408df73c4e3fbc1a5850 Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Sun, 10 May 2026 21:08:13 -0400 Subject: [PATCH 2/4] feat(pipeline): Add on_error policy (enforce | observe | off) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-plugin on_error knob that operators set in YAML to wrap the plugin's behavior without changing the plugin code. Three values: enforce (default) — Reject becomes the HTTP error, pipeline stops. Existing behavior unchanged. observe — plugin runs normally; a Reject return is converted to Continue with Shadow=true on the Invocation. Body mutations under observe are also suppressed. off — plugin is not dispatched at all. Kill-switch without a redeploy. Ships the rollout story for guardrail plugins: a new guardrail that could false-positive (PII detection, prompt-injection scoring, jailbreak classifiers) goes live in observe first, operators watch the shadow-deny rate, then flip to enforce. Scope of this commit: pipeline/policy.go NEW — ErrorPolicy enum + Valid/Resolved pipeline/pipeline.go WithPolicies option, policies slice on Pipeline, off/observe dispatch in Run/RunResponse, policyAt, markShadowAndLog pipeline/context.go currentPolicy field, setCurrent/clearCurrent (unexported; SetCurrentPlugin/ClearCurrent kept for back-compat), SetBody/SetResponseBody no-op under observe with Shadow=true telemetry, recordShadowBodyMutation helper, markLastInvocationShadow helper pipeline/extensions.go Shadow bool on Invocation config/config.go OnError field on PluginEntry, YAML decode, valid-value check at parse time plugins/registry.go Build skips off-policy entries, passes policies through to pipeline.New via WithPolicies; non-off plugins resolve empty→enforce Not in this commit (deliberately deferred from the parent feat/plugin-on-error branch): sealing the built-in registry (reservedBuiltinNames, RegisterBuiltin, IsReservedBuiltin, ChainDirection, BuildChain, chain-placement validation, switching built-ins to RegisterBuiltin). The seal is a different concern — it prevents custom plugins from replacing or shadowing jwt-validation / token-exchange — and can land separately without blocking the rollout mechanism. on_error is the piece guardrail plugins need. Under observe, the plugin's own code is unchanged: Reject intentions are converted to Continue with Shadow=true on the Invocation record, and body mutations are suppressed so the wire sees the original bytes. Dashboards partition on Invocation.Shadow to count rollout candidates (shadow=true) vs enforced outcomes (shadow=false). Tests: 7 new pipeline tests (observe conversion, enforce default, off skipping, response-phase observe, synthesized record when plugin doesn't Record, SetBody observe-is-noop, WithPolicies overflow rejection) + 2 new config tests (on_error YAML parse valid/invalid, omitted-defaults-to-enforce). All existing tests pass. Docs: plugin-reference.md gets an `on_error policy` section with the policy table, observe semantics, shadow timeline query patterns, a note that auth gates shouldn't be shadowed, and what on_error does NOT do (not a circuit breaker, not sampling, not a timeout, not for panics). plugin-tutorial.md gets a "Ship in observe mode first" callout next to the Reject helper. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- authbridge/authlib/config/config.go | 29 ++- authbridge/authlib/config/config_test.go | 77 +++++++ authbridge/authlib/pipeline/context.go | 116 +++++++++- authbridge/authlib/pipeline/extensions.go | 9 + authbridge/authlib/pipeline/pipeline.go | 125 +++++++++-- authbridge/authlib/pipeline/pipeline_test.go | 219 +++++++++++++++++++ authbridge/authlib/pipeline/policy.go | 58 +++++ authbridge/authlib/plugins/registry.go | 10 + authbridge/docs/plugin-reference.md | 79 +++++++ authbridge/docs/plugin-tutorial.md | 20 ++ 10 files changed, 713 insertions(+), 29 deletions(-) create mode 100644 authbridge/authlib/pipeline/policy.go diff --git a/authbridge/authlib/config/config.go b/authbridge/authlib/config/config.go index effda9173..aa15d5525 100644 --- a/authbridge/authlib/config/config.go +++ b/authbridge/authlib/config/config.go @@ -7,6 +7,7 @@ import ( "fmt" "os" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" "gopkg.in/yaml.v3" ) @@ -67,19 +68,25 @@ type PipelineStageConfig struct { // PluginEntry names a plugin and optionally carries per-instance config. // // The YAML accepts both the bare-name form ("jwt-validation") and the -// full form ({name, id, config}). The short form keeps existing pipeline -// configs parsing unchanged; the long form is what plugins that -// implement pipeline.Configurable actually need. See +// full form ({name, id, on_error, config}). The short form keeps +// existing pipeline configs parsing unchanged; the long form is what +// plugins that implement pipeline.Configurable actually need. See // authbridge/docs/plugin-reference.md for the convention plugins // follow when decoding Config. // // Config is captured as a raw subtree via json.RawMessage so the plugin // can do its own DisallowUnknownFields decode against a typed struct — // the framework does not interpret it. +// +// OnError is the framework-owned wrapper policy (see ErrorPolicy). +// Plugin authors do not consume it — it lives outside the plugin's +// own config block so all plugins get the same rollout story without +// each one re-implementing shadow mode. type PluginEntry struct { - Name string `yaml:"name" json:"name"` - ID string `yaml:"id,omitempty" json:"id,omitempty"` - Config json.RawMessage `yaml:"-" json:"config,omitempty"` + Name string `yaml:"name" json:"name"` + ID string `yaml:"id,omitempty" json:"id,omitempty"` + OnError pipeline.ErrorPolicy `yaml:"on_error,omitempty" json:"on_error,omitempty"` + Config json.RawMessage `yaml:"-" json:"config,omitempty"` } // UnmarshalYAML accepts either a bare string or a map. The string form @@ -108,6 +115,16 @@ func (p *PluginEntry) UnmarshalYAML(node *yaml.Node) error { if err := val.Decode(&p.ID); err != nil { return fmt.Errorf("plugin entry id: %w", err) } + case "on_error": + var raw string + if err := val.Decode(&raw); err != nil { + return fmt.Errorf("plugin entry on_error: %w", err) + } + policy := pipeline.ErrorPolicy(raw) + if !policy.Valid() { + return fmt.Errorf("plugin entry on_error: %q is not a valid policy (expected: enforce, observe, off)", raw) + } + p.OnError = policy case "config": // Explicit `config: null` (or `config:` with no value) // decodes to a null-tagged scalar node. Normalize to diff --git a/authbridge/authlib/config/config_test.go b/authbridge/authlib/config/config_test.go index 5599aff5c..1a3c4228d 100644 --- a/authbridge/authlib/config/config_test.go +++ b/authbridge/authlib/config/config_test.go @@ -8,6 +8,8 @@ import ( "strings" "testing" "time" + + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" ) // --- Preset Tests --- @@ -436,3 +438,78 @@ func TestSessionConfig_SessionEnabled(t *testing.T) { }) } } + +// TestPluginEntry_OnError covers the three accepted policy values plus +// the omitted case. An invalid policy must fail loud at YAML parse +// time so operators catch typos before the pod boots. +func TestPluginEntry_OnError(t *testing.T) { + cases := []struct { + name string + yaml string + want pipeline.ErrorPolicy + wantFail bool + }{ + {"enforce explicit", "enforce", pipeline.ErrorPolicyEnforce, false}, + {"observe", "observe", pipeline.ErrorPolicyObserve, false}, + {"off", "off", pipeline.ErrorPolicyOff, false}, + {"typo rejected", "observer", "", true}, + {"upper rejected", "ENFORCE", "", true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "config.yaml") + content := "mode: envoy-sidecar\n" + + "pipeline:\n" + + " inbound:\n" + + " plugins:\n" + + " - name: custom-guardrail\n" + + " on_error: " + c.yaml + "\n" + if err := os.WriteFile(path, []byte(content), 0600); err != nil { + t.Fatal(err) + } + cfg, err := Load(path) + if c.wantFail { + if err == nil { + t.Fatalf("expected parse error for on_error=%q", c.yaml) + } + return + } + if err != nil { + t.Fatalf("Load: %v", err) + } + got := cfg.Pipeline.Inbound.Plugins[0].OnError + if got != c.want { + t.Errorf("OnError = %q, want %q", got, c.want) + } + }) + } +} + +// TestPluginEntry_OnError_Omitted verifies the empty-string default. +// Resolved() should treat an absent on_error as enforce; the parsed +// field itself stays empty so round-tripping YAML doesn't invent a +// value the operator didn't write. +func TestPluginEntry_OnError_Omitted(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "config.yaml") + content := "mode: envoy-sidecar\n" + + "pipeline:\n" + + " inbound:\n" + + " plugins:\n" + + " - name: custom-guardrail\n" + if err := os.WriteFile(path, []byte(content), 0600); err != nil { + t.Fatal(err) + } + cfg, err := Load(path) + if err != nil { + t.Fatalf("Load: %v", err) + } + got := cfg.Pipeline.Inbound.Plugins[0].OnError + if got != "" { + t.Errorf("omitted OnError parsed as %q, want empty", got) + } + if got.Resolved() != pipeline.ErrorPolicyEnforce { + t.Errorf("Resolved() of empty = %q, want enforce", got.Resolved()) + } +} diff --git a/authbridge/authlib/pipeline/context.go b/authbridge/authlib/pipeline/context.go index dedf50114..c130cdbbd 100644 --- a/authbridge/authlib/pipeline/context.go +++ b/authbridge/authlib/pipeline/context.go @@ -104,14 +104,18 @@ type Context struct { Extensions Extensions - // currentPlugin and currentPhase are framework-owned fields set by - // Pipeline.Run / RunResponse around each plugin dispatch. They feed - // the Record / Allow / Skip / Observe / Modify / DenyAndRecord helper - // methods so plugin code doesn't have to repeat its own Name(), - // the phase it's in, or the direction on every Invocation literal. - // Unexported so plugins can only set them indirectly (via the framework). + // currentPlugin, currentPhase, and currentPolicy are framework-owned + // fields set by Pipeline.Run / RunResponse around each plugin + // dispatch. They feed the Record / Allow / Skip / Observe / Modify / + // DenyAndRecord helper methods so plugin code doesn't have to repeat + // its own Name(), the phase it's in, or the direction on every + // Invocation literal. currentPolicy is consulted by SetBody / + // SetResponseBody to suppress mutation propagation when the current + // plugin runs under ErrorPolicyObserve. Unexported so plugins can + // only set them indirectly (via the framework). currentPlugin string currentPhase InvocationPhase + currentPolicy ErrorPolicy // bodyMutated / responseBodyMutated flag that a plugin called // SetBody / SetResponseBody on this context. Listeners read the flag @@ -135,16 +139,35 @@ type Context struct { // pctx in their own dispatch loops (not strictly via Pipeline.Run) need // to set the same fields to get consistent Invocation attribution. // Production plugins should never call this directly. +// +// Sets the policy to the default (enforce). Call setCurrent (unexported) +// to stamp a non-default policy — only Pipeline.Run/RunResponse should +// supply a non-default, and they use the unexported path. func (c *Context) SetCurrentPlugin(name string, phase InvocationPhase) { - c.currentPlugin = name - c.currentPhase = phase + c.setCurrent(name, phase, ErrorPolicyEnforce) } // ClearCurrentPlugin resets the framework-owned attribution fields. // Paired with SetCurrentPlugin. func (c *Context) ClearCurrentPlugin() { + c.clearCurrent() +} + +// setCurrent stamps the per-dispatch attribution including the current +// plugin's on_error policy. Internal to the pipeline package; exported +// SetCurrentPlugin is the listener-facing entry point and defaults +// policy to enforce. +func (c *Context) setCurrent(name string, phase InvocationPhase, policy ErrorPolicy) { + c.currentPlugin = name + c.currentPhase = phase + c.currentPolicy = policy.Resolved() +} + +// clearCurrent zeroes the per-dispatch attribution fields. +func (c *Context) clearCurrent() { c.currentPlugin = "" c.currentPhase = "" + c.currentPolicy = "" } // Record appends an Invocation to pctx under the current pipeline @@ -224,6 +247,13 @@ func (c *Context) DenyAndRecord(reason, code, message string) Action { // SetBody mutate the in-memory Context (readers downstream see the // change), but the wire is unchanged. // +// Under ErrorPolicyObserve (shadow mode) SetBody is a NO-OP on bytes: +// the in-memory body is not replaced, bodyMutated stays false, and +// downstream plugins continue to see the original. A modify +// Invocation is still recorded — with Shadow=true — so operators can +// count "would have redacted" on the rollout dashboard. Plugin code +// therefore looks identical under enforce and observe. +// // SetBody auto-emits a modify-action Invocation with Reason // "body_rewritten" and publishes a plugin-public event under // "body-mutation/event" carrying the before/after length and sha256 — @@ -233,6 +263,10 @@ func (c *Context) DenyAndRecord(reason, code, message string) Action { // Callers should NOT assign pctx.Body directly — the listener wouldn't // know to propagate the change, and the Invocation wouldn't be emitted. func (c *Context) SetBody(newBody []byte) { + if c.currentPolicy == ErrorPolicyObserve { + c.recordShadowBodyMutation("request", c.Body, newBody) + return + } old := c.Body c.Body = newBody c.bodyMutated = true @@ -242,14 +276,44 @@ func (c *Context) SetBody(newBody []byte) { // SetResponseBody is the response-side analogue of SetBody. Used by // plugins that redact or rewrite the upstream response (prompt-safety // guardrails on LLM output, content filters, DLP). Same contract — -// Invocation + body-mutation/event emitted; never logs the body. +// Invocation + body-mutation/event emitted; never logs the body — +// and the same observe-mode suppression: under ErrorPolicyObserve the +// response body is untouched and the Invocation is marked Shadow=true. func (c *Context) SetResponseBody(newBody []byte) { + if c.currentPolicy == ErrorPolicyObserve { + c.recordShadowBodyMutation("response", c.ResponseBody, newBody) + return + } old := c.ResponseBody c.ResponseBody = newBody c.responseBodyMutated = true c.emitBodyMutation("response", old, newBody) } +// recordShadowBodyMutation emits the would-mutate Invocation for a +// SetBody / SetResponseBody call that was suppressed under observe +// mode. Mirrors emitBodyMutation's telemetry (length + sha256 delta) +// so dashboards get the same shape they see under enforce, just with +// Shadow=true and no wire-level effect. +func (c *Context) recordShadowBodyMutation(phase string, oldBody, newBody []byte) { + c.Record(Invocation{ + Action: ActionModify, + Reason: "body_rewritten", + Shadow: true, + }) + if c.Extensions.Custom == nil { + c.Extensions.Custom = map[string]any{} + } + c.Extensions.Custom["body-mutation"+PluginEventSuffix] = bodyMutationEvent{ + Phase: phase, + Plugin: c.currentPlugin, + LengthBefore: len(oldBody), + LengthAfter: len(newBody), + SHA256Before: hashHex(oldBody), + SHA256After: hashHex(newBody), + } +} + // BodyMutated reports whether a plugin called SetBody during this // request. Listeners check this after Run to decide whether to emit a // body mutation on the wire. Stream-scoped — a new Context starts with @@ -353,3 +417,37 @@ type AgentIdentity struct { WorkloadID string TrustDomain string } + +// markLastInvocationShadow finds the most recent Invocation authored +// by pluginName in the given phase (within the current direction +// bucket) and flips its Shadow flag to true. Returns true when a +// matching record was found. Used by the pipeline to retroactively +// tag a plugin's deny record as shadow once Pipeline.Run has observed +// that the plugin returned Reject under ErrorPolicyObserve. +// +// Walking backwards and matching on Plugin+Phase is O(N) in the +// worst case but typically short-circuits at index -1 — plugins +// almost always Record right before returning Reject, so the last +// entry matches. +func (c *Context) markLastInvocationShadow(pluginName string, phase InvocationPhase) bool { + if c.Extensions.Invocations == nil { + return false + } + var list []Invocation + switch c.Direction { + case Inbound: + list = c.Extensions.Invocations.Inbound + case Outbound: + list = c.Extensions.Invocations.Outbound + default: + return false + } + for i := len(list) - 1; i >= 0; i-- { + if list[i].Plugin != pluginName || list[i].Phase != phase { + continue + } + list[i].Shadow = true + return true + } + return false +} diff --git a/authbridge/authlib/pipeline/extensions.go b/authbridge/authlib/pipeline/extensions.go index 73d526e0c..d18359744 100644 --- a/authbridge/authlib/pipeline/extensions.go +++ b/authbridge/authlib/pipeline/extensions.go @@ -295,6 +295,15 @@ type Invocation struct { // The session API has no auth on it — only safe-to-log data // belongs in Invocation.Details. Details map[string]string `json:"details,omitempty"` + + // Shadow reports that the plugin ran under on_error: observe and + // its decision (deny or modify) was NOT applied to the request. + // An operator reading a would-have-blocked timeline filters on + // Shadow=true to count rollout-candidate events; a dashboard that + // aggregates "effective denies" filters Shadow=false. The + // framework, not the plugin, sets this — plugin code looks + // identical under enforce and observe. + Shadow bool `json:"shadow,omitempty"` } // DelegationExtension tracks the token delegation chain across hops. diff --git a/authbridge/authlib/pipeline/pipeline.go b/authbridge/authlib/pipeline/pipeline.go index 8e89aa36b..b1d59d479 100644 --- a/authbridge/authlib/pipeline/pipeline.go +++ b/authbridge/authlib/pipeline/pipeline.go @@ -7,8 +7,13 @@ import ( ) // Pipeline holds an ordered list of plugins and runs them sequentially. +// policies[i] holds the on_error ErrorPolicy that wraps plugins[i]; the +// slice is always the same length as plugins (guaranteed by New) so +// policyAt is a bounds-safe lookup. An empty ErrorPolicy resolves to +// ErrorPolicyEnforce via the Resolved() method. type Pipeline struct { - plugins []Plugin + plugins []Plugin + policies []ErrorPolicy } // defaultSlots lists the built-in extension slot names. @@ -26,6 +31,7 @@ type Option func(*options) type options struct { extraSlots []string + policies []ErrorPolicy } // WithSlots registers additional valid extension slot names beyond the built-in set. @@ -36,6 +42,18 @@ func WithSlots(slots ...string) Option { } } +// WithPolicies attaches per-plugin on_error policies in parallel with +// the plugin slice passed to New. policies[i] belongs to plugins[i]; +// an empty entry defaults to ErrorPolicyEnforce. If fewer policies are +// supplied than plugins, the remaining plugins use the default +// (enforce). Supplying more policies than plugins is a programmer +// error and New returns an error. +func WithPolicies(policies ...ErrorPolicy) Option { + return func(o *options) { + o.policies = append(o.policies, policies...) + } +} + // New creates a Pipeline from the given plugins after validating capability wiring. // Returns an error if any plugin declares a read on a slot that no earlier plugin writes. func New(plugins []Plugin, opts ...Option) (*Pipeline, error) { @@ -53,30 +71,53 @@ func New(plugins []Plugin, opts ...Option) (*Pipeline, error) { if err := validateCapabilities(plugins, validSlots); err != nil { return nil, err } - return &Pipeline{plugins: plugins}, nil + if len(o.policies) > len(plugins) { + return nil, fmt.Errorf("pipeline: WithPolicies has %d entries but only %d plugins", len(o.policies), len(plugins)) + } + policies := make([]ErrorPolicy, len(plugins)) + copy(policies, o.policies) + return &Pipeline{plugins: plugins, policies: policies}, nil } // Run executes the request phase of the pipeline sequentially. // If any plugin returns Reject, the pipeline stops and returns that action // with Violation.PluginName populated. // +// Plugins configured with ErrorPolicyOff are skipped entirely — they +// are not dispatched and contribute no Invocation. Plugins under +// ErrorPolicyObserve are dispatched normally, but a Reject return is +// converted into a pass-through: the Violation is recorded as a +// shadow Invocation and the pipeline continues to the next plugin. +// Body mutations under observe are also suppressed — see +// Context.SetBody / SetResponseBody. +// // Before dispatching into each plugin, Run stamps pctx with the plugin's -// name and the current phase so the plugin's Record / Allow / Skip / -// Observe / Modify / DenyAndRecord helpers can fill Invocation.Plugin -// and Invocation.Phase automatically. The stamp is cleared after each -// plugin returns so a plugin that spawns a goroutine capturing pctx -// won't mis-attribute a late-arriving Record to itself. +// name, the current phase, and the current policy so the plugin's +// Record / Allow / Skip / Observe / Modify / DenyAndRecord helpers can +// fill Invocation.Plugin and Invocation.Phase automatically. The stamp +// is cleared after each plugin returns so a plugin that spawns a +// goroutine capturing pctx won't mis-attribute a late-arriving Record +// to itself. func (p *Pipeline) Run(ctx context.Context, pctx *Context) Action { - for _, plugin := range p.plugins { + for i, plugin := range p.plugins { + policy := p.policyAt(i) + if policy == ErrorPolicyOff { + slog.Debug("pipeline: plugin disabled (on_error: off)", "plugin", plugin.Name()) + continue + } if ctx.Err() != nil { slog.Info("pipeline: request cancelled", "plugin", plugin.Name()) return Deny("pipeline.cancelled", "request cancelled") } - pctx.SetCurrentPlugin(plugin.Name(), InvocationPhaseRequest) + pctx.setCurrent(plugin.Name(), InvocationPhaseRequest, policy) action := plugin.OnRequest(ctx, pctx) - pctx.ClearCurrentPlugin() + pctx.clearCurrent() if action.Type == Reject { stampPluginName(&action, plugin.Name()) + if policy == ErrorPolicyObserve { + markShadowAndLog(pctx, plugin.Name(), InvocationPhaseRequest, action, "request") + continue + } logReject(plugin.Name(), action, "pipeline: plugin rejected request") return action } @@ -88,19 +129,28 @@ func (p *Pipeline) Run(ctx context.Context, pctx *Context) Action { // RunResponse executes the response phase in reverse order. // The last plugin in the chain sees the response first. // -// See Run for the pctx attribution stamping. Same pattern, phase set -// to InvocationPhaseResponse. +// See Run for the pctx attribution stamping, the off-policy skip, and +// the observe-policy shadow conversion. Same pattern, phase set to +// InvocationPhaseResponse. func (p *Pipeline) RunResponse(ctx context.Context, pctx *Context) Action { for i := len(p.plugins) - 1; i >= 0; i-- { + policy := p.policyAt(i) + if policy == ErrorPolicyOff { + continue + } if ctx.Err() != nil { slog.Info("pipeline: response cancelled", "plugin", p.plugins[i].Name()) return Deny("pipeline.cancelled", "request cancelled") } - pctx.SetCurrentPlugin(p.plugins[i].Name(), InvocationPhaseResponse) + pctx.setCurrent(p.plugins[i].Name(), InvocationPhaseResponse, policy) action := p.plugins[i].OnResponse(ctx, pctx) - pctx.ClearCurrentPlugin() + pctx.clearCurrent() if action.Type == Reject { stampPluginName(&action, p.plugins[i].Name()) + if policy == ErrorPolicyObserve { + markShadowAndLog(pctx, p.plugins[i].Name(), InvocationPhaseResponse, action, "response") + continue + } logReject(p.plugins[i].Name(), action, "pipeline: plugin rejected response") return action } @@ -108,6 +158,53 @@ func (p *Pipeline) RunResponse(ctx context.Context, pctx *Context) Action { return Action{Type: Continue} } +// policyAt returns the resolved policy for plugins[i]. The policies +// slice is always the same length as plugins (New guarantees this), +// but we check defensively so a zero-value Pipeline (constructed +// outside New, e.g. in a test) doesn't panic. +func (p *Pipeline) policyAt(i int) ErrorPolicy { + if i < len(p.policies) { + return p.policies[i].Resolved() + } + return ErrorPolicyEnforce +} + +// markShadowAndLog records the would-have-denied Invocation as +// Shadow=true and emits a WARN log. If the plugin already appended a +// deny Invocation (typical for gate plugins that call +// DenyAndRecord / Record before returning Reject), we mark that +// record instead of appending a duplicate — otherwise dashboards +// would double-count a single decision. Synthesize a record only +// when the plugin returned Reject without having recorded its own +// invocation (rare: plugin bug or non-recording denial helper). +func markShadowAndLog(pctx *Context, pluginName string, phase InvocationPhase, action Action, phaseLabel string) { + status, _, _ := action.Violation.Render() + marked := pctx.markLastInvocationShadow(pluginName, phase) + if !marked { + inv := Invocation{ + Plugin: pluginName, + Phase: phase, + Action: ActionDeny, + Reason: "shadow_synthesized", + Path: pctx.Path, + Shadow: true, + } + if action.Violation != nil { + inv.Details = map[string]string{ + "would_deny_code": action.Violation.Code, + "would_deny_reason": action.Violation.Reason, + } + } + pctx.Record(inv) + } + slog.Warn("pipeline: plugin would have denied (shadow)", + "plugin", pluginName, + "phase", phaseLabel, + "status", status, + "code", action.Violation.Code, + "reason", action.Violation.Reason) +} + // stampPluginName annotates a reject action with the plugin that produced // it, so listeners and clients can attribute the denial without the // plugin remembering to set it. diff --git a/authbridge/authlib/pipeline/pipeline_test.go b/authbridge/authlib/pipeline/pipeline_test.go index cacb4e0bd..5c1edd30b 100644 --- a/authbridge/authlib/pipeline/pipeline_test.go +++ b/authbridge/authlib/pipeline/pipeline_test.go @@ -674,3 +674,222 @@ func TestPipelineStop_NoShutdownersIsNoop(t *testing.T) { } p.Stop(context.Background()) // must not panic } +func TestPipelineRun_ObservePolicyConvertsRejectToContinue(t *testing.T) { + var downstreamCalled bool + shadow := &stubPlugin{ + name: "would-deny", + onReq: func(_ context.Context, pctx *Context) Action { + return pctx.DenyAndRecord("guardrail_triggered", "policy.content-blocked", "blocked") + }, + } + downstream := &stubPlugin{ + name: "downstream", + onReq: func(_ context.Context, _ *Context) Action { + downstreamCalled = true + return Action{Type: Continue} + }, + } + p, err := New([]Plugin{shadow, downstream}, WithPolicies(ErrorPolicyObserve, ErrorPolicyEnforce)) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + got := p.Run(context.Background(), pctx) + if got.Type != Continue { + t.Fatalf("action type = %v, want Continue (observe suppresses Reject)", got.Type) + } + if !downstreamCalled { + t.Error("downstream plugin not called; observe mode must not stop the pipeline") + } + if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Inbound) == 0 { + t.Fatal("no invocations recorded") + } + deny := pctx.Extensions.Invocations.Inbound[0] + if deny.Action != ActionDeny { + t.Errorf("shadow invocation action = %q, want deny (plugin's own record is preserved)", deny.Action) + } + if !deny.Shadow { + t.Error("shadow=false on the would-deny record; observe mode must mark it") + } +} + +// TestPipelineRun_EnforceIsDefault confirms that a plugin without a +// specified policy behaves exactly as before: Reject stops the +// pipeline and returns a Reject action upstream. Regression guard — +// the point of adding WithPolicies was to NOT change existing +// deployments. +func TestPipelineRun_EnforceIsDefault(t *testing.T) { + plugin := &stubPlugin{ + name: "denier", + onReq: func(_ context.Context, _ *Context) Action { + return Deny("policy.forbidden", "no") + }, + } + // No WithPolicies — entries default to enforce. + p, err := New([]Plugin{plugin}) + if err != nil { + t.Fatalf("New: %v", err) + } + got := p.Run(context.Background(), &Context{}) + if got.Type != Reject { + t.Errorf("action type = %v, want Reject (default is enforce)", got.Type) + } +} + +// TestPipelineRun_OffPolicySkipsDispatch verifies off-mode plugins +// don't run at all — not even their OnRequest. A kill-switched +// plugin must leave no trace in invocations, matching the "plugin +// absent from config" behavior. +func TestPipelineRun_OffPolicySkipsDispatch(t *testing.T) { + var called bool + disabled := &stubPlugin{ + name: "disabled", + onReq: func(_ context.Context, _ *Context) Action { + called = true + return Action{Type: Continue} + }, + } + p, err := New([]Plugin{disabled}, WithPolicies(ErrorPolicyOff)) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + got := p.Run(context.Background(), pctx) + if got.Type != Continue { + t.Errorf("action = %v, want Continue", got.Type) + } + if called { + t.Error("disabled plugin was dispatched; off must skip it") + } + if pctx.Extensions.Invocations != nil && len(pctx.Extensions.Invocations.Inbound) > 0 { + t.Errorf("off-policy plugin appended invocations: %+v", pctx.Extensions.Invocations.Inbound) + } +} + +// TestPipelineRunResponse_ObservePolicySuppressesReject ensures the +// observe semantics apply symmetrically to the response phase. +// Response-side guardrails (DLP on tool output, jailbreak detection +// on model replies) are exactly the class that most needs shadow +// rollout — they inspect generated content the author can't preview. +func TestPipelineRunResponse_ObservePolicySuppressesReject(t *testing.T) { + shadow := &stubPlugin{ + name: "would-deny-response", + onResp: func(_ context.Context, pctx *Context) Action { + return pctx.DenyAndRecord("pii_leak", "policy.content-blocked", "leak") + }, + } + p, err := New([]Plugin{shadow}, WithPolicies(ErrorPolicyObserve)) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + got := p.RunResponse(context.Background(), pctx) + if got.Type != Continue { + t.Errorf("action type = %v, want Continue (observe on response phase)", got.Type) + } + if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Inbound) == 0 { + t.Fatal("no response-phase invocation recorded") + } + inv := pctx.Extensions.Invocations.Inbound[0] + if inv.Phase != InvocationPhaseResponse { + t.Errorf("phase = %q, want response", inv.Phase) + } + if !inv.Shadow { + t.Error("response-phase shadow flag not set") + } +} + +// TestPipelineRun_ObserveSynthesizesRecordWhenPluginSkipsIt covers +// the defensive path: a plugin that returns Reject without first +// recording its own Invocation (programmer error, or use of the +// bare Deny helper instead of DenyAndRecord). The framework still +// produces a Shadow=true record so the would-have-blocked event is +// visible on the dashboard. +func TestPipelineRun_ObserveSynthesizesRecordWhenPluginSkipsIt(t *testing.T) { + silent := &stubPlugin{ + name: "silent-rejecter", + onReq: func(_ context.Context, _ *Context) Action { + // Note: no pctx.Record / DenyAndRecord. + return Deny("policy.forbidden", "no") + }, + } + p, err := New([]Plugin{silent}, WithPolicies(ErrorPolicyObserve)) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + got := p.Run(context.Background(), pctx) + if got.Type != Continue { + t.Errorf("action = %v, want Continue", got.Type) + } + if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Inbound) != 1 { + t.Fatalf("want exactly one synthesized invocation, got: %+v", pctx.Extensions.Invocations) + } + inv := pctx.Extensions.Invocations.Inbound[0] + if inv.Reason != "shadow_synthesized" { + t.Errorf("reason = %q, want shadow_synthesized", inv.Reason) + } + if !inv.Shadow { + t.Error("Shadow flag not set on synthesized record") + } + if inv.Details["would_deny_code"] != "policy.forbidden" { + t.Errorf("would_deny_code = %q, want policy.forbidden", inv.Details["would_deny_code"]) + } +} + +// TestSetBody_ObserveModeIsNoop verifies the body-mutation side of +// observe mode: SetBody records a shadow Invocation but does not +// replace the in-memory body or flip bodyMutated. Downstream readers +// and the wire both continue to see the original bytes, so a +// redaction plugin running in shadow can be trusted to leave +// production traffic unchanged. +func TestSetBody_ObserveModeIsNoop(t *testing.T) { + mutator := &stubPlugin{ + name: "redactor", + caps: PluginCapabilities{WritesBody: true}, + onReq: func(_ context.Context, pctx *Context) Action { + pctx.SetBody([]byte("REDACTED")) + return Action{Type: Continue} + }, + } + p, err := New([]Plugin{mutator}, WithPolicies(ErrorPolicyObserve)) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{Body: []byte("secret")} + got := p.Run(context.Background(), pctx) + if got.Type != Continue { + t.Fatalf("action = %v, want Continue", got.Type) + } + if string(pctx.Body) != "secret" { + t.Errorf("body = %q, want untouched (observe mode must not replace bytes)", pctx.Body) + } + if pctx.BodyMutated() { + t.Error("BodyMutated=true under observe; wire must see original body") + } + if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Inbound) != 1 { + t.Fatalf("want one invocation, got: %+v", pctx.Extensions.Invocations) + } + inv := pctx.Extensions.Invocations.Inbound[0] + if !inv.Shadow || inv.Reason != "body_rewritten" { + t.Errorf("expected shadow modify invocation, got action=%q reason=%q shadow=%v", + inv.Action, inv.Reason, inv.Shadow) + } +} + +// TestNew_RejectsTooManyPolicies locks the defensive check in New: +// a caller that supplies more policies than plugins is making a +// parallel-slice mistake, not a well-formed pipeline. Fail loud at +// construction rather than silently truncate. +func TestNew_RejectsTooManyPolicies(t *testing.T) { + _, err := New( + []Plugin{&stubPlugin{name: "a"}}, + WithPolicies(ErrorPolicyEnforce, ErrorPolicyObserve), + ) + if err == nil { + t.Fatal("expected error; 2 policies for 1 plugin is a parallel-slice bug") + } + if !strings.Contains(err.Error(), "WithPolicies") { + t.Errorf("error should name WithPolicies: %q", err) + } +} diff --git a/authbridge/authlib/pipeline/policy.go b/authbridge/authlib/pipeline/policy.go new file mode 100644 index 000000000..8b705d433 --- /dev/null +++ b/authbridge/authlib/pipeline/policy.go @@ -0,0 +1,58 @@ +package pipeline + +// ErrorPolicy controls how the pipeline reacts when a plugin returns a +// Reject action (i.e., decides to deny a request or response). The +// policy wraps the plugin — plugin authors do not consume it. +// +// Default is ErrorPolicyEnforce: Reject becomes an HTTP error response +// and pipeline execution stops. Operators roll out a new guardrail in +// ErrorPolicyObserve first: the plugin still evaluates and may still +// return Reject, but the framework converts the Reject into a +// pass-through and records the would-have-denied Invocation as +// Shadow=true. ErrorPolicyOff is a kill-switch — the plugin is +// not dispatched at all. +// +// Reserved built-in gates (jwt-validation, token-exchange) are locked +// to ErrorPolicyEnforce: shadowing auth is an authentication bypass +// dressed as a feature. plugins.BuildChain validates this at startup. +type ErrorPolicy string + +const ( + // ErrorPolicyEnforce is the default. A Reject action becomes the + // HTTP error the Violation describes, and the pipeline stops. + ErrorPolicyEnforce ErrorPolicy = "enforce" + + // ErrorPolicyObserve (shadow mode) evaluates the plugin normally + // but turns a Reject into a pass-through. Used to canary a new + // guardrail: operators watch the shadow-deny counter before + // flipping to enforce. Body mutations (SetBody / SetResponseBody) + // also do not propagate to the wire under observe — the plugin's + // decision logic runs identically, but its effect is muted. + ErrorPolicyObserve ErrorPolicy = "observe" + + // ErrorPolicyOff disables the plugin. The plugin is not dispatched. + // Off is an operator kill-switch without a redeploy. + ErrorPolicyOff ErrorPolicy = "off" +) + +// Valid reports whether p is a recognized ErrorPolicy. The empty +// string is valid and means "use the default" (enforce); callers that +// need the defaulted value use the Resolved method instead. +func (p ErrorPolicy) Valid() bool { + switch p { + case "", ErrorPolicyEnforce, ErrorPolicyObserve, ErrorPolicyOff: + return true + default: + return false + } +} + +// Resolved returns the policy with "" normalized to ErrorPolicyEnforce. +// Callers that dispatch on policy value should use Resolved so they +// don't have to treat "" as a synonym for "enforce" themselves. +func (p ErrorPolicy) Resolved() ErrorPolicy { + if p == "" { + return ErrorPolicyEnforce + } + return p +} diff --git a/authbridge/authlib/plugins/registry.go b/authbridge/authlib/plugins/registry.go index 88b0664de..8ff8c409a 100644 --- a/authbridge/authlib/plugins/registry.go +++ b/authbridge/authlib/plugins/registry.go @@ -99,7 +99,15 @@ func factoryFor(name string) (PluginFactory, bool) { // currently-registered plugin — typo-catching diagnostic. func Build(entries []config.PluginEntry, opts ...pipeline.Option) (*pipeline.Pipeline, error) { ps := make([]pipeline.Plugin, 0, len(entries)) + policies := make([]pipeline.ErrorPolicy, 0, len(entries)) for _, e := range entries { + // ErrorPolicyOff removes the plugin from the running pipeline + // entirely — no Configure, no Init, no dispatch. Operators use + // off as a kill-switch without deleting the entry from YAML, + // which makes re-enabling a one-line edit. + if e.OnError.Resolved() == pipeline.ErrorPolicyOff { + continue + } factory, ok := factoryFor(e.Name) if !ok { return nil, fmt.Errorf("unknown plugin %q (registered: %v)", e.Name, RegisteredPlugins()) @@ -113,6 +121,8 @@ func Build(entries []config.PluginEntry, opts ...pipeline.Option) (*pipeline.Pip return nil, fmt.Errorf("plugin %q does not accept configuration", e.Name) } ps = append(ps, p) + policies = append(policies, e.OnError.Resolved()) } + opts = append(opts, pipeline.WithPolicies(policies...)) return pipeline.New(ps, opts...) } diff --git a/authbridge/docs/plugin-reference.md b/authbridge/docs/plugin-reference.md index 9a773ef86..9aa732b82 100644 --- a/authbridge/docs/plugin-reference.md +++ b/authbridge/docs/plugin-reference.md @@ -40,16 +40,88 @@ pipeline: audience_file: "/shared/client-id.txt" bypass_paths: - "/healthz" + - name: pii-scrubber + on_error: observe # canary: log would-blocks, don't block + config: + patterns: [ssn, credit_card] ``` - **`name`** — required. Must match a key in the plugin registry. - **`id`** — optional. Defaults to `name`. Lets two instances of the same plugin coexist with different config (not yet exercised, but the shape is reserved). +- **`on_error`** — optional. One of `enforce` (default), `observe`, or + `off`. See [`on_error` policy](#on_error-policy) below. - **`config`** — optional. Arbitrary YAML sub-tree owned by the plugin. The framework does not interpret it; it's captured as `json.RawMessage` and handed to `Configure`. +## `on_error` policy + +`on_error` is a **framework-owned** wrapper around the plugin — plugin +authors do not read it, implement it, or branch on it. Its job is to +let operators roll out a new guardrail without risking production. + +| Policy | Plugin dispatched? | Reject → | Body mutation → | Typical use | +|---|---|---|---|---| +| `enforce` (default) | yes | HTTP error, pipeline stops | applied to wire | Production guardrails | +| `observe` | yes | shadow `Invocation`, request passes | suppressed (no-op) | Canarying a new plugin | +| `off` | **no** | n/a | n/a | Kill-switch without redeploy | + +### Observe is plugin-transparent + +Under `observe`, the plugin's `OnRequest` / `OnResponse` runs exactly as +under `enforce`. If it returns `pipeline.Deny(...)`, the framework +intercepts: it marks the plugin's `Invocation` with `Shadow: true`, logs +a `WARN pipeline: plugin would have denied (shadow)` line, and continues +the pipeline. The request is not blocked. Body-mutation calls +(`SetBody` / `SetResponseBody`) likewise record a `Shadow: true` +invocation but do not alter the in-memory body or the wire bytes — +downstream plugins and the upstream see the original. + +The upshot: the same plugin binary, dispatched the same way, is safe to +ship in `observe` for a week while operators watch shadow metrics, +then flipped to `enforce` with confidence. + +### Shadow timeline query + +Operators count would-have-blocked events by filtering Invocations on +`shadow: true`: + +- `count(Invocations where shadow=true)` — rollout candidate volume +- `count(Invocations where shadow=true and action="deny") by plugin` — + per-plugin shadow block rate +- `count(Invocations where shadow=false and action="deny") by plugin` — + enforced denials (unchanged by this feature) + +### Off vs. removing the entry + +Both achieve "don't run this plugin." `off` exists so a single field +flip re-enables the plugin without re-adding the whole block to YAML. +An `off` entry is not `Configure`d and not added to the running +pipeline; its `config:` subtree is not validated. Remove the entry +entirely if you don't anticipate re-enabling it. + +### What `on_error` does not do + +- Not a circuit breaker — a crashing plugin in `observe` still crashes + on every request. Bound crash loops with a separate mechanism. +- Not sampling — `observe` runs the plugin on 100% of traffic. + Percentage rollout is a future feature. +- Not a timeout — a slow plugin still blocks the request. Per-plugin + deadlines are a separate knob. +- Does not cover runtime `error` returns / panics from `OnRequest`. + Policy applies only to intentional `Reject` actions; a panic in + `observe` still surfaces as a 500. + +### Applicability to auth gates + +`on_error` is a generic framework knob that applies to every plugin +including built-in auth gates (`jwt-validation`, `token-exchange`). +Shadowing auth turns authentication into a suggestion — don't do this +in production. Shadow-mode is for third-party guardrails being +canaried; auth gates should stay on `enforce` (the default). + ## The Configurable interface ```go @@ -315,6 +387,13 @@ type Invocation struct { // Plugin-specific diagnostic context. Opaque to the framework; // abctl renders as key=value rows in the detail pane. Details map[string]string + + // Shadow is framework-set; plugins never write it. True when the + // plugin ran under on_error: observe and its decision (deny or + // modify) was NOT applied to the request. Dashboards partition + // on Shadow: enforced outcomes (shadow=false) vs rollout + // candidates (shadow=true). + Shadow bool } ``` diff --git a/authbridge/docs/plugin-tutorial.md b/authbridge/docs/plugin-tutorial.md index 93dbba76e..394622f8f 100644 --- a/authbridge/docs/plugin-tutorial.md +++ b/authbridge/docs/plugin-tutorial.md @@ -129,6 +129,26 @@ When you want to emit an invocation AND reject in one call, use return pctx.DenyAndRecord("caller_not_allowed", "policy.forbidden", "caller not permitted") ``` +### Ship in observe mode first + +For any guardrail or policy plugin that can false-positive — PII +detection, prompt-injection scoring, jailbreak classifiers — roll it +out under `on_error: observe` before flipping to `enforce`: + +```yaml +- name: your-new-guardrail + on_error: observe # log would-blocks, don't actually block + config: { ... } +``` + +In `observe`, the framework converts the plugin's `Deny` into a +pass-through and marks the deny `Invocation` with `Shadow: true`. Your +plugin code is unchanged — the rollout knob lives outside the plugin. +After a soak period, compare the shadow-deny rate against your +acceptable false-positive threshold, then flip to `enforce`. See +[`plugin-reference.md`](./plugin-reference.md#on_error-policy) for +full semantics. + ## Step 4 — Add config If your plugin needs configurable knobs, implement From f7bebe26ea608e6f9a0f2dd750628ff634b574cc Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Sun, 10 May 2026 21:20:03 -0400 Subject: [PATCH 3/4] fixup: PR #395 review feedback Five small fixes from review: 1. Promote filterInvocationsByPhase to (*Invocations).FilteredByPhase in pipeline/. Forwardproxy and reverseproxy had byte-for-byte duplicate helpers; now both call the method and the local helpers are gone. Extproc's snapshotInvocations collapses to a one-line delegation that preserves the snapshotXxx naming at call sites. Aligned the semantics: strict phase match (was inconsistent - forwardproxy/reverseproxy accepted empty Phase, extproc rejected). Strict is correct: the framework always populates Phase via Context.Record. 2. Synthesized shadow Invocation now carries action.Violation.Code as Reason, so dashboards grouping denials by Reason see the plugin's actual deny code across both recorded and synthesized paths. The "synthesized" marker moves to Details["synthesized"] = "true" so debugging can still distinguish "plugin Recorded then Deny'd" from "plugin Deny'd without Recording". Test updated to match. 3. Doc note in plugin-reference.md about NOT echoing matched content into Violation.Reason: the field flows to the session store (unauthenticated) and WARN logs (potentially different retention). Leading example shows a GOOD vs BAD pair and points at Details["match_sha256"] as the place to capture fingerprints when debugging is needed. 4. Lead the on_error section with the naming caveat ("despite the name, this handles intentional Deny, not panics/errors") so operators don't reach for it for the wrong reason. The same point was previously buried under "What on_error does NOT do" - promoted to a blockquote at the top. 5. abctl actionCell appends "*" to the action string when Invocation.Shadow is true (e.g., "deny*"). Operators scanning the timeline can spot would-have-blocked rows at a glance. Detail pane picks up Invocation.Shadow automatically via the JSON marshal tag added in the parent commit. New test case covers the shadow rendering. Not addressed (observation-only in the review): 6. "Missing test: policy references plugin that doesn't exist." The schema can't express this case: PluginEntry.OnError is inline with the plugin entry in YAML, and WithPolicies takes a positional slice indexed parallel to plugins. There's no name-keyed policy map where a missing-plugin reference could arise. Noted in reply thread rather than inventing a test for a non-existent case. Tests: existing pass; 1 new case added to events_pane_test.go for the shadow action cell. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- authbridge/authlib/pipeline/extensions.go | 40 +++++++++++ authbridge/authlib/pipeline/pipeline.go | 14 +++- authbridge/authlib/pipeline/pipeline_test.go | 17 +++-- authbridge/cmd/abctl/tui/events_pane.go | 10 ++- authbridge/cmd/abctl/tui/events_pane_test.go | 17 +++++ .../cmd/authbridge/listener/extproc/server.go | 69 ++++++++----------- .../listener/forwardproxy/server.go | 28 +------- .../listener/reverseproxy/server.go | 26 +------ authbridge/docs/plugin-reference.md | 43 +++++++++++- 9 files changed, 160 insertions(+), 104 deletions(-) diff --git a/authbridge/authlib/pipeline/extensions.go b/authbridge/authlib/pipeline/extensions.go index d18359744..9ab73bb6c 100644 --- a/authbridge/authlib/pipeline/extensions.go +++ b/authbridge/authlib/pipeline/extensions.go @@ -231,6 +231,46 @@ type Invocations struct { Outbound []Invocation `json:"outbound,omitempty"` } +// FilteredByPhase returns a new *Invocations containing only entries +// whose Phase matches the argument. Strict match — untagged entries +// (Phase == "") are dropped because the framework always populates +// Phase via Context.Record; an untagged entry is a plugin bug and +// including it in the wrong phase would double-report in one event +// and be missing from the other. +// +// The underlying Invocation values are copied shallowly; mutating a +// returned entry's Details map mutates the source. Acceptable for +// the per-request flow where the original lives only on pctx and is +// discarded after recording. +// +// Returns nil when no entries match so callers can drop the field +// from the SessionEvent without a null check. +// +// Intended for reject-event recording in listeners and for the +// accept-path phase split (one SessionEvent per phase). Listeners +// that need full independence from pctx's lifecycle layer their own +// snapshot on top. +func (in *Invocations) FilteredByPhase(phase InvocationPhase) *Invocations { + if in == nil { + return nil + } + out := &Invocations{} + for _, inv := range in.Inbound { + if inv.Phase == phase { + out.Inbound = append(out.Inbound, inv) + } + } + for _, inv := range in.Outbound { + if inv.Phase == phase { + out.Outbound = append(out.Outbound, inv) + } + } + if out.Inbound == nil && out.Outbound == nil { + return nil + } + return out +} + // Invocation records one plugin's action on one pipeline pass. Plugin is // the plugin's Name() for traceability. Action is the universal 5-value // verb (see Action). Reason is a stable machine-readable label paired diff --git a/authbridge/authlib/pipeline/pipeline.go b/authbridge/authlib/pipeline/pipeline.go index b1d59d479..0ceec20dc 100644 --- a/authbridge/authlib/pipeline/pipeline.go +++ b/authbridge/authlib/pipeline/pipeline.go @@ -181,17 +181,27 @@ func markShadowAndLog(pctx *Context, pluginName string, phase InvocationPhase, a status, _, _ := action.Violation.Render() marked := pctx.markLastInvocationShadow(pluginName, phase) if !marked { + // Use the Violation's machine-stable code as Reason so + // dashboards grouping denials by reason see the plugin's + // actual deny code for both recorded and synthesized paths. + // The "synthesized" signal lives in Details so operators can + // still distinguish "plugin Recorded then Deny'd" from + // "plugin Deny'd without Recording" when debugging. + reason := "plugin.unspecified" + if action.Violation != nil && action.Violation.Code != "" { + reason = action.Violation.Code + } inv := Invocation{ Plugin: pluginName, Phase: phase, Action: ActionDeny, - Reason: "shadow_synthesized", + Reason: reason, Path: pctx.Path, Shadow: true, } if action.Violation != nil { inv.Details = map[string]string{ - "would_deny_code": action.Violation.Code, + "synthesized": "true", "would_deny_reason": action.Violation.Reason, } } diff --git a/authbridge/authlib/pipeline/pipeline_test.go b/authbridge/authlib/pipeline/pipeline_test.go index 5c1edd30b..ab3f72406 100644 --- a/authbridge/authlib/pipeline/pipeline_test.go +++ b/authbridge/authlib/pipeline/pipeline_test.go @@ -826,14 +826,23 @@ func TestPipelineRun_ObserveSynthesizesRecordWhenPluginSkipsIt(t *testing.T) { t.Fatalf("want exactly one synthesized invocation, got: %+v", pctx.Extensions.Invocations) } inv := pctx.Extensions.Invocations.Inbound[0] - if inv.Reason != "shadow_synthesized" { - t.Errorf("reason = %q, want shadow_synthesized", inv.Reason) + // Reason carries the plugin's machine-stable deny code so + // dashboards grouping by Reason see the actual deny across + // both recorded and synthesized paths. + if inv.Reason != "policy.forbidden" { + t.Errorf("reason = %q, want policy.forbidden (the plugin's deny code)", inv.Reason) } if !inv.Shadow { t.Error("Shadow flag not set on synthesized record") } - if inv.Details["would_deny_code"] != "policy.forbidden" { - t.Errorf("would_deny_code = %q, want policy.forbidden", inv.Details["would_deny_code"]) + // Details["synthesized"] distinguishes "plugin didn't Record then + // returned Deny" from "plugin Recorded then returned Deny" for + // debugging, without hiding the deny code behind a synthetic Reason. + if inv.Details["synthesized"] != "true" { + t.Errorf("Details[synthesized] = %q, want true", inv.Details["synthesized"]) + } + if inv.Details["would_deny_reason"] != "no" { + t.Errorf("would_deny_reason = %q, want %q", inv.Details["would_deny_reason"], "no") } } diff --git a/authbridge/cmd/abctl/tui/events_pane.go b/authbridge/cmd/abctl/tui/events_pane.go index f97586ef2..6f2e9ec41 100644 --- a/authbridge/cmd/abctl/tui/events_pane.go +++ b/authbridge/cmd/abctl/tui/events_pane.go @@ -176,6 +176,15 @@ func (r invocationRow) actionCell() string { if r.inv == nil { return "—" } + // Under on_error: observe the framework converts Reject to a + // pass-through and marks the Invocation Shadow=true. Prefix the + // action with an asterisk so operators scanning the timeline can + // spot would-have-blocked rows at a glance — the request actually + // passed. Width stays within the 8-char column budget (deny* fits, + // observe* fits, modify* fits at 7). + if r.inv.Shadow { + return string(r.inv.Action) + "*" + } return string(r.inv.Action) } @@ -612,4 +621,3 @@ func truncateScopes(scopes []string, n int) string { } return strings.Join(scopes[:n], ", ") + fmt.Sprintf(" +%d more", len(scopes)-n) } - diff --git a/authbridge/cmd/abctl/tui/events_pane_test.go b/authbridge/cmd/abctl/tui/events_pane_test.go index 06e053913..80f98d9af 100644 --- a/authbridge/cmd/abctl/tui/events_pane_test.go +++ b/authbridge/cmd/abctl/tui/events_pane_test.go @@ -63,6 +63,23 @@ func TestInvocationRow_Cells(t *testing.T) { wantAction: "observe", wantPlugin: "a2a-parser", }, + { + name: "shadow deny (observe mode)", + row: invocationRow{ + event: &pipeline.SessionEvent{}, + inv: &pipeline.Invocation{ + Plugin: "pii-scrubber", + Action: pipeline.ActionDeny, + Shadow: true, + }, + direction: pipeline.Inbound, + }, + // Asterisk suffix flags the would-have-blocked event so + // operators can visually separate real denies from shadow + // denies in the timeline. + wantAction: "deny*", + wantPlugin: "pii-scrubber", + }, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { diff --git a/authbridge/cmd/authbridge/listener/extproc/server.go b/authbridge/cmd/authbridge/listener/extproc/server.go index d9c3f5a0a..bf6afc115 100644 --- a/authbridge/cmd/authbridge/listener/extproc/server.go +++ b/authbridge/cmd/authbridge/listener/extproc/server.go @@ -310,31 +310,16 @@ func (s *Server) recordOutboundReject(pctx *pipeline.Context, action pipeline.Ac // cross-phase inspection. At record time each SessionEvent should carry // only the invocations from its own phase, so request events don't // double-report request-phase entries AFTER the response phase has -// already added its own. Each Invocation carries its phase tag (set by -// the producer) — request events pass InvocationPhaseRequest, response -// events pass InvocationPhaseResponse, denied events pass +// already added its own. Request events pass InvocationPhaseRequest, +// response events pass InvocationPhaseResponse, denied events pass // InvocationPhaseRequest (denial terminates the pass before response // runs). Returns nil when no matching entry exists, so the recording // gate can check for "no invocations on this phase" cleanly. +// +// Thin wrapper over (*Invocations).FilteredByPhase — kept as a named +// function so recordXxxSession call sites stay grep-friendly. func snapshotInvocations(ext *pipeline.Invocations, phase pipeline.InvocationPhase) *pipeline.Invocations { - if ext == nil { - return nil - } - var inbound, outbound []pipeline.Invocation - for _, inv := range ext.Inbound { - if inv.Phase == phase { - inbound = append(inbound, inv) - } - } - for _, inv := range ext.Outbound { - if inv.Phase == phase { - outbound = append(outbound, inv) - } - } - if len(inbound) == 0 && len(outbound) == 0 { - return nil - } - return &pipeline.Invocations{Inbound: inbound, Outbound: outbound} + return ext.FilteredByPhase(phase) } // snapshotPlugins collects plugin-public observability events from @@ -416,18 +401,18 @@ func (s *Server) recordOutboundResponseSession(pctx *pipeline.Context) { } plugins := snapshotPlugins(pctx.Extensions.Custom) ev := pipeline.SessionEvent{ - At: time.Now(), - Direction: pipeline.Outbound, - Phase: pipeline.SessionResponse, - MCP: snapshotMCP(pctx.Extensions.MCP), - Inference: snapshotInference(pctx.Extensions.Inference), - Invocations: snapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseResponse), - Plugins: plugins, - Identity: snapshotIdentity(pctx), - StatusCode: pctx.StatusCode, - Error: deriveError(pctx), - Host: pctx.Host, - Duration: durationSince(pctx.StartedAt), + At: time.Now(), + Direction: pipeline.Outbound, + Phase: pipeline.SessionResponse, + MCP: snapshotMCP(pctx.Extensions.MCP), + Inference: snapshotInference(pctx.Extensions.Inference), + Invocations: snapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseResponse), + Plugins: plugins, + Identity: snapshotIdentity(pctx), + StatusCode: pctx.StatusCode, + Error: deriveError(pctx), + Host: pctx.Host, + Duration: durationSince(pctx.StartedAt), } // Auth / Plugins alone qualify for recording; matches the widened // gate in recordInboundSession so outbound denials and plugin-public @@ -515,15 +500,15 @@ func (s *Server) recordOutboundSession(pctx *pipeline.Context) { } plugins := snapshotPlugins(pctx.Extensions.Custom) ev := pipeline.SessionEvent{ - At: time.Now(), - Direction: pipeline.Outbound, - Phase: pipeline.SessionRequest, - MCP: snapshotMCP(pctx.Extensions.MCP), - Inference: snapshotInference(pctx.Extensions.Inference), - Invocations: snapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), - Plugins: plugins, - Identity: snapshotIdentity(pctx), - Host: pctx.Host, + At: time.Now(), + Direction: pipeline.Outbound, + Phase: pipeline.SessionRequest, + MCP: snapshotMCP(pctx.Extensions.MCP), + Inference: snapshotInference(pctx.Extensions.Inference), + Invocations: snapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), + Plugins: plugins, + Identity: snapshotIdentity(pctx), + Host: pctx.Host, } if ev.MCP != nil || ev.Inference != nil || ev.Invocations != nil || plugins != nil { s.Sessions.Append(sid, ev) diff --git a/authbridge/cmd/authbridge/listener/forwardproxy/server.go b/authbridge/cmd/authbridge/listener/forwardproxy/server.go index c7f34acf0..f3c96d43f 100644 --- a/authbridge/cmd/authbridge/listener/forwardproxy/server.go +++ b/authbridge/cmd/authbridge/listener/forwardproxy/server.go @@ -240,7 +240,7 @@ func (s *Server) recordOutboundReject(pctx *pipeline.Context, action pipeline.Ac At: time.Now(), Direction: pipeline.Outbound, Phase: pipeline.SessionDenied, - Invocations: filterInvocationsByPhase(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), + Invocations: pctx.Extensions.Invocations.FilteredByPhase(pipeline.InvocationPhaseRequest), Host: pctx.Host, StatusCode: status, Error: &pipeline.EventError{ @@ -252,32 +252,6 @@ func (s *Server) recordOutboundReject(pctx *pipeline.Context, action pipeline.Ac s.Sessions.Append(sid, ev) } -// filterInvocationsByPhase returns only the invocations that match the -// given phase. Used by reject-event recording to avoid emitting response- -// phase entries on a request-phase denial (the pipeline's Reject -// terminates before RunResponse, so phase filtering is redundant in -// practice, but stays consistent with the extproc listener's helper). -func filterInvocationsByPhase(ext *pipeline.Invocations, phase pipeline.InvocationPhase) *pipeline.Invocations { - if ext == nil { - return nil - } - out := &pipeline.Invocations{} - for _, inv := range ext.Inbound { - if inv.Phase == "" || inv.Phase == phase { - out.Inbound = append(out.Inbound, inv) - } - } - for _, inv := range ext.Outbound { - if inv.Phase == "" || inv.Phase == phase { - out.Outbound = append(out.Outbound, inv) - } - } - if out.Inbound == nil && out.Outbound == nil { - return nil - } - return out -} - func extractBearer(authHeader string) string { if len(authHeader) > 7 && strings.EqualFold(authHeader[:7], "bearer ") { return authHeader[7:] diff --git a/authbridge/cmd/authbridge/listener/reverseproxy/server.go b/authbridge/cmd/authbridge/listener/reverseproxy/server.go index 6af92e66a..66d2409e8 100644 --- a/authbridge/cmd/authbridge/listener/reverseproxy/server.go +++ b/authbridge/cmd/authbridge/listener/reverseproxy/server.go @@ -218,7 +218,7 @@ func (s *Server) recordInboundReject(pctx *pipeline.Context, action pipeline.Act At: time.Now(), Direction: pipeline.Inbound, Phase: pipeline.SessionDenied, - Invocations: filterInvocationsByPhase(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), + Invocations: pctx.Extensions.Invocations.FilteredByPhase(pipeline.InvocationPhaseRequest), Host: pctx.Host, StatusCode: status, Error: &pipeline.EventError{ @@ -230,30 +230,6 @@ func (s *Server) recordInboundReject(pctx *pipeline.Context, action pipeline.Act s.Sessions.Append(sid, ev) } -// filterInvocationsByPhase returns only the invocations matching the -// phase. Mirrors the filter used by the extproc listener so reject -// events carry the request-phase entries that attributed the block. -func filterInvocationsByPhase(ext *pipeline.Invocations, phase pipeline.InvocationPhase) *pipeline.Invocations { - if ext == nil { - return nil - } - out := &pipeline.Invocations{} - for _, inv := range ext.Inbound { - if inv.Phase == "" || inv.Phase == phase { - out.Inbound = append(out.Inbound, inv) - } - } - for _, inv := range ext.Outbound { - if inv.Phase == "" || inv.Phase == phase { - out.Outbound = append(out.Outbound, inv) - } - } - if out.Inbound == nil && out.Outbound == nil { - return nil - } - return out -} - // writeRejection renders a pipeline Reject to the http.ResponseWriter, // preserving the plugin's status, headers, and body. func writeRejection(w http.ResponseWriter, action pipeline.Action) { diff --git a/authbridge/docs/plugin-reference.md b/authbridge/docs/plugin-reference.md index 9aa732b82..eef870890 100644 --- a/authbridge/docs/plugin-reference.md +++ b/authbridge/docs/plugin-reference.md @@ -58,6 +58,15 @@ pipeline: ## `on_error` policy +> **Naming caveat.** Despite the name, `on_error` controls how the +> framework handles **intentional `Deny` actions** returned by the +> plugin — it is not a panic/error handler. A panic or a +> runtime-error return still surfaces as a 500 regardless of this +> setting. Reach for `on_error` to stage the rollout of a new +> guardrail (observe before enforce) or to toggle a plugin off +> without a redeploy; reach for something else when you want to +> bound misbehavior. + `on_error` is a **framework-owned** wrapper around the plugin — plugin authors do not read it, implement it, or branch on it. Its job is to let operators roll out a new guardrail without risking production. @@ -94,6 +103,35 @@ Operators count would-have-blocked events by filtering Invocations on - `count(Invocations where shadow=false and action="deny") by plugin` — enforced denials (unchanged by this feature) +### Don't put matched content in `Violation.Reason` + +`Reason` (the free-text explanation on `pipeline.Deny` / `DenyAndRecord` +and on `Invocation.Reason`) flows to two places: the session store at +`/v1/sessions` and, under `observe`, a `WARN` log line. Both live +outside the authorization domain of the request itself — logs typically +aggregate to a different backend than session events, with different +retention and access policies. + +Plugin authors: **keep `Reason` to a machine-stable short code and/or a +generic description.** Do not echo matched content, raw user input, or +credential-shaped substrings into `Reason`. + +```go +// GOOD +return pctx.DenyAndRecord("ssn_match", "pii.detected", "matched PII pattern") + +// BAD — echoes a matched SSN into logs and the session store: +return pctx.DenyAndRecord("ssn_match", "pii.detected", + fmt.Sprintf("matched SSN %s at offset 42", ssn)) +``` + +If you need to record which pattern matched for debugging, put a +**hash or stable fingerprint** of the content in `Invocation.Details` +(`Details["match_sha256"] = "a1b2c3..."`), never the raw match. The +same rule applies to `Details` values — see the existing "NEVER put +raw tokens, signatures, or client credentials" note on the Invocation +type below. + ### Off vs. removing the entry Both achieve "don't run this plugin." `off` exists so a single field @@ -110,9 +148,8 @@ entirely if you don't anticipate re-enabling it. Percentage rollout is a future feature. - Not a timeout — a slow plugin still blocks the request. Per-plugin deadlines are a separate knob. -- Does not cover runtime `error` returns / panics from `OnRequest`. - Policy applies only to intentional `Reject` actions; a panic in - `observe` still surfaces as a 500. +- Not a panic / runtime-error handler (see the leading caveat at the + top of this section). ### Applicability to auth gates From 7a71aa66fc5861d2b7a3632de3387d76e7c4b544 Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Mon, 11 May 2026 08:07:47 -0400 Subject: [PATCH 4/4] fixup(policy): Remove false claim about startup validation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ErrorPolicy godoc said "Reserved built-in gates are locked to ErrorPolicyEnforce" and "plugins.BuildChain validates this at startup." Both were leftovers from the parent feat/plugin-on-error branch, which included a built-in-registry seal pass. That seal was intentionally dropped when extracting the on_error piece for this PR (see the Gap 3 commit body). The stale claim misleads: an operator or auditor reading the godoc would trust that on_error: observe on jwt-validation cannot be configured, when in fact nothing prevents it today. Rewrite the paragraph as advisory + explicit "not enforced at startup" + pointer to the sealing follow-up. Matches the tone of the §Applicability-to-auth-gates doc section, which was already correctly advisory ("should stay on enforce"). No behavior change; godoc only. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- authbridge/authlib/pipeline/policy.go | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/authbridge/authlib/pipeline/policy.go b/authbridge/authlib/pipeline/policy.go index 8b705d433..e223bb93a 100644 --- a/authbridge/authlib/pipeline/policy.go +++ b/authbridge/authlib/pipeline/policy.go @@ -12,9 +12,14 @@ package pipeline // Shadow=true. ErrorPolicyOff is a kill-switch — the plugin is // not dispatched at all. // -// Reserved built-in gates (jwt-validation, token-exchange) are locked -// to ErrorPolicyEnforce: shadowing auth is an authentication bypass -// dressed as a feature. plugins.BuildChain validates this at startup. +// Shadowing or disabling auth gates (jwt-validation, token-exchange) +// is an authentication bypass dressed as a feature — operators SHOULD +// leave those plugins on ErrorPolicyEnforce (the default). +// +// The framework does NOT enforce this at startup: nothing prevents +// on_error: observe on jwt-validation today. A built-in-registry +// sealing pass is planned as a follow-up to reject a non-enforce +// policy on reserved gates at build time. type ErrorPolicy string const (