diff --git a/authbridge/CLAUDE.md b/authbridge/CLAUDE.md index a9506c403..e680e50fd 100644 --- a/authbridge/CLAUDE.md +++ b/authbridge/CLAUDE.md @@ -333,6 +333,34 @@ curl -N http://localhost:9094/v1/events curl -N "http://localhost:9094/v1/events?session=$SID" ``` +### Event schema + +Every event on `/v1/sessions/{id}` and `/v1/events` carries: + +- `at`, `direction`, `phase` — when, which side, what stage. `phase` is one of `"request"`, `"response"`, or `"denied"` (terminal denial from a pipeline plugin — typically a jwt-validation failure). +- `a2a` / `mcp` / `inference` — protocol parser payloads (one at most). +- `invocations` — per-plugin invocation records for every plugin that ran on the pipeline pass. Structured as `{inbound: [...], outbound: [...]}`; each entry carries `plugin`, `action` (one of 5 values — see below), `reason` (machine-stable code), and optional plugin-specific context (expected issuer, target audience, cache-hit flag, path, etc.). abctl renders one row per invocation, so operators see an explicit per-plugin timeline. +- `plugins` — escape-hatch map for plugin-specific observability. Keys are plugin names; values are the raw JSON each plugin emitted. Unknown plugins render as opaque JSON in abctl. See `authlib/plugins/CONVENTIONS.md` "Emitting session events" for the producer contract. +- `identity`, `host`, `statusCode`, `error`, `durationMs` — request-level context. + +### Invocation action vocabulary + +Every plugin emits one of these 5 action values per invocation, so operators can scan a timeline without memorizing plugin-specific verbs: + +| `action` | Meaning | Example | +|---|---|---| +| `allow` | Gate plugin permitted the request | jwt-validation on valid token | +| `deny` | Gate plugin rejected the request; pipeline stops | jwt-validation on bad token, token-exchange on IdP failure | +| `skip` | Plugin ran but didn't act on this message | jwt-validation on a bypass path; parser whose body didn't match | +| `modify` | Plugin mutated the message | token-exchange replaced the Authorization header | +| `observe` | Plugin attached diagnostic data without changing flow | a2a-parser, mcp-parser, inference-parser when they match | + +Use `reason` to discriminate within an action — e.g. `skip/path_bypass` vs `skip/no_matching_route` tell different stories at the detail-pane level but both scan as "skip" in the at-a-glance timeline. + +### Gotcha: denied requests + +Rejected requests (401 / 503) land as `phase: "denied"` events in `/v1/sessions` when at least one pipeline plugin appended an Invocation before rejecting. If you're debugging an unauthorized-access pattern, the default-session bucket (`GET /v1/sessions/default`) is where denial events aggregate. + ### Disabling Set `session.enabled: false` in the runtime config to turn off the store (and implicitly the API). Setting `listener.session_api_addr: ""` alone is not currently supported as a selective disable — the preset refills it; if you need store-on-API-off, raise an issue. diff --git a/authbridge/authlib/auth/auth.go b/authbridge/authlib/auth/auth.go index a5cc6d39b..23e4ba5b5 100644 --- a/authbridge/authlib/auth/auth.go +++ b/authbridge/authlib/auth/auth.go @@ -282,9 +282,10 @@ func (a *Auth) HandleInbound(ctx context.Context, authHeader, path, audience str a.IncInboundDeny(DENY_NO_HEADER) a.log.Debug("inbound denied: no Authorization header", "path", path) return &InboundResult{ - Action: ActionDeny, - DenyStatus: http.StatusUnauthorized, - DenyReason: "missing Authorization header", + Action: ActionDeny, + DenyStatus: http.StatusUnauthorized, + DenyReason: "missing Authorization header", + DenyReasonCode: DENY_NO_HEADER, } } token := extractBearer(authHeader) @@ -292,9 +293,10 @@ func (a *Auth) HandleInbound(ctx context.Context, authHeader, path, audience str a.IncInboundDeny(DENY_MALFORMED_HEADER) a.log.Debug("inbound denied: malformed Authorization header", "path", path) return &InboundResult{ - Action: ActionDeny, - DenyStatus: http.StatusUnauthorized, - DenyReason: "invalid Authorization header format", + Action: ActionDeny, + DenyStatus: http.StatusUnauthorized, + DenyReason: "invalid Authorization header format", + DenyReasonCode: DENY_MALFORMED_HEADER, } } @@ -302,9 +304,10 @@ func (a *Auth) HandleInbound(ctx context.Context, authHeader, path, audience str if a.verifier == nil { a.IncInboundDeny(DENY_VALIDATOR_MISSING) return &InboundResult{ - Action: ActionDeny, - DenyStatus: http.StatusUnauthorized, - DenyReason: "inbound validation not configured", + Action: ActionDeny, + DenyStatus: http.StatusUnauthorized, + DenyReason: "inbound validation not configured", + DenyReasonCode: DENY_VALIDATOR_MISSING, } } if audience == "" { @@ -312,9 +315,10 @@ func (a *Auth) HandleInbound(ctx context.Context, authHeader, path, audience str } if audience == "" { return &InboundResult{ - Action: ActionDeny, - DenyStatus: http.StatusServiceUnavailable, - DenyReason: "identity not yet configured (credentials pending)", + Action: ActionDeny, + DenyStatus: http.StatusServiceUnavailable, + DenyReason: "identity not yet configured (credentials pending)", + DenyReasonCode: DENY_VALIDATOR_MISSING, } } a.log.Debug("validating inbound JWT", "path", path, "expectedAudience", audience) @@ -330,9 +334,10 @@ func (a *Auth) HandleInbound(ctx context.Context, authHeader, path, audience str "expectedIssuer", a.identity.Load().Audience, "error", err) return &InboundResult{ - Action: ActionDeny, - DenyStatus: http.StatusUnauthorized, - DenyReason: "token validation failed", + Action: ActionDeny, + DenyStatus: http.StatusUnauthorized, + DenyReason: "token validation failed", + DenyReasonCode: DENY_JWT_FAILED, } } @@ -396,7 +401,14 @@ func (a *Auth) HandleOutbound(ctx context.Context, authHeader, host string) *Out if cached, ok := a.cache.Get(subjectToken, audience); ok { a.IncOutboundReplaceToken(OUTBOUND_ACTION_CACHE_HIT) a.log.Debug("outbound cache hit", "host", host, "audience", audience) - return &OutboundResult{Action: ActionReplaceToken, Token: cached} + return &OutboundResult{ + Action: ActionReplaceToken, + Token: cached, + CacheHit: true, + RouteMatched: true, + TargetAudience: audience, + RequestedScopes: scopes, + } } } @@ -437,9 +449,13 @@ func (a *Auth) HandleOutbound(ctx context.Context, authHeader, host string) *Out "tokenEndpoint", resolved.TokenEndpoint, "error", err) return &OutboundResult{ - Action: ActionDeny, - DenyStatus: http.StatusServiceUnavailable, - DenyReason: "token exchange failed", + Action: ActionDeny, + DenyStatus: http.StatusServiceUnavailable, + DenyReason: "token exchange failed", + DenyReasonCode: OUTBOUND_TOKEN_EXCHANGE_FAILED, + RouteMatched: true, + TargetAudience: audience, + RequestedScopes: scopes, } } @@ -453,7 +469,13 @@ func (a *Auth) HandleOutbound(ctx context.Context, authHeader, host string) *Out a.log.Info("outbound token exchanged", "host", host, "audience", audience) a.log.Debug("outbound exchange details", "host", host, "audience", audience, "expiresIn", resp.ExpiresIn) - return &OutboundResult{Action: ActionReplaceToken, Token: resp.AccessToken} + return &OutboundResult{ + Action: ActionReplaceToken, + Token: resp.AccessToken, + RouteMatched: true, + TargetAudience: audience, + RequestedScopes: scopes, + } } func (a *Auth) handleNoToken(ctx context.Context, audience, scopes string) *OutboundResult { @@ -469,9 +491,10 @@ func (a *Auth) handleNoToken(ctx context.Context, audience, scopes string) *Outb a.log.Debug("no token, client_credentials requested but exchanger not configured", "audience", audience) return &OutboundResult{ - Action: ActionDeny, - DenyStatus: http.StatusServiceUnavailable, - DenyReason: "exchanger not configured for client credentials", + Action: ActionDeny, + DenyStatus: http.StatusServiceUnavailable, + DenyReason: "exchanger not configured for client credentials", + DenyReasonCode: OUTBOUND_CREDS_REQUESTED_NO_EXCHANGER, } } a.log.Debug("no token, falling back to client_credentials", @@ -483,9 +506,10 @@ func (a *Auth) handleNoToken(ctx context.Context, audience, scopes string) *Outb a.log.Debug("client credentials failure details", "audience", audience, "scopes", scopes, "error", err) return &OutboundResult{ - Action: ActionDeny, - DenyStatus: http.StatusServiceUnavailable, - DenyReason: "client credentials token acquisition failed", + Action: ActionDeny, + DenyStatus: http.StatusServiceUnavailable, + DenyReason: "client credentials token acquisition failed", + DenyReasonCode: OUTBOUND_CREDENTIALS_GRANT_FAILURE, } } a.IncOutboundReplaceToken(OUTBOUND_ACTION_REPLACE_TOKEN) @@ -496,9 +520,10 @@ func (a *Auth) handleNoToken(ctx context.Context, audience, scopes string) *Outb a.log.Debug("no token, policy denies request", "policy", a.noTokenPolicy, "audience", audience) return &OutboundResult{ - Action: ActionDeny, - DenyStatus: http.StatusUnauthorized, - DenyReason: "missing Authorization header", + Action: ActionDeny, + DenyStatus: http.StatusUnauthorized, + DenyReason: "missing Authorization header", + DenyReasonCode: OUTBOUND_NO_TOKEN, } } } diff --git a/authbridge/authlib/auth/auth_test.go b/authbridge/authlib/auth/auth_test.go index 7e14e6ee3..c48ee13e0 100644 --- a/authbridge/authlib/auth/auth_test.go +++ b/authbridge/authlib/auth/auth_test.go @@ -113,6 +113,64 @@ func TestHandleInbound_NoVerifier_Denies(t *testing.T) { } } +// DenyReasonCode must be populated on every ActionDeny InboundResult. The +// listener/plugin layer uses the code (not the free-form DenyReason string) +// to populate SessionEvent.Auth.Inbound[].Reason for machine-stable +// filtering. A drift here would silently leave the event field empty on +// denied requests. +func TestHandleInbound_PopulatesDenyReasonCode(t *testing.T) { + cases := []struct { + name string + cfg Config + authHeader string + audience string + want InboundDenialReason + }{ + { + name: "no header", + cfg: Config{Verifier: &mockVerifier{}}, + authHeader: "", + want: DENY_NO_HEADER, + }, + { + name: "malformed header", + cfg: Config{Verifier: &mockVerifier{}}, + authHeader: "Basic abc", + want: DENY_MALFORMED_HEADER, + }, + { + name: "validator missing", + cfg: Config{}, // nil Verifier + authHeader: "Bearer tok", + want: DENY_VALIDATOR_MISSING, + }, + { + name: "jwt failed", + cfg: Config{ + Verifier: &mockVerifier{err: fmt.Errorf("bad signature")}, + Identity: IdentityConfig{Audience: "aud"}, + }, + authHeader: "Bearer tok", + want: DENY_JWT_FAILED, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + a := New(tc.cfg) + res := a.HandleInbound(context.Background(), tc.authHeader, "/api", tc.audience) + if res.Action != ActionDeny { + t.Fatalf("expected deny, got %q", res.Action) + } + if res.DenyReasonCode != tc.want { + t.Errorf("DenyReasonCode = %v, want %v", res.DenyReasonCode, tc.want) + } + if res.DenyReason == "" { + t.Error("DenyReason (human string) should still be populated alongside code") + } + }) + } +} + func TestHandleInbound_AudienceOverride(t *testing.T) { mv := &mockVerifier{claims: validClaims()} a := New(Config{ diff --git a/authbridge/authlib/auth/result.go b/authbridge/authlib/auth/result.go index f4e360431..77c77865a 100644 --- a/authbridge/authlib/auth/result.go +++ b/authbridge/authlib/auth/result.go @@ -27,16 +27,28 @@ const ( // InboundResult is the outcome of inbound JWT validation. type InboundResult struct { - Action string // ActionAllow or ActionDeny - Claims *validation.Claims // non-nil when a valid JWT was present - DenyStatus int // HTTP status code (e.g., 401) - DenyReason string // human-readable error + Action string // ActionAllow or ActionDeny + Claims *validation.Claims // non-nil when a valid JWT was present + DenyStatus int // HTTP status code (e.g., 401) + DenyReason string // human-readable error — safe for logs, response bodies + DenyReasonCode InboundDenialReason // machine-stable enum paired with the /stats counter; use for filtering / indexing session events } // OutboundResult is the outcome of outbound token exchange. type OutboundResult struct { - Action string // ActionAllow, ActionReplaceToken, or ActionDeny - Token string // replacement token (when Action == ActionReplaceToken) - DenyStatus int // HTTP status code (e.g., 503) - DenyReason string // human-readable error + Action string // ActionAllow, ActionReplaceToken, or ActionDeny + Token string // replacement token (when Action == ActionReplaceToken) + DenyStatus int // HTTP status code (e.g., 503) + DenyReason string // human-readable error + DenyReasonCode OutboundDenialReason // machine-stable enum paired with the /stats counter + CacheHit bool // true when Token was served from the exchange cache; safe to read on any Action + + // Route context populated when the router matched the request. Gives + // observability callers (token-exchange plugin writing a SessionEvent) + // a way to surface which route was chosen without reaching into + // routing.ResolvedRoute directly. RouteMatched=false means the + // request passed through because no route fired. + RouteMatched bool // true when Router.Resolve returned a non-nil route (regardless of action) + TargetAudience string // resolved route's audience (or deriver's output); empty on passthrough + RequestedScopes string // resolved route's scopes, raw space-separated; empty on passthrough } diff --git a/authbridge/authlib/pipeline/context.go b/authbridge/authlib/pipeline/context.go index 2346993ac..842183a1d 100644 --- a/authbridge/authlib/pipeline/context.go +++ b/authbridge/authlib/pipeline/context.go @@ -88,6 +88,124 @@ type Context struct { ResponseBody []byte 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 string + currentPhase InvocationPhase +} + +// SetCurrentPlugin is called by Pipeline.Run / RunResponse immediately +// before dispatching into a plugin's OnRequest / OnResponse. It stamps +// the plugin name and phase onto pctx so the Record family of helpers +// can fill those fields without plugin-side ceremony. Reset with +// ClearCurrentPlugin after dispatch. +// +// Exported (rather than framework-private) because listeners that embed +// 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. +func (c *Context) SetCurrentPlugin(name string, phase InvocationPhase) { + c.currentPlugin = name + c.currentPhase = phase +} + +// ClearCurrentPlugin resets the framework-owned attribution fields. +// Paired with SetCurrentPlugin. +func (c *Context) ClearCurrentPlugin() { + c.currentPlugin = "" + c.currentPhase = "" +} + +// Record appends an Invocation to pctx under the current pipeline +// direction and framework-stamped plugin + phase. The author supplies +// only what's specific to this call (Action, Reason, plus any +// diagnostic fields like ExpectedIssuer, RouteHost, CacheHit); Plugin, +// Phase, and Path are populated automatically from pctx. +// +// Authors may set Plugin, Phase, or Path on the argument explicitly to +// override the framework defaults — useful for test helpers or for a +// plugin synthesizing an invocation on behalf of a delegated sub-plugin. +// In normal plugin code, leave those fields zero. +// +// For the bare (Action, Reason) case, prefer the convenience wrappers +// (Allow / Skip / Observe / Modify) below — one line each. +func (c *Context) Record(inv Invocation) { + if inv.Plugin == "" { + inv.Plugin = c.currentPlugin + } + if inv.Phase == "" { + inv.Phase = c.currentPhase + } + if inv.Path == "" { + inv.Path = c.Path + } + c.appendInvocation(inv) +} + +// Allow records an Invocation with Action=allow and the given Reason. +// Convenience for gate plugins on the approved branch. +func (c *Context) Allow(reason string) { + c.Record(Invocation{Action: ActionAllow, Reason: reason}) +} + +// Skip records an Invocation with Action=skip. Convenience for plugins +// that ran but didn't act on this message (path bypass, no route match, +// parser skipping a non-matching body). +func (c *Context) Skip(reason string) { + c.Record(Invocation{Action: ActionSkip, Reason: reason}) +} + +// Observe records an Invocation with Action=observe. Convenience for +// parsers that successfully extracted diagnostic data without +// modifying the message. +func (c *Context) Observe(reason string) { + c.Record(Invocation{Action: ActionObserve, Reason: reason}) +} + +// Modify records an Invocation with Action=modify. Convenience for +// plugins that mutated the message (token-exchange replacing the +// Authorization header, a header-rewriter). +func (c *Context) Modify(reason string) { + c.Record(Invocation{Action: ActionModify, Reason: reason}) +} + +// DenyAndRecord records an Invocation with Action=deny AND returns a +// Reject Action. Bundles the two steps a gate plugin always does +// together on the deny path — emit the diagnostic record, then return +// the Action that changes control flow. +// +// code/message become the pipeline.Violation that the listener +// serializes to an HTTP response. Reason becomes the Invocation's +// machine-stable reason code. +// +// If the plugin has richer diagnostic data to attach to the Invocation +// (ExpectedIssuer, TokenScopes, etc.), use the two-step form: call +// pctx.Record(Invocation{...}) explicitly, then return pipeline.Deny. +func (c *Context) DenyAndRecord(reason, code, message string) Action { + c.Record(Invocation{Action: ActionDeny, Reason: reason}) + return Deny(code, message) +} + +// appendInvocation routes an Invocation to the right direction bucket +// based on pctx.Direction. Private — plugins call Record or the +// Allow/Skip/Observe/Modify helpers above. Not exported so external +// plugin authors discover the ergonomic API first and only drop to the +// full Invocation struct when they need diagnostic fields. +func (c *Context) appendInvocation(inv Invocation) { + if c.Extensions.Invocations == nil { + c.Extensions.Invocations = &Invocations{} + } + switch c.Direction { + case Inbound: + c.Extensions.Invocations.Inbound = append(c.Extensions.Invocations.Inbound, inv) + case Outbound: + c.Extensions.Invocations.Outbound = append(c.Extensions.Invocations.Outbound, inv) + } } // AgentIdentity carries the agent's own workload identity. diff --git a/authbridge/authlib/pipeline/extensions.go b/authbridge/authlib/pipeline/extensions.go index a97bdd3b4..e48113158 100644 --- a/authbridge/authlib/pipeline/extensions.go +++ b/authbridge/authlib/pipeline/extensions.go @@ -5,24 +5,52 @@ import "time" // Extensions holds typed extension slots for plugin-to-plugin communication. // Each slot is populated by a specific plugin and consumed by downstream plugins. // -// The named slots (MCP, A2A, Security, Delegation, Inference) are reserved -// for telemetry-worthy extensions — data that flows into SessionEvent, is -// serialized on the wire API, and has a published schema that unrelated -// plugins can rely on. Adding a new named slot is a core-library change. +// The named slots (MCP, A2A, Security, Delegation, Inference, Auth) are +// reserved for telemetry-worthy extensions — data that flows into +// SessionEvent, is serialized on the wire API, and has a published schema +// that unrelated plugins can rely on. Adding a new named slot is a +// core-library change. // -// For plugin-private, per-request state that doesn't need a published -// schema, use the generic GetState / SetState helpers defined below; they -// store values in Custom keyed by plugin name, letting a new plugin land -// without any authlib modification. +// For data that shouldn't drive a core-library change, use Custom. Two +// access patterns share the same map: +// +// - Plugin-PRIVATE state (cross-phase continuity inside one plugin). +// Use the typed SetState / GetState generics. Key is plugin.Name(). +// Value is typically *T for a plugin-internal struct (may contain +// sync primitives, unexported fields, channels). Never flows to +// session events. +// +// - Plugin-PUBLIC observability. Use a key suffixed with "/event" +// (e.g., "rate-limiter/event"). Value must be JSON-marshalable. +// The listener serializes matching entries into SessionEvent.Plugins +// at record time — keyed by the plugin name (suffix stripped). A +// new plugin can surface events to /v1/sessions without any +// authlib modification. See authlib/plugins/CONVENTIONS.md for the +// convention + promotion criteria for named-slot graduation. +// +// The suffix convention keeps the two intents unambiguous at write +// time: a plugin author has to deliberately type "/event" to opt into +// serialization, so private state can never leak by accident. type Extensions struct { - MCP *MCPExtension - A2A *A2AExtension - Security *SecurityExtension - Delegation *DelegationExtension - Inference *InferenceExtension - Custom map[string]any + MCP *MCPExtension + A2A *A2AExtension + Security *SecurityExtension + Delegation *DelegationExtension + Inference *InferenceExtension + Invocations *Invocations + Custom map[string]any } +// PluginEventSuffix is the key suffix that marks a Custom entry as +// plugin-public observability data destined for SessionEvent.Plugins. +// Plugin authors opt into serialization by writing: +// +// pctx.Extensions.Custom["rate-limiter"+pipeline.PluginEventSuffix] = ... +// +// The listener strips the suffix when populating SessionEvent.Plugins, +// so consumers see the plugin name as the map key. +const PluginEventSuffix = "/event" + // SetState stashes a typed value on pctx under key. Intended for plugin- // private per-request state — e.g., a rate-limiter remembering how many // tokens were available when OnRequest saw the call, for OnResponse to @@ -150,6 +178,123 @@ type SecurityExtension struct { BlockReason string `json:"blockReason,omitempty"` } +// InvocationAction is the universal 5-value vocabulary every plugin uses +// to describe what it did on a single pipeline pass. Every plugin — +// gate, parser, rate-limiter, guardrail, whatever we add next — +// MUST emit exactly one of these per Invocation so abctl and /v1/sessions +// can render a consistent per-plugin timeline. +// +// allow — a gate plugin permitted the request. jwt-validation +// returns this on successful signature + issuer + audience. +// deny — a gate plugin rejected the request. Terminal for the +// pipeline pass. jwt-validation on bad token, +// token-exchange on upstream IdP failure. +// skip — the plugin ran but didn't act. jwt-validation on a +// bypass path, token-exchange on a host with no matching +// route, a parser whose body didn't match its format. +// modify — the plugin mutated the message. token-exchange replaced +// the Authorization header with a freshly-issued token. +// observe — the plugin attached diagnostic data without altering +// flow. All parsers use this when they successfully parse. +// +// Reason (the stable machine code alongside Action) can discriminate +// within a value — e.g. skip/path_bypass vs skip/no_matching_route +// tell different stories at the detail-pane level, but both read +// "skip" in the at-a-glance timeline. +// +// Named InvocationAction rather than Action because pipeline.Action is +// already the pipeline-directive struct (Continue / Reject); keeping +// the names distinct avoids a shadowing foot-gun. +type InvocationAction string + +const ( + ActionAllow InvocationAction = "allow" + ActionDeny InvocationAction = "deny" + ActionSkip InvocationAction = "skip" + ActionModify InvocationAction = "modify" + ActionObserve InvocationAction = "observe" +) + +// Invocations carries one record per plugin that ran on a pipeline pass, +// split by direction so a single event's inbound and outbound plugin +// activity stays distinguishable. Multiple plugins can contribute — each +// appends an entry — so chained plugins cooperate without schema churn. +// Directions are disjoint per request: a single listener pass populates +// at most one of Inbound / Outbound. +// +// Replaces the earlier AuthExtension; parsers and any other plugin class +// share the list now. abctl renders one row per Invocation, so operators +// get a per-plugin timeline without guessing which plugins touched each +// event. +type Invocations struct { + Inbound []Invocation `json:"inbound,omitempty"` + Outbound []Invocation `json:"outbound,omitempty"` +} + +// 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 +// with the counters plugins already feed into /stats — use Reason for +// filtering / indexing rather than Action alone when you need to +// distinguish skip/path_bypass from skip/no_matching_route. +// +// Diagnostic fields are populated selectively per plugin. Auth gates +// populate ExpectedIssuer/Audience/Token*; outbound routers populate +// Route* and CacheHit; parsers typically populate only Plugin/Action/ +// Reason because their semantic payload lives on the typed extension +// slots (A2A / MCP / Inference). +// +// NEVER contains the raw bearer token, token signature, or client +// credentials. The session API has no auth on it; only safe-to-log data +// belongs here. +// InvocationPhase identifies whether an Invocation was appended during +// the request pass or the response pass. Without this tag the full list +// on pctx — which is cumulative across both phases — can't be correctly +// partitioned by the listener when it records the request event and the +// response event separately. Keeping the full list on pctx is deliberate +// (plugins may need cross-phase context); the phase tag lets consumers +// filter by pass. +type InvocationPhase string + +const ( + InvocationPhaseRequest InvocationPhase = "request" + InvocationPhaseResponse InvocationPhase = "response" +) + +type Invocation struct { + Plugin string `json:"plugin"` + Action InvocationAction `json:"action"` + // Phase is the pass (request or response) that appended this + // record. The listener uses it to filter invocations per event at + // record time — the request event carries only request-phase + // entries, the response event only response-phase entries, even + // though pctx carries the union. + Phase InvocationPhase `json:"phase,omitempty"` + Reason string `json:"reason,omitempty"` + + // Path is the request path the invocation ran on. Populated so + // operators can disambiguate invocations on the same plugin (e.g. + // a jwt-validation skip on /healthz vs /.well-known/agent.json; + // a mcp-parser observe on tools/call vs tools/list). Left empty + // when the plugin has no path context. + Path string `json:"path,omitempty"` + + // Auth-gate context, populated when applicable. + ExpectedIssuer string `json:"expectedIssuer,omitempty"` + ExpectedAudience string `json:"expectedAudience,omitempty"` + TokenSubject string `json:"tokenSubject,omitempty"` + TokenAudience []string `json:"tokenAudience,omitempty"` + TokenScopes []string `json:"tokenScopes,omitempty"` + + // Outbound routing context. RouteMatched=true means a route rule + // explicitly applied; false means the default policy caught it. + RouteMatched bool `json:"routeMatched,omitempty"` + RouteHost string `json:"routeHost,omitempty"` + TargetAudience string `json:"targetAudience,omitempty"` + RequestedScopes []string `json:"requestedScopes,omitempty"` + CacheHit bool `json:"cacheHit,omitempty"` +} + // DelegationExtension tracks the token delegation chain across hops. // The chain is append-only and unexported to prevent forgery or truncation. type DelegationExtension struct { diff --git a/authbridge/authlib/pipeline/pipeline.go b/authbridge/authlib/pipeline/pipeline.go index f7e4ea1f4..51fe5c18d 100644 --- a/authbridge/authlib/pipeline/pipeline.go +++ b/authbridge/authlib/pipeline/pipeline.go @@ -59,13 +59,22 @@ func New(plugins []Plugin, opts ...Option) (*Pipeline, error) { // 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. +// +// 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. func (p *Pipeline) Run(ctx context.Context, pctx *Context) Action { for _, plugin := range p.plugins { if ctx.Err() != nil { slog.Info("pipeline: request cancelled", "plugin", plugin.Name()) return Deny("pipeline.cancelled", "request cancelled") } + pctx.SetCurrentPlugin(plugin.Name(), InvocationPhaseRequest) action := plugin.OnRequest(ctx, pctx) + pctx.ClearCurrentPlugin() if action.Type == Reject { stampPluginName(&action, plugin.Name()) logReject(plugin.Name(), action, "pipeline: plugin rejected request") @@ -78,13 +87,18 @@ 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. func (p *Pipeline) RunResponse(ctx context.Context, pctx *Context) Action { for i := len(p.plugins) - 1; i >= 0; i-- { 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) action := p.plugins[i].OnResponse(ctx, pctx) + pctx.ClearCurrentPlugin() if action.Type == Reject { stampPluginName(&action, p.plugins[i].Name()) logReject(p.plugins[i].Name(), action, "pipeline: plugin rejected response") diff --git a/authbridge/authlib/pipeline/session.go b/authbridge/authlib/pipeline/session.go index 70d72787a..c87a79114 100644 --- a/authbridge/authlib/pipeline/session.go +++ b/authbridge/authlib/pipeline/session.go @@ -6,12 +6,18 @@ import ( "time" ) -// SessionPhase distinguishes request from response events. +// SessionPhase distinguishes request, response, and terminal-denial events. type SessionPhase int const ( SessionRequest SessionPhase = iota SessionResponse + // SessionDenied is a terminal event for inbound requests a pipeline + // plugin rejected (e.g., jwt-validation failing a token check). The + // listener records this instead of a Request/Response pair so abctl + // and other session consumers can distinguish denials from normal + // request/response flow without scanning StatusCode. + SessionDenied ) func (p SessionPhase) String() string { @@ -20,13 +26,15 @@ func (p SessionPhase) String() string { return "request" case SessionResponse: return "response" + case SessionDenied: + return "denied" default: return "unknown" } } -// MarshalJSON emits the string form ("request"/"response") so the wire -// format stays human-readable. +// MarshalJSON emits the string form ("request"/"response"/"denied") so +// the wire format stays human-readable. func (p SessionPhase) MarshalJSON() ([]byte, error) { return json.Marshal(p.String()) } @@ -46,6 +54,8 @@ func (p *SessionPhase) UnmarshalJSON(data []byte) error { *p = SessionRequest case "response": *p = SessionResponse + case "denied": + *p = SessionDenied default: slog.Debug("pipeline: unknown SessionPhase, defaulting to request", "value", s) *p = SessionRequest @@ -54,7 +64,13 @@ func (p *SessionPhase) UnmarshalJSON(data []byte) error { } // SessionEvent represents a single pipeline event captured by the session store. -// Exactly one of A2A, MCP, or Inference is non-nil. +// At most one of A2A, MCP, or Inference is non-nil on any given event, but +// the event may also carry Invocations (records from any plugin that ran on +// the pass) and/or Plugins (the escape-hatch map from Extensions.Custom) +// regardless of which protocol extension is present. An event with +// Phase=SessionDenied typically carries only Invocations (+ Identity, Host, +// Error) because the request never reached a protocol parser before the +// pipeline rejected it. type SessionEvent struct { // SessionID is the session bucket the event was appended to. Populated by // Store.Append so downstream consumers (particularly the SSE stream @@ -70,6 +86,22 @@ type SessionEvent struct { MCP *MCPExtension Inference *InferenceExtension + // Invocations carries records of every plugin that ran on the + // pipeline pass — gate, parser, rate-limiter, etc. Nil when no + // plugin appended a record, non-nil with at least one Inbound or + // Outbound entry otherwise. See Invocations godoc for the per- + // plugin shape. + Invocations *Invocations + + // Plugins carries plugin-public observability events in JSON form. + // Populated by the listener from Extensions.Custom entries whose keys + // end in PluginEventSuffix ("/event"); the suffix is stripped, so + // consumers see the plugin name as the map key. Value is the plugin- + // provided struct marshaled to JSON — opaque from the listener's + // perspective. Consumers decode each key into their own type. See + // authlib/plugins/CONVENTIONS.md for the producer contract. + Plugins map[string]json.RawMessage + // Identity snapshot at record time. Lets downstream plugins attribute an // event to the caller (Subject) and the handling sidecar (AgentID) // without re-parsing the original request. Nil for events recorded @@ -114,19 +146,21 @@ type SessionEvent struct { // writes to it directly; UnmarshalJSON reads into it and converts back. // Keeping the layout in one place guarantees round-trip symmetry. type sessionEventWire struct { - SessionID string `json:"sessionId,omitempty"` - At time.Time `json:"at"` - Direction Direction `json:"direction"` - Phase SessionPhase `json:"phase"` - A2A *A2AExtension `json:"a2a,omitempty"` - MCP *MCPExtension `json:"mcp,omitempty"` - Inference *InferenceExtension `json:"inference,omitempty"` - Identity *EventIdentity `json:"identity,omitempty"` - StatusCode int `json:"statusCode,omitempty"` - Error *EventError `json:"error,omitempty"` - Host string `json:"host,omitempty"` - TargetAudience string `json:"targetAudience,omitempty"` - DurationMs int64 `json:"durationMs,omitempty"` + SessionID string `json:"sessionId,omitempty"` + At time.Time `json:"at"` + Direction Direction `json:"direction"` + Phase SessionPhase `json:"phase"` + A2A *A2AExtension `json:"a2a,omitempty"` + MCP *MCPExtension `json:"mcp,omitempty"` + Inference *InferenceExtension `json:"inference,omitempty"` + Invocations *Invocations `json:"invocations,omitempty"` + Plugins map[string]json.RawMessage `json:"plugins,omitempty"` + Identity *EventIdentity `json:"identity,omitempty"` + StatusCode int `json:"statusCode,omitempty"` + Error *EventError `json:"error,omitempty"` + Host string `json:"host,omitempty"` + TargetAudience string `json:"targetAudience,omitempty"` + DurationMs int64 `json:"durationMs,omitempty"` } func (e SessionEvent) MarshalJSON() ([]byte, error) { @@ -138,6 +172,8 @@ func (e SessionEvent) MarshalJSON() ([]byte, error) { A2A: e.A2A, MCP: e.MCP, Inference: e.Inference, + Invocations: e.Invocations, + Plugins: e.Plugins, Identity: e.Identity, StatusCode: e.StatusCode, Error: e.Error, @@ -163,6 +199,8 @@ func (e *SessionEvent) UnmarshalJSON(data []byte) error { A2A: w.A2A, MCP: w.MCP, Inference: w.Inference, + Invocations: w.Invocations, + Plugins: w.Plugins, Identity: w.Identity, StatusCode: w.StatusCode, Error: w.Error, diff --git a/authbridge/authlib/pipeline/session_test.go b/authbridge/authlib/pipeline/session_test.go index bc6248368..9663f88e7 100644 --- a/authbridge/authlib/pipeline/session_test.go +++ b/authbridge/authlib/pipeline/session_test.go @@ -107,9 +107,125 @@ func TestSessionEvent_MarshalJSON_OmitsEmpty(t *testing.T) { } s := string(data) - for _, field := range []string{"a2a", "mcp", "inference", "identity", "statusCode", "error", "host", "targetAudience", "durationMs"} { + for _, field := range []string{"a2a", "mcp", "inference", "auth", "plugins", "identity", "statusCode", "error", "host", "targetAudience", "durationMs"} { if strings.Contains(s, `"`+field+`":`) { t.Errorf("expected %q omitted when zero: %s", field, s) } } } + +// SessionDenied must serialize as "denied" on the wire so consumers +// filtering on phase (abctl `/deny`, stats queries) see a stable string +// rather than the numeric enum. +func TestSessionPhase_Denied_SerializesAsString(t *testing.T) { + if got := SessionDenied.String(); got != "denied" { + t.Errorf("SessionDenied.String() = %q, want %q", got, "denied") + } + data, err := json.Marshal(SessionDenied) + if err != nil { + t.Fatalf("Marshal: %v", err) + } + if string(data) != `"denied"` { + t.Errorf("SessionDenied Marshal = %s, want %q", data, `"denied"`) + } + var p SessionPhase + if err := json.Unmarshal([]byte(`"denied"`), &p); err != nil { + t.Fatalf("Unmarshal denied: %v", err) + } + if p != SessionDenied { + t.Errorf("Unmarshal(\"denied\") = %v, want SessionDenied", p) + } +} + +// Round-trip Invocations through JSON including both directions, +// multiple entries per direction, and the optional diagnostic fields. +// Also verifies the Action field serializes as the 5-value string +// vocabulary (not the old "decision"/"action"-per-direction shape). +// Locks the wire schema so a future field addition that's missed in +// sessionEventWire fails this test. +func TestSessionEvent_Invocations_JSONRoundTrip(t *testing.T) { + orig := SessionEvent{ + At: time.Unix(1700000000, 0).UTC(), + Direction: Inbound, + Phase: SessionDenied, + Invocations: &Invocations{ + Inbound: []Invocation{{ + Plugin: "jwt-validation", + Action: ActionDeny, + Reason: "jwt_failed", + ExpectedIssuer: "http://keycloak.localtest.me:8080/realms/kagenti", + ExpectedAudience: "spiffe://localtest.me/ns/team1/sa/weather-tool", + }}, + Outbound: []Invocation{{ + Plugin: "token-exchange", + Action: ActionModify, + Reason: "cache_hit", + RouteMatched: true, + RouteHost: "weather-tool-mcp", + TargetAudience: "spiffe://localtest.me/ns/team1/sa/weather-tool", + RequestedScopes: []string{"openid", "weather-aud"}, + CacheHit: true, + }}, + }, + } + first, err := json.Marshal(orig) + if err != nil { + t.Fatalf("first Marshal: %v", err) + } + if !strings.Contains(string(first), `"phase":"denied"`) { + t.Errorf("expected phase=denied in JSON: %s", first) + } + if !strings.Contains(string(first), `"action":"deny"`) { + t.Errorf("expected action=deny (5-value vocab) in JSON: %s", first) + } + if !strings.Contains(string(first), `"action":"modify"`) { + t.Errorf("expected action=modify in JSON: %s", first) + } + var decoded SessionEvent + if err := json.Unmarshal(first, &decoded); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + second, err := json.Marshal(decoded) + if err != nil { + t.Fatalf("second Marshal: %v", err) + } + if string(first) != string(second) { + t.Errorf("Invocations round-trip drifted:\n first: %s\n second: %s", first, second) + } +} + +// Plugins map should round-trip as keyed RawMessage. The listener is +// expected to have already marshaled each value (Extensions.Plugins uses +// any; SessionEvent.Plugins uses json.RawMessage — it's the wire form). +// This test verifies the consumer side: decode lands RawMessage back in +// place so abctl (or any JSON consumer) can re-decode per plugin. +func TestSessionEvent_PluginsMap_JSONRoundTrip(t *testing.T) { + orig := SessionEvent{ + At: time.Unix(1700000000, 0).UTC(), + Direction: Outbound, + Phase: SessionRequest, + Plugins: map[string]json.RawMessage{ + "rate-limiter": json.RawMessage(`{"allowed":true,"tokensLeft":42}`), + "custom-audit": json.RawMessage(`{"traceId":"abc-123"}`), + }, + } + first, err := json.Marshal(orig) + if err != nil { + t.Fatalf("first Marshal: %v", err) + } + if !strings.Contains(string(first), `"plugins":`) { + t.Errorf("expected plugins in JSON: %s", first) + } + var decoded SessionEvent + if err := json.Unmarshal(first, &decoded); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + if len(decoded.Plugins) != 2 { + t.Fatalf("decoded.Plugins size = %d, want 2", len(decoded.Plugins)) + } + // RawMessage should round-trip byte-identical so downstream consumers + // can decode each plugin's payload into its own type. + if got := string(decoded.Plugins["rate-limiter"]); got != `{"allowed":true,"tokensLeft":42}` { + t.Errorf("rate-limiter payload drifted: %q", got) + } +} diff --git a/authbridge/authlib/plugins/CONVENTIONS.md b/authbridge/authlib/plugins/CONVENTIONS.md index 83efe5988..d65465c64 100644 --- a/authbridge/authlib/plugins/CONVENTIONS.md +++ b/authbridge/authlib/plugins/CONVENTIONS.md @@ -272,6 +272,266 @@ list is empty. The error message names the likely cause so the operator is pointed at the migration, not left wondering why authentication isn't happening. +## Emitting session events + +Every plugin MUST emit at least one `Invocation` record per +`OnRequest` / `OnResponse` call. Plugins may also populate one of the +typed protocol extensions (`MCP`, `A2A`, `Inference`) when they carry +structured semantic payload, and may additionally publish arbitrary +plugin-specific events through the `Custom` escape-hatch map. + +### 1. Invocation record (required for every plugin) + +An `Invocation` says *which* plugin ran and *what* it did, in a +5-value vocabulary shared across all plugins. abctl renders one row +per invocation — without an invocation record, a plugin's work is +invisible to the operator. + +Recording is done through `Context` helpers. The framework fills in +`Plugin`, `Phase`, and `Path` automatically from the currently- +dispatching plugin + phase + request path; the plugin supplies only +what's specific to this call. + +**For the common passive actions, use the one-liner wrappers:** + +```go +pctx.Allow("authorized") // ActionAllow + reason +pctx.Skip("path_bypass") // ActionSkip + reason +pctx.Observe("matched_tools/call") // ActionObserve + reason +pctx.Modify("token_replaced") // ActionModify + reason +``` + +**For rejections (control-flow + record in one call):** + +```go +return pctx.DenyAndRecord("jwt_failed", "auth.unauthorized", "token validation failed") +``` + +**For invocations that carry diagnostic context** (auth-gate fields, +route context, cache-hit flag, etc.) use the full `Record`: + +```go +pctx.Record(pipeline.Invocation{ + Action: pipeline.ActionDeny, + Reason: result.DenyReasonCode.String(), + ExpectedIssuer: p.cfg.Issuer, + ExpectedAudience: audience, +}) +return pipeline.DenyStatus(result.DenyStatus, code, result.DenyReason) +``` + +`Plugin`, `Phase`, and `Path` are left zero — the framework fills +them. A plugin CAN set them to non-zero values to override (useful +for test harnesses synthesizing Invocations outside a pipeline run), +but production plugins should not. + +The 5 actions and when to use them: + +| Action | Meaning | Example | +|---|---|---| +| `allow` | Gate plugin permitted the request | jwt-validation on valid token | +| `deny` | Gate plugin rejected the request; pipeline stops | jwt-validation on bad token, token-exchange on IdP failure | +| `skip` | Plugin ran but didn't act on this message | jwt-validation on a bypass path; parser whose body didn't match | +| `modify` | Plugin mutated the message | token-exchange replaced the Authorization header | +| `observe` | Plugin attached diagnostic data; flow unchanged | parsers extracting MCP / A2A / Inference state | + +`Reason` is a stable machine-readable label (e.g. `path_bypass`, +`no_matching_route`, `jwt_failed`, `matched_tools/call`) that +discriminates within an Action value. Filters in abctl can match on +either — `/skip` shows every skip action regardless of reason; +`/path_bypass` narrows to that specific skip flavour. + +Fields populated selectively: auth gates fill `ExpectedIssuer` / +`ExpectedAudience` / `Token*`; outbound routers fill `Route*` and +`CacheHit`; parsers typically fill only `Plugin` / `Action` / +`Reason` / `Path` because their semantic payload lives on the typed +extension slot. + +NEVER put raw tokens, signatures, or secrets in an `Invocation`. The +session store has no auth. + +### 2. Named protocol extension (optional, for parsers) + +`MCP`, `A2A`, `Inference` are typed slots on `pipeline.Extensions`. +A parser that successfully extracts structured state populates the +matching slot AND emits an `Invocation` with `ActionObserve`. The +slot carries the parsed payload; the Invocation carries the +attribution. + +Adding a new named extension is a core-library change: edit +`pipeline/extensions.go`, `pipeline/session.go` (wire + JSON round- +trip), the listener (snapshot + recorder), and abctl if you want +bespoke rendering. Most new plugins don't need one — they emit an +Invocation and publish extra context through the Custom map +(below). + +### 3. Escape-hatch map (`Custom` with `/event` suffix) + +For plugin-specific observability that doesn't warrant a category yet, +write to `pctx.Extensions.Custom` with a key ending in +`pipeline.PluginEventSuffix` (`"/event"`): + +```go +// Plugin-PUBLIC event. Listener serializes this to SessionEvent.Plugins +// under key "rate-limiter" (suffix stripped). +pctx.Extensions.Custom["rate-limiter"+pipeline.PluginEventSuffix] = rateLimiterEvent{ + Allowed: true, + TokensLeft: 42, +} + +// Plugin-PRIVATE cross-phase state. Never serialized. Used via the +// typed SetState / GetState generics. +pipeline.SetState(pctx, "rate-limiter", &rateLimiterState{Bucket: b}) +``` + +The `/event` suffix is the opt-in marker: the listener only promotes +matching keys into `SessionEvent.Plugins`. Private state stays out. + +Rules for plugin-public events: + +- **Value must be JSON-marshalable.** The listener calls `json.Marshal`; + failures downgrade to `slog.Debug` and skip the entry (a misbehaving + plugin can't break the session stream). +- **NEVER put raw credentials or tokens in the value.** The session + store has no auth on it — only safe-to-log data belongs there. +- **Key prefix MUST be the plugin's `Name()`.** Keeps namespaces clean + so unrelated plugins don't collide. +- **Payload schema is plugin-owned.** No central registry; abctl + treats unknown keys as raw JSON in the detail pane. + +### Graduation: when to promote map → named category + +Graduate to a typed slot when ≥2 of these are true: + +1. **Two or more plugins share the shape.** That's the signal the + "category" concept is worth codifying — it prevents N plugins from + each shipping their own near-identical struct. +2. **abctl or the session API grows conditional logic on the key.** + If consumers already parse the payload, making the schema compile- + checked is a net win. +3. **The data is populated on nearly every deployment.** Core + semantics (auth, protocol) graduate; niche plugins stay in the map. + +Don't graduate speculatively — the map path has no cost if you stay +in it. + +## Registering a plugin + +A plugin advertises itself to the pipeline builder through `RegisterPlugin` +in its package `init()`. The registration is open — any package that +imports `authlib/plugins` can register a plugin, regardless of whether it +lives in this module. The pattern mirrors `database/sql` drivers and +`log/slog` handlers. + +### In-tree plugin (lives in `authbridge/authlib/plugins/`) + +Every built-in plugin has an `init()` at the top of its file: + +```go +// authbridge/authlib/plugins/jwtvalidation.go + +package plugins + +func NewJWTValidation() *JWTValidation { return &JWTValidation{} } + +func init() { + RegisterPlugin("jwt-validation", func() pipeline.Plugin { return NewJWTValidation() }) +} +``` + +Because the file is in the `plugins` package, `init()` runs automatically +when anything imports the package (e.g., `authbridge/cmd/authbridge/main.go` +via `authlib/plugins.Build`). + +### Out-of-tree plugin (separate Go module) + +A plugin maintained outside kagenti-extensions lives in its own package +and registers the same way: + +```go +// github.com/acme/kagenti-rate-limiter/ratelimit.go + +package ratelimit + +import ( + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins" +) + +type RateLimiter struct { ... } + +func (p *RateLimiter) Name() string { return "rate-limiter" } +func (p *RateLimiter) Capabilities() pipeline.PluginCapabilities { ... } +func (p *RateLimiter) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { ... } +func (p *RateLimiter) OnResponse(_ context.Context, pctx *pipeline.Context) pipeline.Action { ... } + +func init() { + plugins.RegisterPlugin("rate-limiter", func() pipeline.Plugin { + return &RateLimiter{} + }) +} +``` + +The authbridge binary picks the plugin up via a single side-effect import: + +```go +// authbridge/cmd/authbridge/plugins_extra.go (or wherever you customize) + +package main + +import _ "github.com/acme/kagenti-rate-limiter/ratelimit" +``` + +With that import, operator YAML can list `rate-limiter` in the pipeline: + +```yaml +pipeline: + inbound: + plugins: + - name: jwt-validation + - name: rate-limiter + config: { requests_per_minute: 100 } +``` + +No fork of kagenti-extensions needed. Plugin is a regular Go module. + +### Rules and guardrails + +- **Double-registration panics.** If two packages both register under the + same name, the second call panics at process start. This is the + correct behaviour: silent last-write-wins would let a version + conflict poison the pipeline composition in ways that only surface as + mysterious runtime behaviour. +- **Empty name panics.** An empty plugin name cannot be referenced from + YAML; registering under one is a programmer bug, not a recoverable + condition. +- **Nil factory panics.** A nil factory would defer the crash until + `Build` tried to call it; panic at registration is closer to the bug. +- **Unknown plugin fails Build.** `Build` rejects entries whose name + isn't in the registry; the error message includes every registered + name so typos are easy to spot. + +### Testing against the registry + +Tests that need a fake plugin use `RegisterPlugin` + `t.Cleanup` with +`UnregisterPlugin`: + +```go +func TestMyScenario(t *testing.T) { + plugins.RegisterPlugin("fake-auth", func() pipeline.Plugin { + return &fakeAuth{} + }) + t.Cleanup(func() { plugins.UnregisterPlugin("fake-auth") }) + + p, err := plugins.Build([]config.PluginEntry{{Name: "fake-auth"}}) + // ... assert on p ... +} +``` + +`UnregisterPlugin` is test-only by convention — production code should +never call it. It exists to keep tests isolated from each other under +`-parallel`. + ## Cross-references - `authbridge/authlib/pipeline/configurable.go` — the interface. @@ -280,3 +540,8 @@ authentication isn't happening. - `authbridge/authlib/config/config.go` — `PluginEntry` YAML shape and parsing. - `authbridge/authlib/plugins/registry.go` — how Build calls Configure. +- `authbridge/authlib/pipeline/extensions.go` — named categories + (`MCP`, `A2A`, `Inference`, `Auth`) + `Custom` map + escape-hatch + convention. +- `authbridge/authlib/pipeline/session.go` — `SessionEvent` wire shape + and the `SessionDenied` phase. diff --git a/authbridge/authlib/plugins/a2aparser.go b/authbridge/authlib/plugins/a2aparser.go index 911d597cc..9dd4db010 100644 --- a/authbridge/authlib/plugins/a2aparser.go +++ b/authbridge/authlib/plugins/a2aparser.go @@ -16,6 +16,10 @@ type A2AParser struct{} func NewA2AParser() *A2AParser { return &A2AParser{} } +func init() { + RegisterPlugin("a2a-parser", func() pipeline.Plugin { return NewA2AParser() }) +} + func (p *A2AParser) Name() string { return "a2a-parser" } func (p *A2AParser) Capabilities() pipeline.PluginCapabilities { @@ -26,6 +30,11 @@ func (p *A2AParser) Capabilities() pipeline.PluginCapabilities { } func (p *A2AParser) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { + // No Invocation recorded when the parser doesn't apply to this + // message (empty body, non-JSON-RPC body) — otherwise every + // unrelated HTTP call through the pipeline would show an "a2a-parser + // skip" row in abctl, which is noise. Operators infer "a2a-parser + // exists in this pipeline" from the pipeline config, not per-event. if len(pctx.Body) == 0 { slog.Debug("a2a-parser: no body, skipping") return pipeline.Action{Type: pipeline.Continue} @@ -75,6 +84,7 @@ func (p *A2AParser) OnRequest(_ context.Context, pctx *pipeline.Context) pipelin for i, part := range ext.Parts { slog.Debug("a2a-parser: part", "index", i, "kind", part.Kind, "content", truncate(part.Content, debugBodyMax)) } + pctx.Observe("matched_" + rpc.Method) return pipeline.Action{Type: pipeline.Continue} } @@ -84,6 +94,9 @@ func (p *A2AParser) OnRequest(_ context.Context, pctx *pipeline.Context) pipelin // // Handles both JSON-RPC responses (message/send) and SSE event streams (message/stream). func (p *A2AParser) OnResponse(_ context.Context, pctx *pipeline.Context) pipeline.Action { + // No Invocation on response when the parser doesn't apply: either + // there's no response body or the matching request wasn't an A2A + // JSON-RPC call. Keeps the response event clean for non-A2A traffic. if len(pctx.ResponseBody) == 0 || pctx.Extensions.A2A == nil { return pipeline.Action{Type: pipeline.Continue} } @@ -110,6 +123,7 @@ func (p *A2AParser) OnResponse(_ context.Context, pctx *pipeline.Context) pipeli "artifactLen", len(pctx.Extensions.A2A.Artifact), "error", pctx.Extensions.A2A.ErrorMessage, ) + pctx.Observe("matched_" + pctx.Extensions.A2A.Method + "_response") return pipeline.Action{Type: pipeline.Continue} } diff --git a/authbridge/authlib/plugins/inferenceparser.go b/authbridge/authlib/plugins/inferenceparser.go index f9c259b21..270089257 100644 --- a/authbridge/authlib/plugins/inferenceparser.go +++ b/authbridge/authlib/plugins/inferenceparser.go @@ -16,6 +16,10 @@ type InferenceParser struct{} func NewInferenceParser() *InferenceParser { return &InferenceParser{} } +func init() { + RegisterPlugin("inference-parser", func() pipeline.Plugin { return NewInferenceParser() }) +} + func (p *InferenceParser) Name() string { return "inference-parser" } func (p *InferenceParser) Capabilities() pipeline.PluginCapabilities { @@ -26,6 +30,11 @@ func (p *InferenceParser) Capabilities() pipeline.PluginCapabilities { } func (p *InferenceParser) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { + // No Invocation recorded when the parser doesn't apply to this + // message — wrong path (anything other than OpenAI chat/completion + // endpoints), empty body, or non-JSON body. Operators infer + // "inference-parser exists in this pipeline" from config, not per- + // event rows. if pctx.Path != "/v1/chat/completions" && pctx.Path != "/v1/completions" { return pipeline.Action{Type: pipeline.Continue} } @@ -76,6 +85,7 @@ func (p *InferenceParser) OnRequest(_ context.Context, pctx *pipeline.Context) p slog.Debug("inference-parser: message", "index", i, "role", m.Role, "content", truncate(m.Content, debugBodyMax)) } + pctx.Observe("matched_" + ext.Model) return pipeline.Action{Type: pipeline.Continue} } @@ -83,6 +93,8 @@ func (p *InferenceParser) OnRequest(_ context.Context, pctx *pipeline.Context) p // token counts) on pctx.Extensions.Inference. Handles both non-streaming // JSON responses and SSE streams from OpenAI-compatible servers. func (p *InferenceParser) OnResponse(_ context.Context, pctx *pipeline.Context) pipeline.Action { + // No Invocation when the parser doesn't apply — request wasn't + // inference or no response body to parse. if len(pctx.ResponseBody) == 0 || pctx.Extensions.Inference == nil { return pipeline.Action{Type: pipeline.Continue} } @@ -101,6 +113,7 @@ func (p *InferenceParser) OnResponse(_ context.Context, pctx *pipeline.Context) "completionTokens", ext.CompletionTokens, ) slog.Debug("inference-parser: completion", "text", truncate(ext.Completion, debugBodyMax)) + pctx.Observe("matched_" + ext.Model + "_response") return pipeline.Action{Type: pipeline.Continue} } diff --git a/authbridge/authlib/plugins/jwtvalidation.go b/authbridge/authlib/plugins/jwtvalidation.go index bee71095d..ffdc7d2f4 100644 --- a/authbridge/authlib/plugins/jwtvalidation.go +++ b/authbridge/authlib/plugins/jwtvalidation.go @@ -160,6 +160,10 @@ type JWTValidation struct { // called before the pipeline accepts traffic. func NewJWTValidation() *JWTValidation { return &JWTValidation{} } +func init() { + RegisterPlugin("jwt-validation", func() pipeline.Plugin { return NewJWTValidation() }) +} + func (p *JWTValidation) Name() string { return "jwt-validation" } func (p *JWTValidation) Capabilities() pipeline.PluginCapabilities { @@ -294,6 +298,19 @@ func (p *JWTValidation) OnRequest(ctx context.Context, pctx *pipeline.Context) p result := p.inner.HandleInbound(ctx, authHeader, path, audience) if result.Action == auth.ActionDeny { + // Surface the decision on pctx BEFORE returning so the listener's + // reject path can record a SessionDenied event with diagnostic + // context (why the token failed, what was expected). Never put + // the raw token here — session store has no auth. The two-step + // form (Record + Deny) is used here because we attach the + // ExpectedIssuer / ExpectedAudience diagnostic fields that the + // one-liner DenyAndRecord doesn't accept. + pctx.Record(pipeline.Invocation{ + Action: pipeline.ActionDeny, + Reason: result.DenyReasonCode.String(), + ExpectedIssuer: p.cfg.Issuer, + ExpectedAudience: audience, + }) // result.DenyReason carries the specific failure (missing header, // audience mismatch, expired, etc.). Pick a code whose default // HTTP status matches what auth returned, so the fallback body is @@ -305,7 +322,27 @@ func (p *JWTValidation) OnRequest(ctx context.Context, pctx *pipeline.Context) p } return pipeline.DenyStatus(result.DenyStatus, code, result.DenyReason) } + + // ActionAllow with nil Claims = bypass path (e.g., /healthz). Record + // as a bypass event so operators can still see the request in the + // session stream — useful for debugging "why is this URL skipping + // JWT?" without hunting through slog lines. + if result.Claims == nil { + pctx.Skip("path_bypass") + return pipeline.Action{Type: pipeline.Continue} + } + + // ActionAllow with Claims = authorized. Surface what the plugin + // VERIFIED in the token — diverges from the top-level Identity + // snapshot if later plugins re-annotate pctx.Claims. pctx.Claims = result.Claims + pctx.Record(pipeline.Invocation{ + Action: pipeline.ActionAllow, + Reason: auth.APPROVE_AUTHORIZED.String(), + TokenSubject: result.Claims.Subject, + TokenAudience: result.Claims.Audience, + TokenScopes: result.Claims.Scopes, + }) return pipeline.Action{Type: pipeline.Continue} } diff --git a/authbridge/authlib/plugins/mcpparser.go b/authbridge/authlib/plugins/mcpparser.go index aae73af87..e41ccc644 100644 --- a/authbridge/authlib/plugins/mcpparser.go +++ b/authbridge/authlib/plugins/mcpparser.go @@ -16,6 +16,10 @@ type MCPParser struct{} func NewMCPParser() *MCPParser { return &MCPParser{} } +func init() { + RegisterPlugin("mcp-parser", func() pipeline.Plugin { return NewMCPParser() }) +} + func (p *MCPParser) Name() string { return "mcp-parser" } func (p *MCPParser) Capabilities() pipeline.PluginCapabilities { @@ -26,6 +30,10 @@ func (p *MCPParser) Capabilities() pipeline.PluginCapabilities { } func (p *MCPParser) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { + // No Invocation recorded when the parser doesn't apply to this + // message — empty body, non-JSON body, or JSON-but-not-JSON-RPC + // (e.g. an OpenAI chat/completions body). Operators infer "mcp- + // parser exists in this pipeline" from config, not per-event rows. if len(pctx.Body) == 0 { slog.Debug("mcp-parser: no body, skipping") return pipeline.Action{Type: pipeline.Continue} @@ -55,10 +63,16 @@ func (p *MCPParser) OnRequest(_ context.Context, pctx *pipeline.Context) pipelin slog.Info("mcp-parser: request", "method", rpc.Method) slog.Debug("mcp-parser: payload", "method", rpc.Method, "body", truncate(string(pctx.Body), debugBodyMax)) + pctx.Observe("matched_" + rpc.Method) return pipeline.Action{Type: pipeline.Continue} } func (p *MCPParser) OnResponse(_ context.Context, pctx *pipeline.Context) pipeline.Action { + // No Invocation when the parser doesn't apply — request wasn't MCP + // JSON-RPC or no response body to parse. The unparseable_response + // case below IS recorded because it's diagnostic: the request WAS + // MCP but the response couldn't be decoded, which usually signals + // an upstream protocol bug worth surfacing. if len(pctx.ResponseBody) == 0 || pctx.Extensions.MCP == nil { return pipeline.Action{Type: pipeline.Continue} } @@ -66,6 +80,7 @@ func (p *MCPParser) OnResponse(_ context.Context, pctx *pipeline.Context) pipeli rpc, ok := parseMCPResponse(pctx.ResponseBody) if !ok { slog.Debug("mcp-parser: response is not valid JSON-RPC or SSE", "bodyLen", len(pctx.ResponseBody)) + pctx.Skip("unparseable_response") return pipeline.Action{Type: pipeline.Continue} } @@ -76,6 +91,7 @@ func (p *MCPParser) OnResponse(_ context.Context, pctx *pipeline.Context) pipeli Data: rpc.Error.Data, } slog.Info("mcp-parser: response error", "method", pctx.Extensions.MCP.Method, "code", rpc.Error.Code, "message", rpc.Error.Message) + pctx.Observe("response_error") return pipeline.Action{Type: pipeline.Continue} } @@ -85,6 +101,7 @@ func (p *MCPParser) OnResponse(_ context.Context, pctx *pipeline.Context) pipeli slog.Debug("mcp-parser: response detail", "method", pctx.Extensions.MCP.Method, "body", truncate(string(pctx.ResponseBody), debugBodyMax)) } + pctx.Observe("matched_" + pctx.Extensions.MCP.Method + "_response") return pipeline.Action{Type: pipeline.Continue} } diff --git a/authbridge/authlib/plugins/plugins_test.go b/authbridge/authlib/plugins/plugins_test.go index 13932fb16..0f29efd6c 100644 --- a/authbridge/authlib/plugins/plugins_test.go +++ b/authbridge/authlib/plugins/plugins_test.go @@ -10,10 +10,33 @@ import ( "strings" "testing" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/bypass" "github.com/kagenti/kagenti-extensions/authbridge/authlib/config" "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/validation" ) +// invokeOnRequest mirrors what Pipeline.Run does around each plugin +// dispatch: set the current-plugin / current-phase attribution fields +// on pctx so pctx.Record / Allow / Skip / Observe / Modify fill in +// Plugin and Phase correctly. Tests that call plugin.OnRequest directly +// (bypassing Pipeline.Run) need this wrapper to exercise the same code +// path as production. Without it, Invocations would land with empty +// Plugin and Phase fields. +func invokeOnRequest(p pipeline.Plugin, pctx *pipeline.Context) pipeline.Action { + pctx.SetCurrentPlugin(p.Name(), pipeline.InvocationPhaseRequest) + defer pctx.ClearCurrentPlugin() + return p.OnRequest(context.Background(), pctx) +} + +// invokeOnResponse is the response-phase twin of invokeOnRequest. +func invokeOnResponse(p pipeline.Plugin, pctx *pipeline.Context) pipeline.Action { + pctx.SetCurrentPlugin(p.Name(), pipeline.InvocationPhaseResponse) + defer pctx.ClearCurrentPlugin() + return p.OnResponse(context.Background(), pctx) +} + // TestAuthbridgeCombinedYAML_Loads asserts that the in-repo default // config consumed by the combined sidecar image // (authbridge/authproxy/authbridge-combined.yaml) parses, env-expands, @@ -302,12 +325,136 @@ func TestJWTValidation_Ready_PerHostAlwaysReady(t *testing.T) { func TestJWTValidation_OnRequest_NotConfigured(t *testing.T) { p := NewJWTValidation() - action := p.OnRequest(context.Background(), &pipeline.Context{Headers: http.Header{}}) + action := invokeOnRequest(p, &pipeline.Context{Headers: http.Header{}}) if action.Type != pipeline.Reject { t.Errorf("got %v, want Reject for unconfigured plugin", action.Type) } } +// --- JWTValidation: Auth extension population --- +// +// These tests verify jwt-validation surfaces its decision on +// pctx.Extensions.Invocations.Inbound so the listener can record a +// SessionEvent reflecting allow/deny/bypass. Plumbed through a +// mockVerifier injected into p.inner (instead of spinning up a real +// JWKS server) — keeps the test focused on plugin behavior, not crypto. + +// mockJWTVerifier lets the tests below dictate what the inner validator +// returns without standing up an httptest JWKS server. It implements the +// validation.Verifier interface. +type mockJWTVerifier struct { + claims *validation.Claims + err error +} + +func (m *mockJWTVerifier) Verify(_ context.Context, _, _ string) (*validation.Claims, error) { + return m.claims, m.err +} + +// newTestJWTValidation constructs a JWTValidation plugin without calling +// Configure — skips file I/O (audience_file polling, bypass pattern +// compile via config) and lets each test wire a tailored inner auth.Auth. +func newTestJWTValidation(t *testing.T, issuer string, inner *auth.Auth) *JWTValidation { + t.Helper() + p := NewJWTValidation() + p.cfg.Issuer = issuer + p.inner = inner + return p +} + +func TestJWTValidation_OnRequest_PopulatesAuth_Bypass(t *testing.T) { + matcher, _ := bypass.NewMatcher(bypass.DefaultPatterns) + inner := auth.New(auth.Config{ + Bypass: matcher, + Verifier: &mockJWTVerifier{claims: &validation.Claims{Subject: "s"}}, + Identity: auth.IdentityConfig{Audience: "agent-aud"}, + }) + p := newTestJWTValidation(t, "http://issuer", inner) + + pctx := &pipeline.Context{Headers: http.Header{}, Path: "/healthz"} + action := invokeOnRequest(p, pctx) + if action.Type != pipeline.Continue { + t.Fatalf("bypass should Continue, got %v", action.Type) + } + if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Inbound) != 1 { + t.Fatalf("expected one Auth.Inbound entry, got %+v", pctx.Extensions.Invocations) + } + got := pctx.Extensions.Invocations.Inbound[0] + if got.Plugin != "jwt-validation" { + t.Errorf("Plugin = %q, want jwt-validation", got.Plugin) + } + if got.Action != pipeline.ActionSkip || got.Reason != "path_bypass" { + t.Errorf("got Action=%q Reason=%q, want skip/path_bypass", got.Action, got.Reason) + } +} + +func TestJWTValidation_OnRequest_PopulatesAuth_Deny_NoHeader(t *testing.T) { + inner := auth.New(auth.Config{ + Verifier: &mockJWTVerifier{}, + Identity: auth.IdentityConfig{Audience: "agent-aud"}, + }) + p := newTestJWTValidation(t, "http://issuer.example", inner) + + pctx := &pipeline.Context{Headers: http.Header{}, Path: "/api/call"} + action := invokeOnRequest(p, pctx) + if action.Type != pipeline.Reject { + t.Fatalf("expected Reject on missing auth header, got %v", action.Type) + } + if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Inbound) != 1 { + t.Fatalf("expected one Auth.Inbound entry, got %+v", pctx.Extensions.Invocations) + } + got := pctx.Extensions.Invocations.Inbound[0] + if got.Action != pipeline.ActionDeny { + t.Errorf("Action = %q, want deny", got.Action) + } + // Reason comes from InboundDenialReason.String() so consumers can + // filter on a machine-stable code without parsing English. + if got.Reason != "no_header" { + t.Errorf("Reason = %q, want no_header", got.Reason) + } + if got.ExpectedIssuer != "http://issuer.example" { + t.Errorf("ExpectedIssuer = %q, want http://issuer.example", got.ExpectedIssuer) + } +} + +func TestJWTValidation_OnRequest_PopulatesAuth_Allow(t *testing.T) { + claims := &validation.Claims{ + Subject: "alice", + Issuer: "http://issuer.example", + Audience: []string{"agent-aud"}, + ClientID: "caller", + Scopes: []string{"openid", "write"}, + } + inner := auth.New(auth.Config{ + Verifier: &mockJWTVerifier{claims: claims}, + Identity: auth.IdentityConfig{Audience: "agent-aud"}, + }) + p := newTestJWTValidation(t, "http://issuer.example", inner) + + pctx := &pipeline.Context{Headers: http.Header{}, Path: "/api/call"} + pctx.Headers.Set("Authorization", "Bearer tok") + action := invokeOnRequest(p, pctx) + if action.Type != pipeline.Continue { + t.Fatalf("expected Continue, got %v (violation=%+v)", action.Type, action.Violation) + } + if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Inbound) != 1 { + t.Fatalf("expected one Auth.Inbound entry, got %+v", pctx.Extensions.Invocations) + } + got := pctx.Extensions.Invocations.Inbound[0] + if got.Action != pipeline.ActionAllow || got.Reason != "authorized" { + t.Errorf("got Action=%q Reason=%q, want allow/authorized", got.Action, got.Reason) + } + if got.TokenSubject != "alice" { + t.Errorf("TokenSubject = %q, want alice", got.TokenSubject) + } + if len(got.TokenScopes) != 2 || got.TokenScopes[0] != "openid" { + t.Errorf("TokenScopes = %v, want [openid write]", got.TokenScopes) + } + if len(got.TokenAudience) != 1 || got.TokenAudience[0] != "agent-aud" { + t.Errorf("TokenAudience = %v, want [agent-aud]", got.TokenAudience) + } +} + // --- TokenExchange: Configure --- func TestTokenExchange_Configure_MissingTokenURL(t *testing.T) { @@ -501,13 +648,31 @@ func TestTokenExchange_Passthrough(t *testing.T) { Host: "some-host", Headers: http.Header{"Authorization": []string{"Bearer user-token"}}, } - action := p.OnRequest(context.Background(), pctx) + action := invokeOnRequest(p, pctx) if action.Type != pipeline.Continue { t.Fatalf("got %v, want Continue", action.Type) } if pctx.Headers.Get("Authorization") != "Bearer user-token" { t.Error("headers should not be modified for passthrough") } + // Passthrough populates Auth.Outbound with Action="passthrough" — + // symmetric with jwt-validation's bypass recording so operators can + // see every outbound host the pod talks to in the session stream. + // RouteHost carries the target so they can spot unexpected egress + // without hunting through slog lines. + if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Outbound) != 1 { + t.Fatalf("expected one Auth.Outbound entry, got %+v", pctx.Extensions.Invocations) + } + ob := pctx.Extensions.Invocations.Outbound[0] + if ob.Plugin != "token-exchange" || ob.Action != pipeline.ActionSkip { + t.Errorf("entry = (%q, %q), want (token-exchange, skip)", ob.Plugin, ob.Action) + } + if ob.RouteHost != "some-host" { + t.Errorf("RouteHost = %q, want some-host", ob.RouteHost) + } + if ob.RouteMatched { + t.Error("RouteMatched should be false on default-policy passthrough") + } } func TestTokenExchange_ExchangeSuccess(t *testing.T) { @@ -535,13 +700,30 @@ func TestTokenExchange_ExchangeSuccess(t *testing.T) { Host: "target-svc", Headers: http.Header{"Authorization": []string{"Bearer user-token"}}, } - action := p.OnRequest(context.Background(), pctx) + action := invokeOnRequest(p, pctx) if action.Type != pipeline.Continue { t.Fatalf("got %v, want Continue", action.Type) } if pctx.Headers.Get("Authorization") != "Bearer new-token" { t.Errorf("token = %q, want Bearer new-token", pctx.Headers.Get("Authorization")) } + // Auth extension must surface the exchange action so it flows into + // SessionEvent.Auth.Outbound once the listener records. Empty route + // (TargetAudience, RequestedScopes) is OK here — this test doesn't + // configure routes, it uses default_policy=exchange. + if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Outbound) != 1 { + t.Fatalf("expected one Auth.Outbound entry, got %+v", pctx.Extensions.Invocations) + } + got := pctx.Extensions.Invocations.Outbound[0] + if got.Plugin != "token-exchange" || got.Action != pipeline.ActionModify { + t.Errorf("got Plugin=%q Action=%q, want token-exchange/modify", got.Plugin, got.Action) + } + if got.RouteHost != "target-svc" { + t.Errorf("RouteHost = %q, want target-svc", got.RouteHost) + } + if got.CacheHit { + t.Error("CacheHit = true on first exchange; should be false") + } } func TestTokenExchange_ExchangeFailure(t *testing.T) { @@ -565,7 +747,7 @@ func TestTokenExchange_ExchangeFailure(t *testing.T) { Host: "target-svc", Headers: http.Header{"Authorization": []string{"Bearer user-token"}}, } - action := p.OnRequest(context.Background(), pctx) + action := invokeOnRequest(p, pctx) if action.Type != pipeline.Reject { t.Fatalf("got %v, want Reject", action.Type) } @@ -573,6 +755,19 @@ func TestTokenExchange_ExchangeFailure(t *testing.T) { if status != http.StatusServiceUnavailable { t.Errorf("status = %d, want 503", status) } + // Deny branch must surface a "denied" Auth.Outbound entry with the + // machine-stable reason — matches what the listener needs to emit a + // SessionDenied event on outbound exchange failure. + if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Outbound) != 1 { + t.Fatalf("expected one Auth.Outbound entry, got %+v", pctx.Extensions.Invocations) + } + got := pctx.Extensions.Invocations.Outbound[0] + if got.Action != pipeline.ActionDeny { + t.Errorf("Action = %q, want deny", got.Action) + } + if got.Reason != "token_exchange_failed" { + t.Errorf("Reason = %q, want token_exchange_failed (from OutboundDenialReason.String)", got.Reason) + } } func TestTokenExchange_NoToken_Deny(t *testing.T) { @@ -591,7 +786,7 @@ func TestTokenExchange_NoToken_Deny(t *testing.T) { Host: "target-svc", Headers: http.Header{}, } - action := p.OnRequest(context.Background(), pctx) + action := invokeOnRequest(p, pctx) if action.Type != pipeline.Reject { t.Fatalf("got %v, want Reject", action.Type) } diff --git a/authbridge/authlib/plugins/registry.go b/authbridge/authlib/plugins/registry.go index a79059c31..88b0664de 100644 --- a/authbridge/authlib/plugins/registry.go +++ b/authbridge/authlib/plugins/registry.go @@ -2,6 +2,8 @@ package plugins import ( "fmt" + "sort" + "sync" "github.com/kagenti/kagenti-extensions/authbridge/authlib/config" "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" @@ -14,12 +16,76 @@ import ( // local config inside Configure. type PluginFactory func() pipeline.Plugin -var registry = map[string]PluginFactory{ - "jwt-validation": func() pipeline.Plugin { return NewJWTValidation() }, - "token-exchange": func() pipeline.Plugin { return NewTokenExchange() }, - "mcp-parser": func() pipeline.Plugin { return NewMCPParser() }, - "a2a-parser": func() pipeline.Plugin { return NewA2AParser() }, - "inference-parser": func() pipeline.Plugin { return NewInferenceParser() }, +// registry is the dynamic plugin table. Populated by RegisterPlugin, +// typically from each plugin package's init() function. Guarded by a +// mutex because init() order across packages isn't guaranteed to be +// serial under every Go build mode, and tests use UnregisterPlugin +// concurrently with t.Parallel. +var ( + registryMu sync.RWMutex + registry = map[string]PluginFactory{} +) + +// RegisterPlugin adds a plugin factory under name. Intended to be +// called from package init() functions of plugin implementations: +// +// func init() { +// plugins.RegisterPlugin("rate-limiter", func() pipeline.Plugin { +// return &RateLimiter{} +// }) +// } +// +// This is the stdlib pattern (database/sql.Register, image codec +// registration, log/slog handler registration): plugins live in their +// own package and advertise themselves by side-effect import: +// +// import _ "github.com/acme/kagenti-rate-limiter/ratelimit" +// +// Double-registration under the same name panics. Silent last-write- +// wins would let a version mismatch or deployment bug poison the +// registry in ways that only surface as mysterious runtime behaviour; +// failing loud at process start is strictly safer. +// +// Empty name or nil factory also panics — both are programmer errors, +// not recoverable conditions. +func RegisterPlugin(name string, factory PluginFactory) { + if name == "" { + panic("plugins: RegisterPlugin called with empty name") + } + if factory == nil { + panic(fmt.Sprintf("plugins: RegisterPlugin(%q) factory is nil", name)) + } + registryMu.Lock() + defer registryMu.Unlock() + if _, exists := registry[name]; exists { + panic(fmt.Sprintf("plugins: %q already registered", name)) + } + registry[name] = factory +} + +// RegisteredPlugins returns the names of every registered plugin in +// sorted order. Intended for diagnostic surfaces (/config, CLI --help, +// Build's "unknown plugin" error message) and for tests that assert a +// plugin is visible to the builder. +func RegisteredPlugins() []string { + registryMu.RLock() + defer registryMu.RUnlock() + names := make([]string, 0, len(registry)) + for n := range registry { + names = append(names, n) + } + sort.Strings(names) + return names +} + +// factoryFor looks up a factory by name. Internal to the package. +// Callers under Build use this to resolve config entries into plugin +// instances. +func factoryFor(name string) (PluginFactory, bool) { + registryMu.RLock() + defer registryMu.RUnlock() + f, ok := registry[name] + return f, ok } // Build constructs a pipeline from an ordered list of plugin entries. @@ -29,13 +95,14 @@ var registry = map[string]PluginFactory{ // stale or misplaced config blocks fail at startup instead of being // silently ignored. // -// Unknown plugin names fail fast. +// Unknown plugin names fail fast with an error that lists every +// currently-registered plugin — typo-catching diagnostic. func Build(entries []config.PluginEntry, opts ...pipeline.Option) (*pipeline.Pipeline, error) { ps := make([]pipeline.Plugin, 0, len(entries)) for _, e := range entries { - factory, ok := registry[e.Name] + factory, ok := factoryFor(e.Name) if !ok { - return nil, fmt.Errorf("unknown plugin %q", e.Name) + return nil, fmt.Errorf("unknown plugin %q (registered: %v)", e.Name, RegisteredPlugins()) } p := factory() if c, ok := p.(pipeline.Configurable); ok { diff --git a/authbridge/authlib/plugins/registry_test.go b/authbridge/authlib/plugins/registry_test.go new file mode 100644 index 000000000..7d2049b1c --- /dev/null +++ b/authbridge/authlib/plugins/registry_test.go @@ -0,0 +1,122 @@ +package plugins + +import ( + "testing" + + "github.com/kagenti/kagenti-extensions/authbridge/authlib/config" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" +) + +// TestBuiltinsRegistered verifies every in-tree plugin is discoverable +// through the new registry — the list is the public contract that +// operator YAML depends on, so a regression here breaks deployments. +func TestBuiltinsRegistered(t *testing.T) { + want := map[string]bool{ + "jwt-validation": true, + "token-exchange": true, + "a2a-parser": true, + "mcp-parser": true, + "inference-parser": true, + } + got := RegisteredPlugins() + gotSet := make(map[string]bool, len(got)) + for _, n := range got { + gotSet[n] = true + } + for name := range want { + if !gotSet[name] { + t.Errorf("built-in plugin %q missing from registry; got: %v", name, got) + } + } +} + +// TestRegisterPlugin_DoubleRegistration_Panics locks the strict-fail +// policy. Silent last-write-wins would let a deployment with two +// incompatible copies of the same plugin corrupt the pipeline +// composition; panic on registration catches it at process start. +func TestRegisterPlugin_DoubleRegistration_Panics(t *testing.T) { + name := "test-double-register" + RegisterPlugin(name, func() pipeline.Plugin { return nil }) + t.Cleanup(func() { UnregisterPlugin(name) }) + + defer func() { + if r := recover(); r == nil { + t.Errorf("expected panic on double-registration") + } + }() + RegisterPlugin(name, func() pipeline.Plugin { return nil }) +} + +func TestRegisterPlugin_EmptyName_Panics(t *testing.T) { + defer func() { + if r := recover(); r == nil { + t.Errorf("expected panic on empty name") + } + }() + RegisterPlugin("", func() pipeline.Plugin { return nil }) +} + +func TestRegisterPlugin_NilFactory_Panics(t *testing.T) { + defer func() { + if r := recover(); r == nil { + t.Errorf("expected panic on nil factory") + } + }() + RegisterPlugin("test-nil-factory", nil) +} + +// TestUnregisterPlugin verifies the test-isolation helper. After +// registering + unregistering, the name is absent from RegisteredPlugins +// and Build rejects it as unknown. +func TestUnregisterPlugin(t *testing.T) { + name := "test-unregister" + RegisterPlugin(name, func() pipeline.Plugin { return nil }) + if !contains(RegisteredPlugins(), name) { + t.Fatalf("plugin not registered after RegisterPlugin") + } + if !UnregisterPlugin(name) { + t.Errorf("UnregisterPlugin returned false for a registered name") + } + if contains(RegisteredPlugins(), name) { + t.Errorf("plugin still in registry after UnregisterPlugin") + } + // Second unregister should be a no-op (returns false). + if UnregisterPlugin(name) { + t.Errorf("UnregisterPlugin returned true for an unregistered name") + } +} + +// TestBuild_UnknownPlugin_ListsRegistered verifies the "unknown plugin" +// error includes the list of registered names so operators get a +// typo-catching diagnostic instead of a generic not-found. +func TestBuild_UnknownPlugin_ListsRegistered(t *testing.T) { + _, err := Build([]config.PluginEntry{{Name: "not-a-real-plugin"}}) + if err == nil { + t.Fatalf("expected error for unknown plugin") + } + msg := err.Error() + if !containsSubstring(msg, "not-a-real-plugin") { + t.Errorf("error should name the unknown plugin: %q", msg) + } + if !containsSubstring(msg, "jwt-validation") { + t.Errorf("error should list registered plugins (for typo diagnostics): %q", msg) + } +} + +func contains(haystack []string, needle string) bool { + for _, s := range haystack { + if s == needle { + return true + } + } + return false +} + +func containsSubstring(s, sub string) bool { + for i := 0; i+len(sub) <= len(s); i++ { + if s[i:i+len(sub)] == sub { + return true + } + } + return false +} diff --git a/authbridge/authlib/plugins/registry_testing.go b/authbridge/authlib/plugins/registry_testing.go new file mode 100644 index 000000000..68df9f3a2 --- /dev/null +++ b/authbridge/authlib/plugins/registry_testing.go @@ -0,0 +1,30 @@ +package plugins + +// This file is intentionally NOT named with a _test.go suffix so that +// UnregisterPlugin is importable by tests in OTHER packages (e.g., +// cmd/authbridge/listener tests that want to register a fake plugin). +// It IS, however, clearly-named and documented as a test affordance — +// callers must not use it in production code paths. + +// UnregisterPlugin removes a plugin factory from the registry. Intended +// for test isolation: a test registers a fake plugin, runs, and uses +// t.Cleanup to unregister so parallel tests aren't poisoned by the +// leftover entry. +// +// Do not call from production code. The registry is intended to be +// written exactly once per plugin per process, at init() time; runtime +// deregistration has no valid use case in a running authbridge binary +// and would make the /config endpoint lie about pipeline composition. +// +// Returns true when the name was registered (and is now removed), false +// when it wasn't. Callers ignoring the return value are common and +// correct — Cleanup doesn't care. +func UnregisterPlugin(name string) bool { + registryMu.Lock() + defer registryMu.Unlock() + if _, ok := registry[name]; !ok { + return false + } + delete(registry, name) + return true +} diff --git a/authbridge/authlib/plugins/tokenexchange.go b/authbridge/authlib/plugins/tokenexchange.go index 887e09938..6b8c98e28 100644 --- a/authbridge/authlib/plugins/tokenexchange.go +++ b/authbridge/authlib/plugins/tokenexchange.go @@ -203,6 +203,10 @@ type TokenExchange struct { // NewTokenExchange constructs an unconfigured plugin. func NewTokenExchange() *TokenExchange { return &TokenExchange{} } +func init() { + RegisterPlugin("token-exchange", func() pipeline.Plugin { return NewTokenExchange() }) +} + func (p *TokenExchange) Name() string { return "token-exchange" } func (p *TokenExchange) Capabilities() pipeline.PluginCapabilities { @@ -463,8 +467,23 @@ func (p *TokenExchange) OnRequest(ctx context.Context, pctx *pipeline.Context) p host := pctx.Host result := p.inner.HandleOutbound(ctx, authHeader, host) + // Record an Auth.Outbound entry on every branch so operators have + // full outbound audit in the session stream — matches the inbound + // side's recording of allow/deny/bypass and mirrors the claim in the + // PLUGIN column that every event is attributable to a plugin. + // Passthrough is the "no route matched, default policy allowed" + // branch and is the noisiest; operators who find it too loud can + // either tighten routes or filter on action=passthrough in abctl. switch result.Action { case auth.ActionDeny: + pctx.Record(pipeline.Invocation{ + Action: pipeline.ActionDeny, + Reason: result.DenyReasonCode.String(), + RouteMatched: result.RouteMatched, + RouteHost: host, + TargetAudience: result.TargetAudience, + RequestedScopes: splitScopes(result.RequestedScopes), + }) // Outbound denials almost always come from failed token exchange // at the IdP (upstream unreachable, bad credentials, audience // refused). The auth layer returns the HTTP status it wants to @@ -476,10 +495,48 @@ func (p *TokenExchange) OnRequest(ctx context.Context, pctx *pipeline.Context) p return pipeline.DenyStatus(result.DenyStatus, code, result.DenyReason) case auth.ActionReplaceToken: pctx.Headers.Set("Authorization", "Bearer "+result.Token) + reason := "token_replaced" + if result.CacheHit { + reason = "cache_hit" + } + pctx.Record(pipeline.Invocation{ + Action: pipeline.ActionModify, + Reason: reason, + RouteMatched: true, + RouteHost: host, + TargetAudience: result.TargetAudience, + RequestedScopes: splitScopes(result.RequestedScopes), + CacheHit: result.CacheHit, + }) + default: + // ActionAllow / unroutable host / default-policy=passthrough all + // land here. Reason discriminates explicit-route-passthrough from + // no-route-match-default-policy; both render as "skip" in the + // 5-value vocab. + reason := "no_matching_route" + if result.RouteMatched { + reason = "route_passthrough" + } + pctx.Record(pipeline.Invocation{ + Action: pipeline.ActionSkip, + Reason: reason, + RouteMatched: result.RouteMatched, + RouteHost: host, + }) } return pipeline.Action{Type: pipeline.Continue} } +// splitScopes turns a space-separated scope string into []string. Returns +// nil for the empty string so the JSON omitempty tag drops the field +// entirely rather than emitting "[]". +func splitScopes(s string) []string { + if s == "" { + return nil + } + return strings.Fields(s) +} + func (p *TokenExchange) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action { return pipeline.Action{Type: pipeline.Continue} } diff --git a/authbridge/authlib/sessionapi/server_test.go b/authbridge/authlib/sessionapi/server_test.go index 3cdb6f0e2..e7ac96d27 100644 --- a/authbridge/authlib/sessionapi/server_test.go +++ b/authbridge/authlib/sessionapi/server_test.go @@ -433,3 +433,120 @@ func scanUntilPrefix(t *testing.T, sc *bufio.Scanner, prefix string, d time.Dura } return "" } + +// TestHandleGet_SerializesInvocations verifies the wire shape of +// SessionEvent.Invocations on /v1/sessions/{id}. A downstream consumer +// (abctl, curl pipes, scripts) must be able to decode the structured +// Inbound / Outbound slices without a side channel — this locks the +// schema, including the 5-value Action vocabulary. +func TestHandleGet_SerializesInvocations(t *testing.T) { + ts, store := newTestServer(t) + store.Append("s-inv", pipeline.SessionEvent{ + Direction: pipeline.Inbound, + Phase: pipeline.SessionDenied, + Invocations: &pipeline.Invocations{ + Inbound: []pipeline.Invocation{{ + Plugin: "jwt-validation", + Action: pipeline.ActionDeny, + Reason: "jwt_failed", + ExpectedIssuer: "http://issuer.example", + ExpectedAudience: "agent-aud", + }}, + }, + }) + + resp, err := http.Get(ts.URL + "/v1/sessions/s-inv") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != 200 { + t.Fatalf("status = %d", resp.StatusCode) + } + body, _ := io.ReadAll(resp.Body) + + // Phase renders as string + if !strings.Contains(string(body), `"phase":"denied"`) { + t.Errorf("missing denied phase in wire: %s", body) + } + // Invocations sub-object present; action rendered as the + // 5-value string, not the old "decision":"deny" shape. + if !strings.Contains(string(body), `"invocations":`) { + t.Errorf("missing invocations field in wire: %s", body) + } + if !strings.Contains(string(body), `"action":"deny"`) { + t.Errorf("missing action=deny in wire: %s", body) + } + + // Structural decode — consumer can unmarshal straight into the + // canonical types. This is the contract abctl relies on. + var view pipeline.SessionView + if err := json.Unmarshal(body, &view); err != nil { + t.Fatalf("SessionView unmarshal: %v", err) + } + if len(view.Events) != 1 { + t.Fatalf("expected 1 event, got %d", len(view.Events)) + } + got := view.Events[0] + if got.Phase != pipeline.SessionDenied { + t.Errorf("Phase = %v, want SessionDenied", got.Phase) + } + if got.Invocations == nil || len(got.Invocations.Inbound) != 1 { + t.Fatalf("Invocations not decoded: %+v", got.Invocations) + } + inv := got.Invocations.Inbound[0] + if inv.Action != pipeline.ActionDeny { + t.Errorf("Action = %q, want deny", inv.Action) + } + if inv.Reason != "jwt_failed" { + t.Errorf("Reason = %q, want jwt_failed", inv.Reason) + } +} + +// TestHandleGet_SerializesPluginsMap verifies the escape-hatch Plugins +// field round-trips as keyed json.RawMessage — abctl consumes each +// plugin's payload by key without needing to know the plugin's schema +// at compile time. +func TestHandleGet_SerializesPluginsMap(t *testing.T) { + ts, store := newTestServer(t) + store.Append("s-plug", pipeline.SessionEvent{ + Direction: pipeline.Outbound, + Phase: pipeline.SessionResponse, + Plugins: map[string]json.RawMessage{ + "rate-limiter": json.RawMessage(`{"allowed":true,"tokensLeft":42}`), + }, + }) + + resp, err := http.Get(ts.URL + "/v1/sessions/s-plug") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + + if !strings.Contains(string(body), `"plugins":{"rate-limiter":`) { + t.Errorf("plugins field missing or reshaped: %s", body) + } + + var view pipeline.SessionView + if err := json.Unmarshal(body, &view); err != nil { + t.Fatalf("unmarshal: %v", err) + } + raw, ok := view.Events[0].Plugins["rate-limiter"] + if !ok { + t.Fatalf("rate-limiter key missing: %+v", view.Events[0].Plugins) + } + // Round-trip the per-plugin payload to a caller-defined type — the + // exact pattern abctl will use to render plugin events it knows + // about, while leaving unknown plugins as raw JSON. + var payload struct { + Allowed bool `json:"allowed"` + TokensLeft int `json:"tokensLeft"` + } + if err := json.Unmarshal(raw, &payload); err != nil { + t.Fatalf("plugin payload decode: %v", err) + } + if !payload.Allowed || payload.TokensLeft != 42 { + t.Errorf("payload drift: %+v", payload) + } +} diff --git a/authbridge/cmd/abctl/tui/app.go b/authbridge/cmd/abctl/tui/app.go index ef4028b4b..9d8a46474 100644 --- a/authbridge/cmd/abctl/tui/app.go +++ b/authbridge/cmd/abctl/tui/app.go @@ -123,6 +123,12 @@ type model struct { detailPlugin *apiclient.PipelinePlugin filterInput textinput.Model + // visibleRows holds the invocationRow spec for each rendered row in + // eventsTbl. Populated by rebuildEventsTable so selectedEvent can + // return the (event, invocation) tuple the cursor is on without + // re-walking the cache. Reset on every rebuild. + visibleRows []invocationRow + // pipeline is the fetched plugin composition. nil until the initial // GetPipeline response arrives; the pipeline pane shows "(loading…)" // until then. diff --git a/authbridge/cmd/abctl/tui/events_pane.go b/authbridge/cmd/abctl/tui/events_pane.go index 4a56e17c6..42ab7f443 100644 --- a/authbridge/cmd/abctl/tui/events_pane.go +++ b/authbridge/cmd/abctl/tui/events_pane.go @@ -2,6 +2,7 @@ package tui import ( "fmt" + "strconv" "strings" "github.com/charmbracelet/bubbles/table" @@ -16,10 +17,12 @@ import ( func newEventsTable() table.Model { t := table.New( table.WithColumns([]table.Column{ + {Title: "#", Width: 4}, {Title: "TIME", Width: 12}, {Title: "DIR", Width: 4}, {Title: "PHASE", Width: 6}, - {Title: "PROTO", Width: 5}, + {Title: "ACTION", Width: 8}, + {Title: "PLUGIN", Width: 18}, {Title: "METHOD", Width: 22}, {Title: "STATUS", Width: 7}, {Title: "DURATION", Width: 10}, @@ -55,34 +58,79 @@ func (m *model) rebuildEventsTable() { prevRow := m.eventsTbl.Cursor() wasAtEnd := prevRow >= len(m.eventsTbl.Rows())-1 - // Compute request↔response pairs up-front so response rows can render - // a visual connector back to their request. - pairs := pairRequestsAndResponses(events) + // Flatten (event, invocation) into row specs up-front so pair-linking + // and filtering can run against the flat row list. Events without + // invocations fall back to a single pseudo-row (unusual — the listener + // only records events that have at least one Invocation or A2A/MCP/ + // Inference extension, but parser-only events can still land here if + // the parser populated its extension without emitting an Invocation). + rowSpecs := flattenInvocations(events) - rows := make([]table.Row, 0, len(events)) - for i, e := range events { - if m.filter != "" && !matchEvent(e, m.filter) { + // Pair request/response rows by (direction, plugin) so each plugin's + // contribution on the request side connects to its contribution on the + // response side, independent of other plugins in the same pipeline. + pairs := pairInvocationRows(rowSpecs) + + // Event-level pair IDs for the # column. Assigns each event a small + // integer; events that match as a (request, response) pair share one + // integer so the operator can scan the column for the repeated + // number. Derived from the same row-level pair map that drives the + // └resp glyph, so the two visual cues stay consistent. + eventIDs := computeEventPairIDs(rowSpecs, pairs) + + rows := make([]table.Row, 0, len(rowSpecs)) + m.visibleRows = m.visibleRows[:0] + var lastEvent *pipeline.SessionEvent // most-recent event already rendered (post-filter) + for i, rs := range rowSpecs { + if m.filter != "" && !matchInvocationRow(rs, m.filter) { continue } - phase := shortPhase(e.Phase) - if e.Phase == pipeline.SessionResponse { - if _, paired := pairs[i]; paired { - // └ prefix visually connects the response to its request - // in the row above (or earlier, if filtered). - phase = "└" + phase + // A "continuation" row is one whose event is the same as the + // previous RENDERED row's event (filtering-aware). We blank the + // event-level columns (#, TIME, DIR, PHASE, STATUS, DURATION, + // TOKENS, HOST) on continuation rows so an event's multi-plugin + // group reads as one visual block — only PLUGIN and ACTION vary. + // METHOD stays populated since a multi-plugin row set can still + // show per-plugin method context (e.g. a2a-parser observes + // message/stream while jwt-validation has no method at all). + continuation := lastEvent == rs.event + + var idCell, timeCell, dirCell, phaseCell, statusC, durCell, tokC, hostC string + if !continuation { + if id, ok := eventIDs[rs.event]; ok { + idCell = strconv.Itoa(id) } + timeCell = rs.event.At.Format("15:04:05.00") + dirCell = shortDirection(rs.event.Direction) + phaseCell = shortPhase(rs.event.Phase) + if rs.event.Phase == pipeline.SessionResponse { + if _, paired := pairs[i]; paired { + // └ prefix visually connects the response row to its + // request row in the same (direction, plugin) pair. + phaseCell = "└" + phaseCell + } + } + statusC = statusCell(*rs.event) + durCell = durationCell(*rs.event) + tokC = tokensCell(*rs.event) + hostC = truncStr(rs.event.Host, 20) } + rows = append(rows, table.Row{ - e.At.Format("15:04:05.00"), - shortDirection(e.Direction), - phase, - shortProto(e), - eventMethod(e), - statusCell(e), - durationCell(e), - tokensCell(e), - truncStr(e.Host, 20), + idCell, + timeCell, + dirCell, + phaseCell, + rs.actionCell(), + truncStr(rs.pluginCell(), 18), + eventMethod(*rs.event), + statusC, + durCell, + tokC, + hostC, }) + m.visibleRows = append(m.visibleRows, rs) + lastEvent = rs.event } m.eventsTbl.SetRows(rows) @@ -95,26 +143,78 @@ func (m *model) rebuildEventsTable() { } } -// selectedEvent returns the event at the cursor row, or nil. +// selectedEvent returns the event at the cursor row, or nil. The cursor +// points into m.visibleRows (the flattened row list), and each row carries +// a reference to its source event. func (m *model) selectedEvent() *pipeline.SessionEvent { - rows := m.eventsTbl.Rows() - if len(rows) == 0 { + if len(m.visibleRows) == 0 { return nil } cur := m.eventsTbl.Cursor() - // Re-walk the cache to find the cur'th filtered event. - events := m.events[m.selectedSess] - idx := 0 + if cur < 0 || cur >= len(m.visibleRows) { + return nil + } + return m.visibleRows[cur].event +} + +// invocationRow is one table row — the cartesian product of SessionEvent +// × Invocation. An event with N plugin invocations produces N rows; an +// event with no invocations produces one row with an empty invocation. +// Rendering and filtering both work off this flat list. +type invocationRow struct { + event *pipeline.SessionEvent + // inv may be nil when the event has no Invocation records. The + // pseudo-row still renders so the event is reachable in the table. + inv *pipeline.Invocation + // direction is the Invocations.{Inbound,Outbound} this row came + // from, disambiguating when a single event somehow carries both + // (doesn't happen today but cheap to be explicit). + direction pipeline.Direction +} + +func (r invocationRow) actionCell() string { + if r.inv == nil { + return "—" + } + return string(r.inv.Action) +} + +func (r invocationRow) pluginCell() string { + if r.inv == nil { + return "—" + } + return r.inv.Plugin +} + +// flattenInvocations walks the event slice in order and, for each event, +// emits one invocationRow per Invocation it carries (Inbound then +// Outbound). Events with no Invocations fall back to a single pseudo-row +// so parser-only events (a SessionEvent carrying just MCP or A2A with no +// matching Invocation) remain reachable. +func flattenInvocations(events []pipeline.SessionEvent) []invocationRow { + out := make([]invocationRow, 0, len(events)) for i := range events { - if m.filter != "" && !matchEvent(events[i], m.filter) { + e := &events[i] + if e.Invocations == nil || (len(e.Invocations.Inbound) == 0 && len(e.Invocations.Outbound) == 0) { + out = append(out, invocationRow{event: e, direction: e.Direction}) continue } - if idx == cur { - return &events[i] + for j := range e.Invocations.Inbound { + out = append(out, invocationRow{ + event: e, + inv: &e.Invocations.Inbound[j], + direction: pipeline.Inbound, + }) + } + for j := range e.Invocations.Outbound { + out = append(out, invocationRow{ + event: e, + inv: &e.Invocations.Outbound[j], + direction: pipeline.Outbound, + }) } - idx++ } - return nil + return out } func shortDirection(d pipeline.Direction) string { @@ -125,31 +225,20 @@ func shortDirection(d pipeline.Direction) string { } func shortPhase(p pipeline.SessionPhase) string { - if p == pipeline.SessionRequest { + switch p { + case pipeline.SessionRequest: return "req" + case pipeline.SessionResponse: + return "resp" + case pipeline.SessionDenied: + return "deny" } - return "resp" + return "?" } -// shortProto classifies an event by which extension carries meaningful -// metadata. Inference wins over MCP when both are present: mcp-parser -// greedily accepts any JSON as JSON-RPC (often with an empty method on -// LLM request bodies) and sets MCPExtension, so an LLM call shows up -// with both MCP{method:""} and Inference{model:...}. Picking inference -// first surfaces the more specific truth. -func shortProto(e pipeline.SessionEvent) string { - switch { - case e.A2A != nil: - return "a2a" - case e.Inference != nil: - return "inf" - case e.MCP != nil && e.MCP.Method != "": - return "mcp" - case e.MCP != nil: - return "—" // empty-method MCP = mcp-parser false-positive - } - return "—" -} +// (authCell and responsiblePlugin are gone — their roles moved onto +// invocationRow's actionCell/pluginCell because each row now corresponds +// to exactly one plugin's invocation rather than a whole event.) func eventMethod(e pipeline.SessionEvent) string { switch { @@ -202,11 +291,153 @@ func truncStr(s string, n int) string { return s[:n-1] + "…" } -// matchEvent does a case-insensitive substring match across every string -// field the operator might reasonably search for. -func matchEvent(e pipeline.SessionEvent, q string) bool { +// computeEventPairIDs assigns a small integer to every SessionEvent, +// sharing one integer across a (request, response) pair and minting a new +// one for unpaired events. The pairing decision is delegated to the row- +// level pair map from pairInvocationRows — if any plugin row on event A +// pairs with a plugin row on event B, then A and B pair at the event +// level too. This keeps the # column's IDs consistent with the `└resp` +// glyph the operator already sees (both derive from the same plugin- +// level (direction, plugin) match). +// +// Deriving from pairInvocationRows rather than recomputing by +// direction+host+method avoids a class of bugs with "featureless" +// requests (no parser matched, so method is empty): multiple concurrent +// passthrough calls to the same host all share the same host+method +// key and a naive matcher claims the wrong response. +// +// IDs are keyed by event pointer so render loops can look up a row's ID +// without knowing the slice index. IDs start at 1 and increment in +// first-seen row order so adjacent pairs get adjacent integers. +func computeEventPairIDs(rowSpecs []invocationRow, pairs map[int]int) map[*pipeline.SessionEvent]int { + // Derive event-level pairs from row-level pairs. pairs is symmetric + // (pairs[i]=j and pairs[j]=i), so iterating either entry sets the + // map symmetrically. Last-write-wins when a single event pairs + // through multiple plugins, but in practice all plugin rows on one + // request event point at the same response event. + eventPair := make(map[*pipeline.SessionEvent]*pipeline.SessionEvent) + for i, j := range pairs { + ei, ej := rowSpecs[i].event, rowSpecs[j].event + if ei != ej { + eventPair[ei] = ej + } + } + + // Event-level fallback for pairs the row-level matcher can't see. + // When a response event has no plugin invocations (e.g. a bypass + // path like /.well-known/agent.json — jwt-validation skipped on the + // request and no parser matched on the response), its pseudo-row + // has no (direction, plugin) key and pairInvocationRows leaves it + // unpaired. Scan the ordered event list and link each unpaired + // response to the closest preceding unpaired request with matching + // direction + host so the # column still reflects the pairing. + // + // Closest-preceding match is sufficient for bypass traffic where + // the response event immediately follows its request in the slice. + // Multiple concurrent bypass requests on the same host could + // theoretically cross-pair, but that's a near-simultaneous + // duplicate-path pattern we don't expect in real traffic. + orderedEvents := orderedUniqueEvents(rowSpecs) + for i, e := range orderedEvents { + if e.Phase != pipeline.SessionResponse { + continue + } + if _, paired := eventPair[e]; paired { + continue + } + for j := i - 1; j >= 0; j-- { + prev := orderedEvents[j] + if prev.Phase != pipeline.SessionRequest { + continue + } + if _, already := eventPair[prev]; already { + continue + } + if prev.Direction != e.Direction || prev.Host != e.Host { + continue + } + eventPair[e] = prev + eventPair[prev] = e + break + } + } + + ids := make(map[*pipeline.SessionEvent]int) + seen := make(map[*pipeline.SessionEvent]bool) + next := 0 + for _, rs := range rowSpecs { + e := rs.event + if seen[e] { + continue + } + seen[e] = true + // If this event's paired partner has already been assigned an + // ID (partner appeared earlier in row order), reuse it. + if partner := eventPair[e]; partner != nil { + if pid, ok := ids[partner]; ok { + ids[e] = pid + continue + } + } + next++ + ids[e] = next + } + return ids +} + +// orderedUniqueEvents returns distinct event pointers in the order they +// first appear in rowSpecs. Used by computeEventPairIDs' event-level +// fallback to walk events sequentially while looking backward for +// unpaired request counterparts. +func orderedUniqueEvents(rowSpecs []invocationRow) []*pipeline.SessionEvent { + seen := make(map[*pipeline.SessionEvent]bool, len(rowSpecs)) + out := make([]*pipeline.SessionEvent, 0, len(rowSpecs)) + for _, rs := range rowSpecs { + if seen[rs.event] { + continue + } + seen[rs.event] = true + out = append(out, rs.event) + } + return out +} + +// matchInvocationRow does a case-insensitive substring match across every +// string field the operator might reasonably search for — the invocation's +// own fields plus the containing event's protocol extensions. Two prefix +// shortcuts: +// +// - `deny` alone matches SessionDenied events and any invocation +// whose Action == ActionDeny — the one-word "show me failures" +// filter. +// - `plugin:` matches rows whose escape-hatch Plugins map on +// the parent event has as a key. +func matchInvocationRow(r invocationRow, q string) bool { q = strings.ToLower(q) - hay := []string{e.Host, e.TargetAudience, shortProto(e), eventMethod(e)} + + if q == "deny" { + if r.event.Phase == pipeline.SessionDenied { + return true + } + if r.inv != nil && r.inv.Action == pipeline.ActionDeny { + return true + } + return false + } + + if after, ok := strings.CutPrefix(q, "plugin:"); ok { + _, present := r.event.Plugins[after] + return present + } + + e := r.event + hay := []string{e.Host, e.TargetAudience, eventMethod(*e)} + if r.inv != nil { + hay = append(hay, + r.inv.Plugin, string(r.inv.Action), r.inv.Reason, r.inv.Path, + r.inv.ExpectedIssuer, r.inv.ExpectedAudience, r.inv.TokenSubject, + r.inv.RouteHost, r.inv.TargetAudience) + } if e.Identity != nil { hay = append(hay, e.Identity.Subject, e.Identity.ClientID) } @@ -230,40 +461,54 @@ func matchEvent(e pipeline.SessionEvent, q string) bool { return false } -// pairRequestsAndResponses returns a map whose keys are the indexes of -// events that participate in a request↔response pair. It walks events in -// order: each SessionRequest is paired with the NEXT SessionResponse that -// matches on direction + protocol + method, within the same session. +// pairInvocationRows pairs request-phase rows with their response-phase +// counterparts by (direction, plugin). Each plugin's contribution on the +// request side connects to its own contribution on the response side, +// independent of other plugins in the same pipeline — so a jwt-validation +// request row pairs with a jwt-validation response row even when several +// other plugins fired on the same event. // -// Sequential pairing is sufficient for AuthBridge's current traffic -// patterns (no overlapping same-method outbound calls per turn). Future -// work: key pairs by MCP.RPCID / A2A.RPCID when available for stricter -// correlation. -func pairRequestsAndResponses(events []pipeline.SessionEvent) map[int]int { +// Sequential pairing is good enough for current traffic: each request +// row is paired with the NEXT response row that shares (direction, plugin) +// and hasn't been claimed. +func pairInvocationRows(rows []invocationRow) map[int]int { pairs := make(map[int]int) - for i := range events { - req := events[i] - if req.Phase != pipeline.SessionRequest { + // Pair key includes plugin + direction + method (from whichever + // parser extension is populated). Without the method component, + // a fire-and-forget request like MCP's notifications/initialized + // would greedily claim the NEXT mcp-parser response — typically + // the response to tools/list — and orphan the actual tools/list + // request from its own response. Method discrimination makes the + // match specific: mcp-parser/out/tools/list only pairs with + // mcp-parser/out/tools/list. Auth plugins have no method; empty + // methods still pair with empty methods (same key), preserving + // pair behaviour for token-exchange and jwt-validation rows. + key := func(r invocationRow) (string, pipeline.Direction, bool) { + if r.inv == nil { + return "", r.direction, false + } + return r.inv.Plugin + "|" + eventMethod(*r.event), r.direction, true + } + for i := range rows { + if rows[i].event.Phase != pipeline.SessionRequest { continue } if _, already := pairs[i]; already { continue } - for j := i + 1; j < len(events); j++ { - resp := events[j] - if resp.Phase != pipeline.SessionResponse { + k, dir, ok := key(rows[i]) + if !ok { + continue + } + for j := i + 1; j < len(rows); j++ { + if rows[j].event.Phase != pipeline.SessionResponse { continue } if _, taken := pairs[j]; taken { continue } - if resp.Direction != req.Direction { - continue - } - if shortProto(resp) != shortProto(req) { - continue - } - if eventMethod(resp) != eventMethod(req) { + rk, rdir, rok := key(rows[j]) + if !rok || rk != k || rdir != dir { continue } pairs[i] = j diff --git a/authbridge/cmd/abctl/tui/events_pane_test.go b/authbridge/cmd/abctl/tui/events_pane_test.go new file mode 100644 index 000000000..06e053913 --- /dev/null +++ b/authbridge/cmd/abctl/tui/events_pane_test.go @@ -0,0 +1,343 @@ +package tui + +import ( + "encoding/json" + "testing" + "time" + + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" +) + +// TestShortPhase_Denied locks the abctl rendering string for the +// denied phase — changing this silently would ripple into the events +// table and teatest snapshots. +func TestShortPhase_Denied(t *testing.T) { + if got := shortPhase(pipeline.SessionDenied); got != "deny" { + t.Errorf("shortPhase(SessionDenied) = %q, want deny", got) + } +} + +// TestInvocationRow_Cells exercises the ACTION and PLUGIN column +// renderers for each shape a row can take: an Invocation with an action, +// multiple invocations (the row is per-invocation, each carries only +// its own plugin), and the pseudo-row fallback when an event has no +// Invocations at all. +func TestInvocationRow_Cells(t *testing.T) { + evWithInv := &pipeline.SessionEvent{ + Invocations: &pipeline.Invocations{ + Inbound: []pipeline.Invocation{ + {Plugin: "jwt-validation", Action: pipeline.ActionAllow}, + {Plugin: "a2a-parser", Action: pipeline.ActionObserve}, + }, + }, + } + cases := []struct { + name string + row invocationRow + wantAction string + wantPlugin string + }{ + { + name: "empty pseudo-row", + row: invocationRow{event: &pipeline.SessionEvent{}}, + wantAction: "—", + wantPlugin: "—", + }, + { + name: "inbound allow", + row: invocationRow{ + event: evWithInv, + inv: &evWithInv.Invocations.Inbound[0], + direction: pipeline.Inbound, + }, + wantAction: "allow", + wantPlugin: "jwt-validation", + }, + { + name: "inbound observe (parser)", + row: invocationRow{ + event: evWithInv, + inv: &evWithInv.Invocations.Inbound[1], + direction: pipeline.Inbound, + }, + wantAction: "observe", + wantPlugin: "a2a-parser", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := tc.row.actionCell(); got != tc.wantAction { + t.Errorf("actionCell = %q, want %q", got, tc.wantAction) + } + if got := tc.row.pluginCell(); got != tc.wantPlugin { + t.Errorf("pluginCell = %q, want %q", got, tc.wantPlugin) + } + }) + } +} + +// TestFlattenInvocations covers the core expansion: an event with N +// invocations should produce N rows; an event with zero invocations +// should still produce one pseudo-row so the event stays reachable. +func TestFlattenInvocations(t *testing.T) { + events := []pipeline.SessionEvent{ + // 2 inbound invocations → 2 rows + { + Direction: pipeline.Inbound, + Invocations: &pipeline.Invocations{ + Inbound: []pipeline.Invocation{ + {Plugin: "jwt-validation", Action: pipeline.ActionAllow}, + {Plugin: "a2a-parser", Action: pipeline.ActionObserve}, + }, + }, + }, + // 1 outbound invocation → 1 row + { + Direction: pipeline.Outbound, + Invocations: &pipeline.Invocations{ + Outbound: []pipeline.Invocation{ + {Plugin: "token-exchange", Action: pipeline.ActionSkip}, + }, + }, + }, + // no invocations → 1 pseudo-row + {Direction: pipeline.Inbound}, + } + got := flattenInvocations(events) + if len(got) != 4 { + t.Fatalf("flattenInvocations returned %d rows, want 4", len(got)) + } + if got[0].inv == nil || got[0].inv.Plugin != "jwt-validation" { + t.Errorf("row 0 = %+v, want jwt-validation", got[0]) + } + if got[1].inv == nil || got[1].inv.Plugin != "a2a-parser" { + t.Errorf("row 1 = %+v, want a2a-parser", got[1]) + } + if got[2].inv == nil || got[2].inv.Plugin != "token-exchange" { + t.Errorf("row 2 = %+v, want token-exchange", got[2]) + } + if got[3].inv != nil { + t.Errorf("row 3 should be pseudo-row with nil inv, got %+v", got[3]) + } +} + +// TestPairInvocationRows verifies that each plugin's request row pairs +// with its own response row independently. A pipeline with +// jwt-validation + a2a-parser on both request and response phases yields +// 4 rows (2 req + 2 resp), and pairing should connect them in-plugin: +// jwt-validation-req ↔ jwt-validation-resp; a2a-parser-req ↔ +// a2a-parser-resp. +func TestPairInvocationRows(t *testing.T) { + inv := func(plugin string, action pipeline.InvocationAction) *pipeline.Invocation { + return &pipeline.Invocation{Plugin: plugin, Action: action} + } + reqEv := &pipeline.SessionEvent{Direction: pipeline.Inbound, Phase: pipeline.SessionRequest} + respEv := &pipeline.SessionEvent{Direction: pipeline.Inbound, Phase: pipeline.SessionResponse} + rows := []invocationRow{ + {event: reqEv, inv: inv("jwt-validation", pipeline.ActionAllow), direction: pipeline.Inbound}, + {event: reqEv, inv: inv("a2a-parser", pipeline.ActionObserve), direction: pipeline.Inbound}, + {event: respEv, inv: inv("jwt-validation", pipeline.ActionAllow), direction: pipeline.Inbound}, + {event: respEv, inv: inv("a2a-parser", pipeline.ActionObserve), direction: pipeline.Inbound}, + } + pairs := pairInvocationRows(rows) + if pairs[0] != 2 || pairs[2] != 0 { + t.Errorf("expected jwt-validation pair 0↔2, got %v", pairs) + } + if pairs[1] != 3 || pairs[3] != 1 { + t.Errorf("expected a2a-parser pair 1↔3, got %v", pairs) + } +} + +// TestMatchInvocationRow_DenyShortcut verifies that typing "deny" in the +// filter box surfaces both the SessionDenied phase AND any invocation +// whose Action is ActionDeny (jwt-validation or token-exchange +// denials). +func TestMatchInvocationRow_DenyShortcut(t *testing.T) { + denied := invocationRow{ + event: &pipeline.SessionEvent{Phase: pipeline.SessionDenied}, + } + if !matchInvocationRow(denied, "deny") { + t.Error("SessionDenied event should match the `deny` shortcut") + } + + inboundDeny := invocationRow{ + event: &pipeline.SessionEvent{Phase: pipeline.SessionRequest}, + inv: &pipeline.Invocation{Action: pipeline.ActionDeny}, + } + if !matchInvocationRow(inboundDeny, "deny") { + t.Error("inbound-deny invocation should match the `deny` shortcut") + } + + clean := invocationRow{ + event: &pipeline.SessionEvent{Phase: pipeline.SessionRequest}, + inv: &pipeline.Invocation{Action: pipeline.ActionAllow}, + } + if matchInvocationRow(clean, "deny") { + t.Error("allow invocation should NOT match the `deny` shortcut") + } +} + +// TestMatchInvocationRow_PluginSubstring verifies that filtering by plugin +// name substring-matches against the Invocation.Plugin field so operators +// can isolate one plugin's rows. +func TestMatchInvocationRow_PluginSubstring(t *testing.T) { + row := invocationRow{ + event: &pipeline.SessionEvent{Phase: pipeline.SessionRequest}, + inv: &pipeline.Invocation{Plugin: "jwt-validation", Action: pipeline.ActionSkip, Reason: "path_bypass", Path: "/healthz"}, + } + if !matchInvocationRow(row, "jwt-validation") { + t.Error("filter jwt-validation should match") + } + if !matchInvocationRow(row, "path_bypass") { + t.Error("filter by reason should match") + } + if !matchInvocationRow(row, "/healthz") { + t.Error("filter by path should match") + } + if matchInvocationRow(row, "token-exchange") { + t.Error("filter token-exchange should NOT match a jwt-validation row") + } +} + +// TestMatchInvocationRow_PluginPrefix tests the `plugin:` escape- +// hatch filter — matches when the event's Plugins map contains . +func TestMatchInvocationRow_PluginPrefix(t *testing.T) { + row := invocationRow{ + event: &pipeline.SessionEvent{ + Plugins: map[string]json.RawMessage{ + "rate-limiter": json.RawMessage(`{"allowed":true}`), + }, + }, + } + if !matchInvocationRow(row, "plugin:rate-limiter") { + t.Error("expected match on plugin:rate-limiter") + } + if matchInvocationRow(row, "plugin:nonexistent") { + t.Error("expected no match for a plugin not in the map") + } +} + +// TestComputeEventPairIDs_BypassResponseWithEmptyInvocations locks the +// event-level fallback pairing: when a response event has no plugin +// invocations at all (e.g. jwt-validation bypass response), it should +// still pair with its preceding request event via direction+host match +// so the # column shows the same ID on both rows. +func TestComputeEventPairIDs_BypassResponseWithEmptyInvocations(t *testing.T) { + events := []pipeline.SessionEvent{ + // Event 0: bypass req — jwt-validation skip invocation + { + Direction: pipeline.Inbound, + Phase: pipeline.SessionRequest, + Invocations: &pipeline.Invocations{Inbound: []pipeline.Invocation{{ + Plugin: "jwt-validation", + Phase: pipeline.InvocationPhaseRequest, + Action: pipeline.ActionSkip, + }}}, + }, + // Event 1: bypass resp — no invocations (response-phase filter returns empty) + {Direction: pipeline.Inbound, Phase: pipeline.SessionResponse, StatusCode: 200}, + // Event 2: bypass req (different bypass path, same direction+host="") + { + Direction: pipeline.Inbound, + Phase: pipeline.SessionRequest, + Invocations: &pipeline.Invocations{Inbound: []pipeline.Invocation{{ + Plugin: "jwt-validation", + Phase: pipeline.InvocationPhaseRequest, + Action: pipeline.ActionSkip, + }}}, + }, + // Event 3: bypass resp + {Direction: pipeline.Inbound, Phase: pipeline.SessionResponse, StatusCode: 200}, + } + + rows := flattenInvocations(events) + pairs := pairInvocationRows(rows) + ids := computeEventPairIDs(rows, pairs) + + id0, id1 := ids[&events[0]], ids[&events[1]] + id2, id3 := ids[&events[2]], ids[&events[3]] + + if id0 != id1 { + t.Errorf("bypass req/resp #1: got ids (%d,%d), want equal", id0, id1) + } + if id2 != id3 { + t.Errorf("bypass req/resp #2: got ids (%d,%d), want equal", id2, id3) + } + if id0 == id2 { + t.Errorf("different bypass pairs should have different ids, both got %d", id0) + } +} + +// TestPairInvocationRows_MethodDiscrimination locks the method-aware +// pairing. Fire-and-forget MCP methods (notifications/initialized) have +// no response; a subsequent tools/list req+resp pair must not be +// disrupted by the notification's mcp-parser row greedily claiming the +// tools/list response row. +func TestPairInvocationRows_MethodDiscrimination(t *testing.T) { + mk := func(phase pipeline.SessionPhase, method string) pipeline.SessionEvent { + return pipeline.SessionEvent{ + Direction: pipeline.Outbound, + Phase: phase, + MCP: &pipeline.MCPExtension{Method: method}, + Invocations: &pipeline.Invocations{Outbound: []pipeline.Invocation{{ + Plugin: "mcp-parser", + Phase: invocationPhaseFor(phase), + Action: pipeline.ActionObserve, + }}}, + } + } + events := []pipeline.SessionEvent{ + mk(pipeline.SessionRequest, "notifications/initialized"), // no resp (fire and forget) + mk(pipeline.SessionRequest, "tools/list"), + mk(pipeline.SessionResponse, "tools/list"), + } + rows := flattenInvocations(events) + pairs := pairInvocationRows(rows) + ids := computeEventPairIDs(rows, pairs) + + if ids[&events[1]] != ids[&events[2]] { + t.Errorf("tools/list req and resp must share ID, got %d vs %d", + ids[&events[1]], ids[&events[2]]) + } + if ids[&events[0]] == ids[&events[1]] { + t.Errorf("notifications/initialized (orphan) must not share ID with tools/list, both got %d", + ids[&events[0]]) + } +} + +func invocationPhaseFor(p pipeline.SessionPhase) pipeline.InvocationPhase { + if p == pipeline.SessionResponse { + return pipeline.InvocationPhaseResponse + } + return pipeline.InvocationPhaseRequest +} + +// Build a realistic auth-only request/response pair and assert that the +// flatten → pair pipeline connects them end-to-end. Regression-protects +// the chart-default case (jwt-validation only, no parsers). +func TestFlattenPair_AuthOnlyEndToEnd(t *testing.T) { + now := time.Date(2026, 5, 8, 14, 22, 5, 0, time.UTC) + invs := &pipeline.Invocations{Inbound: []pipeline.Invocation{{Plugin: "jwt-validation", Action: pipeline.ActionAllow}}} + events := []pipeline.SessionEvent{ + {At: now, Direction: pipeline.Inbound, Phase: pipeline.SessionRequest, Invocations: invs, Host: "weather-agent"}, + {At: now.Add(12 * time.Millisecond), Direction: pipeline.Inbound, Phase: pipeline.SessionResponse, Invocations: invs, Host: "weather-agent", StatusCode: 200, Duration: 12 * time.Millisecond}, + } + + rows := flattenInvocations(events) + if len(rows) != 2 { + t.Fatalf("expected 2 rows, got %d", len(rows)) + } + pairs := pairInvocationRows(rows) + if pairs[0] != 1 || pairs[1] != 0 { + t.Errorf("expected auth-only req/resp to pair: got %v", pairs) + } + if got := rows[0].actionCell(); got != "allow" { + t.Errorf("req actionCell = %q, want allow", got) + } + if got := rows[1].pluginCell(); got != "jwt-validation" { + t.Errorf("resp pluginCell = %q, want jwt-validation", got) + } + if got := statusCell(*rows[1].event); got != "200" { + t.Errorf("statusCell = %q, want 200", got) + } +} diff --git a/authbridge/cmd/authbridge/listener/extproc/server.go b/authbridge/cmd/authbridge/listener/extproc/server.go index 08a50c623..f343389f7 100644 --- a/authbridge/cmd/authbridge/listener/extproc/server.go +++ b/authbridge/cmd/authbridge/listener/extproc/server.go @@ -130,6 +130,7 @@ func (s *Server) handleInbound(stream extprocv3.ExternalProcessor_ProcessServer, action := s.InboundPipeline.Run(ctx, pctx) if action.Type == pipeline.Reject { + s.recordInboundReject(pctx, action) return rejectFromAction(action), nil } @@ -149,6 +150,7 @@ func (s *Server) handleInboundBody(stream extprocv3.ExternalProcessor_ProcessSer action := s.InboundPipeline.Run(ctx, pctx) if action.Type == pipeline.Reject { + s.recordInboundReject(pctx, action) return rejectFromAction(action), nil } @@ -164,42 +166,176 @@ func (s *Server) handleInboundBody(stream extprocv3.ExternalProcessor_ProcessSer // the previous conversation's rekeyed bucket, stranding the current turn's // request events in the prior bucket and creating an orphan 1-event session // for the response. +// +// Auth-only events (no A2A parser match — e.g. a rejected request that +// never reached the parser) route to DefaultSessionID. This is where +// operators will look for unauthorized-access events in abctl. func inboundSessionID(pctx *pipeline.Context) string { - if sid := pctx.Extensions.A2A.SessionID; sid != "" { - return sid + if pctx.Extensions.A2A != nil && pctx.Extensions.A2A.SessionID != "" { + return pctx.Extensions.A2A.SessionID } return session.DefaultSessionID } func (s *Server) recordInboundSession(pctx *pipeline.Context) { - if s.Sessions == nil || pctx.Extensions.A2A == nil { + if s.Sessions == nil { + return + } + // Widened gate (was: A2A == nil). Any of A2A / Auth / plugin-public + // Custom entries qualify. Keeps traffic with no protocol parser but + // meaningful auth state visible in the session stream. + plugins := snapshotPlugins(pctx.Extensions.Custom) + if pctx.Extensions.A2A == nil && pctx.Extensions.Invocations == nil && plugins == nil { return } sid := inboundSessionID(pctx) ev := pipeline.SessionEvent{ At: time.Now(), Direction: pipeline.Inbound, - Phase: pipeline.SessionRequest, - A2A: snapshotA2A(pctx.Extensions.A2A), + Phase: pipeline.SessionRequest, + A2A: snapshotA2A(pctx.Extensions.A2A), + Invocations: snapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), + Plugins: plugins, Identity: snapshotIdentity(pctx), Host: pctx.Host, } s.Sessions.Append(sid, ev) } +// recordInboundReject emits a SessionDenied event for requests a pipeline +// plugin rejected. Called from the Reject path BEFORE rejectFromAction +// returns, so denied requests appear in the session stream rather than +// silently vanishing (which was the pre-Auth-extension behavior — denials +// only surfaced via /stats counters, invisible to abctl). Fires only when +// at least one plugin populated Auth — otherwise we wouldn't have +// diagnostic context worth recording and would just be logging an HTTP +// status. +func (s *Server) recordInboundReject(pctx *pipeline.Context, action pipeline.Action) { + if s.Sessions == nil || pctx.Extensions.Invocations == nil { + return + } + var status int + var code, message string + if action.Violation != nil { + // Use the structured fields directly — Render() produces the HTTP + // wire payload (status, headers, JSON body) which is the wrong + // shape for a session event. We want the semantic Code + Reason. + 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: 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(inboundSessionID(pctx), 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 +// 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 +// 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. +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} +} + +// snapshotPlugins collects plugin-public observability events from +// pctx.Extensions.Custom entries whose keys end in PluginEventSuffix. +// Each matching value is json.Marshaled into the wire-form map under +// the plugin name (suffix stripped). Marshal errors downgrade to slog +// Debug and skip the entry rather than aborting recording — that keeps +// a misbehaving plugin from taking out the whole session stream. +func snapshotPlugins(custom map[string]any) map[string]json.RawMessage { + if len(custom) == 0 { + return nil + } + var out map[string]json.RawMessage + for k, v := range custom { + if !strings.HasSuffix(k, pipeline.PluginEventSuffix) { + continue + } + raw, err := json.Marshal(v) + if err != nil { + slog.Debug("session: skipping non-marshalable plugin event", + "key", k, "error", err) + continue + } + if out == nil { + out = make(map[string]json.RawMessage) + } + pluginName := strings.TrimSuffix(k, pipeline.PluginEventSuffix) + out[pluginName] = raw + } + return out +} + // recordInboundResponseSession appends a Phase:SessionResponse event for the -// inbound A2A direction. Called after RunResponse completes so the event -// carries the updated SessionID (from the response body's contextId). +// inbound direction. Called after RunResponse completes so the event carries +// the updated SessionID (from the response body's contextId, when an A2A +// parser ran) or the default bucket (when the pipeline is auth-only). +// +// Recording gate parallels the request-phase gate in recordInboundSession +// and the outbound-response gate in recordOutboundResponseSession: A2A, +// Auth, or plugin-public Custom entries all qualify. The earlier gate that +// required A2A silently dropped response events for auth-only pipelines +// (jwt-validation without any parser) — the request phase recorded, the +// response phase didn't, so operators saw one-sided conversations in abctl. func (s *Server) recordInboundResponseSession(pctx *pipeline.Context) { - if s.Sessions == nil || pctx.Extensions.A2A == nil { + if s.Sessions == nil { + return + } + plugins := snapshotPlugins(pctx.Extensions.Custom) + if pctx.Extensions.A2A == nil && pctx.Extensions.Invocations == nil && plugins == nil { return } sid := inboundSessionID(pctx) ev := pipeline.SessionEvent{ At: time.Now(), Direction: pipeline.Inbound, - Phase: pipeline.SessionResponse, - A2A: snapshotA2A(pctx.Extensions.A2A), + Phase: pipeline.SessionResponse, + A2A: snapshotA2A(pctx.Extensions.A2A), + Invocations: snapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseResponse), + Plugins: plugins, Identity: snapshotIdentity(pctx), StatusCode: pctx.StatusCode, Error: deriveError(pctx), @@ -220,12 +356,15 @@ func (s *Server) recordOutboundResponseSession(pctx *pipeline.Context) { if sid == "" { sid = session.DefaultSessionID } + 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), @@ -233,7 +372,11 @@ func (s *Server) recordOutboundResponseSession(pctx *pipeline.Context) { TargetAudience: routeAudience(pctx), Duration: durationSince(pctx.StartedAt), } - if ev.MCP != nil || ev.Inference != nil { + // Auth / Plugins alone qualify for recording; matches the widened + // gate in recordInboundSession so outbound denials and plugin-public + // observability aren't dropped just because the response carried no + // MCP/Inference payload. + if ev.MCP != nil || ev.Inference != nil || ev.Invocations != nil || plugins != nil { s.Sessions.Append(sid, ev) } } @@ -318,17 +461,20 @@ func (s *Server) recordOutboundSession(pctx *pipeline.Context) { if sid == "" { sid = session.DefaultSessionID } + 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, TargetAudience: routeAudience(pctx), } - if ev.MCP != nil || ev.Inference != nil { + if ev.MCP != nil || ev.Inference != nil || ev.Invocations != nil || plugins != nil { s.Sessions.Append(sid, ev) } } diff --git a/authbridge/cmd/authbridge/listener/extproc/server_test.go b/authbridge/cmd/authbridge/listener/extproc/server_test.go index 903f21dd3..42b08d2a9 100644 --- a/authbridge/cmd/authbridge/listener/extproc/server_test.go +++ b/authbridge/cmd/authbridge/listener/extproc/server_test.go @@ -688,15 +688,63 @@ func TestRecordInboundResponseSession(t *testing.T) { } } -func TestRecordInboundResponseSession_NoA2A(t *testing.T) { - // No A2A extension — nothing to record. +func TestRecordInboundResponseSession_EmptyPctx(t *testing.T) { + // No A2A, no Auth, no plugin-public Custom entries — nothing to + // record (parallel to the empty-request gate). Auth-only and plugin- + // only cases are covered separately below. store := session.New(5*time.Minute, 100, 0) defer store.Close() s := &Server{Sessions: store} s.recordInboundResponseSession(&pipeline.Context{}) if store.View(session.DefaultSessionID) != nil { - t.Error("no session should have been created without A2A extension") + t.Error("no session should have been created with empty pctx") + } +} + +// TestRecordInboundResponseSession_AuthOnly covers the exact scenario +// Option 2 activates for the chart default pipeline (jwt-validation only, +// no A2A parser). The request-phase gate was widened in d55524b but the +// response-phase gate kept the old A2A-only check, so auth-only response +// events were silently dropped — operators saw request rows without their +// paired response rows. Locks the fix. +func TestRecordInboundResponseSession_AuthOnly(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + s := &Server{Sessions: store} + + // Record side filters by phase, so this test passes a response-phase + // invocation. In production jwt-validation's OnResponse is a no-op, + // but the test exercises the gate: any response-phase entry is + // sufficient to record a SessionResponse event. + pctx := &pipeline.Context{ + Extensions: pipeline.Extensions{ + Invocations: &pipeline.Invocations{ + Inbound: []pipeline.Invocation{{ + Plugin: "jwt-validation", + Phase: pipeline.InvocationPhaseResponse, + Action: pipeline.ActionAllow, + Reason: "authorized", + }}, + }, + }, + StatusCode: 200, + } + s.recordInboundResponseSession(pctx) + + v := store.View(session.DefaultSessionID) + if v == nil || len(v.Events) != 1 { + t.Fatalf("expected 1 event under default, got %v", v) + } + e := v.Events[0] + if e.Direction != pipeline.Inbound || e.Phase != pipeline.SessionResponse { + t.Errorf("event fields = (%v, %v), want (Inbound, SessionResponse)", e.Direction, e.Phase) + } + if e.Invocations == nil || len(e.Invocations.Inbound) != 1 || e.Invocations.Inbound[0].Action != pipeline.ActionAllow { + t.Errorf("Invocations not attached correctly: %+v", e.Invocations) + } + if e.StatusCode != 200 { + t.Errorf("StatusCode = %d, want 200", e.StatusCode) } } @@ -1091,3 +1139,162 @@ func TestRecordOutboundResponseSession_CapturesHostAndRoute(t *testing.T) { t.Errorf("Duration = %v, want >= 25ms", resp.Duration) } } + +// TestRecordInboundSession_AuthOnly verifies the widened gate: a request +// that never reached a protocol parser (A2A is nil) but populated the +// Auth extension still gets recorded. This is the session-stream +// visibility path for auth decisions that don't carry conversation +// payload — e.g., an authorized inbound ping to a non-A2A endpoint. +func TestRecordInboundSession_AuthOnly(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + s := &Server{Sessions: store} + + pctx := &pipeline.Context{ + Extensions: pipeline.Extensions{ + Invocations: &pipeline.Invocations{ + Inbound: []pipeline.Invocation{{ + Plugin: "jwt-validation", + Phase: pipeline.InvocationPhaseRequest, + Action: pipeline.ActionAllow, + Reason: "authorized", + }}, + }, + }, + } + s.recordInboundSession(pctx) + + v := store.View(session.DefaultSessionID) + if v == nil || len(v.Events) != 1 { + t.Fatalf("expected 1 event under default session, got %v", v) + } + ev := v.Events[0] + if ev.Invocations == nil || len(ev.Invocations.Inbound) != 1 { + t.Fatalf("Invocations.Inbound not snapshotted: %+v", ev.Invocations) + } + if ev.Invocations.Inbound[0].Action != pipeline.ActionAllow { + t.Errorf("Action lost in snapshot: %+v", ev.Invocations.Inbound[0]) + } +} + +// TestRecordInboundReject_EmitsDeniedPhase verifies the new denial +// recording path: when the pipeline rejects an inbound request, a +// SessionDenied event appears with the Auth diagnostic context and the +// Violation mapped onto StatusCode + EventError. Before this, denied +// requests were invisible in /v1/sessions. +func TestRecordInboundReject_EmitsDeniedPhase(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + s := &Server{Sessions: store} + + pctx := &pipeline.Context{ + StartedAt: time.Now().Add(-10 * time.Millisecond), + Extensions: pipeline.Extensions{ + Invocations: &pipeline.Invocations{ + Inbound: []pipeline.Invocation{{ + Plugin: "jwt-validation", + Phase: pipeline.InvocationPhaseRequest, + Action: pipeline.ActionDeny, + Reason: "jwt_failed", + ExpectedIssuer: "http://issuer.example", + ExpectedAudience: "agent-aud", + }}, + }, + }, + } + action := pipeline.DenyStatus(401, "auth.unauthorized", "token validation failed") + s.recordInboundReject(pctx, action) + + v := store.View(session.DefaultSessionID) + if v == nil || len(v.Events) != 1 { + t.Fatalf("expected 1 event under default session, got %v", v) + } + ev := v.Events[0] + if ev.Phase != pipeline.SessionDenied { + t.Errorf("Phase = %v, want SessionDenied", ev.Phase) + } + if ev.StatusCode != 401 { + t.Errorf("StatusCode = %d, want 401", ev.StatusCode) + } + if ev.Error == nil || ev.Error.Code != "auth.unauthorized" { + t.Errorf("Error = %+v, want code=auth.unauthorized", ev.Error) + } + if ev.Invocations == nil || len(ev.Invocations.Inbound) != 1 || ev.Invocations.Inbound[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) + } +} + +// TestRecordInboundReject_SkipsWithoutAuth ensures the denial recording +// path is gated on Auth being populated — otherwise every plugin reject +// (including those unrelated to auth, e.g. body-size-exceeded) would +// land in the session stream with no useful context. Stats counters are +// the right place for those; session denials are for auth-class events. +func TestRecordInboundReject_SkipsWithoutAuth(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + s := &Server{Sessions: store} + + action := pipeline.DenyStatus(413, "request.too-large", "body too large") + s.recordInboundReject(&pipeline.Context{}, 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. +// Plugin-private state (Custom entries without the suffix, used by +// SetState / GetState) stays out of the session stream. +func TestSnapshotPlugins_FiltersByEventSuffix(t *testing.T) { + type rateLimiterPrivate struct { + TokenBucket int + } + type rateLimiterEvent struct { + Allowed bool `json:"allowed"` + TokensLeft int `json:"tokensLeft"` + } + custom := map[string]any{ + // Private state — stored by SetState for cross-phase continuity. + // Must NOT appear in SessionEvent.Plugins. + "rate-limiter": &rateLimiterPrivate{TokenBucket: 17}, + // Public event — stored with the "/event" suffix. Must appear, + // keyed by "rate-limiter" (suffix stripped) in the output map. + "rate-limiter" + pipeline.PluginEventSuffix: rateLimiterEvent{ + Allowed: true, TokensLeft: 42, + }, + } + out := snapshotPlugins(custom) + if _, private := out["rate-limiter"]; !private { + // Key exists in out because the /event entry WAS promoted. + // Clarifying: we want exactly one entry, keyed "rate-limiter". + } + if len(out) != 1 { + t.Fatalf("expected 1 promoted plugin event, got %d: %+v", len(out), out) + } + raw, ok := out["rate-limiter"] + if !ok { + t.Fatalf("expected key 'rate-limiter' (suffix stripped), got keys %v", + keysOf(out)) + } + // Round-trip JSON to verify the payload is intact. + var got rateLimiterEvent + if err := json.Unmarshal(raw, &got); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if !got.Allowed || got.TokensLeft != 42 { + t.Errorf("payload drifted: %+v", got) + } +} + +func keysOf(m map[string]json.RawMessage) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + return out +}