From 4306fbdf54bcd5f3a60f7bce7eea4efca24e6e24 Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Mon, 11 May 2026 22:37:14 -0400 Subject: [PATCH 1/3] feat(pipeline): Add Finisher hook for stateful plugin cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduces pipeline.Finisher, an optional interface that lets a plugin register a release point paired with its OnRequest acquire — fired once per request regardless of whether the pipeline allowed, denied, or errored. Closes the "cleanup leaks on denial" class of bugs flagged by Evaline Ju: a rate-limiter that reserves a slot in OnRequest and releases in OnResponse silently leaks whenever a later plugin denies, because OnResponse is only walked on pipelines that returned Continue. Same shape hits audit plugins (orphaned "request started" events), distributed tracers (unclosed spans), and any plugin that holds a lease across phases. Today no in-tree plugin is stateful on response so the gap is latent, but that makes it a framework correctness trap to resolve before a second stateful plugin ships. type Finisher interface { OnFinish(ctx context.Context, pctx *Context) } type Outcome struct { FinalAction OutcomeAction // allow | deny | error StatusCode int DenyingPlugin string Duration time.Duration } Design — answers to the 9 questions raised during discussion: 1. Timing: fires AFTER OnResponse if it ran, on every request end. Not an error-only hook; stateful plugins get one release site. 2. API shape: separate opt-in interface (type assertion). Symmetric with Initializer / Shutdowner / Readier. 3. Participation: only plugins whose OnRequest actually ran, including the denier. Pipeline.Run tracks the dispatched set on pctx as each plugin enters OnRequest. 4. Outcome info: hybrid. Consistent (ctx, pctx) signature; outcome reached via pctx.Outcome() which returns non-nil ONLY during OnFinish. Nil everywhere else so accidental reads in the wrong phase surface as a nil deref rather than a stale zero value. 5. Context: fresh ctx derived from context.Background() with a framework-set deadline (DefaultFinishTimeout = 2s, overridable via WithFinishTimeout). Client disconnect during the request does not cancel OnFinish I/O. The listener-supplied ctx argument is reserved for future parent-cancel needs but intentionally not wired. 6. Order: LIFO, symmetric with Shutdowner and RunResponse. 7. State: full pctx including response body. Framework holds the buffer across the finish window. 8. Errors: best-effort. Each plugin's OnFinish runs under its own recover so a panic is logged and the LIFO chain continues. One misbehaving plugin cannot leak state in every other plugin. 9. Invocations: OnFinish is silent. pctx.Record / SetBody / SetResponseBody called during OnFinish are dropped with WARN logs (SessionEvent published, response on the wire). Implementation surface: authlib/pipeline/plugin.go Finisher interface + detailed godoc authlib/pipeline/outcome.go Outcome struct, OutcomeAction enum, Context.Outcome() getter authlib/pipeline/context.go dispatched []int ledger, outcome *Outcome, inFinish bool, Record/SetBody guards authlib/pipeline/pipeline.go DefaultFinishTimeout, WithFinishTimeout Option, Pipeline.finishTimeout, Pipeline.RunFinish + dispatchFinish authlib/pipeline/finisher_test.go 13 tests covering LIFO, only- dispatched, outcome visibility, nil- outside-OnFinish, panic recovery, fresh-ctx-not-cancelled, deadline, default timeout, off-policy skip, non-Finisher skip, Record-drop, SetBody-drop. Docs: framework-architecture.md §6 — full "Per-request finish hook" section next to the Start/Stop lifecycle. §12 — versioning entry. §13 — cross-reference updated. plugin-reference.md — new "Finishing requests (stateful plugins)" section with rate-limiter example + the four "can't break" rules. No listener wiring yet. Listeners call Pipeline.RunFinish in a future commit; until then no plugin implements Finisher and the dispatcher remains dormant (empty dispatched list = immediate return from RunFinish). Existing plugins and tests unchanged. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- authbridge/authlib/pipeline/context.go | 46 +++ authbridge/authlib/pipeline/finisher_test.go | 410 +++++++++++++++++++ authbridge/authlib/pipeline/outcome.go | 77 ++++ authbridge/authlib/pipeline/pipeline.go | 113 ++++- authbridge/authlib/pipeline/plugin.go | 52 +++ authbridge/docs/framework-architecture.md | 75 +++- authbridge/docs/plugin-reference.md | 93 +++++ 7 files changed, 854 insertions(+), 12 deletions(-) create mode 100644 authbridge/authlib/pipeline/finisher_test.go create mode 100644 authbridge/authlib/pipeline/outcome.go diff --git a/authbridge/authlib/pipeline/context.go b/authbridge/authlib/pipeline/context.go index ffede67cf..646abb835 100644 --- a/authbridge/authlib/pipeline/context.go +++ b/authbridge/authlib/pipeline/context.go @@ -139,6 +139,33 @@ type Context struct { // "tried to redact, nothing matched" is valid telemetry. bodyMutated bool responseBodyMutated bool + + // dispatched lists the pipeline indices whose OnRequest was actually + // invoked (including the plugin that denied, if any). Populated by + // Pipeline.Run before each plugin's OnRequest call; consumed by + // Pipeline.RunFinish to dispatch OnFinish in LIFO to exactly the set + // of plugins that "reserved" per-request state. + // + // Indices (not plugin pointers) because the pipeline slice is the + // authoritative owner of plugin identity; pctx avoids holding a back + // reference so a pctx is safe to pass across pipelines in tests. + dispatched []int + + // outcome is populated exactly once by Pipeline.RunFinish immediately + // before dispatching OnFinish on the first plugin. Nil during OnRequest + // and OnResponse. Read via Outcome() so the "nil outside OnFinish" + // invariant is enforced at the call site rather than via a documented + // zero-value contract. + outcome *Outcome + + // inFinish is true while RunFinish is dispatching OnFinish on one + // of the Finisher plugins. Read by Record / SetBody / SetResponseBody + // to enforce the "SessionEvent is frozen during OnFinish" contract + // chosen for the finish hook: plugins that accidentally call + // Record / SetBody during cleanup hit a WARN log + no-op rather + // than silently mutating a SessionEvent that has already been + // published or a response that is already on the wire. + inFinish bool } // SetCurrentPlugin is called by Pipeline.Run / RunResponse immediately @@ -196,6 +223,13 @@ func (c *Context) clearCurrent() { // For the bare (Action, Reason) case, prefer the convenience wrappers // (Allow / Skip / Observe / Modify) below — one line each. func (c *Context) Record(inv Invocation) { + if c.inFinish { + slog.Warn("pipeline: plugin called pctx.Record during OnFinish — dropped", + "plugin", inv.Plugin, + "action", inv.Action, + "reason", inv.Reason) + return + } if inv.Plugin == "" { inv.Plugin = c.currentPlugin } @@ -275,6 +309,12 @@ func (c *Context) DenyAndRecord(reason, code, message string) Action { // Callers should NOT assign pctx.Body directly — the listener wouldn't // know to propagate the change, and the Invocation wouldn't be emitted. func (c *Context) SetBody(newBody []byte) { + if c.inFinish { + slog.Warn("pipeline: plugin called pctx.SetBody during OnFinish — dropped (response already sent)", + "plugin", c.currentPlugin, + "new_len", len(newBody)) + return + } if c.currentPolicy == ErrorPolicyObserve { c.recordShadowBodyMutation("request", c.Body, newBody) return @@ -292,6 +332,12 @@ func (c *Context) SetBody(newBody []byte) { // and the same observe-mode suppression: under ErrorPolicyObserve the // response body is untouched and the Invocation is marked Shadow=true. func (c *Context) SetResponseBody(newBody []byte) { + if c.inFinish { + slog.Warn("pipeline: plugin called pctx.SetResponseBody during OnFinish — dropped (response already sent)", + "plugin", c.currentPlugin, + "new_len", len(newBody)) + return + } if c.currentPolicy == ErrorPolicyObserve { c.recordShadowBodyMutation("response", c.ResponseBody, newBody) return diff --git a/authbridge/authlib/pipeline/finisher_test.go b/authbridge/authlib/pipeline/finisher_test.go new file mode 100644 index 000000000..1f316e8da --- /dev/null +++ b/authbridge/authlib/pipeline/finisher_test.go @@ -0,0 +1,410 @@ +package pipeline + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +// finisherPlugin embeds stubPlugin and adds OnFinish. Using embedding +// rather than a brand-new struct so the tests reuse the existing +// Name/Capabilities/OnRequest/OnResponse surface. +type finisherPlugin struct { + *stubPlugin + onFinish func(ctx context.Context, pctx *Context) +} + +func (f *finisherPlugin) OnFinish(ctx context.Context, pctx *Context) { + if f.onFinish != nil { + f.onFinish(ctx, pctx) + } +} + +func newFinisher(name string, onFinish func(ctx context.Context, pctx *Context)) *finisherPlugin { + return &finisherPlugin{ + stubPlugin: &stubPlugin{name: name}, + onFinish: onFinish, + } +} + +// TestRunFinish_NilWhenNoDispatch verifies that a pctx with an empty +// dispatched list (no Run was ever called, or the pipeline was empty) +// is a safe no-op. Prevents an accidental double-free on the edge case +// where a listener defers RunFinish but Run never reached dispatch. +func TestRunFinish_NilWhenNoDispatch(t *testing.T) { + p, err := New([]Plugin{newFinisher("f", nil)}) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + p.RunFinish(context.Background(), pctx, Outcome{FinalAction: OutcomeAllow}) + // No panic, no OnFinish invocation (nothing in dispatched). +} + +// TestRunFinish_LIFO verifies dispatch order is reverse of OnRequest +// order, symmetric with Shutdowner and RunResponse. +func TestRunFinish_LIFO(t *testing.T) { + var order []string + mk := func(name string) *finisherPlugin { + return newFinisher(name, func(_ context.Context, _ *Context) { + order = append(order, name) + }) + } + p, err := New([]Plugin{mk("a"), mk("b"), mk("c")}) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + if act := p.Run(context.Background(), pctx); act.Type != Continue { + t.Fatalf("Run action = %v, want Continue", act.Type) + } + p.RunFinish(context.Background(), pctx, Outcome{FinalAction: OutcomeAllow}) + if got, want := order, []string{"c", "b", "a"}; !equal(got, want) { + t.Errorf("finish order = %v, want %v (LIFO)", got, want) + } +} + +// TestRunFinish_OnlyDispatched verifies that OnFinish is NOT called on +// plugins whose OnRequest never ran because an earlier plugin denied. +// Plugin A reserves, B denies, C+D never dispatch. Only A and B (the +// denier) should see OnFinish; C and D must not. +func TestRunFinish_OnlyDispatched(t *testing.T) { + var seen []string + markSeen := func(name string) *finisherPlugin { + return newFinisher(name, func(_ context.Context, _ *Context) { + seen = append(seen, name) + }) + } + a := markSeen("a") + b := markSeen("b") + b.stubPlugin.onReq = func(ctx context.Context, pctx *Context) Action { + return Action{Type: Reject, Violation: &Violation{Code: "test.deny", Reason: "b denied"}} + } + c := markSeen("c") + d := markSeen("d") + p, err := New([]Plugin{a, b, c, d}) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + action := p.Run(context.Background(), pctx) + if action.Type != Reject { + t.Fatalf("Run action = %v, want Reject", action.Type) + } + p.RunFinish(context.Background(), pctx, Outcome{ + FinalAction: OutcomeDeny, + DenyingPlugin: "b", + }) + // LIFO over [a, b]: b then a. c, d absent. + if got, want := seen, []string{"b", "a"}; !equal(got, want) { + t.Errorf("finish plugins = %v, want %v (only dispatched, LIFO)", got, want) + } +} + +// TestRunFinish_OutcomeVisible verifies pctx.Outcome() returns the +// supplied outcome during OnFinish. +func TestRunFinish_OutcomeVisible(t *testing.T) { + var gotAction OutcomeAction + var gotStatus int + var gotDenier string + var gotDurNonZero bool + f := newFinisher("f", func(_ context.Context, pctx *Context) { + o := pctx.Outcome() + if o == nil { + t.Fatal("pctx.Outcome() nil during OnFinish") + } + gotAction = o.FinalAction + gotStatus = o.StatusCode + gotDenier = o.DenyingPlugin + gotDurNonZero = o.Duration > 0 + }) + p, err := New([]Plugin{f}) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{StartedAt: time.Now().Add(-10 * time.Millisecond)} + p.Run(context.Background(), pctx) + p.RunFinish(context.Background(), pctx, Outcome{ + FinalAction: OutcomeDeny, + StatusCode: 503, + DenyingPlugin: "upstream-gate", + }) + if gotAction != OutcomeDeny { + t.Errorf("FinalAction = %q, want deny", gotAction) + } + if gotStatus != 503 { + t.Errorf("StatusCode = %d, want 503", gotStatus) + } + if gotDenier != "upstream-gate" { + t.Errorf("DenyingPlugin = %q, want upstream-gate", gotDenier) + } + if !gotDurNonZero { + t.Error("Duration should have been auto-derived from StartedAt and non-zero") + } +} + +// TestRunFinish_OutcomeNilOutsideFinish verifies the "nil everywhere +// except OnFinish" contract for pctx.Outcome(). +func TestRunFinish_OutcomeNilOutsideFinish(t *testing.T) { + var seenInRequest *Outcome + var seenInResponse *Outcome + var seenInFinish *Outcome + f := newFinisher("f", func(_ context.Context, pctx *Context) { + seenInFinish = pctx.Outcome() + }) + f.stubPlugin.onReq = func(_ context.Context, pctx *Context) Action { + seenInRequest = pctx.Outcome() + return Action{Type: Continue} + } + f.stubPlugin.onResp = func(_ context.Context, pctx *Context) Action { + seenInResponse = pctx.Outcome() + return Action{Type: Continue} + } + p, err := New([]Plugin{f}) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + p.Run(context.Background(), pctx) + p.RunResponse(context.Background(), pctx) + p.RunFinish(context.Background(), pctx, Outcome{FinalAction: OutcomeAllow}) + if seenInRequest != nil { + t.Errorf("pctx.Outcome() in OnRequest = %v, want nil", seenInRequest) + } + if seenInResponse != nil { + t.Errorf("pctx.Outcome() in OnResponse = %v, want nil", seenInResponse) + } + if seenInFinish == nil { + t.Error("pctx.Outcome() in OnFinish = nil, want non-nil") + } + // And cleared after RunFinish returns: + if pctx.Outcome() != nil { + t.Error("pctx.Outcome() after RunFinish returned != nil") + } +} + +// TestRunFinish_PanicRecovered verifies a panicking OnFinish doesn't +// prevent later plugins in the LIFO chain from running. +func TestRunFinish_PanicRecovered(t *testing.T) { + var aRan, cRan atomic.Bool + a := newFinisher("a", func(_ context.Context, _ *Context) { + aRan.Store(true) + }) + b := newFinisher("b", func(_ context.Context, _ *Context) { + panic("b is a bug") + }) + c := newFinisher("c", func(_ context.Context, _ *Context) { + cRan.Store(true) + }) + p, err := New([]Plugin{a, b, c}) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + p.Run(context.Background(), pctx) + p.RunFinish(context.Background(), pctx, Outcome{FinalAction: OutcomeAllow}) + if !cRan.Load() { + t.Error("plugin c (first in LIFO) should have run, didn't") + } + if !aRan.Load() { + t.Error("plugin a (after panicking b) should have run, didn't") + } +} + +// TestRunFinish_FreshContextNotCancelled verifies OnFinish receives a +// ctx that is NOT derived from a cancelled request ctx. The request +// ctx is cancelled (simulating client disconnect); the framework must +// still dispatch OnFinish with a live ctx so plugin I/O doesn't fail +// immediately. +func TestRunFinish_FreshContextNotCancelled(t *testing.T) { + var ctxErrAtFinish error + f := newFinisher("f", func(ctx context.Context, _ *Context) { + ctxErrAtFinish = ctx.Err() + }) + p, err := New([]Plugin{f}) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + + // Simulate client disconnect by pre-cancelling the request ctx. + reqCtx, cancel := context.WithCancel(context.Background()) + cancel() + + // Run Run with a fresh ctx so the pipeline doesn't short-circuit + // on ctx.Err() before even dispatching — that's a separate path. + p.Run(context.Background(), pctx) + + // But RunFinish is called with the (cancelled) request ctx. The + // fresh-ctx contract means OnFinish's ctx is NOT this cancelled one. + p.RunFinish(reqCtx, pctx, Outcome{FinalAction: OutcomeAllow}) + + if ctxErrAtFinish != nil { + t.Errorf("OnFinish ctx.Err() = %v, want nil (framework must supply a fresh ctx)", ctxErrAtFinish) + } +} + +// TestRunFinish_FreshContextHasDeadline verifies the fresh ctx carries +// the framework-set finish timeout so a hung plugin is eventually +// unblocked. +func TestRunFinish_FreshContextHasDeadline(t *testing.T) { + var hadDeadline bool + var deadlineRemaining time.Duration + f := newFinisher("f", func(ctx context.Context, _ *Context) { + d, ok := ctx.Deadline() + hadDeadline = ok + if ok { + deadlineRemaining = time.Until(d) + } + }) + // Pick an override small enough to see but large enough to reliably + // be positive-remaining when OnFinish reads it. + p, err := New([]Plugin{f}, WithFinishTimeout(250*time.Millisecond)) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + p.Run(context.Background(), pctx) + p.RunFinish(context.Background(), pctx, Outcome{FinalAction: OutcomeAllow}) + if !hadDeadline { + t.Fatal("OnFinish ctx had no deadline; expected framework-set timeout") + } + if deadlineRemaining <= 0 || deadlineRemaining > 250*time.Millisecond { + t.Errorf("deadline remaining = %v, want in (0, 250ms]", deadlineRemaining) + } +} + +// TestRunFinish_DefaultTimeout verifies WithFinishTimeout defaults to +// DefaultFinishTimeout when not supplied or when supplied as zero. +func TestRunFinish_DefaultTimeout(t *testing.T) { + p1, err := New(nil) + if err != nil { + t.Fatalf("New: %v", err) + } + if p1.finishTimeout != DefaultFinishTimeout { + t.Errorf("unset finishTimeout = %v, want default %v", p1.finishTimeout, DefaultFinishTimeout) + } + p2, err := New(nil, WithFinishTimeout(0)) + if err != nil { + t.Fatalf("New: %v", err) + } + if p2.finishTimeout != DefaultFinishTimeout { + t.Errorf("explicit 0 finishTimeout = %v, want default %v", p2.finishTimeout, DefaultFinishTimeout) + } +} + +// TestRunFinish_OffPolicyPluginNotDispatched verifies a plugin wired +// with ErrorPolicyOff is not invoked by OnRequest, therefore not in +// pctx.dispatched, therefore its OnFinish never runs either. +func TestRunFinish_OffPolicyPluginNotDispatched(t *testing.T) { + var offRan atomic.Bool + off := newFinisher("off-plugin", func(_ context.Context, _ *Context) { + offRan.Store(true) + }) + on := newFinisher("on-plugin", nil) + p, err := New([]Plugin{off, on}, WithPolicies(ErrorPolicyOff, ErrorPolicyEnforce)) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + p.Run(context.Background(), pctx) + p.RunFinish(context.Background(), pctx, Outcome{FinalAction: OutcomeAllow}) + if offRan.Load() { + t.Error("off-policy plugin OnFinish ran; should have been skipped (never dispatched in Run)") + } +} + +// TestRunFinish_NonFinisherSkipped verifies plugins that ran OnRequest +// but don't implement Finisher are silently skipped — no panic, no +// effect. Typical mix: stateful plugin + stateless parser in same +// chain. +func TestRunFinish_NonFinisherSkipped(t *testing.T) { + var finisherRan atomic.Bool + parser := &stubPlugin{name: "parser"} // doesn't implement Finisher + f := newFinisher("stateful", func(_ context.Context, _ *Context) { + finisherRan.Store(true) + }) + p, err := New([]Plugin{parser, f}) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + p.Run(context.Background(), pctx) + p.RunFinish(context.Background(), pctx, Outcome{FinalAction: OutcomeAllow}) + if !finisherRan.Load() { + t.Error("Finisher-implementing plugin didn't run") + } +} + +// TestRunFinish_RecordDroppedDuringFinish verifies pctx.Record called +// from within OnFinish is a no-op (SessionEvent frozen), not a panic +// or a silent append. +func TestRunFinish_RecordDroppedDuringFinish(t *testing.T) { + f := newFinisher("f", func(_ context.Context, pctx *Context) { + pctx.Record(Invocation{Action: ActionObserve, Reason: "should_be_dropped"}) + }) + p, err := New([]Plugin{f}) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + p.Run(context.Background(), pctx) + p.RunFinish(context.Background(), pctx, Outcome{FinalAction: OutcomeAllow}) + // No Invocation should have landed from OnFinish. + if pctx.Extensions.Invocations != nil { + for _, inv := range pctx.Extensions.Invocations.Inbound { + if inv.Reason == "should_be_dropped" { + t.Errorf("OnFinish-emitted Invocation leaked to session (inbound): %+v", inv) + } + } + for _, inv := range pctx.Extensions.Invocations.Outbound { + if inv.Reason == "should_be_dropped" { + t.Errorf("OnFinish-emitted Invocation leaked to session (outbound): %+v", inv) + } + } + } +} + +// TestRunFinish_SetBodyDroppedDuringFinish verifies pctx.SetBody and +// pctx.SetResponseBody called from within OnFinish are no-ops (response +// already on the wire) — neither panic nor mutate pctx state. +func TestRunFinish_SetBodyDroppedDuringFinish(t *testing.T) { + f := newFinisher("f", func(_ context.Context, pctx *Context) { + pctx.SetBody([]byte("should-not-apply")) + pctx.SetResponseBody([]byte("also-not-apply")) + }) + p, err := New([]Plugin{f}) + if err != nil { + t.Fatalf("New: %v", err) + } + original := []byte("original-req") + originalResp := []byte("original-resp") + pctx := &Context{Body: original, ResponseBody: originalResp} + p.Run(context.Background(), pctx) + p.RunFinish(context.Background(), pctx, Outcome{FinalAction: OutcomeAllow}) + if string(pctx.Body) != "original-req" { + t.Errorf("Body mutated during OnFinish = %q, want unchanged", pctx.Body) + } + if string(pctx.ResponseBody) != "original-resp" { + t.Errorf("ResponseBody mutated during OnFinish = %q, want unchanged", pctx.ResponseBody) + } + if pctx.BodyMutated() { + t.Error("BodyMutated() true after OnFinish-only mutation; should have been suppressed") + } + if pctx.ResponseBodyMutated() { + t.Error("ResponseBodyMutated() true after OnFinish-only mutation; should have been suppressed") + } +} + +func equal(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} diff --git a/authbridge/authlib/pipeline/outcome.go b/authbridge/authlib/pipeline/outcome.go new file mode 100644 index 000000000..30800f8b6 --- /dev/null +++ b/authbridge/authlib/pipeline/outcome.go @@ -0,0 +1,77 @@ +package pipeline + +import "time" + +// OutcomeAction classifies the terminal state of a request. Intentionally +// a small 3-value vocabulary distinct from the 5-value InvocationAction: +// Outcome describes the request as a whole (for OnFinish accounting), +// while InvocationAction describes what one plugin did in one phase. +// +// A rate-limiter refunding a slot cares about "was this call +// successful" (OutcomeAllow) vs "did a plugin choose to deny" +// (OutcomeDeny) vs "did the upstream / framework fail" (OutcomeError) +// — three distinct accounting buckets. The 5-value InvocationAction +// (allow / deny / skip / modify / observe) doesn't answer that +// question because skip / modify / observe are mid-pipeline, not +// terminal states. +type OutcomeAction string + +const ( + // OutcomeAllow — every plugin returned Continue; response was + // produced and sent to the client. + OutcomeAllow OutcomeAction = "allow" + + // OutcomeDeny — a plugin (request-side or response-side) returned + // Reject. Outcome.DenyingPlugin names it. + OutcomeDeny OutcomeAction = "deny" + + // OutcomeError — the request terminated without a plugin denial: + // upstream transport failure, context cancellation, a panic + // recovered inside the dispatcher. DenyingPlugin is empty. + OutcomeError OutcomeAction = "error" +) + +// Outcome carries the terminal state of a request — what final action +// the pipeline took, the resulting HTTP status, which plugin denied (if +// any), and how long the request took end-to-end. Populated by the +// framework exactly once per request, immediately before dispatching +// OnFinish on any Finisher-implementing plugins. +// +// Read via pctx.Outcome(). The getter returns nil during OnRequest and +// OnResponse — plugins that accidentally inspect Outcome in those +// phases observe a nil pointer rather than a stale zero value, so the +// "this field is only meaningful in OnFinish" contract is enforced at +// read time rather than documented and forgotten. +// +// Outcome is deliberately a small struct. If a future need demands +// more context (upstream response headers, body sha256, external +// request ID, etc.) add a field here rather than inventing a parallel +// mechanism — plugins already reach for pctx.Outcome(). +type Outcome struct { + // FinalAction classifies the request as Allow / Deny / Error — + // three accounting buckets useful for per-outcome metrics and + // stateful-plugin cleanup. + FinalAction OutcomeAction + + // StatusCode is the final HTTP status written to the downstream + // client. Zero for errors that never produced a response. + StatusCode int + + // DenyingPlugin names the plugin whose Reject action stopped the + // pipeline. Empty when FinalAction != OutcomeDeny. + DenyingPlugin string + + // Duration is wall-clock time between pctx.StartedAt and the + // moment OnFinish dispatches (after the response is on the wire). + // Useful for per-outcome latency accounting. + Duration time.Duration +} + +// Outcome returns the terminal outcome of the request. Valid only +// during OnFinish — returns nil in OnRequest and OnResponse. Plugins +// implementing Finisher can rely on the return being non-nil; there is +// no path in the dispatcher that calls OnFinish with an unpopulated +// outcome. +func (c *Context) Outcome() *Outcome { + return c.outcome +} diff --git a/authbridge/authlib/pipeline/pipeline.go b/authbridge/authlib/pipeline/pipeline.go index 0ceec20dc..37a4f2e99 100644 --- a/authbridge/authlib/pipeline/pipeline.go +++ b/authbridge/authlib/pipeline/pipeline.go @@ -4,16 +4,28 @@ import ( "context" "fmt" "log/slog" + "time" ) +// DefaultFinishTimeout bounds how long each plugin's OnFinish may run. +// OnFinish commonly does network I/O (flush audits, release +// distributed leases) so the budget needs to be realistic, not +// minimal. 2s matches the order of magnitude of typical side-effect +// I/O without blocking the finish chain indefinitely when a plugin's +// sink hangs. Configurable per-listener via WithFinishTimeout; the +// per-plugin ctx is derived from context.Background() so client +// disconnect during the request never cancels OnFinish. +const DefaultFinishTimeout = 2 * time.Second + // Pipeline holds an ordered list of plugins and runs them sequentially. // policies[i] holds the on_error ErrorPolicy that wraps plugins[i]; the // slice is always the same length as plugins (guaranteed by New) so // policyAt is a bounds-safe lookup. An empty ErrorPolicy resolves to // ErrorPolicyEnforce via the Resolved() method. type Pipeline struct { - plugins []Plugin - policies []ErrorPolicy + plugins []Plugin + policies []ErrorPolicy + finishTimeout time.Duration } // defaultSlots lists the built-in extension slot names. @@ -30,8 +42,9 @@ var defaultSlots = map[string]bool{ type Option func(*options) type options struct { - extraSlots []string - policies []ErrorPolicy + extraSlots []string + policies []ErrorPolicy + finishTimeout time.Duration } // WithSlots registers additional valid extension slot names beyond the built-in set. @@ -42,6 +55,18 @@ func WithSlots(slots ...string) Option { } } +// WithFinishTimeout overrides the per-plugin OnFinish timeout. Each +// plugin's OnFinish runs under a fresh ctx derived from +// context.Background() with this timeout applied; a zero or negative +// value falls back to DefaultFinishTimeout. Listeners that know their +// deployment's OnFinish I/O patterns can tighten (fast local sinks) or +// relax (remote lease service) this knob. +func WithFinishTimeout(d time.Duration) Option { + return func(o *options) { + o.finishTimeout = d + } +} + // WithPolicies attaches per-plugin on_error policies in parallel with // the plugin slice passed to New. policies[i] belongs to plugins[i]; // an empty entry defaults to ErrorPolicyEnforce. If fewer policies are @@ -76,7 +101,11 @@ func New(plugins []Plugin, opts ...Option) (*Pipeline, error) { } policies := make([]ErrorPolicy, len(plugins)) copy(policies, o.policies) - return &Pipeline{plugins: plugins, policies: policies}, nil + finishTimeout := o.finishTimeout + if finishTimeout <= 0 { + finishTimeout = DefaultFinishTimeout + } + return &Pipeline{plugins: plugins, policies: policies, finishTimeout: finishTimeout}, nil } // Run executes the request phase of the pipeline sequentially. @@ -110,6 +139,7 @@ func (p *Pipeline) Run(ctx context.Context, pctx *Context) Action { return Deny("pipeline.cancelled", "request cancelled") } pctx.setCurrent(plugin.Name(), InvocationPhaseRequest, policy) + pctx.dispatched = append(pctx.dispatched, i) action := plugin.OnRequest(ctx, pctx) pctx.clearCurrent() if action.Type == Reject { @@ -383,6 +413,79 @@ func (p *Pipeline) Stop(ctx context.Context) { } } +// RunFinish dispatches the OnFinish hook on every Finisher-implementing +// plugin whose OnRequest was invoked during this request (Pipeline.Run +// tracks the dispatched set on pctx). Iteration is LIFO — reverse of +// OnRequest order — symmetric with Shutdowner and RunResponse. +// +// Called by the listener after the response has been written to the +// wire (or after the terminal error has been recorded, for denied or +// errored requests). Before the first plugin's OnFinish runs, the +// framework populates pctx.outcome from the supplied Outcome so +// pctx.Outcome() returns non-nil for the duration of the finish chain +// and nil everywhere else. +// +// Each plugin's OnFinish runs under a FRESH context derived from +// context.Background() with p.finishTimeout applied — not from the +// caller's ctx — so a client disconnect or earlier cancellation does +// not abort OnFinish's I/O. Callers pass a bare ctx for future parent- +// cancellation needs (process-wide Stop signal) but it is intentionally +// not wired into the per-plugin ctx. +// +// OnFinish is best-effort: a panicking plugin is recovered and logged, +// a returning plugin's errors (there is no error return on the +// interface by design — see Finisher godoc) are not observed by the +// framework. The LIFO chain continues regardless so one misbehaving +// plugin does not leak state in earlier plugins. +// +// RunFinish is safe to call at most once per request. Calling it twice +// on the same pctx would double-release state in every Finisher. +// Listeners MUST call it in a defer wrapping the response-produce +// block so a panic in response-writing still reaches cleanup. +func (p *Pipeline) RunFinish(_ context.Context, pctx *Context, outcome Outcome) { + if len(pctx.dispatched) == 0 { + return + } + // Derive Duration from pctx.StartedAt if the caller didn't set it. + if outcome.Duration == 0 && !pctx.StartedAt.IsZero() { + outcome.Duration = time.Since(pctx.StartedAt) + } + pctx.outcome = &outcome + defer func() { pctx.outcome = nil }() + + // LIFO over the dispatched indices. Skip the off-policy check: + // plugins configured on_error: off never have their OnRequest + // invoked so their index will not be in pctx.dispatched. + for i := len(pctx.dispatched) - 1; i >= 0; i-- { + idx := pctx.dispatched[i] + plugin := p.plugins[idx] + finisher, ok := plugin.(Finisher) + if !ok { + continue + } + p.dispatchFinish(plugin.Name(), finisher, pctx) + } +} + +// dispatchFinish runs OnFinish on one plugin under a fresh ctx with +// timeout, recovering any panic into a WARN log so later plugins in +// the LIFO chain still run. Isolated in its own method so the recover +// block's scope is exactly one plugin's dispatch. +func (p *Pipeline) dispatchFinish(name string, f Finisher, pctx *Context) { + ctx, cancel := context.WithTimeout(context.Background(), p.finishTimeout) + defer cancel() + defer func() { + if r := recover(); r != nil { + slog.Warn("pipeline: plugin OnFinish panicked", + "plugin", name, + "panic", r) + } + }() + pctx.inFinish = true + defer func() { pctx.inFinish = false }() + f.OnFinish(ctx, pctx) +} + // validateCapabilities checks that every slot a plugin reads has been written // by an earlier plugin in the chain, and applies the body-mutation rules: // - At most one WritesBody plugin per pipeline (direction-scoped). diff --git a/authbridge/authlib/pipeline/plugin.go b/authbridge/authlib/pipeline/plugin.go index cdcd71502..9428828ea 100644 --- a/authbridge/authlib/pipeline/plugin.go +++ b/authbridge/authlib/pipeline/plugin.go @@ -146,6 +146,58 @@ type Shutdowner interface { Shutdown(ctx context.Context) error } +// Finisher is an optional interface a plugin may implement when it +// reserves per-request state in OnRequest and needs a guaranteed +// release point — regardless of whether the request was allowed, +// denied by a later plugin, or errored at the upstream. The canonical +// shape is acquire-in-OnRequest / release-in-OnFinish: +// +// func (p *RateLimiter) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { +// tenant := pctx.Identity.ClientID() +// p.slots.Reserve(tenant) +// pipeline.SetState(pctx, "rl", &rlState{tenant: tenant}) +// return pipeline.Action{Type: pipeline.Continue} +// } +// +// func (p *RateLimiter) OnFinish(ctx context.Context, pctx *pipeline.Context) { +// s, ok := pipeline.GetState[*rlState](pctx, "rl") +// if !ok { return } +// p.slots.Release(s.tenant) +// } +// +// OnFinish runs once per request, after OnResponse has completed (if +// it ran), on every plugin whose OnRequest was dispatched — including +// the plugin that denied, if any. The dispatcher walks in LIFO order, +// symmetric with Shutdowner and OnResponse, so a plugin's cleanup can +// still rely on resources set up by earlier plugins. +// +// The ctx passed to OnFinish is a FRESH context with a framework-set +// deadline (default 2s). It is NOT derived from the original request +// ctx, so a client disconnect during the request does not cancel +// OnFinish's I/O. Plugins that perform network work (flushing audits, +// releasing distributed leases) see a usable ctx by default. +// +// pctx carries the full request + response state observed by +// OnResponse, plus pctx.Outcome() which returns a non-nil *Outcome +// describing the request's terminal outcome (allow / deny / error, +// status code, denying plugin, duration). pctx.Outcome() returns nil +// during OnRequest and OnResponse — the field is populated by the +// framework only before OnFinish dispatches. +// +// OnFinish runs best-effort: panics are recovered and logged, errors +// in one plugin's OnFinish do not prevent later plugins in the LIFO +// chain from running. OnFinish must not call pctx.SetBody / +// SetResponseBody — the response is already on the wire; mutations +// are dropped with a WARN log. +// +// OnFinish emits no automatic Invocation records. Plugins that want +// observability on cleanup publish through their own sink +// (Prometheus, external audit service) or via the pctx.Extensions.Custom +// escape-hatch map documented in plugin-reference.md. +type Finisher interface { + OnFinish(ctx context.Context, pctx *Context) +} + // Readier is an optional interface a plugin may implement when it has // deferred initialization that matters to a /readyz probe. The host // ANDs Ready() across all implementers to decide whether the pipeline diff --git a/authbridge/docs/framework-architecture.md b/authbridge/docs/framework-architecture.md index 15d3f5e75..c2391de56 100644 --- a/authbridge/docs/framework-architecture.md +++ b/authbridge/docs/framework-architecture.md @@ -376,12 +376,13 @@ There is no "soft error" channel today — a plugin that wants to fail open logs ```go func New(plugins []Plugin, opts ...Option) (*Pipeline, error) -func (p *Pipeline) Run(ctx context.Context, pctx *Context) Action // request phase -func (p *Pipeline) RunResponse(ctx context.Context, pctx *Context) Action // response phase (reverse) -func (p *Pipeline) Start(ctx context.Context) error // invoke Init on Initializer plugins -func (p *Pipeline) Stop(ctx context.Context) // invoke Shutdown on Shutdowner plugins -func (p *Pipeline) Plugins() []Plugin // defensive copy -func (p *Pipeline) NeedsBody() bool // OR over all plugins' BodyAccess +func (p *Pipeline) Run(ctx context.Context, pctx *Context) Action // request phase +func (p *Pipeline) RunResponse(ctx context.Context, pctx *Context) Action // response phase (reverse) +func (p *Pipeline) RunFinish(ctx context.Context, pctx *Context, outcome Outcome) // finish phase (reverse) +func (p *Pipeline) Start(ctx context.Context) error // invoke Init on Initializer plugins +func (p *Pipeline) Stop(ctx context.Context) // invoke Shutdown on Shutdowner plugins +func (p *Pipeline) Plugins() []Plugin // defensive copy +func (p *Pipeline) NeedsBody() bool // OR over all plugins' BodyAccess ``` `New` validates capability wiring at startup: every `Read` must be satisfied by some earlier plugin's `Write`. `plugins.Build` additionally validates the cross-plugin relationship declarations — `Requires`, `RequiresAny`, `After`, `Claims` — before returning the pipeline to the listener. See [`plugin-reference.md` "Declaring plugin relationships"](./plugin-reference.md#declaring-plugin-relationships). @@ -466,6 +467,64 @@ func (p *RateLimiter) OnResponse(context.Context, *pipeline.Context) pipeline.Ac } ``` +### Per-request finish hook (`Finisher`) + +Plugins that reserve per-request state in `OnRequest` need a guaranteed release point — whether the request was allowed, denied by a later plugin, or errored at the upstream. Without such a hook, a rate-limiter that reserves a slot in `OnRequest` and releases it in `OnResponse` silently leaks the slot whenever a later plugin denies (because `OnResponse` is only walked when the pipeline returned `Continue`). The `Finisher` optional interface closes this: + +```go +type Finisher interface { + OnFinish(ctx context.Context, pctx *Context) +} +``` + +`OnFinish` fires **once per request**, after `OnResponse` if it ran, on every plugin whose `OnRequest` was actually invoked — including the plugin that denied (if any). Dispatch is LIFO, symmetric with `Shutdowner` and `RunResponse`, so a plugin's cleanup can still depend on earlier plugins' resources. The listener calls `Pipeline.RunFinish` in a `defer` wrapping the response-produce block: + +```go +pctx := newContext(...) +defer func() { + pipeline.RunFinish(ctx, pctx, pipeline.Outcome{ + FinalAction: finalAction, // OutcomeAllow | OutcomeDeny | OutcomeError + StatusCode: statusCode, + DenyingPlugin: denyingPlugin, // "" unless FinalAction == OutcomeDeny + }) +}() +// ... Run, write response, RunResponse ... +``` + +Key contract points, chosen to make the plugin-author surface small and hard to misuse: + +- **Fresh ctx with a framework-set deadline.** `OnFinish` does not receive the request `ctx` — that one may be cancelled (client disconnect) by the time the response is on the wire. The framework derives a fresh ctx from `context.Background()` with `DefaultFinishTimeout` (2s) applied, overridable via `WithFinishTimeout`. Plugins doing network I/O (flushing audits, releasing a distributed lease) see a live ctx by default. +- **`pctx.Outcome()` returns non-nil only during OnFinish.** The `*Outcome` carries `FinalAction` (three values: `OutcomeAllow` / `OutcomeDeny` / `OutcomeError`), `StatusCode`, `DenyingPlugin`, and `Duration`. In `OnRequest` / `OnResponse` the getter returns nil, so the "only meaningful in OnFinish" contract is enforced at read time rather than via documented zero-value semantics. +- **Full pctx is available**, including response body — the framework buffers through the finish window so audit / metrics plugins can read the final payload. +- **Best-effort dispatch.** A panic is recovered and logged; later plugins in the LIFO chain still run. One misbehaving plugin does not leak state in every other plugin. +- **OnFinish is silent.** The framework does NOT auto-emit `Invocation` records for this phase. `pctx.Record` called from within `OnFinish` is dropped with a WARN log (SessionEvent is frozen once the response is published). Plugins that want observability on cleanup publish through their own sink (Prometheus, external audit service) or the `pctx.Extensions.Custom` escape-hatch map. +- **Body mutation is forbidden.** `pctx.SetBody` / `SetResponseBody` called from `OnFinish` are dropped with a WARN log — the response is already on the wire. +- **`on_error: off` plugins are never dispatched in any phase**, so their `OnFinish` never runs either (consistent with the "only plugins whose OnRequest actually ran" participation rule). + +Canonical rate-limiter: + +```go +type rlState struct{ tenant string } + +func (p *RateLimiter) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { + tenant := pctx.Identity.ClientID() + p.slots.Reserve(tenant) + pipeline.SetState(pctx, "rate-limiter", &rlState{tenant: tenant}) + return pipeline.Action{Type: pipeline.Continue} +} + +func (p *RateLimiter) OnFinish(ctx context.Context, pctx *pipeline.Context) { + s, ok := pipeline.GetState[*rlState](pctx, "rate-limiter") + if !ok { + return + } + p.slots.Release(s.tenant) + p.metrics.RecordOutcome(pctx.Outcome().FinalAction) // allow / deny / error buckets +} +``` + +Contrast with the pre-Finisher workaround: reserve in `OnRequest`, release in `OnResponse` — leaks on every denied request. Or: reserve in `OnRequest`, spawn a goroutine with a TTL-based release — leaks goroutines on process shutdown and runs on a timer, not on the actual response. + ### Extension slots known to the validator Built-in: `mcp`, `a2a`, `security`, `delegation`, `inference`, `custom`. @@ -680,6 +739,7 @@ The plugin interface is **not** semver-stable yet (AuthBridge is pre-1.0). Chang - **Detyped framework**: `pipeline/` no longer imports plugin-specific packages. **Breaking**: `Context.Claims *validation.Claims` → `Context.Identity Identity` (interface with `Subject()`/`ClientID()`/`Scopes()`); plugins publish adapters. `Context.Route` removed (was dead code). `Invocation`'s nine jwt-validation + token-exchange specific fields (`ExpectedIssuer`, `TokenSubject`, `RouteHost`, `CacheHit`, etc.) collapsed into `Details map[string]string`; built-in plugins migrated to `Details["expected_issuer"]` etc. `SessionEvent.TargetAudience` removed (was only populated from dead `pctx.Route`). Third-party plugins get a clean diagnostic slot they can populate without framework edits. - **Single-owner packages relocated**: `authlib/validation` → `authlib/plugins/jwtvalidation/validation`. `authlib/exchange` / `authlib/cache` / `authlib/spiffe` → `authlib/plugins/tokenexchange/{exchange,cache,spiffe}`. Each plugin now lives in its own directory (`plugins/jwtvalidation/plugin.go`, `plugins/tokenexchange/plugin.go`) and self-registers via its own init(). `authlib/bypass`, `authlib/routing`, `authlib/auth` stay shared. - **Plugin relationship declarations**: `PluginCapabilities` extended with four chain-scoped fields — `Requires` (all-must-be-earlier), `RequiresAny` (at-least-one-earlier), `After` (soft ordering), `Claims` (mutex on a semantic resource). Validated at `plugins.Build` time (startup + hot-reload); all errors per chain are collected into one report. `authlib/contracts/claims.go` ships `ClaimAuthorizationHeader` as the initial canonical claim constant. `token-exchange` and `token-broker` migrated to declare it, so configuring both on the same outbound chain now fails startup instead of silently clobbering each other's Authorization header. See [`plugin-reference.md` "Declaring plugin relationships"](./plugin-reference.md#declaring-plugin-relationships). +- **Per-request finish hook (`Finisher`)**: new optional interface `Finisher { OnFinish(ctx, pctx) }` gives stateful plugins a guaranteed release point that fires after every request end — regardless of whether the pipeline allowed, denied, or errored. Dispatch is LIFO across Finisher-implementing plugins whose OnRequest was invoked (including the denier). `pctx.Outcome()` returns a non-nil `*Outcome{FinalAction, StatusCode, DenyingPlugin, Duration}` during OnFinish and nil in all other phases. OnFinish runs under a fresh ctx derived from `context.Background()` with a framework-set deadline (default 2s, `WithFinishTimeout` overrides) so client disconnect during the request doesn't cancel cleanup I/O. Best-effort: panics are recovered; OnFinish is silent (no auto-emitted Invocations); `pctx.Record` / `SetBody` / `SetResponseBody` called during OnFinish are dropped with WARN logs since the SessionEvent is published and the response is on the wire. Closes the rate-limiter / audit / lease class of "cleanup leaks on denial" bugs. See §6 "Per-request finish hook". Breaking changes will be announced in `authbridge/CHANGELOG.md` (TBD) before a 1.0 tag. @@ -696,7 +756,8 @@ Breaking changes will be announced in `authbridge/CHANGELOG.md` (TBD) before a 1 - `pipeline.go` — `Pipeline` type, `New`, `Run`, `RunResponse`, `Start`, `Stop`, `Plugins`, `NeedsBody`, `WritesBody`. - `holder.go` — `Holder`, the atomic slot listeners hold in place of a raw `*Pipeline`. -- `plugin.go` — `Plugin` interface, `PluginCapabilities` (with `ReadsBody` / `WritesBody` / deprecated `BodyAccess` + `Normalize()`; chain-scoped relationship fields `Requires` / `RequiresAny` / `After` / `Claims`), `Configurable`, `Initializer`, `Shutdowner`, `Readier`. +- `plugin.go` — `Plugin` interface, `PluginCapabilities` (with `ReadsBody` / `WritesBody` / deprecated `BodyAccess` + `Normalize()`; chain-scoped relationship fields `Requires` / `RequiresAny` / `After` / `Claims`), `Configurable`, `Initializer`, `Shutdowner`, `Readier`, `Finisher`. +- `outcome.go` — `Outcome` struct + `OutcomeAction` (allow / deny / error) for `Finisher` consumers; `Context.Outcome()` getter. - `action.go` — `Action`, `ActionType`, `Violation`, helper constructors (`Deny`, `DenyStatus`, `DenyWithDetails`, `Challenge`, `RateLimited`), `StatusFromCode`. - `context.go` — `Context`, `Direction`, `AgentIdentity`, the `pctx.Record` / `Allow` / `Skip` / `Observe` / `Modify` / `DenyAndRecord` helpers, and `pctx.SetBody` / `SetResponseBody` / `BodyMutated` / `ResponseBodyMutated` for body mutation. - `extensions.go` — `Extensions` struct, `Invocation`, `Invocations`, `InvocationAction`, named protocol extensions, `GetState` / `SetState`. diff --git a/authbridge/docs/plugin-reference.md b/authbridge/docs/plugin-reference.md index eee3e1a02..9a7501e76 100644 --- a/authbridge/docs/plugin-reference.md +++ b/authbridge/docs/plugin-reference.md @@ -740,6 +740,99 @@ session store is unauthenticated. Plugin-private debug logs may include body bytes at DEBUG level, but never publish them to the session stream or Custom map. +## Finishing requests (stateful plugins) + +Plugins that **reserve per-request state** in `OnRequest` — rate-limiter +slots, distributed leases, audit "request started" events, in-flight +trace spans — need a guaranteed release point, regardless of whether +the request was allowed, denied by a later plugin, or errored at the +upstream. Releasing in `OnResponse` alone is a trap: `OnResponse` is +only walked for plugins on a `Continue` pipeline; a deny from a later +plugin leaves earlier plugins' state leaked. + +The `Finisher` optional interface closes this: + +```go +type Finisher interface { + OnFinish(ctx context.Context, pctx *pipeline.Context) +} +``` + +`OnFinish` fires **once per request, after `OnResponse` if it ran**, on +every Finisher-implementing plugin whose `OnRequest` was actually +invoked — including the plugin that denied, if any. Stateless plugins +don't implement the interface and see nothing new; the framework skips +them. + +> The full contract (dispatch order, context rules, error handling, +> the silent-phase invariants) lives in +> [`framework-architecture.md` §6 "Per-request finish hook"](./framework-architecture.md#per-request-finish-hook-finisher). +> This section is the plugin-author usage surface. + +### Canonical shape: acquire in OnRequest, release in OnFinish + +```go +type rlState struct{ tenant string } + +func (p *RateLimiter) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { + tenant := pctx.Identity.ClientID() + p.slots.Reserve(tenant) + pipeline.SetState(pctx, "rate-limiter", &rlState{tenant: tenant}) + return pipeline.Action{Type: pipeline.Continue} +} + +func (p *RateLimiter) OnFinish(ctx context.Context, pctx *pipeline.Context) { + s, ok := pipeline.GetState[*rlState](pctx, "rate-limiter") + if !ok { // OnRequest didn't reach the reservation; nothing to release + return + } + p.slots.Release(s.tenant) + p.metrics.RecordOutcome(pctx.Outcome().FinalAction) +} +``` + +`pipeline.SetState` / `GetState` is the typed cross-phase state API — it +keeps the state private to one plugin (unlike `Extensions.Custom`, +which is shared). The key convention is the plugin's `Name()`. + +### Reading the outcome + +`pctx.Outcome()` returns non-nil **only during OnFinish**: + +```go +type Outcome struct { + FinalAction OutcomeAction // OutcomeAllow | OutcomeDeny | OutcomeError + StatusCode int // final HTTP status written downstream + DenyingPlugin string // name of rejecting plugin, "" if allowed + Duration time.Duration // wall-clock request duration +} +``` + +- `OutcomeAllow` — pipeline returned Continue end-to-end; response produced. +- `OutcomeDeny` — a plugin (request-side or response-side) denied. `DenyingPlugin` names it. +- `OutcomeError` — non-deny termination: upstream failure, framework panic, context cancellation. `DenyingPlugin` is empty. + +Plugins that don't care about the outcome just ignore it — `OnFinish` +still fires and state still gets released. + +### Rules you can't break + +- **Don't call `pctx.SetBody` / `pctx.SetResponseBody` in OnFinish.** The response is already on the wire. Both calls are dropped with a WARN log. +- **Don't call `pctx.Record` or its `Allow` / `Skip` / `Observe` / `Modify` / `DenyAndRecord` siblings in OnFinish.** The SessionEvent is frozen once OnFinish starts. Recorded Invocations are dropped with a WARN log. +- **Don't spawn goroutines from OnFinish that outlive it.** Use `Initializer` / `Shutdowner` for process-lifetime background work. OnFinish-scoped goroutines leak under `context.WithTimeout`-based cleanup semantics. +- **Do use the `ctx` the framework provides** — it's a fresh `context.Background()`-derived ctx with a ~2s deadline, specifically so client disconnect during the request doesn't cancel your I/O. Don't reach for pctx-carried cancellation tokens. + +### Publishing observability from OnFinish + +OnFinish doesn't auto-emit Invocations. Plugins that want per-request +cleanup telemetry go to their own sinks: + +- **Prometheus / OTEL metrics** — fire-and-forget counters and histograms. Don't need session-event plumbing. +- **External audit service** — POST the event over HTTP using the ctx the framework supplies. +- **`pctx.Extensions.Custom["my-plugin/event"]` (escape hatch)** — NO. The SessionEvent is already published once OnFinish runs; writes to Custom after that are visible to nothing. + +If you find yourself wanting OnFinish-phase Invocations in session events, the design choice behind OnFinish being silent (explicit in the [versioning entry](./framework-architecture.md#12-versioning)) is worth re-reading before asking for a framework change. + ## Exposing content to guardrails Parser plugins whose extensions carry user-visible text (message bodies, From d723827c8368c39c3ee4439b0e30a46b7e065654 Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Mon, 11 May 2026 22:46:59 -0400 Subject: [PATCH 2/3] feat(listener): Wire RunFinish in all four listeners MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the missing half of the Finisher contract from 4306fbd: until the listeners call Pipeline.RunFinish on each request's exit path, the hook is dormant (no plugin's OnFinish can ever fire). This commit threads the dispatch through all four listener implementations and adds one end-to-end integration test. Design keeps the per-listener footprint minimal: authlib/pipeline/outcome.go OutcomeFromContext(pctx) derives a best-effort Outcome from pctx state: deny Invocation wins, else StatusCode > 0 = Allow, else Error. Each listener is a one-liner (defer RunFinish + OutcomeFromContext). authlib/pipeline/holder.go Holder.RunFinish delegate so listeners that hold *Holder (all of them, post-reloader) can call it without an .Load() indirection. authlib/pipeline/finisher_test.go TestOutcomeFromContext covering allow / error / deny (inbound, outbound-wins-over-inbound, shadow-ignored) cases. Per-listener wiring: reverseproxy/server.go defer in handleRequest after pctx is constructed. Populates StartedAt so Outcome.Duration is derived. forwardproxy/server.go same pattern as reverseproxy. extauthz/server.go two defers — inbound first (early, to fire even if inbound denies), outbound second (only if inbound allowed). Explicit StatusCode = 200 on allow paths because ext_authz has no HTTP status concept and OutcomeFromContext defaults StatusCode 0 to OutcomeError. extproc/server.go one defer at the top of Process() that reads pctx + requestDirection from the stream's local state. pctx may be nil if the stream never reached Run (early Recv error); skip in that case. Integration test: reverseproxy/finisher_test.go Allow path: OnFinish fires once with OutcomeAllow + StatusCode 200. Deny path: pipeline [before, denier, after]. OnFinish fires on before AND denier with OutcomeDeny + DenyingPlugin="denier"; after's OnFinish does NOT fire (its OnRequest never ran, so it's not in pctx.dispatched). Risk assessment: none of the in-tree plugins implement Finisher, so RunFinish walks an empty Finisher subset of pctx.dispatched and returns. Behavior for existing chains is unchanged. The only new cost is the deferred function call per request (one atomic Load from Holder + the no-op RunFinish body). Full authlib + cmd/authbridge test suites pass under -race. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- authbridge/authlib/pipeline/finisher_test.go | 88 ++++++++++ authbridge/authlib/pipeline/holder.go | 9 + authbridge/authlib/pipeline/outcome.go | 62 +++++++ .../authbridge/listener/extauthz/server.go | 22 +++ .../cmd/authbridge/listener/extproc/server.go | 16 ++ .../listener/forwardproxy/server.go | 9 + .../listener/reverseproxy/finisher_test.go | 160 ++++++++++++++++++ .../listener/reverseproxy/server.go | 10 ++ 8 files changed, 376 insertions(+) create mode 100644 authbridge/cmd/authbridge/listener/reverseproxy/finisher_test.go diff --git a/authbridge/authlib/pipeline/finisher_test.go b/authbridge/authlib/pipeline/finisher_test.go index 1f316e8da..e94b822e4 100644 --- a/authbridge/authlib/pipeline/finisher_test.go +++ b/authbridge/authlib/pipeline/finisher_test.go @@ -397,6 +397,94 @@ func TestRunFinish_SetBodyDroppedDuringFinish(t *testing.T) { } } +// TestOutcomeFromContext covers the listener-facing derivation helper: +// deny Invocations win, StatusCode 0 means Error, non-zero + no deny +// means Allow. Shadow denials don't count. +func TestOutcomeFromContext(t *testing.T) { + tests := []struct { + name string + setup func(*Context) + wantAct OutcomeAction + wantDeny string + }{ + { + name: "no invocations, status 200 → allow", + setup: func(pctx *Context) { + pctx.StatusCode = 200 + }, + wantAct: OutcomeAllow, + }, + { + name: "no invocations, status 0 → error", + setup: func(pctx *Context) { + pctx.StatusCode = 0 + }, + wantAct: OutcomeError, + }, + { + name: "no invocations, upstream 502 → allow (pipeline didn't deny)", + setup: func(pctx *Context) { + pctx.StatusCode = 502 + }, + wantAct: OutcomeAllow, + }, + { + name: "inbound deny invocation → deny + plugin name", + setup: func(pctx *Context) { + pctx.StatusCode = 401 + pctx.Extensions.Invocations = &Invocations{ + Inbound: []Invocation{ + {Plugin: "allow-plugin", Action: ActionAllow}, + {Plugin: "jwt-validation", Action: ActionDeny, Reason: "bad_token"}, + }, + } + }, + wantAct: OutcomeDeny, + wantDeny: "jwt-validation", + }, + { + name: "outbound deny wins over inbound deny (most recent)", + setup: func(pctx *Context) { + pctx.StatusCode = 503 + pctx.Extensions.Invocations = &Invocations{ + Inbound: []Invocation{{Plugin: "inbound-gate", Action: ActionDeny}}, + Outbound: []Invocation{{Plugin: "token-exchange", Action: ActionDeny, Reason: "idp_down"}}, + } + }, + wantAct: OutcomeDeny, + wantDeny: "token-exchange", + }, + { + name: "shadow deny ignored (observe mode — not an actual deny)", + setup: func(pctx *Context) { + pctx.StatusCode = 200 + pctx.Extensions.Invocations = &Invocations{ + Inbound: []Invocation{ + {Plugin: "canary", Action: ActionDeny, Shadow: true}, + }, + } + }, + wantAct: OutcomeAllow, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + pctx := &Context{} + tc.setup(pctx) + out := OutcomeFromContext(pctx) + if out.FinalAction != tc.wantAct { + t.Errorf("FinalAction = %q, want %q", out.FinalAction, tc.wantAct) + } + if out.DenyingPlugin != tc.wantDeny { + t.Errorf("DenyingPlugin = %q, want %q", out.DenyingPlugin, tc.wantDeny) + } + if out.StatusCode != pctx.StatusCode { + t.Errorf("StatusCode = %d, want %d", out.StatusCode, pctx.StatusCode) + } + }) + } +} + func equal(a, b []string) bool { if len(a) != len(b) { return false diff --git a/authbridge/authlib/pipeline/holder.go b/authbridge/authlib/pipeline/holder.go index 95639e926..2ac8df525 100644 --- a/authbridge/authlib/pipeline/holder.go +++ b/authbridge/authlib/pipeline/holder.go @@ -56,6 +56,15 @@ func (h *Holder) RunResponse(ctx context.Context, pctx *Context) Action { return h.p.Load().RunResponse(ctx, pctx) } +// RunFinish is equivalent to h.Load().RunFinish(ctx, pctx, outcome). +// Listeners call this in a defer at request entry to guarantee +// Finisher dispatch on every exit path. See Pipeline.RunFinish for +// the contract (LIFO over dispatched plugins, fresh ctx with timeout, +// pctx.Outcome() populated, silent phase). +func (h *Holder) RunFinish(ctx context.Context, pctx *Context, outcome Outcome) { + h.p.Load().RunFinish(ctx, pctx, outcome) +} + // NeedsBody is equivalent to h.Load().NeedsBody(). Hot path on listeners // that decide whether to buffer the request/response body. func (h *Holder) NeedsBody() bool { return h.p.Load().NeedsBody() } diff --git a/authbridge/authlib/pipeline/outcome.go b/authbridge/authlib/pipeline/outcome.go index 30800f8b6..703d506d2 100644 --- a/authbridge/authlib/pipeline/outcome.go +++ b/authbridge/authlib/pipeline/outcome.go @@ -75,3 +75,65 @@ type Outcome struct { func (c *Context) Outcome() *Outcome { return c.outcome } + +// OutcomeFromContext derives a best-effort Outcome from a pctx's final +// state, intended for listeners that want a one-liner finish call +// without threading outcome state through nested response / error +// callbacks. The derivation rules: +// +// - A deny Invocation on either phase → OutcomeDeny, DenyingPlugin = +// the most recent deny's Plugin name. Most-recent rather than +// first so a response-side deny (e.g. an output filter) is +// correctly attributed over any earlier record. +// - No deny Invocation AND StatusCode > 0 → OutcomeAllow. The HTTP +// status itself is not sufficient to classify error vs allow — a +// legitimate 500 from the upstream is still a pipeline Allow. +// - No deny Invocation AND StatusCode == 0 → OutcomeError (no +// response was written: upstream transport failure, listener +// panic, etc.). +// +// Listeners with more precise information at hand (a typed error from +// the upstream transport, a listener-level reject distinct from any +// plugin deny) should construct Outcome explicitly rather than call +// this helper. StatusCode is taken from pctx.StatusCode verbatim. +// Duration is left zero; RunFinish auto-fills from pctx.StartedAt +// when left unset. +func OutcomeFromContext(pctx *Context) Outcome { + out := Outcome{StatusCode: pctx.StatusCode} + if denier, ok := lastDenyingPlugin(pctx); ok { + out.FinalAction = OutcomeDeny + out.DenyingPlugin = denier + return out + } + if pctx.StatusCode == 0 { + out.FinalAction = OutcomeError + return out + } + out.FinalAction = OutcomeAllow + return out +} + +// lastDenyingPlugin walks pctx.Extensions.Invocations in reverse (both +// directions) looking for the most-recent deny-action record. Returns +// the plugin name and true if found; "", false otherwise. +func lastDenyingPlugin(pctx *Context) (string, bool) { + if pctx.Extensions.Invocations == nil { + return "", false + } + // Check outbound first, then inbound — outbound Invocations are + // produced on a chain that runs after inbound on dual-listener + // deployments, so "most recent" under wall-clock is outbound. + for i := len(pctx.Extensions.Invocations.Outbound) - 1; i >= 0; i-- { + inv := pctx.Extensions.Invocations.Outbound[i] + if inv.Action == ActionDeny && !inv.Shadow { + return inv.Plugin, true + } + } + for i := len(pctx.Extensions.Invocations.Inbound) - 1; i >= 0; i-- { + inv := pctx.Extensions.Invocations.Inbound[i] + if inv.Action == ActionDeny && !inv.Shadow { + return inv.Plugin, true + } + } + return "", false +} diff --git a/authbridge/cmd/authbridge/listener/extauthz/server.go b/authbridge/cmd/authbridge/listener/extauthz/server.go index af43b405b..fbf16724d 100644 --- a/authbridge/cmd/authbridge/listener/extauthz/server.go +++ b/authbridge/cmd/authbridge/listener/extauthz/server.go @@ -7,6 +7,7 @@ import ( "context" "net/http" "strings" + "time" corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" authv3 "github.com/envoyproxy/go-control-plane/envoy/service/auth/v3" @@ -53,7 +54,15 @@ func (s *Server) Check(ctx context.Context, req *authv3.CheckRequest) (*authv3.C Host: host, Path: path, Headers: mapToHTTPHeader(headers), + StartedAt: time.Now(), } + // Finisher dispatch for inbound. Deferred before Run so the hook + // fires whether the pipeline allows, denies, or the Check returns + // early. Outbound's defer (added after the outbound pctx is + // created) runs first under LIFO. + defer func() { + s.InboundPipeline.RunFinish(ctx, inPctx, pipeline.OutcomeFromContext(inPctx)) + }() inAction := s.InboundPipeline.Run(ctx, inPctx) if inAction.Type == pipeline.Reject { return deniedFromAction(codes.Unauthenticated, inAction), nil @@ -66,13 +75,26 @@ func (s *Server) Check(ctx context.Context, req *authv3.CheckRequest) (*authv3.C Host: host, Path: path, Headers: mapToHTTPHeader(headers), + StartedAt: time.Now(), } + // Finisher dispatch for outbound. Only created/deferred if inbound + // allowed — mirrors the two-pipeline control flow. + defer func() { + s.OutboundPipeline.RunFinish(ctx, outPctx, pipeline.OutcomeFromContext(outPctx)) + }() originalAuth := outPctx.Headers.Get("Authorization") outAction := s.OutboundPipeline.Run(ctx, outPctx) if outAction.Type == pipeline.Reject { return deniedFromAction(codes.PermissionDenied, outAction), nil } + // Mark both pctxes as "allow" so OutcomeFromContext returns + // OutcomeAllow (StatusCode 0 would otherwise be classified as + // OutcomeError — ext_authz doesn't model an HTTP status, so we + // pick 200 as the sentinel for "Check returned OK"). + inPctx.StatusCode = 200 + outPctx.StatusCode = 200 + newAuth := outPctx.Headers.Get("Authorization") if newAuth != originalAuth { return allowedWithToken(extractBearer(newAuth)), nil diff --git a/authbridge/cmd/authbridge/listener/extproc/server.go b/authbridge/cmd/authbridge/listener/extproc/server.go index c1a400e4d..30c48bf1b 100644 --- a/authbridge/cmd/authbridge/listener/extproc/server.go +++ b/authbridge/cmd/authbridge/listener/extproc/server.go @@ -54,6 +54,22 @@ func (s *Server) Process(stream extprocv3.ExternalProcessor_ProcessServer) error var pctx *pipeline.Context var requestDirection string + // Finisher dispatch runs once when Process returns — stream end is + // Envoy's signal that the request is finalized (response sent or + // abandoned). A stream that never reached Run (no RequestHeaders + // ever arrived) leaves pctx nil, in which case we have no chain + // to finish on; skip. + defer func() { + if pctx == nil { + return + } + p := s.OutboundPipeline + if requestDirection == "inbound" { + p = s.InboundPipeline + } + p.RunFinish(ctx, pctx, pipeline.OutcomeFromContext(pctx)) + }() + for { select { case <-ctx.Done(): diff --git a/authbridge/cmd/authbridge/listener/forwardproxy/server.go b/authbridge/cmd/authbridge/listener/forwardproxy/server.go index c41945743..4866d3f9c 100644 --- a/authbridge/cmd/authbridge/listener/forwardproxy/server.go +++ b/authbridge/cmd/authbridge/listener/forwardproxy/server.go @@ -60,8 +60,17 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { Host: r.Host, Path: r.URL.Path, Headers: r.Header.Clone(), + StartedAt: time.Now(), } + // Finisher dispatch runs after every exit path. RunFinish is a + // no-op when pctx.dispatched is empty (pre-pipeline rejects), so + // this defer is safe on every path including the body-too-large + // early return. + defer func() { + s.OutboundPipeline.RunFinish(r.Context(), pctx, pipeline.OutcomeFromContext(pctx)) + }() + if s.OutboundPipeline.NeedsBody() && r.Body != nil { r.Body = http.MaxBytesReader(w, r.Body, maxBodySize) body, err := io.ReadAll(r.Body) diff --git a/authbridge/cmd/authbridge/listener/reverseproxy/finisher_test.go b/authbridge/cmd/authbridge/listener/reverseproxy/finisher_test.go new file mode 100644 index 000000000..944913a9b --- /dev/null +++ b/authbridge/cmd/authbridge/listener/reverseproxy/finisher_test.go @@ -0,0 +1,160 @@ +package reverseproxy + +import ( + "context" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/plugintesting" +) + +// finisherStub is a minimal Plugin + Finisher used by these tests to +// observe OnFinish dispatch and the Outcome the listener derived. +type finisherStub struct { + name string + onReq func(*pipeline.Context) pipeline.Action + seen atomic.Bool + outcome atomic.Pointer[pipeline.Outcome] +} + +func (p *finisherStub) Name() string { return p.name } +func (p *finisherStub) Capabilities() pipeline.PluginCapabilities { + return pipeline.PluginCapabilities{} +} +func (p *finisherStub) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { + if p.onReq != nil { + return p.onReq(pctx) + } + return pipeline.Action{Type: pipeline.Continue} +} +func (p *finisherStub) OnResponse(context.Context, *pipeline.Context) pipeline.Action { + return pipeline.Action{Type: pipeline.Continue} +} +func (p *finisherStub) OnFinish(_ context.Context, pctx *pipeline.Context) { + p.seen.Store(true) + if o := pctx.Outcome(); o != nil { + cp := *o + p.outcome.Store(&cp) + } +} + +func pipelineWith(t *testing.T, plugins ...pipeline.Plugin) *pipeline.Holder { + t.Helper() + p, err := plugintesting.BuildPipeline(plugins) + if err != nil { + t.Fatalf("BuildPipeline: %v", err) + } + return pipeline.NewHolder(p) +} + +// TestReverseProxy_Finisher_Allow verifies OnFinish fires with +// OutcomeAllow on the happy path. +func TestReverseProxy_Finisher_Allow(t *testing.T) { + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte("ok")) + })) + defer backend.Close() + + f := &finisherStub{name: "f-allow"} + srv, err := NewServer(pipelineWith(t, f), nil, backend.URL) + if err != nil { + t.Fatal(err) + } + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + resp, err := http.Get(proxy.URL + "/anything") + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + + if !f.seen.Load() { + t.Fatal("OnFinish did not fire") + } + o := f.outcome.Load() + if o == nil { + t.Fatal("Outcome was nil in OnFinish") + } + if o.FinalAction != pipeline.OutcomeAllow { + t.Errorf("FinalAction = %q, want allow", o.FinalAction) + } + if o.StatusCode != http.StatusOK { + t.Errorf("StatusCode = %d, want 200", o.StatusCode) + } + if o.DenyingPlugin != "" { + t.Errorf("DenyingPlugin = %q, want empty", o.DenyingPlugin) + } +} + +// TestReverseProxy_Finisher_Deny verifies OnFinish fires on the plugin +// that denied AND on earlier plugins whose OnRequest ran. Both see +// OutcomeDeny with the correct DenyingPlugin attribution. +func TestReverseProxy_Finisher_Deny(t *testing.T) { + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + t.Error("backend should not be called on deny") + })) + defer backend.Close() + + before := &finisherStub{name: "before-deny"} + denier := &finisherStub{ + name: "denier", + onReq: func(pctx *pipeline.Context) pipeline.Action { + pctx.Record(pipeline.Invocation{ + Plugin: "denier", + Action: pipeline.ActionDeny, + Reason: "test_deny", + }) + return pipeline.Action{ + Type: pipeline.Reject, + Violation: &pipeline.Violation{Status: 403, Code: "test.deny", Reason: "denied for test"}, + } + }, + } + after := &finisherStub{name: "after-deny"} + + srv, err := NewServer(pipelineWith(t, before, denier, after), nil, backend.URL) + if err != nil { + t.Fatal(err) + } + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + resp, err := http.Get(proxy.URL + "/anything") + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusForbidden { + t.Errorf("HTTP status = %d, want 403", resp.StatusCode) + } + + if !before.seen.Load() { + t.Error("before-deny.OnFinish should have fired (OnRequest ran before denial)") + } + if !denier.seen.Load() { + t.Error("denier.OnFinish should have fired (OnRequest ran and produced the deny)") + } + if after.seen.Load() { + t.Error("after-deny.OnFinish should NOT have fired (OnRequest never ran)") + } + + // Both should report FinalAction=Deny, DenyingPlugin=denier. + for _, stub := range []*finisherStub{before, denier} { + o := stub.outcome.Load() + if o == nil { + t.Errorf("%s: outcome nil", stub.name) + continue + } + if o.FinalAction != pipeline.OutcomeDeny { + t.Errorf("%s: FinalAction = %q, want deny", stub.name, o.FinalAction) + } + if o.DenyingPlugin != "denier" { + t.Errorf("%s: DenyingPlugin = %q, want %q", stub.name, o.DenyingPlugin, "denier") + } + } +} diff --git a/authbridge/cmd/authbridge/listener/reverseproxy/server.go b/authbridge/cmd/authbridge/listener/reverseproxy/server.go index 89beeee6e..21c521108 100644 --- a/authbridge/cmd/authbridge/listener/reverseproxy/server.go +++ b/authbridge/cmd/authbridge/listener/reverseproxy/server.go @@ -79,8 +79,18 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { Host: r.Host, Path: r.URL.Path, Headers: r.Header.Clone(), + StartedAt: time.Now(), } + // Finisher dispatch runs after every exit path from this handler — + // allowed requests, plugin denials, upstream errors. RunFinish is + // a no-op when pctx.dispatched is empty (e.g. body-too-large + // rejected before Run), so this defer is safe on the pre-pipeline + // error paths too. + defer func() { + s.InboundPipeline.RunFinish(r.Context(), pctx, pipeline.OutcomeFromContext(pctx)) + }() + if s.InboundPipeline.NeedsBody() && r.Body != nil { r.Body = http.MaxBytesReader(w, r.Body, maxBodySize) body, err := io.ReadAll(r.Body) From 668489e19e03c87b84fa58f1d2ab1326ed11f448 Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Mon, 11 May 2026 22:56:34 -0400 Subject: [PATCH 3/3] =?UTF-8?q?fixup(pipeline):=20Address=20PR=20#401=20re?= =?UTF-8?q?view=20=E2=80=94=20robustness=20+=20ergonomics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses five review comments from PR #401, four code changes plus one comment-only tweak: #1 — Framework-stamped rejecting plugin (robustness). OutcomeFromContext previously inferred the denying plugin from pctx.Extensions.Invocations, making correctness dependent on plugin authors pairing Action{Type: Reject} with a pctx.Record call. A plugin that returned Reject without recording was misclassified as OutcomeAllow or OutcomeError. Pipeline.Run / RunResponse now stamp pctx.rejectingPlugin on every enforce-policy Reject. OutcomeFromContext reads that first; the Invocation walk stays as defense-in-depth for bespoke dispatchers. Shadow rejections (on_error: observe) do not stamp the field — matches "the pipeline effectively allowed" semantics already used by abctl. Exposed via pctx.RejectingPlugin() getter for listeners that build Outcome explicitly. #2 — context.WithoutCancel in dispatchFinish (ergonomics). OnFinish previously ran under context.Background()-derived ctx, discarding any values (slog fields, request ID, tracing span) the caller carried. Now uses context.WithoutCancel(parent) as the base: preserves values while still detaching cancellation so client disconnect doesn't abort cleanup I/O. The caller-supplied ctx is wired through RunFinish; its parameter is no longer vestigial. #3 — Drop ext_authz StatusCode=200 sentinel (cleanup). OutcomeFromContext classifies pctx.StatusCode == 0 as OutcomeError, which is correct for HTTP listeners where 0 means "no response written." ext_authz has no HTTP status concept, so it was setting StatusCode=200 as a sentinel to coerce the helper into returning OutcomeAllow. With #1 landed, ext_authz now calls a local authzOutcome(pctx) helper that reads pctx.RejectingPlugin() directly and defaults to OutcomeAllow — no sentinel, intent is explicit. #4 — RunFinish double-call guard (defense-in-depth). A pctx.finished flag, set at RunFinish entry, causes a second call to log a WARN and early-return rather than double-release every Finisher's per-request state. All current listeners call RunFinish exactly once; this is purely protection against a refactor accident (two defers registered) or dispatch bug. #5 — Reword ext_authz LIFO comment (readability). Moved the LIFO note to the outbound defer's block; the inbound defer now just says "fires whether the pipeline allows, denies, or Check returns early." The prior wording made the inbound defer's comment easy to misread as "inbound runs first." New tests: TestRunFinish_DoubleCallGuard second call is no-op TestRunFinish_CtxValuePropagation values carry through detach TestRun_StampsRejectingPlugin allow/enforce-deny/shadow-deny TestOutcomeFromContext_PrefersRejectingPlugin plugins that return Reject without recording are still Deny Full authlib + cmd/authbridge test suites pass under -race. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- authbridge/authlib/pipeline/context.go | 40 ++++++ authbridge/authlib/pipeline/finisher_test.go | 123 ++++++++++++++++++ authbridge/authlib/pipeline/outcome.go | 44 ++++--- authbridge/authlib/pipeline/pipeline.go | 51 +++++--- .../authbridge/listener/extauthz/server.go | 37 ++++-- 5 files changed, 248 insertions(+), 47 deletions(-) diff --git a/authbridge/authlib/pipeline/context.go b/authbridge/authlib/pipeline/context.go index 646abb835..f5bcb6ed4 100644 --- a/authbridge/authlib/pipeline/context.go +++ b/authbridge/authlib/pipeline/context.go @@ -166,6 +166,22 @@ type Context struct { // than silently mutating a SessionEvent that has already been // published or a response that is already on the wire. inFinish bool + + // rejectingPlugin is populated by Pipeline.Run / RunResponse when + // a plugin returns Action{Type: Reject} under the enforce policy. + // It is the framework-authoritative source of "which plugin denied + // this request," freeing OutcomeFromContext from having to walk + // Invocations (and freeing plugin authors from the obligation to + // pair Reject with a pctx.Record call). Stays empty on shadow + // denials (policy == observe) and on allow paths. + rejectingPlugin string + + // finished is set at RunFinish entry to prevent accidental double + // dispatch from a buggy listener (two defers registered, a + // refactor that routes the finish call through two paths). Second + // call hits a WARN log and early-returns rather than + // double-releasing every Finisher's state. + finished bool } // SetCurrentPlugin is called by Pipeline.Run / RunResponse immediately @@ -209,6 +225,30 @@ func (c *Context) clearCurrent() { c.currentPolicy = "" } +// RejectingPlugin returns the name of the plugin whose Reject action +// stopped the pipeline, or "" if the pipeline allowed end-to-end. +// Populated by Pipeline.Run / RunResponse before they return; callable +// from OnFinish, listener code, or anywhere else that needs to know +// the denier without walking Invocations. +// +// Shadow-mode denials (policy == observe, where the plugin's Reject +// was converted to a pass-through) do not set this field — the +// framework treats shadow rejections as "the pipeline effectively +// allowed," which matches how abctl and the session store classify +// them. +func (c *Context) RejectingPlugin() string { return c.rejectingPlugin } + +// setRejectingPlugin records the name of the plugin that returned +// Reject. Framework-internal; callers in Pipeline.Run / RunResponse +// set this once per request, never overwrite (first rejection wins, +// but in practice no plugin runs after Reject so the check is +// defensive). +func (c *Context) setRejectingPlugin(name string) { + if c.rejectingPlugin == "" { + c.rejectingPlugin = name + } +} + // Record appends an Invocation to pctx under the current pipeline // direction and framework-stamped plugin + phase. The author supplies // only what's specific to this call (Action, Reason, plus any diff --git a/authbridge/authlib/pipeline/finisher_test.go b/authbridge/authlib/pipeline/finisher_test.go index e94b822e4..0e01c8cef 100644 --- a/authbridge/authlib/pipeline/finisher_test.go +++ b/authbridge/authlib/pipeline/finisher_test.go @@ -397,6 +397,129 @@ func TestRunFinish_SetBodyDroppedDuringFinish(t *testing.T) { } } +// TestRunFinish_DoubleCallGuard verifies the second RunFinish on a +// pctx is rejected with a WARN and early-returns, so a buggy listener +// (two defers registered, refactor accident) doesn't double-release +// Finisher state. +func TestRunFinish_DoubleCallGuard(t *testing.T) { + var n atomic.Int32 + f := newFinisher("f", func(_ context.Context, _ *Context) { + n.Add(1) + }) + p, err := New([]Plugin{f}) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + p.Run(context.Background(), pctx) + p.RunFinish(context.Background(), pctx, Outcome{FinalAction: OutcomeAllow}) + p.RunFinish(context.Background(), pctx, Outcome{FinalAction: OutcomeAllow}) + if got := n.Load(); got != 1 { + t.Errorf("OnFinish fired %d times, want 1 (second RunFinish must no-op)", got) + } +} + +// TestRunFinish_CtxValuePropagation verifies OnFinish sees values +// from the caller's ctx (WithoutCancel preserves values), even though +// cancellation is detached. Uses a per-test context key. +func TestRunFinish_CtxValuePropagation(t *testing.T) { + type key struct{} + var seen any + f := newFinisher("f", func(ctx context.Context, _ *Context) { + seen = ctx.Value(key{}) + }) + p, err := New([]Plugin{f}) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + parent := context.WithValue(context.Background(), key{}, "req-abc-123") + // Cancel parent to prove cancellation is detached. + cancelCtx, cancel := context.WithCancel(parent) + cancel() + p.Run(context.Background(), pctx) + p.RunFinish(cancelCtx, pctx, Outcome{FinalAction: OutcomeAllow}) + if seen != "req-abc-123" { + t.Errorf("OnFinish ctx value = %v, want req-abc-123 (WithoutCancel must preserve values)", seen) + } +} + +// TestRun_StampsRejectingPlugin verifies Pipeline.Run populates +// pctx.RejectingPlugin() on an enforce-policy deny but leaves it +// empty on shadow-mode denies and on allow paths. +func TestRun_StampsRejectingPlugin(t *testing.T) { + tests := []struct { + name string + onReq func(*Context) Action + policy ErrorPolicy + wantName string + }{ + { + name: "allow: RejectingPlugin empty", + onReq: func(*Context) Action { return Action{Type: Continue} }, + policy: ErrorPolicyEnforce, + wantName: "", + }, + { + name: "enforce deny: RejectingPlugin populated", + onReq: func(*Context) Action { + return Action{Type: Reject, Violation: &Violation{Code: "t", Reason: "r"}} + }, + policy: ErrorPolicyEnforce, + wantName: "denier", + }, + { + name: "observe deny (shadow): RejectingPlugin empty", + onReq: func(*Context) Action { + return Action{Type: Reject, Violation: &Violation{Code: "t", Reason: "r"}} + }, + policy: ErrorPolicyObserve, + wantName: "", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + s := &stubPlugin{name: "denier", onReq: func(_ context.Context, pctx *Context) Action { + return tc.onReq(pctx) + }} + p, err := New([]Plugin{s}, WithPolicies(tc.policy)) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + p.Run(context.Background(), pctx) + if got := pctx.RejectingPlugin(); got != tc.wantName { + t.Errorf("RejectingPlugin = %q, want %q", got, tc.wantName) + } + }) + } +} + +// TestOutcomeFromContext_PrefersRejectingPlugin verifies the helper +// reads pctx.RejectingPlugin() first, so a plugin that returns Reject +// without calling pctx.Record is still classified as OutcomeDeny. +func TestOutcomeFromContext_PrefersRejectingPlugin(t *testing.T) { + s := &stubPlugin{name: "no-record-denier", onReq: func(_ context.Context, _ *Context) Action { + // Return Reject without recording. Pre-fix this path + // classified as OutcomeError (StatusCode 0) or OutcomeAllow + // (StatusCode > 0) depending on downstream state. + return Action{Type: Reject, Violation: &Violation{Code: "t", Reason: "r"}} + }} + p, err := New([]Plugin{s}) + if err != nil { + t.Fatalf("New: %v", err) + } + pctx := &Context{} + p.Run(context.Background(), pctx) + out := OutcomeFromContext(pctx) + if out.FinalAction != OutcomeDeny { + t.Errorf("FinalAction = %q, want deny (framework-stamped rejectingPlugin must drive classification)", out.FinalAction) + } + if out.DenyingPlugin != "no-record-denier" { + t.Errorf("DenyingPlugin = %q, want %q", out.DenyingPlugin, "no-record-denier") + } +} + // TestOutcomeFromContext covers the listener-facing derivation helper: // deny Invocations win, StatusCode 0 means Error, non-zero + no deny // means Allow. Shadow denials don't count. diff --git a/authbridge/authlib/pipeline/outcome.go b/authbridge/authlib/pipeline/outcome.go index 703d506d2..306804f7b 100644 --- a/authbridge/authlib/pipeline/outcome.go +++ b/authbridge/authlib/pipeline/outcome.go @@ -79,27 +79,36 @@ func (c *Context) Outcome() *Outcome { // OutcomeFromContext derives a best-effort Outcome from a pctx's final // state, intended for listeners that want a one-liner finish call // without threading outcome state through nested response / error -// callbacks. The derivation rules: +// callbacks. The derivation rules, checked in order: // -// - A deny Invocation on either phase → OutcomeDeny, DenyingPlugin = -// the most recent deny's Plugin name. Most-recent rather than -// first so a response-side deny (e.g. an output filter) is -// correctly attributed over any earlier record. -// - No deny Invocation AND StatusCode > 0 → OutcomeAllow. The HTTP -// status itself is not sufficient to classify error vs allow — a +// - pctx.RejectingPlugin() is non-empty → OutcomeDeny, DenyingPlugin +// names the rejector. Populated by the framework (Pipeline.Run / +// RunResponse) whenever a plugin returned Action{Type: Reject} +// under the enforce policy. Independent of whether the plugin +// paired Reject with a pctx.Record call. +// - Fallback: a deny Invocation on either phase → OutcomeDeny, +// DenyingPlugin = the most-recent deny's Plugin. Kept as +// defense-in-depth for bespoke dispatchers that populate +// Invocations directly. +// - No deny AND StatusCode > 0 → OutcomeAllow. The HTTP status +// itself is not sufficient to classify error vs allow — a // legitimate 500 from the upstream is still a pipeline Allow. -// - No deny Invocation AND StatusCode == 0 → OutcomeError (no -// response was written: upstream transport failure, listener -// panic, etc.). +// - No deny AND StatusCode == 0 → OutcomeError (no response was +// written: upstream transport failure, listener panic, etc.). // -// Listeners with more precise information at hand (a typed error from -// the upstream transport, a listener-level reject distinct from any -// plugin deny) should construct Outcome explicitly rather than call -// this helper. StatusCode is taken from pctx.StatusCode verbatim. -// Duration is left zero; RunFinish auto-fills from pctx.StartedAt -// when left unset. +// Listeners with no HTTP-status concept (check-only protocols like +// ext_authz) should construct Outcome explicitly using +// pctx.RejectingPlugin() — the "StatusCode == 0 means error" rule is +// HTTP-listener specific. StatusCode is taken from pctx.StatusCode +// verbatim. Duration is left zero; RunFinish auto-fills from +// pctx.StartedAt when left unset. func OutcomeFromContext(pctx *Context) Outcome { out := Outcome{StatusCode: pctx.StatusCode} + if denier := pctx.RejectingPlugin(); denier != "" { + out.FinalAction = OutcomeDeny + out.DenyingPlugin = denier + return out + } if denier, ok := lastDenyingPlugin(pctx); ok { out.FinalAction = OutcomeDeny out.DenyingPlugin = denier @@ -116,6 +125,9 @@ func OutcomeFromContext(pctx *Context) Outcome { // lastDenyingPlugin walks pctx.Extensions.Invocations in reverse (both // directions) looking for the most-recent deny-action record. Returns // the plugin name and true if found; "", false otherwise. +// +// Secondary to pctx.RejectingPlugin(); kept as defense-in-depth for +// bespoke dispatchers that bypass Pipeline.Run. func lastDenyingPlugin(pctx *Context) (string, bool) { if pctx.Extensions.Invocations == nil { return "", false diff --git a/authbridge/authlib/pipeline/pipeline.go b/authbridge/authlib/pipeline/pipeline.go index 37a4f2e99..e3cc0cd0d 100644 --- a/authbridge/authlib/pipeline/pipeline.go +++ b/authbridge/authlib/pipeline/pipeline.go @@ -148,6 +148,7 @@ func (p *Pipeline) Run(ctx context.Context, pctx *Context) Action { markShadowAndLog(pctx, plugin.Name(), InvocationPhaseRequest, action, "request") continue } + pctx.setRejectingPlugin(plugin.Name()) logReject(plugin.Name(), action, "pipeline: plugin rejected request") return action } @@ -181,6 +182,7 @@ func (p *Pipeline) RunResponse(ctx context.Context, pctx *Context) Action { markShadowAndLog(pctx, p.plugins[i].Name(), InvocationPhaseResponse, action, "response") continue } + pctx.setRejectingPlugin(p.plugins[i].Name()) logReject(p.plugins[i].Name(), action, "pipeline: plugin rejected response") return action } @@ -425,12 +427,14 @@ func (p *Pipeline) Stop(ctx context.Context) { // pctx.Outcome() returns non-nil for the duration of the finish chain // and nil everywhere else. // -// Each plugin's OnFinish runs under a FRESH context derived from -// context.Background() with p.finishTimeout applied — not from the -// caller's ctx — so a client disconnect or earlier cancellation does -// not abort OnFinish's I/O. Callers pass a bare ctx for future parent- -// cancellation needs (process-wide Stop signal) but it is intentionally -// not wired into the per-plugin ctx. +// Each plugin's OnFinish runs under a context derived from +// context.WithoutCancel(ctx) with p.finishTimeout applied. That means: +// - Cancellation of the caller-supplied ctx (client disconnect, +// listener shutdown signal) does NOT abort OnFinish's I/O. +// - Values carried on the caller-supplied ctx (slog fields, request +// ID, tracing span) ARE propagated into OnFinish. +// - Deadlines from the caller-supplied ctx are NOT inherited; the +// per-plugin timeout is authoritative. // // OnFinish is best-effort: a panicking plugin is recovered and logged, // a returning plugin's errors (there is no error return on the @@ -438,11 +442,19 @@ func (p *Pipeline) Stop(ctx context.Context) { // framework. The LIFO chain continues regardless so one misbehaving // plugin does not leak state in earlier plugins. // -// RunFinish is safe to call at most once per request. Calling it twice -// on the same pctx would double-release state in every Finisher. -// Listeners MUST call it in a defer wrapping the response-produce -// block so a panic in response-writing still reaches cleanup. -func (p *Pipeline) RunFinish(_ context.Context, pctx *Context, outcome Outcome) { +// RunFinish is safe to call at most once per request. A second call +// on the same pctx is rejected with a WARN log rather than double- +// releasing Finisher state (defensive against a listener bug where +// two defers end up registered, or a handler refactor routes the +// finish call through two paths). Listeners MUST call it in a defer +// wrapping the response-produce block so a panic in response-writing +// still reaches cleanup. +func (p *Pipeline) RunFinish(ctx context.Context, pctx *Context, outcome Outcome) { + if pctx.finished { + slog.Warn("pipeline: RunFinish called twice on the same pctx — second call dropped") + return + } + pctx.finished = true if len(pctx.dispatched) == 0 { return } @@ -463,16 +475,19 @@ func (p *Pipeline) RunFinish(_ context.Context, pctx *Context, outcome Outcome) if !ok { continue } - p.dispatchFinish(plugin.Name(), finisher, pctx) + p.dispatchFinish(ctx, plugin.Name(), finisher, pctx) } } -// dispatchFinish runs OnFinish on one plugin under a fresh ctx with -// timeout, recovering any panic into a WARN log so later plugins in -// the LIFO chain still run. Isolated in its own method so the recover -// block's scope is exactly one plugin's dispatch. -func (p *Pipeline) dispatchFinish(name string, f Finisher, pctx *Context) { - ctx, cancel := context.WithTimeout(context.Background(), p.finishTimeout) +// dispatchFinish runs OnFinish on one plugin under a detached ctx +// (context.WithoutCancel(parent) + finishTimeout) so the parent's +// cancellation does not abort cleanup I/O but values and tracing +// spans propagate. Panics are recovered into a WARN log so later +// plugins in the LIFO chain still run. Isolated in its own method so +// the recover block's scope is exactly one plugin's dispatch. +func (p *Pipeline) dispatchFinish(parent context.Context, name string, f Finisher, pctx *Context) { + base := context.WithoutCancel(parent) + ctx, cancel := context.WithTimeout(base, p.finishTimeout) defer cancel() defer func() { if r := recover(); r != nil { diff --git a/authbridge/cmd/authbridge/listener/extauthz/server.go b/authbridge/cmd/authbridge/listener/extauthz/server.go index fbf16724d..e3e66c930 100644 --- a/authbridge/cmd/authbridge/listener/extauthz/server.go +++ b/authbridge/cmd/authbridge/listener/extauthz/server.go @@ -57,11 +57,10 @@ func (s *Server) Check(ctx context.Context, req *authv3.CheckRequest) (*authv3.C StartedAt: time.Now(), } // Finisher dispatch for inbound. Deferred before Run so the hook - // fires whether the pipeline allows, denies, or the Check returns - // early. Outbound's defer (added after the outbound pctx is - // created) runs first under LIFO. + // fires whether the pipeline allows, denies, or Check returns + // early. defer func() { - s.InboundPipeline.RunFinish(ctx, inPctx, pipeline.OutcomeFromContext(inPctx)) + s.InboundPipeline.RunFinish(ctx, inPctx, authzOutcome(inPctx)) }() inAction := s.InboundPipeline.Run(ctx, inPctx) if inAction.Type == pipeline.Reject { @@ -78,9 +77,11 @@ func (s *Server) Check(ctx context.Context, req *authv3.CheckRequest) (*authv3.C StartedAt: time.Now(), } // Finisher dispatch for outbound. Only created/deferred if inbound - // allowed — mirrors the two-pipeline control flow. + // allowed — mirrors the two-pipeline control flow. Registered + // AFTER the inbound defer so under LIFO this outbound finish runs + // first, then inbound. defer func() { - s.OutboundPipeline.RunFinish(ctx, outPctx, pipeline.OutcomeFromContext(outPctx)) + s.OutboundPipeline.RunFinish(ctx, outPctx, authzOutcome(outPctx)) }() originalAuth := outPctx.Headers.Get("Authorization") outAction := s.OutboundPipeline.Run(ctx, outPctx) @@ -88,13 +89,6 @@ func (s *Server) Check(ctx context.Context, req *authv3.CheckRequest) (*authv3.C return deniedFromAction(codes.PermissionDenied, outAction), nil } - // Mark both pctxes as "allow" so OutcomeFromContext returns - // OutcomeAllow (StatusCode 0 would otherwise be classified as - // OutcomeError — ext_authz doesn't model an HTTP status, so we - // pick 200 as the sentinel for "Check returned OK"). - inPctx.StatusCode = 200 - outPctx.StatusCode = 200 - newAuth := outPctx.Headers.Get("Authorization") if newAuth != originalAuth { return allowedWithToken(extractBearer(newAuth)), nil @@ -102,6 +96,23 @@ func (s *Server) Check(ctx context.Context, req *authv3.CheckRequest) (*authv3.C return allowed(), nil } +// authzOutcome builds a Finisher Outcome from pctx state. ext_authz is +// a check-only protocol with no HTTP status concept, so +// pipeline.OutcomeFromContext (which classifies StatusCode == 0 as +// OutcomeError) doesn't fit — the absence of a status here means +// "Check returned OK," not "error." We rely on +// pctx.RejectingPlugin(), the framework-stamped deny source, to +// distinguish allow from deny. +func authzOutcome(pctx *pipeline.Context) pipeline.Outcome { + if denier := pctx.RejectingPlugin(); denier != "" { + return pipeline.Outcome{ + FinalAction: pipeline.OutcomeDeny, + DenyingPlugin: denier, + } + } + return pipeline.Outcome{FinalAction: pipeline.OutcomeAllow} +} + func mapToHTTPHeader(m map[string]string) http.Header { h := make(http.Header) for k, v := range m {