diff --git a/authbridge/CLAUDE.md b/authbridge/CLAUDE.md index 7ff3482d1..88ed0f200 100644 --- a/authbridge/CLAUDE.md +++ b/authbridge/CLAUDE.md @@ -406,6 +406,8 @@ When `session.enabled` is true (default) and `listener.session_api_addr` is non- | `GET /v1/sessions` | `application/json` | List active sessions: `{sessions: [{id, createdAt, updatedAt, eventCount, active}]}`. | | `GET /v1/sessions/{id}` | `application/json` | Full snapshot of one session's events. 404 if unknown/expired. | | `GET /v1/events` | `text/event-stream` | SSE stream of new events. Optional `?session=` filters to one session. Heartbeat every 30s. | +| `GET /v1/pipeline` | `application/json` | Active pipeline composition: `{inbound: [...], outbound: [...]}`. Each plugin entry carries `name`, `direction`, `position`, `bodyAccess`, `writes`, `reads`, plus the static metadata (`requires`, `requiresAny`, `after`, `claims`, `description`) and runtime `config` when present. abctl renders this as the Pipeline pane. | +| `GET /v1/plugins` | `application/json` | Catalog of every registered plugin (whether or not in the active pipeline): `{plugins: [{name, requires, requiresAny, after, claims, description, ...}]}`. abctl renders this as the Catalog pane (`P` key). 404s when the binary's session API was constructed without `WithCatalog`. | | `GET /healthz` | text | Liveness probe. | ### Quick examples diff --git a/authbridge/authlib/listener/reverseproxy/server.go b/authbridge/authlib/listener/reverseproxy/server.go index e47b93b85..d42632d14 100644 --- a/authbridge/authlib/listener/reverseproxy/server.go +++ b/authbridge/authlib/listener/reverseproxy/server.go @@ -230,13 +230,7 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { // not A2A-only. A non-A2A inbound, or an A2A request that fails to // parse, is intentionally not recorded here. if s.Sessions != nil && pctx.Extensions.A2A != nil { - sid := pctx.Extensions.A2A.SessionID - if sid == "" { - sid = s.Sessions.ActiveSession() - } - if sid == "" { - sid = session.DefaultSessionID - } + sid := inboundSessionID(pctx) // Snapshot-copy the protocol extension and use the shared helpers // for plugin invocations / observability / identity. Mirrors what // extproc does so request events don't pick up response-phase @@ -320,13 +314,7 @@ func (s *Server) modifyResponse(resp *http.Response) error { // pipeline saw at this point (may be empty for streamed bodies), // but the status code and plugin invocations are always meaningful. if s.Sessions != nil && pctx.Extensions.A2A != nil { - sid := pctx.Extensions.A2A.SessionID - if sid == "" { - sid = s.Sessions.ActiveSession() - } - if sid == "" { - sid = session.DefaultSessionID - } + sid := inboundSessionID(pctx) s.Sessions.Append(sid, pipeline.SessionEvent{ At: time.Now(), Direction: pipeline.Inbound, @@ -367,18 +355,10 @@ func (s *Server) recordInboundReject(pctx *pipeline.Context, action pipeline.Act return } // Inbound uses the A2A-stated contextId when available; otherwise - // falls through to the default bucket. Matches the accept path's - // bucketing rule (A2A request event at line 112-125). - sid := "" - if pctx.Extensions.A2A != nil { - sid = pctx.Extensions.A2A.SessionID - } - if sid == "" { - sid = s.Sessions.ActiveSession() - } - if sid == "" { - sid = session.DefaultSessionID - } + // the default bucket. Same rule as the accept path's + // inboundSessionID helper, kept consistent so denial events land + // in the same bucket the accepted request would have. + sid := inboundSessionID(pctx) var status int var code, message string if action.Violation != nil { @@ -430,3 +410,19 @@ func requestScheme(r *http.Request) string { } return "http" } + +// inboundSessionID returns the bucket ID for an inbound event. Mirrors +// extproc's inboundSessionID: trusts the A2A-stated contextId when +// non-empty, otherwise routes to DefaultSessionID. Does NOT fall back +// to ActiveSession() — that fallback was a cross-conversation +// contamination vector (a new conversation's first turn would inherit +// the previous conversation's rekeyed bucket, stranding the current +// turn's events in the prior bucket and creating an orphan one-event +// session for the response). Rekey on response migrates the Default +// bucket into the contextId once the agent reveals it. +func inboundSessionID(pctx *pipeline.Context) string { + if pctx.Extensions.A2A != nil && pctx.Extensions.A2A.SessionID != "" { + return pctx.Extensions.A2A.SessionID + } + return session.DefaultSessionID +} diff --git a/authbridge/authlib/listener/reverseproxy/server_test.go b/authbridge/authlib/listener/reverseproxy/server_test.go index 7f026895b..a0e00def0 100644 --- a/authbridge/authlib/listener/reverseproxy/server_test.go +++ b/authbridge/authlib/listener/reverseproxy/server_test.go @@ -562,3 +562,82 @@ func TestReverseProxy_ModifyResponse_RekeyAndResponseEvent(t *testing.T) { t.Errorf("response event has no Invocations; expected one a2a-stamp observe entry") } } + +// TestInboundSessionID_NoActiveSessionFallback locks in the +// reverseproxy's bucketing rule: when the A2A request has no +// contextId (first turn of a conversation), inboundSessionID returns +// DefaultSessionID — never falling back to ActiveSession(). The +// previous fallback was a cross-conversation contamination vector: +// a fresh chat's first turn would silently inherit a previous +// conversation's still-alive session bucket, stranding all events +// in the prior bucket and creating an orphan one-event session for +// the rekeyed response. +func TestInboundSessionID_NoActiveSessionFallback(t *testing.T) { + cases := []struct { + name string + ctx string // pctx.Extensions.A2A.SessionID + want string + }{ + {"first turn — empty contextId", "", session.DefaultSessionID}, + {"subsequent turn — explicit contextId", "ctx-abc", "ctx-abc"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + pctx := &pipeline.Context{ + Direction: pipeline.Inbound, + Extensions: pipeline.Extensions{ + A2A: &pipeline.A2AExtension{SessionID: tc.ctx}, + }, + } + if got := inboundSessionID(pctx); got != tc.want { + t.Errorf("inboundSessionID = %q, want %q", got, tc.want) + } + }) + } +} + +// TestInboundSessionID_NoA2AExtension exercises the auth-only path: +// non-A2A inbound (e.g. a denied request that never reached the +// parser) routes to DefaultSessionID, where operators expect to find +// unauthorized-access events. +func TestInboundSessionID_NoA2AExtension(t *testing.T) { + pctx := &pipeline.Context{Direction: pipeline.Inbound} + if got := inboundSessionID(pctx); got != session.DefaultSessionID { + t.Errorf("inboundSessionID = %q, want %q", got, session.DefaultSessionID) + } +} + +// TestRecordInbound_NewChatDoesntLeakIntoPriorBucket is the +// integration regression for the original bug: with a previous +// conversation's session still alive (so ActiveSession() returns it), +// a new chat's first-turn inbound request lands in DefaultSessionID +// rather than the previous bucket. The rekey-on-response path then +// migrates Default → server-assigned contextId, keeping the new +// conversation in its own bucket. +func TestRecordInbound_NewChatDoesntLeakIntoPriorBucket(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + + // Seed a "previous conversation" bucket and make it the active session. + store.Append("prev-conversation", pipeline.SessionEvent{ + At: time.Now(), Direction: pipeline.Inbound, Phase: pipeline.SessionRequest, + }) + if got := store.ActiveSession(); got != "prev-conversation" { + t.Fatalf("setup: ActiveSession = %q, want prev-conversation", got) + } + + // New chat's first turn arrives — empty contextId. + pctx := &pipeline.Context{ + Direction: pipeline.Inbound, + Extensions: pipeline.Extensions{ + A2A: &pipeline.A2AExtension{Method: "message/stream"}, + }, + } + sid := inboundSessionID(pctx) + if sid == "prev-conversation" { + t.Fatal("regression: new chat's first turn leaked into previous bucket") + } + if sid != session.DefaultSessionID { + t.Errorf("first-turn sid = %q, want %q", sid, session.DefaultSessionID) + } +} diff --git a/authbridge/authlib/pipeline/plugin.go b/authbridge/authlib/pipeline/plugin.go index 199fe1aea..e149feb4c 100644 --- a/authbridge/authlib/pipeline/plugin.go +++ b/authbridge/authlib/pipeline/plugin.go @@ -95,6 +95,19 @@ type PluginCapabilities struct { // list." See authlib/contracts/claims.go for the canonical // vocabulary. Claims []string + + // Description is operator-facing prose, one line, ≤80 chars, + // describing what this plugin does. Surfaces in `abctl`'s + // plugin-detail and catalog panes, and in /v1/plugins. Empty for + // plugins that haven't opted in; new plugins should populate it. + // + // Capabilities are static type-level metadata: Capabilities() must + // return the same value for any instance produced by a given + // factory. The catalog endpoint relies on this — varying capabilities + // by instance state silently produces wrong catalog entries. If a + // plugin's behavior varies enough that its capabilities differ, + // register it under multiple names. + Description string } // Normalize applies compatibility rules to a PluginCapabilities: diff --git a/authbridge/authlib/plugins/a2aparser/plugin.go b/authbridge/authlib/plugins/a2aparser/plugin.go index 8d7b74314..113d481dc 100644 --- a/authbridge/authlib/plugins/a2aparser/plugin.go +++ b/authbridge/authlib/plugins/a2aparser/plugin.go @@ -26,8 +26,9 @@ func (p *A2AParser) Name() string { return "a2a-parser" } func (p *A2AParser) Capabilities() pipeline.PluginCapabilities { return pipeline.PluginCapabilities{ - Writes: []string{"a2a"}, - ReadsBody: true, + Writes: []string{"a2a"}, + ReadsBody: true, + Description: "Parses A2A messages into pctx.Extensions.A2A for downstream plugins.", } } @@ -109,10 +110,19 @@ 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 { + // Stay silent when the request side never participated — the parser + // recorded nothing on request, so recording on response would orphan + // the row. + if pctx.Extensions.A2A == nil { + return pipeline.Action{Type: pipeline.Continue} + } + // We DID process the request but the response has no body — record + // a Skip so abctl can pair the response row with the request row. + // Without this, the request invocation orphans and the events table + // shows req+resp as unpaired (no ┌/└ glyphs, plugin/action columns + // blank on the response side). + if len(pctx.ResponseBody) == 0 { + pctx.Skip("no_response_body") return pipeline.Action{Type: pipeline.Continue} } diff --git a/authbridge/authlib/plugins/a2aparser/plugin_test.go b/authbridge/authlib/plugins/a2aparser/plugin_test.go index d8b89ef99..fc6371531 100644 --- a/authbridge/authlib/plugins/a2aparser/plugin_test.go +++ b/authbridge/authlib/plugins/a2aparser/plugin_test.go @@ -429,9 +429,13 @@ func TestA2AParser_OnResponse_NoRequestContext(t *testing.T) { } } +// TestA2AParser_OnResponse_EmptyBody locks the regression: when the +// request side parsed (Extensions.A2A populated) but response body is +// empty, the parser MUST record a Skip so abctl pairs the timeline rows. func TestA2AParser_OnResponse_EmptyBody(t *testing.T) { p := NewA2AParser() pctx := &pipeline.Context{ + Direction: pipeline.Inbound, Extensions: pipeline.Extensions{A2A: &pipeline.A2AExtension{Method: "message/send"}}, } action := p.OnResponse(context.Background(), pctx) @@ -441,6 +445,13 @@ func TestA2AParser_OnResponse_EmptyBody(t *testing.T) { if pctx.Extensions.A2A.SessionID != "" { t.Errorf("SessionID should remain empty, got %q", pctx.Extensions.A2A.SessionID) } + if pctx.Extensions.Invocations == nil { + t.Fatal("expected a Skip Invocation, got none") + } + invs := pctx.Extensions.Invocations.Inbound + if len(invs) != 1 || invs[0].Action != pipeline.ActionSkip || invs[0].Reason != "no_response_body" { + t.Fatalf("expected single Skip/no_response_body, got %+v", invs) + } } func TestA2AParser_OnResponse_JSONRPC_ContextID(t *testing.T) { diff --git a/authbridge/authlib/plugins/ibac/plugin.go b/authbridge/authlib/plugins/ibac/plugin.go index 653850555..386cc26b6 100644 --- a/authbridge/authlib/plugins/ibac/plugin.go +++ b/authbridge/authlib/plugins/ibac/plugin.go @@ -184,8 +184,9 @@ func (p *IBAC) Capabilities() pipeline.PluginCapabilities { // IBAC must come after so it can read the parsed tool name // and args. If it's absent, IBAC still functions on raw // HTTP — the judge sees method+host+path+body excerpt. - After: []string{"mcp-parser"}, - ReadsBody: true, + After: []string{"mcp-parser"}, + ReadsBody: true, + Description: "LLM-judge intent-based access control for outbound tool calls.", } } diff --git a/authbridge/authlib/plugins/inferenceparser/plugin.go b/authbridge/authlib/plugins/inferenceparser/plugin.go index 733b5edd0..5218983d7 100644 --- a/authbridge/authlib/plugins/inferenceparser/plugin.go +++ b/authbridge/authlib/plugins/inferenceparser/plugin.go @@ -26,8 +26,9 @@ func (p *InferenceParser) Name() string { return "inference-parser" } func (p *InferenceParser) Capabilities() pipeline.PluginCapabilities { return pipeline.PluginCapabilities{ - Writes: []string{"inference"}, - ReadsBody: true, + Writes: []string{"inference"}, + ReadsBody: true, + Description: "Parses LLM completions into pctx.Extensions.Inference.", } } @@ -95,9 +96,16 @@ 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 { + // Stay silent when the request side never participated — the parser + // recorded nothing on request, so recording on response would orphan + // the row. + if pctx.Extensions.Inference == nil { + return pipeline.Action{Type: pipeline.Continue} + } + // We DID process the request but the response has no body — record + // a Skip so abctl can pair the response row with the request row. + if len(pctx.ResponseBody) == 0 { + pctx.Skip("no_response_body") return pipeline.Action{Type: pipeline.Continue} } diff --git a/authbridge/authlib/plugins/inferenceparser/plugin_test.go b/authbridge/authlib/plugins/inferenceparser/plugin_test.go index 4bf08addf..7f8eb532f 100644 --- a/authbridge/authlib/plugins/inferenceparser/plugin_test.go +++ b/authbridge/authlib/plugins/inferenceparser/plugin_test.go @@ -395,9 +395,14 @@ func TestInferenceParser_OnResponse_NoRequestContext(t *testing.T) { } } +// TestInferenceParser_OnResponse_EmptyBody locks the regression: when +// the request side parsed (Extensions.Inference populated) but the +// response body is empty, the parser MUST record a Skip so abctl pairs +// the timeline rows. func TestInferenceParser_OnResponse_EmptyBody(t *testing.T) { p := NewInferenceParser() pctx := &pipeline.Context{ + Direction: pipeline.Outbound, Extensions: pipeline.Extensions{Inference: &pipeline.InferenceExtension{Model: "gpt-4"}}, } action := p.OnResponse(context.Background(), pctx) @@ -407,6 +412,13 @@ func TestInferenceParser_OnResponse_EmptyBody(t *testing.T) { if pctx.Extensions.Inference.Completion != "" { t.Errorf("Completion = %q, want empty", pctx.Extensions.Inference.Completion) } + if pctx.Extensions.Invocations == nil { + t.Fatal("expected a Skip Invocation, got none") + } + invs := pctx.Extensions.Invocations.Outbound + if len(invs) != 1 || invs[0].Action != pipeline.ActionSkip || invs[0].Reason != "no_response_body" { + t.Fatalf("expected single Skip/no_response_body, got %+v", invs) + } } func TestInferenceParser_OnResponse_NonStreaming(t *testing.T) { diff --git a/authbridge/authlib/plugins/jwtvalidation/plugin.go b/authbridge/authlib/plugins/jwtvalidation/plugin.go index 4d4069a4e..f3f263de7 100644 --- a/authbridge/authlib/plugins/jwtvalidation/plugin.go +++ b/authbridge/authlib/plugins/jwtvalidation/plugin.go @@ -209,7 +209,10 @@ func init() { func (p *JWTValidation) Name() string { return "jwt-validation" } func (p *JWTValidation) Capabilities() pipeline.PluginCapabilities { - return pipeline.PluginCapabilities{Writes: []string{"security"}} + return pipeline.PluginCapabilities{ + Writes: []string{"security"}, + Description: "Inbound JWT validation (signature, issuer, audience) against JWKS.", + } } // Configure decodes the plugin's config subtree, applies defaults, diff --git a/authbridge/authlib/plugins/mcpparser/plugin.go b/authbridge/authlib/plugins/mcpparser/plugin.go index a3b33ab63..f0ac2ebea 100644 --- a/authbridge/authlib/plugins/mcpparser/plugin.go +++ b/authbridge/authlib/plugins/mcpparser/plugin.go @@ -26,8 +26,9 @@ func (p *MCPParser) Name() string { return "mcp-parser" } func (p *MCPParser) Capabilities() pipeline.PluginCapabilities { return pipeline.PluginCapabilities{ - Writes: []string{"mcp"}, - ReadsBody: true, + Writes: []string{"mcp"}, + ReadsBody: true, + Description: "Parses MCP tool calls/results into pctx.Extensions.MCP.", } } @@ -70,12 +71,19 @@ func (p *MCPParser) OnRequest(_ context.Context, pctx *pipeline.Context) pipelin } 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 { + // Stay silent when the request side never participated — the parser + // recorded nothing on request, so recording on response would orphan + // the row. + if pctx.Extensions.MCP == nil { + return pipeline.Action{Type: pipeline.Continue} + } + // We DID process the request but the response has no body — typical + // for JSON-RPC notifications that ack with HTTP 202. Record a Skip + // so abctl can pair the response row with the request row in the + // timeline (pairing keys on plugin+method+direction; an empty + // invocation slot orphans both ends). + if len(pctx.ResponseBody) == 0 { + pctx.Skip("no_response_body") return pipeline.Action{Type: pipeline.Continue} } diff --git a/authbridge/authlib/plugins/mcpparser/plugin_test.go b/authbridge/authlib/plugins/mcpparser/plugin_test.go index 7f1601cb5..406c9ba84 100644 --- a/authbridge/authlib/plugins/mcpparser/plugin_test.go +++ b/authbridge/authlib/plugins/mcpparser/plugin_test.go @@ -201,10 +201,12 @@ func TestMCPParser_MissingParams(t *testing.T) { } func TestMCPParser_OnResponse_NoRequestContext(t *testing.T) { - // If the request phase never ran (no MCP extension populated), OnResponse - // should skip — there's nothing to correlate the response to. + // Request phase didn't run (no MCP extension populated): OnResponse + // stays silent — no Invocation recorded — so the response event + // doesn't appear as an MCP row at all. p := NewMCPParser() pctx := &pipeline.Context{ + Direction: pipeline.Outbound, ResponseBody: []byte(`{"jsonrpc":"2.0","id":1,"result":{"tools":[]}}`), } action := p.OnResponse(context.Background(), pctx) @@ -214,12 +216,23 @@ func TestMCPParser_OnResponse_NoRequestContext(t *testing.T) { if pctx.Extensions.MCP != nil { t.Error("MCP extension should remain nil when request was not parsed") } + if pctx.Extensions.Invocations != nil && + (len(pctx.Extensions.Invocations.Inbound)+len(pctx.Extensions.Invocations.Outbound)) > 0 { + t.Errorf("non-MCP response should not record any Invocation; got %+v", + pctx.Extensions.Invocations) + } } +// TestMCPParser_OnResponse_EmptyBody is the regression test for the +// notifications/initialized pairing bug: when the request side parsed +// the message (Extensions.MCP populated) but the response body is empty +// (HTTP 202 ack), the parser must record a Skip so abctl can pair the +// response row with the request row in the events timeline. func TestMCPParser_OnResponse_EmptyBody(t *testing.T) { p := NewMCPParser() pctx := &pipeline.Context{ - Extensions: pipeline.Extensions{MCP: &pipeline.MCPExtension{Method: "tools/list"}}, + Direction: pipeline.Outbound, + Extensions: pipeline.Extensions{MCP: &pipeline.MCPExtension{Method: "notifications/initialized"}}, } action := p.OnResponse(context.Background(), pctx) if action.Type != pipeline.Continue { @@ -228,6 +241,19 @@ func TestMCPParser_OnResponse_EmptyBody(t *testing.T) { if pctx.Extensions.MCP.Result != nil { t.Error("Result should remain nil when response body is empty") } + if pctx.Extensions.Invocations == nil { + t.Fatal("expected a Skip Invocation, got none") + } + invs := pctx.Extensions.Invocations.Outbound + if len(invs) != 1 { + t.Fatalf("expected 1 Invocation, got %d", len(invs)) + } + if invs[0].Action != pipeline.ActionSkip { + t.Errorf("Action = %q, want skip", invs[0].Action) + } + if invs[0].Reason != "no_response_body" { + t.Errorf("Reason = %q, want no_response_body", invs[0].Reason) + } } func TestMCPParser_OnResponse_ToolsList(t *testing.T) { diff --git a/authbridge/authlib/plugins/registry.go b/authbridge/authlib/plugins/registry.go index 11ba93182..61532fdd0 100644 --- a/authbridge/authlib/plugins/registry.go +++ b/authbridge/authlib/plugins/registry.go @@ -80,6 +80,125 @@ func RegisteredPlugins() []string { return names } +// CatalogEntry pairs a registered plugin's name with the capabilities +// it advertises. Surfaces in `abctl`'s catalog pane and in the +// /v1/plugins endpoint so operators can see what plugins exist and +// what each one needs without reading source. +type CatalogEntry struct { + Name string + Capabilities pipeline.PluginCapabilities +} + +// catalogCache memoizes the Catalog() result on first call. Constructors +// run once at first invocation; the throwaway instance is unreferenced +// after Capabilities() is read but never explicitly torn down. +// +// Caching bounds the constructor side-effect surface to a one-shot per +// process — even a misbehaving plugin that allocates a goroutine in its +// constructor leaks one goroutine, not one per /v1/plugins request. +var ( + catalogCacheMu sync.RWMutex + catalogCacheVal []CatalogEntry +) + +// Catalog returns a sorted snapshot of every registered plugin's +// capabilities. The result is computed on first call and cached; +// subsequent calls return the cached slice (do not mutate it). +// +// First-call mechanics: each registered factory is invoked once with no +// config; the resulting instance's Capabilities() is the static +// type-level metadata that the rest of the framework also reads. +// +// CONSTRUCTOR CONTRACT: factories called from Catalog MUST NOT allocate +// goroutines, network connections, file handles, or other resources +// that need explicit teardown. Allocate the plugin struct and nothing +// more — heavy work belongs in Init() (after Configure()), where the +// framework owns the lifecycle. The throwaway instance Catalog +// constructs is never Shutdown'd; anything it leaks is process-wide. +// +// The godoc on PluginCapabilities documents the parallel constraint +// that Capabilities() must be instance-state-independent (so the +// cached snapshot from one instance describes every instance the +// factory produces). +func Catalog() []CatalogEntry { + catalogCacheMu.RLock() + if catalogCacheVal != nil { + out := cloneCatalog(catalogCacheVal) + catalogCacheMu.RUnlock() + return out + } + catalogCacheMu.RUnlock() + + catalogCacheMu.Lock() + defer catalogCacheMu.Unlock() + if catalogCacheVal != nil { // double-check under write lock + return cloneCatalog(catalogCacheVal) + } + + registryMu.RLock() + defer registryMu.RUnlock() + out := make([]CatalogEntry, 0, len(registry)) + for name, factory := range registry { + out = append(out, CatalogEntry{ + Name: name, + Capabilities: factory().Capabilities().Normalize(), + }) + } + sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) + catalogCacheVal = out + return cloneCatalog(out) +} + +// cloneCatalog returns a deep copy of in: each CatalogEntry is copied +// and every []string field on its Capabilities is freshly allocated. +// Catalog returns a clone so callers can mutate the slice (sort, filter, +// extend per-entry slices) without tainting the cached snapshot — and +// without that tainted view leaking into future /v1/plugins responses. +func cloneCatalog(in []CatalogEntry) []CatalogEntry { + if len(in) == 0 { + return nil + } + out := make([]CatalogEntry, len(in)) + for i := range in { + caps := in[i].Capabilities + out[i] = CatalogEntry{ + Name: in[i].Name, + Capabilities: pipeline.PluginCapabilities{ + ReadsBody: caps.ReadsBody, + WritesBody: caps.WritesBody, + BodyAccess: caps.BodyAccess, + Description: caps.Description, + Writes: append([]string(nil), caps.Writes...), + Reads: append([]string(nil), caps.Reads...), + Requires: append([]string(nil), caps.Requires...), + RequiresAny: append([]string(nil), caps.RequiresAny...), + After: append([]string(nil), caps.After...), + Claims: append([]string(nil), caps.Claims...), + }, + } + } + return out +} + +// resetCatalogCache clears the memoized Catalog result. Intended for +// tests that register/unregister plugins and need a fresh view. +func resetCatalogCache() { + catalogCacheMu.Lock() + catalogCacheVal = nil + catalogCacheMu.Unlock() +} + +// WarmCatalog triggers Catalog() at boot so any plugin whose factory +// violates the constructor contract (panics, allocates goroutines / +// connections / file handles) surfaces during startup rather than +// silently caching a faulty throwaway instance on the first /v1/plugins +// request. Cheap insurance — the result is already memoized, so calling +// this at boot moves "first call" out of the request hot path. +// +// Intended for the binary's main() to invoke once after all plugin +// init() registrations have run (typically right before sessionapi.New). +func WarmCatalog() { _ = Catalog() } + // factoryFor looks up a factory by name. Internal to the package. // Callers under Build use this to resolve config entries into plugin // instances. diff --git a/authbridge/authlib/plugins/registry_test.go b/authbridge/authlib/plugins/registry_test.go index 9ac508490..20b82d286 100644 --- a/authbridge/authlib/plugins/registry_test.go +++ b/authbridge/authlib/plugins/registry_test.go @@ -3,6 +3,7 @@ package plugins import ( "context" "encoding/json" + "reflect" "testing" "github.com/kagenti/kagenti-extensions/authbridge/authlib/config" @@ -432,8 +433,10 @@ type consumerPlugin struct { setProvider *spiffe.Provider } -func (p *consumerPlugin) Name() string { return p.name } -func (p *consumerPlugin) Capabilities() pipeline.PluginCapabilities { return pipeline.PluginCapabilities{} } +func (p *consumerPlugin) Name() string { return p.name } +func (p *consumerPlugin) Capabilities() pipeline.PluginCapabilities { + return pipeline.PluginCapabilities{} +} func (p *consumerPlugin) OnRequest(context.Context, *pipeline.Context) pipeline.Action { return pipeline.Action{Type: pipeline.Continue} } @@ -613,3 +616,150 @@ func TestBuildWithSPIFFEWrapsConfigurablePluginsForRawConfig(t *testing.T) { t.Fatal("non-Configurable plugin should NOT be wrapped") } } + +// TestCatalog_IncludesRegisteredPlugins verifies Catalog() walks the +// registry, calls each factory once, and returns sorted CatalogEntries +// carrying the static capabilities each plugin advertises. +func TestCatalog_IncludesRegisteredPlugins(t *testing.T) { + resetCatalogCache() + t.Cleanup(resetCatalogCache) + const a, b = "test-catalog-a", "test-catalog-b" + RegisterPlugin(a, func() pipeline.Plugin { + return &relPlugin{ + name: a, + caps: pipeline.PluginCapabilities{ + Description: "A plugin", + Writes: []string{"out-a"}, + }, + } + }) + t.Cleanup(func() { UnregisterPlugin(a) }) + RegisterPlugin(b, func() pipeline.Plugin { + return &relPlugin{ + name: b, + caps: pipeline.PluginCapabilities{ + Description: "B plugin", + Reads: []string{"out-a"}, + Requires: []string{a}, + }, + } + }) + t.Cleanup(func() { UnregisterPlugin(b) }) + + entries := Catalog() + got := map[string]pipeline.PluginCapabilities{} + for _, e := range entries { + got[e.Name] = e.Capabilities + } + + if got[a].Description != "A plugin" { + t.Fatalf("Catalog missing %s with Description: %+v", a, got[a]) + } + if got[b].Description != "B plugin" { + t.Fatalf("Catalog missing %s with Description: %+v", b, got[b]) + } + if len(got[b].Requires) != 1 || got[b].Requires[0] != a { + t.Fatalf("Catalog %s.Requires lost: %+v", b, got[b].Requires) + } + + // Sorted order: walk entries and confirm a < b appear in that order + // (relative position only — other tests may register plugins). + idxA, idxB := -1, -1 + for i, e := range entries { + if e.Name == a { + idxA = i + } + if e.Name == b { + idxB = i + } + } + if idxA == -1 || idxB == -1 || idxA > idxB { + t.Fatalf("Catalog not sorted: a@%d b@%d", idxA, idxB) + } +} + +// TestCatalog_FactoryInvariant locks the load-bearing claim that every +// plugin's factory produces instances with byte-identical Capabilities(). +// Catalog() reads metadata once (from the first instance) and caches the +// result, so a future plugin that varies Capabilities() based on instance +// state would silently produce wrong catalog entries. +// +// Walks every registered plugin, calls its factory twice, and asserts the +// two Capabilities() return values are equal. Failure points to a plugin +// that needs to either (a) make its Capabilities() purely static, or +// (b) split into multiple registered names per behavioral variant. +func TestCatalog_FactoryInvariant(t *testing.T) { + registryMu.RLock() + names := make([]string, 0, len(registry)) + for n := range registry { + names = append(names, n) + } + registryMu.RUnlock() + + for _, name := range names { + name := name + t.Run(name, func(t *testing.T) { + factory, ok := factoryFor(name) + if !ok { + t.Fatalf("factory missing") + } + capsA := factory().Capabilities().Normalize() + capsB := factory().Capabilities().Normalize() + if !reflect.DeepEqual(capsA, capsB) { + t.Fatalf("Capabilities() differs across factory instances:\n A: %+v\n B: %+v", + capsA, capsB) + } + }) + } +} + +// TestCatalog_ReturnsDefensiveCopy verifies callers can mutate the +// returned slice (and its nested capability slices) without tainting +// the cached snapshot or future Catalog() reads. +func TestCatalog_ReturnsDefensiveCopy(t *testing.T) { + resetCatalogCache() + t.Cleanup(resetCatalogCache) + const name = "test-defensive-copy" + RegisterPlugin(name, func() pipeline.Plugin { + return &relPlugin{name: name, caps: pipeline.PluginCapabilities{ + Writes: []string{"slot-a"}, + }} + }) + t.Cleanup(func() { UnregisterPlugin(name) }) + + first := Catalog() + // Find our entry. + var idx int = -1 + for i, e := range first { + if e.Name == name { + idx = i + break + } + } + if idx < 0 { + t.Fatal("seeded plugin missing from Catalog") + } + + // Mutate the returned slice and the nested Writes slice. + first[idx].Capabilities.Writes[0] = "MUTATED" + first[idx].Capabilities.Description = "MUTATED" + + // A second call must still see the original values. + second := Catalog() + var idx2 int = -1 + for i, e := range second { + if e.Name == name { + idx2 = i + break + } + } + if idx2 < 0 { + t.Fatal("seeded plugin missing from second Catalog call") + } + if got := second[idx2].Capabilities.Writes[0]; got != "slot-a" { + t.Errorf("cache tainted by caller mutation: Writes[0] = %q, want slot-a", got) + } + if got := second[idx2].Capabilities.Description; got != "" { + t.Errorf("cache tainted by caller mutation: Description = %q, want empty", got) + } +} diff --git a/authbridge/authlib/plugins/tokenbroker/plugin.go b/authbridge/authlib/plugins/tokenbroker/plugin.go index e7f74a587..0c88ddd07 100644 --- a/authbridge/authlib/plugins/tokenbroker/plugin.go +++ b/authbridge/authlib/plugins/tokenbroker/plugin.go @@ -154,7 +154,9 @@ func NewTokenBroker() *TokenBroker { return &TokenBroker{} } func (p *TokenBroker) Name() string { return "token-broker" } func (p *TokenBroker) Capabilities() pipeline.PluginCapabilities { - return pipeline.PluginCapabilities{} + return pipeline.PluginCapabilities{ + Description: "Token broker: exchanges incoming tokens against the configured IdP.", + } } func (p *TokenBroker) Configure(raw json.RawMessage) error { diff --git a/authbridge/authlib/plugins/tokenexchange/plugin.go b/authbridge/authlib/plugins/tokenexchange/plugin.go index 8179fbfdd..6183554df 100644 --- a/authbridge/authlib/plugins/tokenexchange/plugin.go +++ b/authbridge/authlib/plugins/tokenexchange/plugin.go @@ -273,7 +273,9 @@ func init() { func (p *TokenExchange) Name() string { return "token-exchange" } func (p *TokenExchange) Capabilities() pipeline.PluginCapabilities { - return pipeline.PluginCapabilities{} + return pipeline.PluginCapabilities{ + Description: "RFC 8693 outbound token exchange against Keycloak per route.", + } } func (p *TokenExchange) Configure(raw json.RawMessage) error { diff --git a/authbridge/authlib/sessionapi/catalog_adapter.go b/authbridge/authlib/sessionapi/catalog_adapter.go new file mode 100644 index 000000000..dfb678e65 --- /dev/null +++ b/authbridge/authlib/sessionapi/catalog_adapter.go @@ -0,0 +1,32 @@ +package sessionapi + +import "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins" + +// PluginsCatalog adapts plugins.Catalog() into the wire-shaped +// CatalogEntry slice WithCatalog expects. Three binaries (authbridge- +// proxy, -envoy, -lite) plug it in identically; centralizing the +// conversion here keeps the field list one-place. +// +// Direction is left empty: the catalog describes plugin TYPES, and +// most plugins can be configured into either chain (parsers especially). +// abctl renders direction only for the active pipeline, where the +// answer is positional, not type-level. +func PluginsCatalog() []CatalogEntry { + src := plugins.Catalog() + out := make([]CatalogEntry, len(src)) + for i, e := range src { + n := e.Capabilities.Normalize() + out[i] = CatalogEntry{ + Name: e.Name, + ReadsBody: n.ReadsBody, + Writes: n.Writes, + Reads: n.Reads, + Requires: n.Requires, + RequiresAny: n.RequiresAny, + After: n.After, + Claims: n.Claims, + Description: n.Description, + } + } + return out +} diff --git a/authbridge/authlib/sessionapi/server.go b/authbridge/authlib/sessionapi/server.go index 0d2a7d991..db6a2fe8f 100644 --- a/authbridge/authlib/sessionapi/server.go +++ b/authbridge/authlib/sessionapi/server.go @@ -37,8 +37,38 @@ type Server struct { inbound *pipeline.Holder outbound *pipeline.Holder heartbeat time.Duration + // catalog returns the registered-plugin metadata for /v1/plugins. + // nil disables the endpoint (returns 404). The binary wires this to + // plugins.Catalog; tests inject a stub provider. + catalog CatalogProvider } +// CatalogEntry is the wire shape for one plugin in /v1/plugins. Mirrors +// pipelinePluginView's metadata fields so abctl can use the same +// rendering paths for the active pipeline and the catalog browser. +// +// Uses readsBody (the modern field name) instead of pipelinePluginView's +// legacy bodyAccess: this is a new wire shape introduced in the same PR +// that documents bodyAccess as deprecated, so there's no compat cost to +// emit the right name from day one. +type CatalogEntry struct { + Name string `json:"name"` + Direction string `json:"direction,omitempty"` + ReadsBody bool `json:"readsBody,omitempty"` + Writes []string `json:"writes,omitempty"` + Reads []string `json:"reads,omitempty"` + Requires []string `json:"requires,omitempty"` + RequiresAny []string `json:"requiresAny,omitempty"` + After []string `json:"after,omitempty"` + Claims []string `json:"claims,omitempty"` + Description string `json:"description,omitempty"` +} + +// CatalogProvider is the function the binary supplies to expose +// registered-plugin metadata to /v1/plugins. Decoupled so the +// sessionapi package doesn't import authlib/plugins. +type CatalogProvider func() []CatalogEntry + // Option configures a Server at construction time. type Option func(*Server) @@ -58,6 +88,14 @@ func WithPipelines(inbound, outbound *pipeline.Holder) Option { } } +// WithCatalog attaches a CatalogProvider so the server exposes the +// registered-plugin catalog at GET /v1/plugins. Without this option the +// endpoint returns 404 — useful in tests, harmless in production +// because plugins.Catalog is always available to the binary. +func WithCatalog(c CatalogProvider) Option { + return func(s *Server) { s.catalog = c } +} + // New constructs an HTTP server serving the session API at addr. store must // be non-nil; callers should only instantiate when session tracking is on. func New(addr string, store *session.Store, opts ...Option) *Server { @@ -74,6 +112,7 @@ func New(addr string, store *session.Store, opts ...Option) *Server { mux.HandleFunc("GET /v1/sessions/{id}", s.handleGet) mux.HandleFunc("GET /v1/events", s.handleStream) mux.HandleFunc("GET /v1/pipeline", s.handlePipeline) + mux.HandleFunc("GET /v1/plugins", s.handlePluginCatalog) mux.HandleFunc("GET /healthz", s.handleHealthz) s.server = &http.Server{ @@ -103,14 +142,25 @@ func (s *Server) handleHealthz(w http.ResponseWriter, _ *http.Request) { } // pipelinePluginView is the wire shape for one plugin in /v1/pipeline. +// +// The capability fields below (Reads/Writes/Requires/RequiresAny/After/ +// Claims/Description) are static type-level metadata: same for every +// instance produced by a given factory. abctl uses them to render the +// plugin-detail pane and to compute the "deps satisfied" indicator on +// the Pipeline pane without needing a separate /v1/plugins call. type pipelinePluginView struct { - Name string `json:"name"` - Direction string `json:"direction"` - Position int `json:"position"` // 1-based order within its direction - BodyAccess bool `json:"bodyAccess"` - Writes []string `json:"writes,omitempty"` - Reads []string `json:"reads,omitempty"` - Config json.RawMessage `json:"config,omitempty"` + Name string `json:"name"` + Direction string `json:"direction"` + Position int `json:"position"` // 1-based order within its direction + BodyAccess bool `json:"bodyAccess"` + Writes []string `json:"writes,omitempty"` + Reads []string `json:"reads,omitempty"` + Requires []string `json:"requires,omitempty"` + RequiresAny []string `json:"requiresAny,omitempty"` + After []string `json:"after,omitempty"` + Claims []string `json:"claims,omitempty"` + Description string `json:"description,omitempty"` + Config json.RawMessage `json:"config,omitempty"` } // handlePipeline returns the composition of the inbound and outbound @@ -129,6 +179,30 @@ func (s *Server) handlePipeline(w http.ResponseWriter, _ *http.Request) { } } +// handlePluginCatalog returns every registered plugin's metadata — +// not just the ones in the active pipeline. abctl renders this in +// the catalog browser pane so operators can see what's available +// before adding one to the pipeline. +// +// Auth: none, consistent with the rest of /v1/* (the package-level +// trust model gates this server to in-cluster networking only). The +// catalog reveals plugin metadata — names, dependency declarations, +// descriptions — never user content or secrets, so this is fine for +// the current posture. Revisit if sessionapi ever gates auth. +func (s *Server) handlePluginCatalog(w http.ResponseWriter, _ *http.Request) { + if s.catalog == nil { + http.NotFound(w, nil) + return + } + body := struct { + Plugins []CatalogEntry `json:"plugins"` + }{Plugins: s.catalog()} + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(body); err != nil { + slog.Debug("sessionapi: catalog encode failed", "error", err) + } +} + // describePipeline turns a *pipeline.Holder into its wire form, or an // empty slice when nil. Loads through the Holder so a hot-swap that // landed between requests is reflected immediately. @@ -139,14 +213,22 @@ func describePipeline(h *pipeline.Holder, direction string) []pipelinePluginView plugins := h.Plugins() out := make([]pipelinePluginView, len(plugins)) for i, pl := range plugins { - caps := pl.Capabilities() + caps := pl.Capabilities().Normalize() view := pipelinePluginView{ - Name: pl.Name(), - Direction: direction, - Position: i + 1, - BodyAccess: caps.BodyAccess, - Writes: caps.Writes, - Reads: caps.Reads, + Name: pl.Name(), + Direction: direction, + Position: i + 1, + // Normalize folds BodyAccess (deprecated) into ReadsBody; + // emit ReadsBody as the wire's BodyAccess field for backward + // compatibility with abctl < the catalog PR. + BodyAccess: caps.ReadsBody, + Writes: caps.Writes, + Reads: caps.Reads, + Requires: caps.Requires, + RequiresAny: caps.RequiresAny, + After: caps.After, + Claims: caps.Claims, + Description: caps.Description, } // Surface raw config when the plugin was wrapped by the registry. // Non-Configurable plugins don't satisfy RawConfigProvider; Config diff --git a/authbridge/authlib/sessionapi/server_test.go b/authbridge/authlib/sessionapi/server_test.go index 6193e57a5..f2ca67985 100644 --- a/authbridge/authlib/sessionapi/server_test.go +++ b/authbridge/authlib/sessionapi/server_test.go @@ -611,3 +611,126 @@ func TestHandleGet_SerializesPluginsMap(t *testing.T) { t.Errorf("payload drift: %+v", payload) } } + +// TestHandlePipelineSurfacesCapabilityMetadata verifies the new +// metadata fields (Requires/RequiresAny/After/Claims/Description) +// flow through to /v1/pipeline and are omitted when empty. +func TestHandlePipelineSurfacesCapabilityMetadata(t *testing.T) { + rich := pipeline.PluginCapabilities{ + Writes: []string{"mcp"}, + Requires: []string{"a2a-parser"}, + RequiresAny: []string{"jwt-validation", "token-broker"}, + After: []string{"mcp-parser"}, + Claims: []string{"authorization-header"}, + Description: "Test plugin description", + } + inbound, err := pipeline.New([]pipeline.Plugin{ + &fakePlugin{name: "rich-plugin", caps: rich}, + &fakePlugin{name: "bare-plugin"}, + }) + if err != nil { + t.Fatal(err) + } + + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + srv := New(":0", store, WithPipelines(pipeline.NewHolder(inbound), nil)) + ts := httptest.NewServer(srv.server.Handler) + defer ts.Close() + + resp, err := http.Get(ts.URL + "/v1/pipeline") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + // Decode into raw JSON to verify omitempty for the bare plugin. + var raw struct { + Inbound []map[string]any `json:"inbound"` + } + if err := json.NewDecoder(resp.Body).Decode(&raw); err != nil { + t.Fatal(err) + } + if len(raw.Inbound) != 2 { + t.Fatalf("inbound = %d, want 2", len(raw.Inbound)) + } + + got := raw.Inbound[0] + if got["description"] != "Test plugin description" { + t.Errorf("description = %v", got["description"]) + } + if reqs, _ := got["requires"].([]any); len(reqs) != 1 || reqs[0] != "a2a-parser" { + t.Errorf("requires = %v", got["requires"]) + } + if reqA, _ := got["requiresAny"].([]any); len(reqA) != 2 { + t.Errorf("requiresAny = %v", got["requiresAny"]) + } + if a, _ := got["after"].([]any); len(a) != 1 || a[0] != "mcp-parser" { + t.Errorf("after = %v", got["after"]) + } + if c, _ := got["claims"].([]any); len(c) != 1 || c[0] != "authorization-header" { + t.Errorf("claims = %v", got["claims"]) + } + + bare := raw.Inbound[1] + for _, k := range []string{"requires", "requiresAny", "after", "claims", "description"} { + if _, present := bare[k]; present { + t.Errorf("bare plugin should omit %q, got %v", k, bare[k]) + } + } +} + +func TestHandlePluginCatalog_ListsRegisteredPlugins(t *testing.T) { + stub := func() []CatalogEntry { + return []CatalogEntry{ + {Name: "alpha", Description: "First plugin", Writes: []string{"x"}}, + {Name: "beta", Requires: []string{"alpha"}}, + } + } + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + srv := New(":0", store, WithCatalog(stub)) + ts := httptest.NewServer(srv.server.Handler) + defer ts.Close() + + resp, err := http.Get(ts.URL + "/v1/plugins") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != 200 { + t.Fatalf("status = %d", resp.StatusCode) + } + var body struct { + Plugins []CatalogEntry `json:"plugins"` + } + if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { + t.Fatal(err) + } + if len(body.Plugins) != 2 { + t.Fatalf("plugins = %d, want 2", len(body.Plugins)) + } + if body.Plugins[0].Name != "alpha" || body.Plugins[0].Description != "First plugin" { + t.Errorf("plugins[0] = %+v", body.Plugins[0]) + } + if len(body.Plugins[1].Requires) != 1 || body.Plugins[1].Requires[0] != "alpha" { + t.Errorf("plugins[1].Requires = %v", body.Plugins[1].Requires) + } +} + +func TestHandlePluginCatalog_NoProvider404(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + srv := New(":0", store) // no WithCatalog + ts := httptest.NewServer(srv.server.Handler) + defer ts.Close() + + resp, err := http.Get(ts.URL + "/v1/plugins") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != 404 { + t.Fatalf("status = %d, want 404", resp.StatusCode) + } +} diff --git a/authbridge/cmd/abctl/README.md b/authbridge/cmd/abctl/README.md index 26cdf87d6..fefda3476 100644 --- a/authbridge/cmd/abctl/README.md +++ b/authbridge/cmd/abctl/README.md @@ -59,7 +59,7 @@ session APIs that aren't in your kube context. ## Panes -The UI has three panes. `Enter` drills in; `Esc` backs out. +The UI has these top-level panes. `Enter` drills in; `Esc` backs out. - **Sessions** (default): table of active sessions in the store, most recently updated first. Columns: ID, updated (relative), event count, @@ -71,6 +71,18 @@ The UI has three panes. `Enter` drills in; `Esc` backs out. - **Detail**: pretty-printed JSON of a single event. Scroll with arrow keys; `y` yanks to `/tmp/abctl-event-.json` and flashes the path in the footer. +- **Pipeline**: the active plugin chain in inbound + outbound order. + Columns: position, direction, plugin name, DEPS (✓/✗ — see "Plugin + dependencies" below), writes, body access, event count. `e` opens + the editor. +- **Plugin detail**: drill-into-row for Pipeline or Catalog. Shows + description, position, reads/writes, body access, plugin config, and + per-dependency satisfaction status against the active chain. +- **Catalog**: registered-plugin browser, opened by `P` from anywhere. + Lists every plugin the running binary knows how to construct, + including ones not in the active pipeline. Useful for discovering + what's available before adding to the pipeline. Sourced from + `/v1/plugins`. ## Keybindings @@ -89,6 +101,8 @@ The UI has three panes. `Enter` drills in; `Esc` backs out. | `p` | any | pause/resume stream | | `y` | detail | yank event JSON to `/tmp` | | `g` / `G` | lists | jump to top / bottom | +| `P` | sessions, pipeline, plugin-detail | open the registered-plugin catalog | +| `r` | catalog | refresh the catalog from `/v1/plugins` | | `e` | pipeline | edit pipeline subtree in `$EDITOR` | | `y` | edit/diff | apply the edit | | `N` | edit/diff | abort the edit | @@ -122,6 +136,27 @@ pipeline subtree. fields needed to fetch and apply aren't populated; pressing `e` flashes a hint instead of opening a broken edit. +### Pre-apply validation + +After save, abctl runs the same Requires/RequiresAny/After/Claims +checks the framework runs at reload-time, against the cached +`/v1/plugins` catalog. Issues land as a red banner above the diff +in ~50ms instead of waiting through the kubelet sync (~60s) to +discover them at hot-reload: + +```text +⚠ 1 validation issue — framework reload will reject: + • [outbound] ibac pos 1: Requires "mcp-parser", but it is not in the outbound chain +``` + +The y/N prompt becomes "apply anyway? (y/N)" — abctl's check is +non-blocking. The framework's own validateRelationships is the +source of truth and will fire again at reload regardless. + +Validation is silently skipped when the catalog isn't loaded +(operator hasn't pressed `P` yet). Visit the catalog pane once to +populate it for the rest of the session. + ### Agent-name resolution The per-agent ConfigMap is named `authbridge-config-`. abctl @@ -198,6 +233,33 @@ The poller terminates with one of: auto-rollback so the on-disk ConfigMap doesn't drift from the running pipeline. +## Plugin dependencies + +Plugins declare dependencies in their `Capabilities()`: + +- **Requires**: hard dependency. The named plugin MUST be in the same + chain at a strictly-lower position; otherwise framework reload fails. +- **RequiresAny**: soft OR. At least one of the listed plugins must + appear upstream; each one that IS present must be earlier. +- **After**: ordering hint. If the named plugin IS present, it must + appear earlier; absent is OK. +- **Claims**: exclusive ownership. Within one chain, two plugins + cannot both declare the same claim string. + +abctl surfaces these in three places: + +- **Pipeline pane DEPS column**: ✓ when all declared deps satisfied, + ✗ when any fail, blank when no deps declared. The footer hint + reports the count of plugins with unmet deps. +- **Plugin detail pane**: per-dependency rows with ✓/✗ and the + satisfying upstream's position when applicable. +- **Pre-apply validation in the editor**: catches missing/misordered + Requires before kubectl apply (~50ms vs ~60s framework roundtrip). + See the "Pre-apply validation" subsection above. + +The framework's own validateRelationships is the source of truth and +runs at every reload. abctl's checks are the fast-feedback layer. + ## Trust model `abctl` does no authentication — same as the server. Use only against diff --git a/authbridge/cmd/abctl/apiclient/client.go b/authbridge/cmd/abctl/apiclient/client.go index dffc405d1..1fe4ac168 100644 --- a/authbridge/cmd/abctl/apiclient/client.go +++ b/authbridge/cmd/abctl/apiclient/client.go @@ -83,13 +83,18 @@ type PipelineView struct { // PipelinePlugin describes one plugin's position, direction, and // capabilities. Mirrors the server's pipelinePluginView exactly. type PipelinePlugin struct { - Name string `json:"name"` - Direction string `json:"direction"` - Position int `json:"position"` - BodyAccess bool `json:"bodyAccess"` - Writes []string `json:"writes"` - Reads []string `json:"reads"` - Config json.RawMessage `json:"config,omitempty"` + Name string `json:"name"` + Direction string `json:"direction"` + Position int `json:"position"` + BodyAccess bool `json:"bodyAccess"` + Writes []string `json:"writes"` + Reads []string `json:"reads"` + Requires []string `json:"requires,omitempty"` + RequiresAny []string `json:"requiresAny,omitempty"` + After []string `json:"after,omitempty"` + Claims []string `json:"claims,omitempty"` + Description string `json:"description,omitempty"` + Config json.RawMessage `json:"config,omitempty"` } // GetPipeline fetches /v1/pipeline. @@ -101,6 +106,42 @@ func (c *Client) GetPipeline(ctx context.Context) (*PipelineView, error) { return &view, nil } +// PluginCatalog is the decoded shape of GET /v1/plugins. +type PluginCatalog struct { + Plugins []PluginCatalogEntry `json:"plugins"` +} + +// PluginCatalogEntry mirrors the server's sessionapi.CatalogEntry. +// Describes a registered plugin's static type-level metadata; the +// catalog includes plugins not currently in the active pipeline. +// +// readsBody is the modern field name (matches pipeline.PluginCapabilities +// post-Normalize); the older bodyAccess alias is intentionally NOT +// emitted on this new wire shape. +type PluginCatalogEntry struct { + Name string `json:"name"` + Direction string `json:"direction,omitempty"` + ReadsBody bool `json:"readsBody,omitempty"` + Writes []string `json:"writes,omitempty"` + Reads []string `json:"reads,omitempty"` + Requires []string `json:"requires,omitempty"` + RequiresAny []string `json:"requiresAny,omitempty"` + After []string `json:"after,omitempty"` + Claims []string `json:"claims,omitempty"` + Description string `json:"description,omitempty"` +} + +// GetPluginCatalog fetches /v1/plugins. Returns ErrNotFound when the +// server is too old to serve the endpoint (no WithCatalog option) so +// callers can degrade gracefully. +func (c *Client) GetPluginCatalog(ctx context.Context) (*PluginCatalog, error) { + var cat PluginCatalog + if err := c.getJSON(ctx, "/v1/plugins", &cat); err != nil { + return nil, err + } + return &cat, nil +} + func (c *Client) getJSON(ctx context.Context, path string, out any) error { req, err := http.NewRequestWithContext(ctx, "GET", c.endpoint+path, nil) if err != nil { diff --git a/authbridge/cmd/abctl/apiclient/client_test.go b/authbridge/cmd/abctl/apiclient/client_test.go index 5589eadc0..428e489e1 100644 --- a/authbridge/cmd/abctl/apiclient/client_test.go +++ b/authbridge/cmd/abctl/apiclient/client_test.go @@ -146,3 +146,85 @@ func TestPipelinePluginDecodesConfig(t *testing.T) { string(view.Inbound[1].Config)) } } + +func TestGetPluginCatalog(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/plugins" { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{ + "plugins": [ + {"name": "alpha", "description": "A plugin", "writes": ["x"]}, + {"name": "beta", "requires": ["alpha"]} + ] + }`)) + })) + defer ts.Close() + + c := New(ts.URL) + cat, err := c.GetPluginCatalog(context.Background()) + if err != nil { + t.Fatalf("GetPluginCatalog: %v", err) + } + if len(cat.Plugins) != 2 { + t.Fatalf("plugins = %d, want 2", len(cat.Plugins)) + } + if cat.Plugins[0].Description != "A plugin" { + t.Errorf("plugins[0].Description = %q", cat.Plugins[0].Description) + } + if len(cat.Plugins[1].Requires) != 1 || cat.Plugins[1].Requires[0] != "alpha" { + t.Errorf("plugins[1].Requires = %v", cat.Plugins[1].Requires) + } +} + +// TestPipelinePluginDecodesCapabilityMetadata verifies the new +// metadata fields decode correctly through the apiclient. +func TestPipelinePluginDecodesCapabilityMetadata(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/pipeline" { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{ + "inbound": [ + { + "name": "rich", + "direction": "inbound", + "position": 1, + "requires": ["a"], + "requiresAny": ["b","c"], + "after": ["d"], + "claims": ["sec"], + "description": "Rich plugin" + } + ], + "outbound": [] + }`)) + })) + defer ts.Close() + + c := New(ts.URL) + view, err := c.GetPipeline(context.Background()) + if err != nil { + t.Fatal(err) + } + p := view.Inbound[0] + if len(p.Requires) != 1 || p.Requires[0] != "a" { + t.Errorf("Requires = %v", p.Requires) + } + if len(p.RequiresAny) != 2 { + t.Errorf("RequiresAny = %v", p.RequiresAny) + } + if len(p.After) != 1 || p.After[0] != "d" { + t.Errorf("After = %v", p.After) + } + if len(p.Claims) != 1 || p.Claims[0] != "sec" { + t.Errorf("Claims = %v", p.Claims) + } + if p.Description != "Rich plugin" { + t.Errorf("Description = %q", p.Description) + } +} diff --git a/authbridge/cmd/abctl/edit/validate.go b/authbridge/cmd/abctl/edit/validate.go new file mode 100644 index 000000000..08bc05b82 --- /dev/null +++ b/authbridge/cmd/abctl/edit/validate.go @@ -0,0 +1,203 @@ +package edit + +import ( + "fmt" + + "gopkg.in/yaml.v3" + + "github.com/kagenti/kagenti-extensions/authbridge/cmd/abctl/apiclient" +) + +// ValidationError describes one problem with a proposed pipeline, +// detected by abctl before kubectl apply. The framework's own +// validateRelationships is the source of truth (and runs again after +// reload); this is the fast-feedback layer. +type ValidationError struct { + // Direction is "inbound" or "outbound". + Direction string + // PluginName is the offending plugin's name. + PluginName string + // Position is the offending plugin's 1-based position in its chain. + Position int + // Message is operator-facing: "Requires mcp-parser, missing in + // outbound chain" / "Unknown plugin name" / etc. + Message string +} + +// pipelineDoc mirrors the runtime YAML's pipeline subtree. Only the +// fields the validator needs. +type pipelineDoc struct { + Inbound pipelineChain `yaml:"inbound"` + Outbound pipelineChain `yaml:"outbound"` +} + +type pipelineChain struct { + Plugins []pluginEntry `yaml:"plugins"` +} + +type pluginEntry struct { + Name string `yaml:"name"` +} + +// pipelineRoot wraps pipelineDoc under the top-level "pipeline:" key, +// which is what callers pass in (the inner subtree). +type pipelineRoot struct { + Pipeline pipelineDoc `yaml:"pipeline"` +} + +// ValidatePipeline parses subtree YAML and checks Requires / +// RequiresAny / After / Claims against the catalog. Catalog comes +// from /v1/plugins; passing nil disables validation (no errors +// returned). Unknown plugin names produce errors so a typo gets +// caught before apply. +// +// The Claims check enforces cross-plugin uniqueness within a single +// chain — same as plugins.validateRelationships. +// +// Returns nil when all checks pass. +func ValidatePipeline(subtree []byte, catalog []apiclient.PluginCatalogEntry) []ValidationError { + if catalog == nil { + return nil + } + var root pipelineRoot + if err := yaml.Unmarshal(subtree, &root); err != nil { + // YAML errors are surfaced separately by the caller; this + // validator's job is the dependency layer only. + return nil + } + byName := make(map[string]apiclient.PluginCatalogEntry, len(catalog)) + for _, e := range catalog { + byName[e.Name] = e + } + + var errs []ValidationError + errs = append(errs, validateChain("inbound", root.Pipeline.Inbound, byName)...) + errs = append(errs, validateChain("outbound", root.Pipeline.Outbound, byName)...) + return errs +} + +// validateChain runs the Requires/RequiresAny/After/Claims/unknown-name +// checks for one direction. +func validateChain(direction string, chain pipelineChain, byName map[string]apiclient.PluginCatalogEntry) []ValidationError { + var errs []ValidationError + // Track positions of each name for ordering checks. Using lowest + // position wins on duplicates — same as the framework. + positions := map[string]int{} + for i, p := range chain.Plugins { + if _, seen := positions[p.Name]; !seen { + positions[p.Name] = i + 1 + } + } + // Track claims; first-declarer wins for the diagnostic. + claimedBy := map[string]string{} + + for i, p := range chain.Plugins { + pos := i + 1 + entry, known := byName[p.Name] + if !known { + // abctl caches /v1/plugins for the session; a freshly-installed + // plugin in-cluster won't appear until refresh. Hint at the + // staleness path so operators don't get stuck in confusion when + // the framework would actually accept the edit. + errs = append(errs, ValidationError{ + Direction: direction, + PluginName: p.Name, + Position: pos, + Message: fmt.Sprintf("Unknown plugin %q (not in cached /v1/plugins; "+ + "catalog may be stale, press P then r to refresh)", p.Name), + }) + continue + } + + // Requires: every name MUST appear at strictly-lower position. + for _, req := range entry.Requires { + rp, present := positions[req] + if !present { + errs = append(errs, ValidationError{ + Direction: direction, + PluginName: p.Name, + Position: pos, + Message: fmt.Sprintf("Requires %q, but it is not in the %s chain", + req, direction), + }) + continue + } + if rp >= pos { + errs = append(errs, ValidationError{ + Direction: direction, + PluginName: p.Name, + Position: pos, + Message: fmt.Sprintf("Requires %q upstream, but it is at position %d (must be < %d)", + req, rp, pos), + }) + } + } + + // RequiresAny: at least one of the listed names must appear at + // lower position. Each named one that IS present must be earlier. + if len(entry.RequiresAny) > 0 { + anyOK := false + for _, opt := range entry.RequiresAny { + rp, present := positions[opt] + if !present { + continue + } + if rp < pos { + anyOK = true + } + if rp >= pos { + errs = append(errs, ValidationError{ + Direction: direction, + PluginName: p.Name, + Position: pos, + Message: fmt.Sprintf("RequiresAny lists %q which is at position %d (must be < %d)", + opt, rp, pos), + }) + } + } + if !anyOK { + errs = append(errs, ValidationError{ + Direction: direction, + PluginName: p.Name, + Position: pos, + Message: fmt.Sprintf("RequiresAny %v: none present upstream in %s chain", + entry.RequiresAny, direction), + }) + } + } + + // After: present-at-or-after-this-position is a misorder. Matches + // the framework's validateRelationships rule (j >= i, not j > i) + // so a plugin listing itself in After or having a duplicate at + // the same index is flagged identically by abctl and the + // framework. + for _, name := range entry.After { + rp, present := positions[name] + if present && rp >= pos { + errs = append(errs, ValidationError{ + Direction: direction, + PluginName: p.Name, + Position: pos, + Message: fmt.Sprintf("After %q expects it earlier; it's at position %d (must be < %d)", + name, rp, pos), + }) + } + } + + // Claims: at most one declarer per claim string per chain. + for _, claim := range entry.Claims { + if other, taken := claimedBy[claim]; taken && other != p.Name { + errs = append(errs, ValidationError{ + Direction: direction, + PluginName: p.Name, + Position: pos, + Message: fmt.Sprintf("Claim %q already declared by %q in this chain", + claim, other), + }) + } else { + claimedBy[claim] = p.Name + } + } + } + return errs +} diff --git a/authbridge/cmd/abctl/edit/validate_test.go b/authbridge/cmd/abctl/edit/validate_test.go new file mode 100644 index 000000000..e97a4f32d --- /dev/null +++ b/authbridge/cmd/abctl/edit/validate_test.go @@ -0,0 +1,150 @@ +package edit + +import ( + "strings" + "testing" + + "github.com/kagenti/kagenti-extensions/authbridge/cmd/abctl/apiclient" +) + +func validateFixtureCatalog() []apiclient.PluginCatalogEntry { + return []apiclient.PluginCatalogEntry{ + {Name: "jwt-validation", Description: "Inbound JWT"}, + {Name: "a2a-parser", Description: "Parser"}, + {Name: "mcp-parser", Description: "MCP parser"}, + {Name: "ibac", Description: "IBAC", Requires: []string{"mcp-parser"}, After: []string{"a2a-parser"}}, + {Name: "claim-a", Claims: []string{"authorization-header"}}, + {Name: "claim-b", Claims: []string{"authorization-header"}}, + } +} + +func TestValidatePipeline_Empty(t *testing.T) { + errs := ValidatePipeline([]byte("pipeline:\n inbound:\n plugins: []\n"), validateFixtureCatalog()) + if len(errs) != 0 { + t.Fatalf("empty chains: got %v", errs) + } +} + +func TestValidatePipeline_HappyPath(t *testing.T) { + yaml := `pipeline: + inbound: + plugins: + - name: jwt-validation + - name: a2a-parser + outbound: + plugins: + - name: mcp-parser + - name: ibac +` + errs := ValidatePipeline([]byte(yaml), validateFixtureCatalog()) + if len(errs) != 0 { + t.Fatalf("happy path produced errors: %+v", errs) + } +} + +func TestValidatePipeline_MissingRequires(t *testing.T) { + yaml := `pipeline: + outbound: + plugins: + - name: ibac +` + errs := ValidatePipeline([]byte(yaml), validateFixtureCatalog()) + if len(errs) == 0 { + t.Fatal("expected error for missing mcp-parser") + } + found := false + for _, e := range errs { + if strings.Contains(e.Message, "Requires \"mcp-parser\"") { + found = true + } + } + if !found { + t.Fatalf("error should mention mcp-parser; got %+v", errs) + } +} + +func TestValidatePipeline_MisorderedRequires(t *testing.T) { + yaml := `pipeline: + outbound: + plugins: + - name: ibac + - name: mcp-parser +` + errs := ValidatePipeline([]byte(yaml), validateFixtureCatalog()) + if len(errs) == 0 { + t.Fatal("expected misorder error") + } + found := false + for _, e := range errs { + if strings.Contains(e.Message, "must be <") { + found = true + } + } + if !found { + t.Fatalf("error should call out misorder; got %+v", errs) + } +} + +func TestValidatePipeline_AfterMisorder(t *testing.T) { + yaml := `pipeline: + outbound: + plugins: + - name: ibac + - name: a2a-parser + - name: mcp-parser +` + errs := ValidatePipeline([]byte(yaml), validateFixtureCatalog()) + // ibac.After=[a2a-parser]; a2a-parser is at position 2 > 1. + found := false + for _, e := range errs { + if e.PluginName == "ibac" && strings.Contains(e.Message, "After \"a2a-parser\"") { + found = true + } + } + if !found { + t.Fatalf("expected After-misorder for ibac; got %+v", errs) + } +} + +func TestValidatePipeline_UnknownPlugin(t *testing.T) { + yaml := `pipeline: + inbound: + plugins: + - name: definitely-not-a-real-plugin +` + errs := ValidatePipeline([]byte(yaml), validateFixtureCatalog()) + if len(errs) != 1 || !strings.Contains(errs[0].Message, "Unknown plugin") { + t.Fatalf("expected Unknown plugin error; got %+v", errs) + } +} + +func TestValidatePipeline_ClaimsConflict(t *testing.T) { + yaml := `pipeline: + outbound: + plugins: + - name: claim-a + - name: claim-b +` + errs := ValidatePipeline([]byte(yaml), validateFixtureCatalog()) + found := false + for _, e := range errs { + if strings.Contains(e.Message, "already declared") { + found = true + } + } + if !found { + t.Fatalf("expected Claims conflict; got %+v", errs) + } +} + +func TestValidatePipeline_NilCatalogSkips(t *testing.T) { + yaml := `pipeline: + outbound: + plugins: + - name: bogus +` + errs := ValidatePipeline([]byte(yaml), nil) + if errs != nil { + t.Fatalf("nil catalog should disable validation, got %+v", errs) + } +} diff --git a/authbridge/cmd/abctl/tui/app.go b/authbridge/cmd/abctl/tui/app.go index aab775fdf..6c4fe6534 100644 --- a/authbridge/cmd/abctl/tui/app.go +++ b/authbridge/cmd/abctl/tui/app.go @@ -36,8 +36,15 @@ const ( paneDetail panePipeline panePluginDetail + paneCatalog ) +// paneNone is the explicit "no previous pane recorded" sentinel for +// model.previousPane. Using paneNamespaces (the zero value) as a +// sentinel would conflict with a future feature that wanted to open +// the catalog from the picker. -1 is unambiguous. +const paneNone paneID = -1 + // Connection state for the SSE stream. type connPhase int @@ -206,6 +213,7 @@ type model struct { sessionsTbl table.Model eventsTbl table.Model pipelineTbl table.Model + catalogTbl table.Model detailVp viewport.Model detailEvent *pipeline.SessionEvent detailPlugin *apiclient.PipelinePlugin @@ -222,6 +230,15 @@ type model struct { // until then. pipeline *apiclient.PipelineView + // catalog is the registered-plugin catalog from /v1/plugins, + // fetched lazily when the user first opens the catalog pane via + // `P`. Cached for the session; `r` from the catalog pane refreshes. + // nil before the first fetch; the catalog pane shows "(loading…)". + catalog *apiclient.PluginCatalog + // previousPane lets `Esc` from the catalog pane return to whichever + // pane the user came from instead of always defaulting to one. + previousPane paneID + // streamCh is the single SSE channel from the apiclient. Opened once // in Init; re-pumped on every streamMsg until it closes. streamCh <-chan apiclient.StreamEvent @@ -272,19 +289,21 @@ func New(ctx context.Context, c *apiclient.Client) tea.Model { ti.Prompt = "/ " return &model{ - endpoint: c.Endpoint(), - client: c, - ctx: ctx, - cancel: cancel, - events: make(map[string][]pipeline.SessionEvent), - pane: paneSessions, - sessionsTbl: newSessionsTable(), - eventsTbl: newEventsTable(), - pipelineTbl: newPipelineTable(), - detailVp: viewport.New(0, 0), - filterInput: ti, - lastTick: time.Now(), - connState: connStateInfo{phase: connConnecting}, + endpoint: c.Endpoint(), + client: c, + ctx: ctx, + cancel: cancel, + events: make(map[string][]pipeline.SessionEvent), + pane: paneSessions, + sessionsTbl: newSessionsTable(), + eventsTbl: newEventsTable(), + pipelineTbl: newPipelineTable(), + catalogTbl: newCatalogTable(), + previousPane: paneNone, + detailVp: viewport.New(0, 0), + filterInput: ti, + lastTick: time.Now(), + connState: connStateInfo{phase: connConnecting}, } } @@ -328,6 +347,12 @@ func (m *model) backToPodsPane() { m.rate = 0 m.drops = 0 m.pipeline = nil + // Drop the cached /v1/plugins snapshot too — a different pod is a + // different framework instance with potentially different plugin + // versions registered. The next `P` press refetches. + m.catalog = nil + m.catalogTbl.SetRows(nil) + m.previousPane = paneNone m.detailEvent = nil m.detailPlugin = nil m.selectedSess = "" @@ -365,6 +390,24 @@ func (m *model) loadPipelineCmd() tea.Cmd { } } +// catalogLoadedMsg carries the result of /v1/plugins. Distinct from +// pipelineLoadedMsg because the catalog is the registered set, not +// the active chain. +type catalogLoadedMsg struct { + catalog *apiclient.PluginCatalog + err error +} + +// loadCatalogCmd fetches /v1/plugins. Called lazily when the user +// presses `P` to enter the catalog pane; the result is cached on the +// model and refreshed on demand. +func (m *model) loadCatalogCmd() tea.Cmd { + return func() tea.Msg { + cat, err := m.client.GetPluginCatalog(m.ctx) + return catalogLoadedMsg{catalog: cat, err: err} + } +} + func tickCmd() tea.Cmd { return tea.Tick(time.Second, func(t time.Time) tea.Msg { return tickMsg(t) }) } @@ -480,6 +523,15 @@ func (m *model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { m.rebuildPipelineTable() return m, nil + case catalogLoadedMsg: + if msg.err != nil { + m.setFlash("catalog fetch failed: " + msg.err.Error()) + return m, nil + } + m.catalog = msg.catalog + m.rebuildCatalogTable() + return m, nil + case snapshotLoadedMsg: // Only update if we're still focused on this session. m.events[msg.id] = trim(msg.events, maxEventsPerSession) @@ -624,6 +676,33 @@ func (m *model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { m.editState.err = "invalid YAML: " + err.Error() return m, nil } + // The user's edited subtree parses standalone, but the framework + // loads the WHOLE inner YAML. An indentation mismatch between + // the user's edit and the splice site can produce a subtree that + // parses on its own yet breaks the combined doc. Splice + reparse + // to catch that here, instead of after a 60s kubelet round-trip. + previewInner := edit.Splice( + m.editState.fetched.InnerYAML, + m.editState.fetched.PipelineStart, + m.editState.fetched.PipelineEnd, + edited, + ) + var combinedVal any + if err := yaml.Unmarshal(previewInner, &combinedVal); err != nil { + m.editState.phase = editPhaseError + m.editState.err = "invalid YAML after splice: " + err.Error() + + "\n(probably an indentation mismatch — the pipeline: subtree must start at column 0)" + return m, nil + } + // Pre-apply validation against the catalog. Skipped silently + // when the catalog hasn't been fetched yet (operator hasn't + // pressed P); the framework's validateRelationships is the + // source of truth and runs again on reload regardless. + var catalog []apiclient.PluginCatalogEntry + if m.catalog != nil { + catalog = m.catalog.Plugins + } + m.editState.validationErrs = edit.ValidatePipeline(edited, catalog) m.editState.diff = edit.Diff(originalSubtree, edited) m.editState.phase = editPhaseDiff return m, nil @@ -879,6 +958,15 @@ func (m *model) View() string { } title = fmt.Sprintf("abctl · pipeline · %s", name) body = m.detailVp.View() + case paneCatalog: + title = fmt.Sprintf("abctl · %s · catalog", m.endpoint) + if m.catalog == nil { + body = styleHint.Render("(loading catalog…)") + } else if len(m.catalog.Plugins) == 0 { + body = styleHint.Render("(no registered plugins reported by /v1/plugins)") + } else { + body = m.catalogTbl.View() + } } header := styleTitle.Render(title) diff --git a/authbridge/cmd/abctl/tui/catalog_pane.go b/authbridge/cmd/abctl/tui/catalog_pane.go new file mode 100644 index 000000000..b6c9b331f --- /dev/null +++ b/authbridge/cmd/abctl/tui/catalog_pane.go @@ -0,0 +1,84 @@ +package tui + +import ( + "strings" + + "github.com/charmbracelet/bubbles/table" + + "github.com/kagenti/kagenti-extensions/authbridge/cmd/abctl/apiclient" +) + +// newCatalogTable builds the registered-plugin catalog table. Same +// styling as the pipeline table for visual consistency when switching +// between them via the `P` keybind. +func newCatalogTable() table.Model { + t := table.New( + table.WithColumns([]table.Column{ + {Title: "NAME", Width: 22}, + {Title: "REQUIRES", Width: 28}, + {Title: "DESCRIPTION", Width: 60}, + }), + table.WithFocused(true), + ) + t.SetStyles(tableStyles()) + return t +} + +// rebuildCatalogTable populates the catalog table from m.catalog. +// Empty rows when the catalog hasn't loaded yet — caller renders +// "(loading…)" in the title. +func (m *model) rebuildCatalogTable() { + if m.catalog == nil { + m.catalogTbl.SetRows(nil) + return + } + rows := make([]table.Row, 0, len(m.catalog.Plugins)) + for _, e := range m.catalog.Plugins { + // Combine Requires and RequiresAny (the "any" group joined by " | ") + // so operators see the full dependency picture in one column. + reqs := []string{} + reqs = append(reqs, e.Requires...) + if len(e.RequiresAny) > 0 { + reqs = append(reqs, strings.Join(e.RequiresAny, "|")) + } + rows = append(rows, table.Row{ + e.Name, + strings.Join(reqs, ", "), + e.Description, + }) + } + m.catalogTbl.SetRows(rows) +} + +// selectedCatalogEntry returns the catalog entry under the cursor as +// a synthetic PipelinePlugin so showPluginDetail can render it. Direction +// is left blank and Position is 0 — showPluginDetail elides those fields +// when empty so the detail view degrades gracefully for catalog entries. +func (m *model) selectedCatalogEntry() *apiclient.PipelinePlugin { + if m.catalog == nil { + return nil + } + rows := m.catalogTbl.Rows() + i := m.catalogTbl.Cursor() + if i < 0 || i >= len(rows) { + return nil + } + name := rows[i][0] + for _, e := range m.catalog.Plugins { + if e.Name == name { + p := apiclient.PipelinePlugin{ + Name: e.Name, + BodyAccess: e.ReadsBody, + Writes: e.Writes, + Reads: e.Reads, + Requires: e.Requires, + RequiresAny: e.RequiresAny, + After: e.After, + Claims: e.Claims, + Description: e.Description, + } + return &p + } + } + return nil +} diff --git a/authbridge/cmd/abctl/tui/catalog_pane_test.go b/authbridge/cmd/abctl/tui/catalog_pane_test.go new file mode 100644 index 000000000..c618f476b --- /dev/null +++ b/authbridge/cmd/abctl/tui/catalog_pane_test.go @@ -0,0 +1,94 @@ +package tui + +import ( + "context" + "strings" + "testing" + + "github.com/kagenti/kagenti-extensions/authbridge/cmd/abctl/apiclient" +) + +func TestRebuildCatalogTable_RendersEntries(t *testing.T) { + m := newPickerModel(context.Background(), nil, nil) + m.catalog = &apiclient.PluginCatalog{ + Plugins: []apiclient.PluginCatalogEntry{ + {Name: "alpha", Description: "First plugin"}, + {Name: "beta", Requires: []string{"alpha"}, Description: "Second"}, + }, + } + m.rebuildCatalogTable() + rows := m.catalogTbl.Rows() + if len(rows) != 2 { + t.Fatalf("rows = %d, want 2", len(rows)) + } + if rows[0][0] != "alpha" || !strings.Contains(rows[0][2], "First plugin") { + t.Errorf("rows[0] = %+v", rows[0]) + } + if rows[1][0] != "beta" || !strings.Contains(rows[1][1], "alpha") { + t.Errorf("rows[1] = %+v (expected requires=alpha)", rows[1]) + } +} + +func TestSelectedCatalogEntry_ReturnsCursorRow(t *testing.T) { + m := newPickerModel(context.Background(), nil, nil) + m.catalog = &apiclient.PluginCatalog{ + Plugins: []apiclient.PluginCatalogEntry{ + {Name: "alpha", Description: "First", Requires: []string{"x"}}, + {Name: "beta", Description: "Second"}, + }, + } + m.rebuildCatalogTable() + m.catalogTbl.SetCursor(1) + got := m.selectedCatalogEntry() + if got == nil { + t.Fatal("expected entry, got nil") + } + if got.Name != "beta" { + t.Fatalf("Name = %q, want beta", got.Name) + } +} + +func TestRebuildCatalogTable_NilCatalogClearsRows(t *testing.T) { + m := newPickerModel(context.Background(), nil, nil) + m.catalog = nil + m.rebuildCatalogTable() + if rows := m.catalogTbl.Rows(); len(rows) != 0 { + t.Fatalf("rows should be empty when catalog nil, got %d", len(rows)) + } +} + +// TestSelectedCatalogEntry_AsPipelinePlugin verifies the converter +// preserves the metadata fields the plugin-detail pane reads, so +// pressing Enter on a catalog row renders the full detail. +func TestSelectedCatalogEntry_AsPipelinePlugin(t *testing.T) { + m := newPickerModel(context.Background(), nil, nil) + m.catalog = &apiclient.PluginCatalog{ + Plugins: []apiclient.PluginCatalogEntry{ + { + Name: "ibac", + Description: "Judge", + Requires: []string{"mcp-parser"}, + After: []string{"a2a-parser"}, + }, + }, + } + m.rebuildCatalogTable() + got := m.selectedCatalogEntry() + if got == nil { + t.Fatal("nil entry") + } + if got.Description != "Judge" { + t.Errorf("Description = %q", got.Description) + } + if len(got.Requires) != 1 || got.Requires[0] != "mcp-parser" { + t.Errorf("Requires = %v", got.Requires) + } + if len(got.After) != 1 { + t.Errorf("After = %v", got.After) + } + // Direction and Position deliberately blank for catalog entries — + // showPluginDetail elides them. + if got.Direction != "" || got.Position != 0 { + t.Errorf("Direction/Position should be zero for catalog entry, got %q/%d", got.Direction, got.Position) + } +} diff --git a/authbridge/cmd/abctl/tui/deps.go b/authbridge/cmd/abctl/tui/deps.go new file mode 100644 index 000000000..3d9a9fa52 --- /dev/null +++ b/authbridge/cmd/abctl/tui/deps.go @@ -0,0 +1,170 @@ +package tui + +import ( + "fmt" + + "github.com/kagenti/kagenti-extensions/authbridge/cmd/abctl/apiclient" +) + +// depCheck describes whether one declared dependency is satisfied by +// the rest of a same-direction chain. Used by both the Pipeline pane's +// per-row indicator and the Plugin-detail pane's per-Requires section. +type depCheck struct { + Name string // the dependency-target plugin name + Satisfied bool + // UpstreamPosition is the 1-based position of the satisfying + // upstream plugin when Satisfied is true, otherwise 0. + UpstreamPosition int +} + +// sameDirectionChain returns the slice of plugins in p's chain. Caller +// supplies the full PipelineView so we can look up the right side +// without re-fetching. +func sameDirectionChain(p *apiclient.PipelinePlugin, view *apiclient.PipelineView) []apiclient.PipelinePlugin { + if view == nil { + return nil + } + if p.Direction == "inbound" { + return view.Inbound + } + if p.Direction == "outbound" { + return view.Outbound + } + return nil +} + +// requiresStatus returns one depCheck per entry in p.Requires. +// Satisfied iff the named plugin is present at a strictly-lower +// position. (RequiresAny semantics differ; see requiresAnyStatus.) +func requiresStatus(p *apiclient.PipelinePlugin, chain []apiclient.PipelinePlugin) []depCheck { + out := make([]depCheck, 0, len(p.Requires)) + for _, name := range p.Requires { + c := depCheck{Name: name} + for _, q := range chain { + if q.Name == name && q.Position < p.Position { + c.Satisfied = true + c.UpstreamPosition = q.Position + break + } + } + out = append(out, c) + } + return out +} + +// requiresAnyOK matches the framework's validateRelationships rule +// exactly: at least one named plugin must be present upstream, AND +// every name that IS present must be earlier (a downstream presence +// is a misorder). Earlier this returned true on at-least-one-upstream +// without checking the misorder condition, which let the Pipeline +// pane's DEPS column show ✓ for chains the framework would reject. +func requiresAnyOK(p *apiclient.PipelinePlugin, chain []apiclient.PipelinePlugin) bool { + if len(p.RequiresAny) == 0 { + return true + } + anyUpstream := false + for _, name := range p.RequiresAny { + for _, q := range chain { + if q.Name != name { + continue + } + if q.Position < p.Position { + anyUpstream = true + } else if q.Position > p.Position { + // Present-but-downstream violates the ordering rule. + return false + } + break + } + } + return anyUpstream +} + +// requiresAnyStatus returns one depCheck per entry in p.RequiresAny. +// Satisfied means "present at lower position." Used by the detail +// pane to show which alternatives currently satisfy the OR-group. +func requiresAnyStatus(p *apiclient.PipelinePlugin, chain []apiclient.PipelinePlugin) []depCheck { + out := make([]depCheck, 0, len(p.RequiresAny)) + for _, name := range p.RequiresAny { + c := depCheck{Name: name} + for _, q := range chain { + if q.Name == name && q.Position < p.Position { + c.Satisfied = true + c.UpstreamPosition = q.Position + break + } + } + out = append(out, c) + } + return out +} + +// afterStatus returns one depCheck per entry in p.After. Satisfied +// when the named plugin is absent (After is a soft hint) or present +// at a lower position. Misorder (present at higher position) is the +// only failure case. +func afterStatus(p *apiclient.PipelinePlugin, chain []apiclient.PipelinePlugin) []depCheck { + out := make([]depCheck, 0, len(p.After)) + for _, name := range p.After { + c := depCheck{Name: name, Satisfied: true} + for _, q := range chain { + if q.Name == name { + if q.Position < p.Position { + c.UpstreamPosition = q.Position + } else if q.Position > p.Position { + c.Satisfied = false + } + break + } + } + out = append(out, c) + } + return out +} + +// pluginDepsAllSatisfied returns true iff Requires, RequiresAny, and +// After are all OK for p in chain. Drives the Pipeline pane's per-row +// ✓/✗ indicator. Plugins with no declared deps are always ✓. +func pluginDepsAllSatisfied(p *apiclient.PipelinePlugin, chain []apiclient.PipelinePlugin) bool { + for _, c := range requiresStatus(p, chain) { + if !c.Satisfied { + return false + } + } + if !requiresAnyOK(p, chain) { + return false + } + for _, c := range afterStatus(p, chain) { + if !c.Satisfied { + return false + } + } + return true +} + +// pluginHasAnyDeps reports whether p declares any dependency that the +// indicator can render. Plugins without any declarations get a blank +// indicator (no false-positive ✓). +func pluginHasAnyDeps(p *apiclient.PipelinePlugin) bool { + return len(p.Requires) > 0 || len(p.RequiresAny) > 0 || len(p.After) > 0 +} + +// formatDepCheck returns a one-line description of a dependency check. +// Catalog-mode (no chain to check against) skips the ✓/✗ prefix and +// just shows the name — operators see the declared dependency without +// a misleading "satisfied" claim. +func formatDepCheck(c depCheck, withStatus bool) string { + if !withStatus { + return c.Name + } + if c.Satisfied { + if c.UpstreamPosition > 0 { + return styleOK.Render("✓ ") + c.Name + + styleHint.Render(fmt.Sprintf(" — at position %d", c.UpstreamPosition)) + } + return styleOK.Render("✓ ") + c.Name + + styleHint.Render(" — absent (soft)") + } + return styleError.Render("✗ ") + c.Name + + styleHint.Render(" — NOT in this chain") +} diff --git a/authbridge/cmd/abctl/tui/deps_test.go b/authbridge/cmd/abctl/tui/deps_test.go new file mode 100644 index 000000000..394e183a9 --- /dev/null +++ b/authbridge/cmd/abctl/tui/deps_test.go @@ -0,0 +1,134 @@ +package tui + +import ( + "testing" + + "github.com/kagenti/kagenti-extensions/authbridge/cmd/abctl/apiclient" +) + +func TestPluginDepsAllSatisfied_RequiresMet(t *testing.T) { + chain := []apiclient.PipelinePlugin{ + {Name: "a", Direction: "outbound", Position: 1}, + {Name: "b", Direction: "outbound", Position: 2, Requires: []string{"a"}}, + } + if !pluginDepsAllSatisfied(&chain[1], chain) { + t.Fatal("Requires=['a'] should be satisfied when 'a' is at position 1") + } +} + +func TestPluginDepsAllSatisfied_RequiresMissing(t *testing.T) { + chain := []apiclient.PipelinePlugin{ + {Name: "b", Direction: "outbound", Position: 1, Requires: []string{"a"}}, + } + if pluginDepsAllSatisfied(&chain[0], chain) { + t.Fatal("Requires=['a'] should NOT be satisfied when 'a' is absent") + } +} + +func TestPluginDepsAllSatisfied_RequiresMisordered(t *testing.T) { + chain := []apiclient.PipelinePlugin{ + {Name: "b", Direction: "outbound", Position: 1, Requires: []string{"a"}}, + {Name: "a", Direction: "outbound", Position: 2}, + } + if pluginDepsAllSatisfied(&chain[0], chain) { + t.Fatal("Requires=['a'] should NOT be satisfied when 'a' is downstream") + } +} + +func TestPluginDepsAllSatisfied_RequiresAnyMet(t *testing.T) { + chain := []apiclient.PipelinePlugin{ + {Name: "b", Direction: "outbound", Position: 1}, + {Name: "c", Direction: "outbound", Position: 2, + RequiresAny: []string{"a", "b"}}, + } + if !pluginDepsAllSatisfied(&chain[1], chain) { + t.Fatal("RequiresAny should be satisfied when one alternative is upstream") + } +} + +func TestPluginDepsAllSatisfied_RequiresAnyAllMissing(t *testing.T) { + chain := []apiclient.PipelinePlugin{ + {Name: "c", Direction: "outbound", Position: 1, + RequiresAny: []string{"a", "b"}}, + } + if pluginDepsAllSatisfied(&chain[0], chain) { + t.Fatal("RequiresAny should NOT be satisfied when no alternative is present") + } +} + +func TestPluginDepsAllSatisfied_AfterAbsentIsOK(t *testing.T) { + chain := []apiclient.PipelinePlugin{ + {Name: "ibac", Direction: "outbound", Position: 1, After: []string{"mcp-parser"}}, + } + if !pluginDepsAllSatisfied(&chain[0], chain) { + t.Fatal("After=['mcp-parser'] should be satisfied when mcp-parser is absent (soft hint)") + } +} + +func TestPluginDepsAllSatisfied_AfterMisordered(t *testing.T) { + chain := []apiclient.PipelinePlugin{ + {Name: "ibac", Direction: "outbound", Position: 1, After: []string{"mcp-parser"}}, + {Name: "mcp-parser", Direction: "outbound", Position: 2}, + } + if pluginDepsAllSatisfied(&chain[0], chain) { + t.Fatal("After=['mcp-parser'] should NOT be satisfied when mcp-parser is downstream") + } +} + +func TestPluginHasAnyDeps(t *testing.T) { + if pluginHasAnyDeps(&apiclient.PipelinePlugin{Name: "x"}) { + t.Fatal("plugin with no deps should report false") + } + if !pluginHasAnyDeps(&apiclient.PipelinePlugin{Name: "x", Requires: []string{"a"}}) { + t.Fatal("plugin with Requires should report true") + } + if !pluginHasAnyDeps(&apiclient.PipelinePlugin{Name: "x", After: []string{"a"}}) { + t.Fatal("plugin with After should report true") + } +} + +func TestUnmetDepsCount_Zero(t *testing.T) { + m := &model{ + pipeline: &apiclient.PipelineView{ + Inbound: []apiclient.PipelinePlugin{ + {Name: "a", Direction: "inbound", Position: 1}, + {Name: "b", Direction: "inbound", Position: 2, Requires: []string{"a"}}, + }, + }, + } + if got := m.unmetDepsCount(); got != 0 { + t.Fatalf("unmetDepsCount = %d, want 0", got) + } +} + +func TestUnmetDepsCount_Two(t *testing.T) { + m := &model{ + pipeline: &apiclient.PipelineView{ + Inbound: []apiclient.PipelinePlugin{ + {Name: "needy-a", Direction: "inbound", Position: 1, Requires: []string{"missing"}}, + }, + Outbound: []apiclient.PipelinePlugin{ + {Name: "needy-b", Direction: "outbound", Position: 1, RequiresAny: []string{"x"}}, + }, + }, + } + if got := m.unmetDepsCount(); got != 2 { + t.Fatalf("unmetDepsCount = %d, want 2", got) + } +} + +// TestPluginDepsAllSatisfied_RequiresAnyMisorderOnAlternate locks the +// stricter semantics: when one RequiresAny target is upstream (good) +// but ANOTHER named target is downstream (misorder), the overall +// result is NOT satisfied. Earlier this incorrectly returned true. +func TestPluginDepsAllSatisfied_RequiresAnyMisorderOnAlternate(t *testing.T) { + chain := []apiclient.PipelinePlugin{ + {Name: "a", Direction: "outbound", Position: 1}, + {Name: "c", Direction: "outbound", Position: 2, + RequiresAny: []string{"a", "b"}}, + {Name: "b", Direction: "outbound", Position: 3}, + } + if pluginDepsAllSatisfied(&chain[1], chain) { + t.Fatal("RequiresAny should NOT be satisfied when an alternative is downstream") + } +} diff --git a/authbridge/cmd/abctl/tui/edit_e2e_test.go b/authbridge/cmd/abctl/tui/edit_e2e_test.go index 13b5e4f1f..c3a1447f9 100644 --- a/authbridge/cmd/abctl/tui/edit_e2e_test.go +++ b/authbridge/cmd/abctl/tui/edit_e2e_test.go @@ -467,3 +467,41 @@ func TestEditFlow_StaleMsgFromAbandonedEditDropped(t *testing.T) { // Quiet "cmd is unused after Edit 2 dispatch" by referencing it. _ = cmd } + +// TestEditFlow_PostSpliceYAMLError verifies the post-splice YAML +// validation: an edit that parses standalone but produces an invalid +// combined inner YAML when spliced back routes to editPhaseError +// before the diff phase, instead of waiting for kubelet sync to +// surface the framework's reload error. +func TestEditFlow_PostSpliceYAMLError(t *testing.T) { + runner := &editFakeRunner{getResponse: []byte(editFixtureCMYAML)} + m := newPickerModel(context.Background(), nil, nil) + m.editRunner = runner.run + m.statusURL = "http://stub" + m.selectedNamespace = "team1" + m.selectedPod = "email-agent" + m.pane = panePipeline + + updated, cmd := m.Update(tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune{'e'}}) + mm := updated.(*model) + fetchedMsg := cmd().(genFetchedMsg) + defer os.Remove(fetchedMsg.TempPath) + + // Edit with leading whitespace on the pipeline: line. The subtree + // parses standalone, but splicing it back produces broken indent + // against the surrounding mode:/session: keys. + bad := []byte(" pipeline:\n inbound:\n - name: jwt-validation\n") + if err := os.WriteFile(fetchedMsg.TempPath, bad, 0o600); err != nil { + t.Fatal(err) + } + updated, _ = mm.Update(fetchedMsg) + mm = updated.(*model) + updated, _ = mm.Update(editorExitedMsg{gen: mm.editState.generation, err: nil}) + mm = updated.(*model) + if mm.editState.phase != editPhaseError { + t.Fatalf("phase = %v, want editPhaseError (post-splice YAML should fail)", mm.editState.phase) + } + if !strings.Contains(mm.editState.err, "after splice") { + t.Fatalf("error should mention splice; got %q", mm.editState.err) + } +} diff --git a/authbridge/cmd/abctl/tui/edit_overlay.go b/authbridge/cmd/abctl/tui/edit_overlay.go index d0113f817..8665f7fc6 100644 --- a/authbridge/cmd/abctl/tui/edit_overlay.go +++ b/authbridge/cmd/abctl/tui/edit_overlay.go @@ -40,6 +40,13 @@ type editState struct { diff string // colorized output from edit.Diff err string // single-line message in editPhaseError applyTime time.Time + // validationErrs are dependency/claim issues abctl detected before + // apply by checking the proposed pipeline against the plugin + // catalog. Empty when validation passed or the catalog isn't loaded. + // Rendered above the diff in the editPhaseDiff overlay so operators + // see them before deciding to apply. Non-blocking — the framework's + // own validateRelationships is still the source of truth at reload. + validationErrs []edit.ValidationError // generation is bumped each time a fresh edit cycle begins (initial // `e`, retry from error, restart after abort). Each tea.Cmd captures // the value at issue time; handlers drop messages whose captured @@ -77,9 +84,27 @@ func renderEditOverlay(s editState, width, height int) string { case editPhaseDiff: b.WriteString(styleTitle.Render("Edit pipeline — review diff")) b.WriteString("\n\n") + // Validation banner: render BEFORE the diff so operators see + // dependency issues at first glance. Non-blocking — apply still + // works. + if len(s.validationErrs) > 0 { + b.WriteString(styleError.Render(fmt.Sprintf( + "⚠ %d validation issue%s — framework reload will reject:", + len(s.validationErrs), plural(len(s.validationErrs))))) + b.WriteString("\n") + for _, ve := range s.validationErrs { + b.WriteString(fmt.Sprintf(" • [%s] %s pos %d: %s\n", + ve.Direction, ve.PluginName, ve.Position, ve.Message)) + } + b.WriteString("\n") + } b.WriteString(s.diff) b.WriteString("\n") - b.WriteString(styleHint.Render("apply this change? (y/N)")) + if len(s.validationErrs) > 0 { + b.WriteString(styleHint.Render("apply anyway? (y/N)")) + } else { + b.WriteString(styleHint.Render("apply this change? (y/N)")) + } case editPhaseApplying: b.WriteString(styleTitle.Render("Edit pipeline")) b.WriteString("\n\n") diff --git a/authbridge/cmd/abctl/tui/events_pane.go b/authbridge/cmd/abctl/tui/events_pane.go index d159c4f0d..015b775ff 100644 --- a/authbridge/cmd/abctl/tui/events_pane.go +++ b/authbridge/cmd/abctl/tui/events_pane.go @@ -92,12 +92,21 @@ func (m *model) rebuildEventsTable() { if m.filter != "" && !matchInvocationRow(rs, m.filter) { continue } - // Hide plugin-didn't-act rows by default. The footer hint - // shows the count + the toggle key so an operator who - // expected more rows can press `s` to surface them. + // Hide plugin-didn't-act rows by default. Exception: if a skip + // row's pair partner (the other side of the req/resp span) is + // NOT a skip, keep this row visible — the partner's ┌/└ glyph + // would otherwise orphan, making a successful exchange read + // as a missing response. This is exactly what + // notifications/initialized produces: a request observe paired + // with a response skip ("no_response_body"). Both visible + // preserves the pair semantics; both hidden when both are + // skips keeps the noise down. if !m.showSkips && isSkipRow(rs) { - m.hiddenSkips++ - continue + partnerIdx, paired := pairs[i] + if !paired || isSkipRow(rowSpecs[partnerIdx]) { + m.hiddenSkips++ + continue + } } // A "continuation" row is one whose event is the same as the // previous RENDERED row's event (filtering-aware). We blank the @@ -116,22 +125,22 @@ func (m *model) rebuildEventsTable() { // budget: outer (1) + inner (1) + base phase ("resp"/"deny" = // 4) → 6 max. prefix := spanGlyphs[i].prefix() + // PHASE is always populated (prefix + "req"/"resp") so a quick + // scan down the column reveals lifecycle position even on + // continuation rows. Earlier the continuation case rendered + // only the prefix, which left operators wondering whether the + // row was a request or response invocation. + phaseCell = prefix + shortPhase(rs.event.Phase) 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 = prefix + shortPhase(rs.event.Phase) statusC = statusCell(*rs.event) durCell = durationCell(*rs.event) tokC = tokensCell(*rs.event) hostC = truncStr(rs.event.Host, 20) - } else if prefix != "" { - // Continuation row inside a pair span: render only the - // connectors (no base phase text) so the operator's eye - // can follow the pair across multi-invocation events. - phaseCell = prefix } rows = append(rows, table.Row{ diff --git a/authbridge/cmd/abctl/tui/keys.go b/authbridge/cmd/abctl/tui/keys.go index 05dfaf95a..798eb0c9e 100644 --- a/authbridge/cmd/abctl/tui/keys.go +++ b/authbridge/cmd/abctl/tui/keys.go @@ -141,13 +141,28 @@ func (m *model) handleKey(msg tea.KeyMsg) tea.Cmd { return nil case "esc", "left", "h": - // Back-out: plugin-detail → pipeline; detail → events; events → sessions. + // Back-out: plugin-detail → pipeline (or catalog if we came from + // there); detail → events; events → sessions; catalog → previous. // In picker mode, the top-level session tabs (paneSessions and // panePipeline are siblings) back out further to the Pods picker, // tearing down PF + SSE. switch m.pane { case panePluginDetail: - m.pane = panePipeline + // Return to whichever pane invoked the detail (Pipeline or Catalog). + if m.previousPane == paneCatalog { + m.pane = paneCatalog + m.previousPane = paneNone + } else { + m.pane = panePipeline + } + case paneCatalog: + // Return to whichever pane the user pressed P from. + if m.previousPane != paneNone { + m.pane = m.previousPane + m.previousPane = paneNone + } else { + m.pane = panePipeline + } case paneDetail: m.pane = paneEvents case paneEvents: @@ -187,6 +202,16 @@ func (m *model) handleKey(msg tea.KeyMsg) tea.Cmd { if p == nil { return nil } + m.previousPane = panePipeline + m.showPluginDetail(p) + m.pane = panePluginDetail + return nil + case paneCatalog: + p := m.selectedCatalogEntry() + if p == nil { + return nil + } + m.previousPane = paneCatalog m.showPluginDetail(p) m.pane = panePluginDetail return nil @@ -232,6 +257,27 @@ func (m *model) handleKey(msg tea.KeyMsg) tea.Cmd { m.goBottom() return nil + case "P": + // Open the registered-plugin catalog. Available from any + // session-view pane; in --endpoint mode the picker fields + // don't matter — the catalog comes via the same /v1/* endpoint + // abctl is already pointed at. + if m.client == nil { + return nil + } + switch m.pane { + case paneNamespaces, panePods: + return nil + } + m.previousPane = m.pane + m.pane = paneCatalog + // Fetch on first open; cached afterward (refresh via `r`). + if m.catalog == nil { + return m.loadCatalogCmd() + } + m.rebuildCatalogTable() + return nil + // Dispatch j/k/up/down to the active component's Update. } @@ -262,6 +308,16 @@ func (m *model) handleKey(msg tea.KeyMsg) tea.Cmd { } } return cmd + case paneCatalog: + // `r` here refreshes the catalog (in the catalog pane only — the + // top-level `r` is reserved for the picker). All other keys go to + // the table for navigation. + if msg.String() == "r" { + return m.loadCatalogCmd() + } + var cmd tea.Cmd + m.catalogTbl, cmd = m.catalogTbl.Update(msg) + return cmd } return nil } @@ -280,6 +336,8 @@ func (m *model) refreshActivePane() { func (m *model) goTop() { switch m.pane { + case paneCatalog: + m.catalogTbl.SetCursor(0) case paneSessions: m.sessionsTbl.SetCursor(0) case paneEvents: @@ -305,6 +363,10 @@ func (m *model) goBottom() { if n := len(m.pipelineTbl.Rows()); n > 0 { m.pipelineTbl.SetCursor(n - 1) } + case paneCatalog: + if n := len(m.catalogTbl.Rows()); n > 0 { + m.catalogTbl.SetCursor(n - 1) + } case paneDetail, panePluginDetail: m.detailVp.GotoBottom() } @@ -345,12 +407,26 @@ func (m *model) helpView() string { case paneDetail: return "[↑↓] scroll [y] yank [esc] back [q] quit" case panePipeline: + var base string if m.parentCtx != nil { - return "[↑↓] nav [↵] plugin detail [e] edit [tab] sessions [esc] pods [q] quit" + base = "[↑↓] nav [↵] plugin detail [e] edit [tab] sessions [esc] pods [q] quit" + } else { + base = "[↑↓] nav [↵] plugin detail [e] edit [tab] sessions [q] quit" } - return "[↑↓] nav [↵] plugin detail [e] edit [tab] sessions [q] quit" + // Surface a count of plugins with unmet dependencies so a single + // "✗" in the DEPS column doesn't get lost in a long list. + if n := m.unmetDepsCount(); n > 0 { + base = fmt.Sprintf("%s · %d plugin%s with unmet deps", + base, n, plural(n)) + } + return base case panePluginDetail: return "[↑↓] scroll [esc] back [q] quit" + case paneCatalog: + if m.catalog == nil { + return "loading catalog… [esc] back [q] quit" + } + return "[↑↓] nav [↵] plugin detail [r] refresh [esc] back [q] quit" } return "[q] quit" } diff --git a/authbridge/cmd/abctl/tui/namespaces_pane.go b/authbridge/cmd/abctl/tui/namespaces_pane.go index 2c9c34427..7a99ffc53 100644 --- a/authbridge/cmd/abctl/tui/namespaces_pane.go +++ b/authbridge/cmd/abctl/tui/namespaces_pane.go @@ -61,18 +61,20 @@ func newPickerModel(ctx context.Context, lister cluster.Lister, pf cluster.PortF return &model{ // endpoint and client are set later, when portForwardReadyMsg arrives. - parentCtx: parentCtx, - ctx: ctx, - cancel: cancel, - events: make(map[string][]pipeline.SessionEvent), - pane: paneNamespaces, - sessionsTbl: newSessionsTable(), - eventsTbl: newEventsTable(), - pipelineTbl: newPipelineTable(), - detailVp: viewport.New(0, 0), - filterInput: ti, - lastTick: time.Now(), - connState: connStateInfo{phase: connConnecting}, + parentCtx: parentCtx, + ctx: ctx, + cancel: cancel, + events: make(map[string][]pipeline.SessionEvent), + pane: paneNamespaces, + sessionsTbl: newSessionsTable(), + eventsTbl: newEventsTable(), + pipelineTbl: newPipelineTable(), + catalogTbl: newCatalogTable(), + previousPane: paneNone, + detailVp: viewport.New(0, 0), + filterInput: ti, + lastTick: time.Now(), + connState: connStateInfo{phase: connConnecting}, // Picker-only: lister: lister, diff --git a/authbridge/cmd/abctl/tui/pipeline_pane.go b/authbridge/cmd/abctl/tui/pipeline_pane.go index c7efbf037..4bfb27d4f 100644 --- a/authbridge/cmd/abctl/tui/pipeline_pane.go +++ b/authbridge/cmd/abctl/tui/pipeline_pane.go @@ -18,6 +18,7 @@ func newPipelineTable() table.Model { {Title: "#", Width: 3}, {Title: "DIRECTION", Width: 10}, {Title: "PLUGIN", Width: 22}, + {Title: "DEPS", Width: 5}, {Title: "WRITES", Width: 18}, {Title: "BODY", Width: 6}, {Title: "EVENTS", Width: 8}, @@ -41,12 +42,12 @@ func (m *model) rebuildPipelineTable() { rows := make([]table.Row, 0, len(m.pipeline.Inbound)+len(m.pipeline.Outbound)+1) for _, p := range m.pipeline.Inbound { - rows = append(rows, pipelineRow(p, counts[p.Name])) + rows = append(rows, pipelineRow(p, counts[p.Name], m.pipeline.Inbound)) } // Divider between inbound and outbound. - rows = append(rows, table.Row{"", "", "── (app) ──", "", "", ""}) + rows = append(rows, table.Row{"", "", "── (app) ──", "", "", "", ""}) for _, p := range m.pipeline.Outbound { - rows = append(rows, pipelineRow(p, counts[p.Name])) + rows = append(rows, pipelineRow(p, counts[p.Name], m.pipeline.Outbound)) } m.pipelineTbl.SetRows(rows) // If cursor is on the divider row, nudge to the next plugin. @@ -55,7 +56,7 @@ func (m *model) rebuildPipelineTable() { } } -func pipelineRow(p apiclient.PipelinePlugin, events int) table.Row { +func pipelineRow(p apiclient.PipelinePlugin, events int, chain []apiclient.PipelinePlugin) table.Row { body := "no" if p.BodyAccess { body = "yes" @@ -64,6 +65,18 @@ func pipelineRow(p apiclient.PipelinePlugin, events int) table.Row { if events > 0 { eventsStr = fmt.Sprintf("%d", events) } + // DEPS column: ✓ when all declared dependencies are met, ✗ when any + // fail, blank when the plugin declares no Requires/RequiresAny/After. + // Blank vs ✓ avoids a misleading "looks fine" mark on plugins that + // have nothing to verify in the first place. + deps := "" + if pluginHasAnyDeps(&p) { + if pluginDepsAllSatisfied(&p, chain) { + deps = "✓" + } else { + deps = "✗" + } + } // Plugin names used to be colored by protocol but bubbles v1's // runewidth.Truncate miscounts ANSI escape bytes as visible width, // which truncated the closing \x1b[0m reset for longer names and @@ -72,6 +85,7 @@ func pipelineRow(p apiclient.PipelinePlugin, events int) table.Row { fmt.Sprintf("%d", p.Position), p.Direction, p.Name, + deps, strings.Join(p.Writes, ","), body, eventsStr, @@ -116,6 +130,30 @@ func (m *model) selectedPlugin() *apiclient.PipelinePlugin { return nil } +// unmetDepsCount returns how many active plugins have at least one +// unsatisfied Requires/RequiresAny/After dependency. Pulls from the +// active pipeline view so a hot-reload that lands a new chain +// immediately reflects in the count. +func (m *model) unmetDepsCount() int { + if m.pipeline == nil { + return 0 + } + n := 0 + for i := range m.pipeline.Inbound { + p := &m.pipeline.Inbound[i] + if pluginHasAnyDeps(p) && !pluginDepsAllSatisfied(p, m.pipeline.Inbound) { + n++ + } + } + for i := range m.pipeline.Outbound { + p := &m.pipeline.Outbound[i] + if pluginHasAnyDeps(p) && !pluginDepsAllSatisfied(p, m.pipeline.Outbound) { + n++ + } + } + return n +} + // countEventsPerPlugin counts how many times each plugin actually ran // across all cached events, by walking every event's Invocations list. // This includes auth-gate plugins (jwt-validation, token-exchange, ibac) @@ -139,4 +177,3 @@ func (m *model) countEventsPerPlugin() map[string]int { } return counts } - diff --git a/authbridge/cmd/abctl/tui/plugin_detail_pane.go b/authbridge/cmd/abctl/tui/plugin_detail_pane.go index e9e64a41d..dec1896ed 100644 --- a/authbridge/cmd/abctl/tui/plugin_detail_pane.go +++ b/authbridge/cmd/abctl/tui/plugin_detail_pane.go @@ -10,14 +10,29 @@ import ( // showPluginDetail loads the focused plugin into the detail viewport. // Uses a simple labelled block rather than JSON — the values are short // and human-readable. +// +// When m.pipeline is non-nil and the plugin's direction is "inbound" or +// "outbound", the Requires/RequiresAny/After sections render with ✓/✗ +// indicators against the active chain. For catalog-pane invocations +// (no live pipeline / direction), those sections render as informational +// lists without satisfaction status. func (m *model) showPluginDetail(p *apiclient.PipelinePlugin) { m.detailPlugin = p counts := m.countEventsPerPlugin() + chain := sameDirectionChain(p, m.pipeline) var b strings.Builder - fmt.Fprintf(&b, "%s %s\n\n", styleTitle.Render("Plugin:"), p.Name) - fmt.Fprintf(&b, "%s %s\n", styleMuted.Render("Direction:"), p.Direction) - fmt.Fprintf(&b, "%s %d\n", styleMuted.Render("Position: "), p.Position) + fmt.Fprintf(&b, "%s %s\n", styleTitle.Render("Plugin:"), p.Name) + if p.Description != "" { + fmt.Fprintf(&b, "%s\n", styleHint.Render(p.Description)) + } + fmt.Fprintln(&b) + if p.Direction != "" { + fmt.Fprintf(&b, "%s %s\n", styleMuted.Render("Direction:"), p.Direction) + } + if p.Position > 0 { + fmt.Fprintf(&b, "%s %d\n", styleMuted.Render("Position: "), p.Position) + } if len(p.Writes) > 0 { fmt.Fprintf(&b, "%s %s\n", styleMuted.Render("Writes: "), strings.Join(p.Writes, ", ")) } @@ -29,7 +44,46 @@ func (m *model) showPluginDetail(p *apiclient.PipelinePlugin) { body = "yes" } fmt.Fprintf(&b, "%s %s\n", styleMuted.Render("Body: "), body) - fmt.Fprintf(&b, "%s %d events in cached sessions\n", styleMuted.Render("Activity: "), counts[p.Name]) + if p.Position > 0 { + fmt.Fprintf(&b, "%s %d events in cached sessions\n", styleMuted.Render("Activity: "), counts[p.Name]) + } + + // Dependency sections. Render only when the plugin declares them. + if len(p.Requires) > 0 { + fmt.Fprintln(&b) + b.WriteString(styleMuted.Render("Requires:")) + b.WriteString("\n") + for _, c := range requiresStatus(p, chain) { + b.WriteString(" ") + b.WriteString(formatDepCheck(c, chain != nil)) + b.WriteString("\n") + } + } + if len(p.RequiresAny) > 0 { + fmt.Fprintln(&b) + b.WriteString(styleMuted.Render("Requires any of:")) + b.WriteString("\n") + for _, c := range requiresAnyStatus(p, chain) { + b.WriteString(" ") + b.WriteString(formatDepCheck(c, chain != nil)) + b.WriteString("\n") + } + } + if len(p.After) > 0 { + fmt.Fprintln(&b) + b.WriteString(styleMuted.Render("After (soft):")) + b.WriteString("\n") + for _, c := range afterStatus(p, chain) { + b.WriteString(" ") + b.WriteString(formatDepCheck(c, chain != nil)) + b.WriteString("\n") + } + } + if len(p.Claims) > 0 { + fmt.Fprintln(&b) + fmt.Fprintf(&b, "%s %s\n", styleMuted.Render("Claims: "), strings.Join(p.Claims, ", ")) + } + fmt.Fprintln(&b) // Always-newline format keeps the visual layout consistent whether // the plugin is Configurable (JSON body, multi-line) or not ("(none)", diff --git a/authbridge/cmd/abctl/tui/plugin_detail_pane_test.go b/authbridge/cmd/abctl/tui/plugin_detail_pane_test.go index b8dbcfa25..e7d54a2ad 100644 --- a/authbridge/cmd/abctl/tui/plugin_detail_pane_test.go +++ b/authbridge/cmd/abctl/tui/plugin_detail_pane_test.go @@ -87,3 +87,75 @@ func TestShowPluginDetailHandlesMalformedConfig(t *testing.T) { t.Fatalf("rendered view missing raw config fallback:\n%s", view) } } + +// TestShowPluginDetailRendersRequiresSatisfied renders ibac with +// mcp-parser already at a lower outbound position; the After-section +// indicator should report ✓. +func TestShowPluginDetailRendersRequiresSatisfied(t *testing.T) { + m := newPickerModel(context.Background(), nil, nil) + m.detailVp.Width = 80 + m.detailVp.Height = 30 + m.pipeline = &apiclient.PipelineView{ + Outbound: []apiclient.PipelinePlugin{ + {Name: "mcp-parser", Direction: "outbound", Position: 1}, + {Name: "ibac", Direction: "outbound", Position: 2, After: []string{"mcp-parser"}}, + }, + } + plugin := &m.pipeline.Outbound[1] + m.showPluginDetail(plugin) + view := m.detailVp.View() + if !strings.Contains(view, "After (soft)") { + t.Fatalf("After section missing:\n%s", view) + } + if !strings.Contains(view, "mcp-parser") { + t.Fatalf("After should mention mcp-parser:\n%s", view) + } + if !strings.Contains(view, "position 1") { + t.Fatalf("After should report upstream position 1:\n%s", view) + } +} + +// TestShowPluginDetailRendersRequiresUnmet sets up ibac without +// mcp-parser anywhere; After should still pass (soft hint, absent +// is OK), but the helper render path is exercised. +func TestShowPluginDetailRendersDescription(t *testing.T) { + m := newPickerModel(context.Background(), nil, nil) + m.detailVp.Width = 80 + m.detailVp.Height = 30 + plugin := &apiclient.PipelinePlugin{ + Name: "test", + Direction: "inbound", + Position: 1, + Description: "Operator-facing description", + } + m.showPluginDetail(plugin) + view := m.detailVp.View() + if !strings.Contains(view, "Operator-facing description") { + t.Fatalf("Description not rendered:\n%s", view) + } +} + +// TestShowPluginDetailRendersUnmetRequires verifies the ✗ branch +// triggers when a Required upstream is absent. +func TestShowPluginDetailRendersUnmetRequires(t *testing.T) { + m := newPickerModel(context.Background(), nil, nil) + m.detailVp.Width = 80 + m.detailVp.Height = 30 + m.pipeline = &apiclient.PipelineView{ + Outbound: []apiclient.PipelinePlugin{ + {Name: "needy", Direction: "outbound", Position: 1, Requires: []string{"missing-dep"}}, + }, + } + plugin := &m.pipeline.Outbound[0] + m.showPluginDetail(plugin) + view := m.detailVp.View() + if !strings.Contains(view, "Requires:") { + t.Fatalf("Requires section missing:\n%s", view) + } + if !strings.Contains(view, "missing-dep") { + t.Fatalf("Requires should name missing-dep:\n%s", view) + } + if !strings.Contains(view, "NOT in this chain") { + t.Fatalf("Requires should call out the missing dep:\n%s", view) + } +} diff --git a/authbridge/cmd/authbridge-envoy/main.go b/authbridge/cmd/authbridge-envoy/main.go index 85aea2f86..faeddda7f 100644 --- a/authbridge/cmd/authbridge-envoy/main.go +++ b/authbridge/cmd/authbridge-envoy/main.go @@ -221,12 +221,18 @@ func main() { } statSrv := startStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler()) + // Warm the plugin catalog at boot so any factory that violates the + // constructor contract surfaces here rather than on the first + // /v1/plugins request. + plugins.WarmCatalog() + var sessionAPISrv *sessionapi.Server if cfg.Listener.SessionAPIAddr != "" && sessions != nil { sessionAPISrv = sessionapi.New( cfg.Listener.SessionAPIAddr, sessions, sessionapi.WithPipelines(inboundH, outboundH), + sessionapi.WithCatalog(sessionapi.PluginsCatalog), ) go func() { slog.Warn("session API listening — UNAUTHENTICATED; contains raw user content; never expose via ingress", diff --git a/authbridge/cmd/authbridge-lite/main.go b/authbridge/cmd/authbridge-lite/main.go index 1da213ad8..69fd597d1 100644 --- a/authbridge/cmd/authbridge-lite/main.go +++ b/authbridge/cmd/authbridge-lite/main.go @@ -247,12 +247,18 @@ func main() { } statSrv := startStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler()) + // Warm the plugin catalog at boot so any factory that violates the + // constructor contract surfaces here rather than on the first + // /v1/plugins request. + plugins.WarmCatalog() + var sessionAPISrv *sessionapi.Server if cfg.Listener.SessionAPIAddr != "" && sessions != nil { sessionAPISrv = sessionapi.New( cfg.Listener.SessionAPIAddr, sessions, sessionapi.WithPipelines(inboundH, outboundH), + sessionapi.WithCatalog(sessionapi.PluginsCatalog), ) go func() { slog.Warn("session API listening — UNAUTHENTICATED; contains raw user content; never expose via ingress", diff --git a/authbridge/cmd/authbridge-proxy/main.go b/authbridge/cmd/authbridge-proxy/main.go index 58020dbb8..5ae4358cc 100644 --- a/authbridge/cmd/authbridge-proxy/main.go +++ b/authbridge/cmd/authbridge-proxy/main.go @@ -261,12 +261,18 @@ func main() { } statSrv := startStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler()) + // Warm the plugin catalog at boot so any factory that violates the + // constructor contract surfaces here rather than on the first + // /v1/plugins request. + plugins.WarmCatalog() + var sessionAPISrv *sessionapi.Server if cfg.Listener.SessionAPIAddr != "" && sessions != nil { sessionAPISrv = sessionapi.New( cfg.Listener.SessionAPIAddr, sessions, sessionapi.WithPipelines(inboundH, outboundH), + sessionapi.WithCatalog(sessionapi.PluginsCatalog), ) go func() { slog.Warn("session API listening — UNAUTHENTICATED; contains raw user content; never expose via ingress",