From e629c80e2ab20acfee54e5e47a91ddabb601654b Mon Sep 17 00:00:00 2001 From: cwiklik Date: Tue, 6 Oct 2026 15:54:04 -0400 Subject: [PATCH] fix: record the same fields on a denied event from every listener MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The four recorders that emit a phase:"denied" event — one per direction per listener — grew apart from the same copy-pasted body: Plugins Identity Duration TLS extproc in/out yes yes yes no reverseproxy in no no no yes forwardproxy out no no no no So which fields /v1/sessions published was a property of the listener that served the request rather than of the request. The forward proxy was the worst case, dropping all four: an operator reading a denial in proxy-sidecar mode could not see who was denied, which plugin denied them, or how long it took. Collapse all four onto one builder in listener/internal/sessionevent, which core/listener/... can import and which can therefore reach both core/pipeline and core/tlsconfig without the cycle that keeps pipeline.NewEventTLS taking its SPIFFE ID as a parameter. Each field is derived from pctx and absent when pctx does not carry it, so a listener that cannot know something records nothing for it rather than a zero that reads as real — extproc still publishes no TLS, because Envoy terminates the handshake and ext_proc never sees the client connection. The callers keep the two decisions that genuinely differ per listener: whether to record at all, and which session bucket to append to. Why the parity suite missed it: its observation struct excluded Identity, Duration and TLS by construction, and the deny fixture authenticated as nobody and emitted no plugin event, so the three dropped fields compared nil to nil and PluginKeys compared empty to empty. The fixture now authenticates and emits, and the comparison carries Identity (full) and Duration (presence — it is a live time.Since, so an equality check across legs would flake). Both are also asserted absolutely per listener, because the pairwise diff cannot see a gap that both legs share, and these two recorders were copied from each other. Verified failing against the pre-fix call sites on both the inbound and outbound pair. TLS stays out of parity deliberately, and the struct now says why rather than calling it a follow-up: cross-listener equality on TLS would assert something false, and all three drivers run plaintext, so it would pass for the wrong reason. Its derivation is unit-tested against a real tls.ConnectionState instead. Fixes #936 Assisted-By: Claude (Anthropic AI) Signed-off-by: cwiklik --- core/listener/extproc/server.go | 68 +----- core/listener/forwardproxy/server.go | 30 +-- .../internal/sessionevent/sessionevent.go | 86 +++++++ .../sessionevent/sessionevent_test.go | 229 ++++++++++++++++++ core/listener/parity/drivers_test.go | 61 ++++- core/listener/parity/parity_test.go | 55 +++++ core/listener/parity/spy_test.go | 43 ++++ core/listener/reverseproxy/server.go | 45 +--- core/pipeline/session.go | 4 +- 9 files changed, 489 insertions(+), 132 deletions(-) create mode 100644 core/listener/internal/sessionevent/sessionevent.go create mode 100644 core/listener/internal/sessionevent/sessionevent_test.go diff --git a/core/listener/extproc/server.go b/core/listener/extproc/server.go index 69a08cd52..4f08f08c7 100644 --- a/core/listener/extproc/server.go +++ b/core/listener/extproc/server.go @@ -23,6 +23,7 @@ import ( "google.golang.org/grpc/status" "github.com/rossoctl/cortex/core/listener/httpx" + "github.com/rossoctl/cortex/core/listener/internal/sessionevent" "github.com/rossoctl/cortex/core/listener/internal/sseframe" "github.com/rossoctl/cortex/core/listener/skiphost" "github.com/rossoctl/cortex/core/pipeline" @@ -381,40 +382,7 @@ func (s *Server) recordInboundReject(pctx *pipeline.Context, action pipeline.Act if s.Sessions == nil || pctx.Extensions.Invocations == nil { return } - var status int - var code, message string - if action.Violation != nil { - // Use the structured fields directly — Render() produces the HTTP - // wire payload (status, headers, JSON body) which is the wrong - // shape for a session event. We want the semantic Code + Reason. - status = action.Violation.Status - if status == 0 { - status = pipeline.StatusFromCode(action.Violation.Code) - } - code = action.Violation.Code - message = action.Violation.Reason - } - ev := pipeline.SessionEvent{ - At: time.Now(), - Direction: pipeline.Inbound, - Phase: pipeline.SessionDenied, - RequestID: pctx.RequestID(), - Invocations: pipeline.SnapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), - Plugins: pipeline.SnapshotPlugins(pctx.Extensions.Custom), - Identity: pipeline.SnapshotIdentity(pctx), - Host: pctx.Host, - HTTPMethod: pctx.Method, - HTTPPath: pctx.Path, - StatusCode: status, - Error: &pipeline.EventError{ - Kind: "policy", - Code: code, - Message: message, - }, - Duration: pipeline.DurationSince(pctx.StartedAt), - Client: pctx.ClientInfo(), - } - s.Sessions.Append(inboundSessionID(pctx), ev) + s.Sessions.Append(inboundSessionID(pctx), sessionevent.Deny(pctx, action, pipeline.Inbound)) } // recordOutboundReject emits a SessionDenied event for outbound requests @@ -441,37 +409,7 @@ func (s *Server) recordOutboundReject(pctx *pipeline.Context, action pipeline.Ac if sid == "" { sid = session.DefaultSessionID } - var status int - var code, message string - if action.Violation != nil { - status = action.Violation.Status - if status == 0 { - status = pipeline.StatusFromCode(action.Violation.Code) - } - code = action.Violation.Code - message = action.Violation.Reason - } - ev := pipeline.SessionEvent{ - At: time.Now(), - Direction: pipeline.Outbound, - Phase: pipeline.SessionDenied, - RequestID: pctx.RequestID(), - Invocations: pipeline.SnapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), - Plugins: pipeline.SnapshotPlugins(pctx.Extensions.Custom), - Identity: pipeline.SnapshotIdentity(pctx), - Host: pctx.Host, - HTTPMethod: pctx.Method, - HTTPPath: pctx.Path, - StatusCode: status, - Error: &pipeline.EventError{ - Kind: "policy", - Code: code, - Message: message, - }, - Duration: pipeline.DurationSince(pctx.StartedAt), - Client: pctx.ClientInfo(), - } - s.Sessions.Append(sid, ev) + s.Sessions.Append(sid, sessionevent.Deny(pctx, action, pipeline.Outbound)) } // recordInboundResponseSession appends a Phase:SessionResponse event for the diff --git a/core/listener/forwardproxy/server.go b/core/listener/forwardproxy/server.go index 25b13e475..2592adb07 100644 --- a/core/listener/forwardproxy/server.go +++ b/core/listener/forwardproxy/server.go @@ -25,6 +25,7 @@ import ( "errors" "github.com/rossoctl/cortex/core/listener/httpx" "github.com/rossoctl/cortex/core/listener/internal/bodyread" + "github.com/rossoctl/cortex/core/listener/internal/sessionevent" "github.com/rossoctl/cortex/core/listener/internal/sseframe" "github.com/rossoctl/cortex/core/listener/skiphost" "github.com/rossoctl/cortex/core/peerproc" @@ -1520,34 +1521,7 @@ func (s *Server) recordOutboundRejectIn(tl *tunnelLog, pctx *pipeline.Context, a if s.Sessions == nil || pctx.Extensions.Invocations == nil { return } - var status int - var code, message string - if action.Violation != nil { - status = action.Violation.Status - if status == 0 { - status = pipeline.StatusFromCode(action.Violation.Code) - } - code = action.Violation.Code - message = action.Violation.Reason - } - ev := pipeline.SessionEvent{ - At: time.Now(), - Direction: pipeline.Outbound, - Phase: pipeline.SessionDenied, - RequestID: pctx.RequestID(), - Invocations: pipeline.SnapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), - Host: pctx.Host, - HTTPMethod: pctx.Method, - HTTPPath: pctx.Path, - StatusCode: status, - Error: &pipeline.EventError{ - Kind: "policy", - Code: code, - Message: message, - }, - Client: pctx.ClientInfo(), - } - s.appendOutbound(tl, sid, ev) + s.appendOutbound(tl, sid, sessionevent.Deny(pctx, action, pipeline.Outbound)) } // connectDialTimeout bounds the upstream TCP dial for a CONNECT tunnel. diff --git a/core/listener/internal/sessionevent/sessionevent.go b/core/listener/internal/sessionevent/sessionevent.go new file mode 100644 index 000000000..0762bfbc0 --- /dev/null +++ b/core/listener/internal/sessionevent/sessionevent.go @@ -0,0 +1,86 @@ +// Package sessionevent builds the parts of a session event that every +// listener should record identically. +// +// It exists because the four recorders that emit a phase:"denied" event — one +// per direction per listener — grew apart from the same copy-pasted body. +// extproc recorded Plugins, Identity and Duration on both directions; the +// reverse proxy recorded TLS and none of those three; the forward proxy +// recorded none of the four. So an operator reading /v1/sessions saw a +// different event shape depending on which listener served the request, and +// which fields were populated was a property of the serving listener rather +// than of the request — the one thing a parity suite exists to prevent. +// +// Every field is derived from pctx and is absent when pctx does not carry it, +// so a listener that cannot know something records nothing for it rather than +// a zero that reads as a real value. +package sessionevent + +import ( + "time" + + "github.com/rossoctl/cortex/core/pipeline" + authtls "github.com/rossoctl/cortex/core/tlsconfig" +) + +// Deny builds the phase:"denied" event for a request a pipeline plugin +// rejected on OnRequest. +// +// dir is a parameter rather than read from pctx.Direction because the +// recorders are already one-per-direction and several of their unit tests +// construct a bare pctx with no Direction set. Reading it from pctx would +// relabel those events silently — Inbound is the zero value — which is a +// worse failure than passing the constant the caller already knows. +// +// The caller keeps two decisions. Whether to record at all: a deny from a +// plugin that appended no Invocation carries no attribution, and a +// content-free denial event is noise. And which session bucket to append to: +// that genuinely differs per listener, inbound keying on the A2A contextId +// while outbound inherits the most-recently-updated session. +func Deny(pctx *pipeline.Context, action pipeline.Action, dir pipeline.Direction) pipeline.SessionEvent { + var status int + var code, message string + if action.Violation != nil { + // Read the structured fields directly. Render() produces the HTTP wire + // payload — status, headers, JSON body — which is the wrong shape for a + // session event; what belongs here is the semantic Code and Reason. + status = action.Violation.Status + if status == 0 { + status = pipeline.StatusFromCode(action.Violation.Code) + } + code = action.Violation.Code + message = action.Violation.Reason + } + return pipeline.SessionEvent{ + At: time.Now(), + Direction: dir, + Phase: pipeline.SessionDenied, + RequestID: pctx.RequestID(), + Invocations: pipeline.SnapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), + Plugins: pipeline.SnapshotPlugins(pctx.Extensions.Custom), + Identity: pipeline.SnapshotIdentity(pctx), + Host: pctx.Host, + HTTPMethod: pctx.Method, + HTTPPath: pctx.Path, + StatusCode: status, + Error: &pipeline.EventError{ + Kind: "policy", + Code: code, + Message: message, + }, + Duration: pipeline.DurationSince(pctx.StartedAt), + Client: pctx.ClientInfo(), + TLS: TLS(pctx), + } +} + +// TLS summarizes the connection the request arrived on, and is nil whenever +// there is nothing true to say: a plaintext caller, or a listener that never +// sees the client's connection at all. In envoy-sidecar mode Envoy terminates +// the TLS and ext_proc receives the request over gRPC, so pctx.TLS is unset +// there and this records nothing rather than inventing a handshake. +func TLS(pctx *pipeline.Context) *pipeline.EventTLS { + if pctx == nil || pctx.TLS == nil { + return nil + } + return pipeline.NewEventTLS(pctx.TLS, authtls.PeerSPIFFEID(pctx.PeerCertificate())) +} diff --git a/core/listener/internal/sessionevent/sessionevent_test.go b/core/listener/internal/sessionevent/sessionevent_test.go new file mode 100644 index 000000000..0f5824993 --- /dev/null +++ b/core/listener/internal/sessionevent/sessionevent_test.go @@ -0,0 +1,229 @@ +package sessionevent + +import ( + "crypto/tls" + "crypto/x509" + "net/http" + "net/url" + "testing" + "time" + + "github.com/rossoctl/cortex/core/pipeline" +) + +// stubIdentity is a pipeline.Identity with fixed answers. +type stubIdentity struct { + subject string + clientID string + scopes []string +} + +func (s stubIdentity) Subject() string { return s.subject } +func (s stubIdentity) ClientID() string { return s.clientID } +func (s stubIdentity) Scopes() []string { return s.scopes } + +// richContext is a pctx carrying every input Deny reads, so a field that +// stops being populated shows up as a zero rather than as an equal-to-equal +// comparison between two listeners that both dropped it. +func richContext() *pipeline.Context { + spiffeURI, _ := url.Parse("spiffe://example.org/ns/team1/sa/weather-agent") + hdr := http.Header{} + hdr.Set("User-Agent", "claude-cli/1.2.3") + return &pipeline.Context{ + Direction: pipeline.Inbound, + Method: "POST", + Path: "/parity/deny", + Host: "weather-agent.team1.svc.cluster.local", + Headers: hdr, + StartedAt: time.Now().Add(-10 * time.Millisecond), + Identity: stubIdentity{ + subject: "alice", + clientID: "weather-agent", + scopes: []string{"openid", "weather-aud"}, + }, + TLS: &tls.ConnectionState{ + Version: tls.VersionTLS13, + CipherSuite: tls.TLS_AES_128_GCM_SHA256, + PeerCertificates: []*x509.Certificate{{URIs: []*url.URL{spiffeURI}}}, + }, + Extensions: pipeline.Extensions{ + Invocations: &pipeline.Invocations{ + Inbound: []pipeline.Invocation{{ + Plugin: "parity-spy-a", + Phase: pipeline.InvocationPhaseRequest, + Action: pipeline.ActionDeny, + Reason: "spy.denied", + }}, + }, + Custom: map[string]any{ + "parity-spy-a" + pipeline.PluginEventSuffix: map[string]string{"marker": "denied"}, + }, + }, + } +} + +// TestDeny_PopulatesEveryFieldTheContextCarries is the assertion the four +// recorders could not make for themselves. Each had drifted to its own +// subset — Plugins, Identity and Duration on extproc; TLS and none of those +// three on the reverse proxy; none of the four on the forward proxy — and +// because the fields were simply absent rather than wrong, every listener's +// own tests passed. This pins the whole set in one place. +func TestDeny_PopulatesEveryFieldTheContextCarries(t *testing.T) { + pctx := richContext() + action := pipeline.DenyStatus(429, "spy.denied", "quota exhausted") + + ev := Deny(pctx, action, pipeline.Inbound) + + if ev.Phase != pipeline.SessionDenied { + t.Errorf("Phase = %v, want SessionDenied", ev.Phase) + } + if ev.Direction != pipeline.Inbound { + t.Errorf("Direction = %v, want Inbound", ev.Direction) + } + if ev.StatusCode != 429 { + t.Errorf("StatusCode = %d, want 429", ev.StatusCode) + } + if ev.Host != "weather-agent.team1.svc.cluster.local" { + t.Errorf("Host = %q", ev.Host) + } + if ev.HTTPMethod != "POST" || ev.HTTPPath != "/parity/deny" { + t.Errorf("HTTPMethod/HTTPPath = %q/%q, want POST//parity/deny", ev.HTTPMethod, ev.HTTPPath) + } + if ev.Error == nil || ev.Error.Kind != "policy" || ev.Error.Code != "spy.denied" || ev.Error.Message != "quota exhausted" { + t.Errorf("Error = %+v, want kind=policy code=spy.denied message=quota exhausted", ev.Error) + } + + // The four fields #936 is about. + if ev.Identity == nil { + t.Error("Identity not snapshotted — the field the reverse and forward proxies dropped") + } else { + if ev.Identity.Subject != "alice" || ev.Identity.ClientID != "weather-agent" { + t.Errorf("Identity = %+v, want subject=alice clientId=weather-agent", ev.Identity) + } + if len(ev.Identity.Scopes) != 2 { + t.Errorf("Identity.Scopes = %v, want 2 scopes", ev.Identity.Scopes) + } + } + if _, ok := ev.Plugins["parity-spy-a"]; !ok { + t.Errorf("Plugins missing the spy's event, got keys %v", keysOf(ev.Plugins)) + } + if ev.Duration <= 0 { + t.Errorf("Duration = %v, want > 0 — a denial ends the request, so it has one", ev.Duration) + } + if ev.TLS == nil { + t.Error("TLS not snapshotted — the field extproc and the forward proxy dropped") + } else { + if ev.TLS.Version != "TLS 1.3" { + t.Errorf("TLS.Version = %q, want TLS 1.3", ev.TLS.Version) + } + if ev.TLS.PeerSPIFFEID != "spiffe://example.org/ns/team1/sa/weather-agent" { + t.Errorf("TLS.PeerSPIFFEID = %q", ev.TLS.PeerSPIFFEID) + } + } + + // Carried over from the original bodies, so the refactor is not a + // silent trade of new fields for old ones. + if ev.Invocations == nil || len(ev.Invocations.Inbound) != 1 { + t.Fatalf("Invocations = %+v, want 1 inbound record", ev.Invocations) + } + if ev.Invocations.Inbound[0].Action != pipeline.ActionDeny { + t.Errorf("Invocations.Inbound[0].Action = %v, want deny", ev.Invocations.Inbound[0].Action) + } + if ev.Client == nil || ev.Client.Name == "" { + t.Errorf("Client = %+v, want the User-Agent resolved", ev.Client) + } + if ev.At.IsZero() { + t.Error("At is zero") + } +} + +// TestDeny_OmitsWhatTheContextDoesNotCarry is the other half of the +// contract: a listener that cannot know something must record nothing for +// it, not a zero value that a consumer reads as real. extproc is the live +// case — Envoy terminates the client's TLS, so pctx.TLS is never set there +// and the event must carry no handshake rather than an empty one. +func TestDeny_OmitsWhatTheContextDoesNotCarry(t *testing.T) { + pctx := &pipeline.Context{ + Extensions: pipeline.Extensions{ + Invocations: &pipeline.Invocations{ + Inbound: []pipeline.Invocation{{Plugin: "p", Phase: pipeline.InvocationPhaseRequest}}, + }, + }, + } + + ev := Deny(pctx, pipeline.DenyStatus(403, "p.denied", "no"), pipeline.Outbound) + + if ev.Identity != nil { + t.Errorf("Identity = %+v, want nil with no identity on pctx", ev.Identity) + } + if ev.TLS != nil { + t.Errorf("TLS = %+v, want nil for a plaintext request", ev.TLS) + } + if ev.Plugins != nil { + t.Errorf("Plugins = %v, want nil with no plugin events", ev.Plugins) + } + if ev.Duration != 0 { + t.Errorf("Duration = %v, want 0 with a zero StartedAt", ev.Duration) + } + if ev.Direction != pipeline.Outbound { + t.Errorf("Direction = %v, want the dir the caller passed", ev.Direction) + } +} + +// TestDeny_DerivesStatusFromCode covers the one piece of logic in here that +// is not a copy: a plugin that set no HTTP status on its Violation gets one +// mapped from the semantic code, as the wire path does via Render(). +func TestDeny_DerivesStatusFromCode(t *testing.T) { + pctx := &pipeline.Context{ + Extensions: pipeline.Extensions{ + Invocations: &pipeline.Invocations{Inbound: []pipeline.Invocation{{Plugin: "p"}}}, + }, + } + action := pipeline.Action{Type: pipeline.Reject, Violation: &pipeline.Violation{ + Code: "auth.unauthorized", + Reason: "token validation failed", + }} + + ev := Deny(pctx, action, pipeline.Inbound) + + if want := pipeline.StatusFromCode("auth.unauthorized"); ev.StatusCode != want { + t.Errorf("StatusCode = %d, want %d from StatusFromCode", ev.StatusCode, want) + } +} + +// TestDeny_NoViolation records the event anyway rather than panicking. A +// Reject with no Violation is a plugin bug, but the denial still happened +// and an event with no code beats no event at all. +func TestDeny_NoViolation(t *testing.T) { + pctx := &pipeline.Context{ + Extensions: pipeline.Extensions{ + Invocations: &pipeline.Invocations{Inbound: []pipeline.Invocation{{Plugin: "p"}}}, + }, + } + + ev := Deny(pctx, pipeline.Action{Type: pipeline.Reject}, pipeline.Inbound) + + if ev.StatusCode != 0 { + t.Errorf("StatusCode = %d, want 0 when there is no violation to read one from", ev.StatusCode) + } + if ev.Error == nil || ev.Error.Kind != "policy" { + t.Errorf("Error = %+v, want a policy error with empty code", ev.Error) + } +} + +func TestTLS_NilCases(t *testing.T) { + if got := TLS(nil); got != nil { + t.Errorf("TLS(nil) = %+v, want nil", got) + } + if got := TLS(&pipeline.Context{}); got != nil { + t.Errorf("TLS(plaintext pctx) = %+v, want nil", got) + } +} + +func keysOf[V any](m map[string]V) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + return out +} diff --git a/core/listener/parity/drivers_test.go b/core/listener/parity/drivers_test.go index 6bbb99e77..d6fe03315 100644 --- a/core/listener/parity/drivers_test.go +++ b/core/listener/parity/drivers_test.go @@ -84,6 +84,20 @@ type fixture struct { // assert content beyond the pairwise diff." Fixtures that want // bug-catching (not just drift-catching) fill this in. expectedPluginEvents map[string]string + + // expectedIdentity, when non-nil, is asserted against EVERY listener + // rather than only compared between them. The pairwise diff cannot see + // a gap both legs share — two listeners that each record nothing agree + // perfectly — and for Identity the shared case is the likely one, + // since the two recorders that dropped it were copy-pasted from each + // other. Same argument the Inference comment in observationDiff makes. + expectedIdentity *identitySummary + + // expectDuration asserts every listener reported a non-zero duration. + // Absolute for the same reason: a denial is a completed request, so + // "how long did it take" has an answer on every listener, and all four + // already stamp pctx.StartedAt at entry. + expectDuration bool } // contentType returns the fixture's response content-type or a sensible @@ -111,8 +125,21 @@ func spyEntry(name string, cfg spyConfig) config.PluginEntry { // observation is the parity-comparable snapshot of a listener's session // event. Covers the operator-facing wire surface; per-listener locals -// (Host casing, timestamps, RequestID, Duration, TLS, Identity) are -// excluded — expanding coverage there is a follow-up fixture pass. +// (Host casing, timestamps, RequestID) are excluded. +// +// Identity and Duration used to be excluded too, as "a follow-up fixture +// pass". They were the follow-up: #936 found Identity, Duration and Plugins +// recorded by extproc and dropped by both HTTP listeners on the deny path, +// invisible here precisely because this struct did not carry them. +// +// TLS stays out, and not as a deferral. extproc legitimately has no client +// TLS to report — Envoy terminates the handshake and ext_proc receives the +// request over gRPC — so a cross-listener equality check on TLS would be +// asserting something false, and it would pass here for the wrong reason +// anyway: all three drivers run plaintext httptest servers, so every leg +// records nil. TLS derivation is pinned by unit test in +// listener/internal/sessionevent instead, which is where it can be given a +// real tls.ConnectionState. type observation struct { // PipelineRan is false when the listener rejected the request before // the pipeline (e.g. request body too large). Overflow fixtures then @@ -128,6 +155,17 @@ type observation struct { // PluginEventJSON pins the raw JSON per plugin key so a snapshot // difference between listeners surfaces as a value diff. PluginEventJSON map[string]string + // Identity is who the request authenticated as, nil when the event + // carried none. Compared in full: a listener that drops the subject or + // the scopes publishes an event an operator cannot attribute, and an + // unattributable denial is the one denial that matters most. + Identity *identitySummary + // HasDuration is presence, not value. Every listener computes it as a + // live time.Since(pctx.StartedAt), so the figures legitimately differ + // between two legs of the same fixture and an equality check here + // would flake. What is comparable — and what was actually wrong — is + // whether a listener reports a duration at all. + HasDuration bool // Inference is the token report the event carried, nil when it carried none. // // IT IS NOT A RESTATEMENT OF THE COST RECORD, which travels separately in @@ -157,6 +195,16 @@ type inferenceSummary struct { PresentKinds uint8 } +// identitySummary mirrors pipeline.EventIdentity. Scopes compared as an +// ordered slice because SnapshotIdentity copies them in the order the auth +// plugin supplied, and nothing between there and the event reorders them. +type identitySummary struct { + Subject string + Scopes []string + ClientID string + AgentID string +} + // errorSummary mirrors pipeline.EventError so listener drift in the // deny reason fails parity instead of remaining invisible. type errorSummary struct { @@ -218,8 +266,17 @@ func observe(t *testing.T, store *session.Store, wantDir pipeline.Direction, wan obs := &observation{ Phase: ev.Phase.String(), StatusCode: ev.StatusCode, + HasDuration: ev.Duration > 0, PluginEventJSON: map[string]string{}, } + if ev.Identity != nil { + obs.Identity = &identitySummary{ + Subject: ev.Identity.Subject, + Scopes: ev.Identity.Scopes, + ClientID: ev.Identity.ClientID, + AgentID: ev.Identity.AgentID, + } + } if ev.Error != nil { obs.Error = &errorSummary{Kind: ev.Error.Kind, Code: ev.Error.Code, Message: ev.Error.Message} } diff --git a/core/listener/parity/parity_test.go b/core/listener/parity/parity_test.go index 9df832bca..65281c77a 100644 --- a/core/listener/parity/parity_test.go +++ b/core/listener/parity/parity_test.go @@ -34,6 +34,14 @@ var outboundListeners = []listenerRun{ // TestParity_DenyOnRequest: pctx.Record before Reject must produce the // same phase:"denied" event on both listeners. +// +// The fixture authenticates and emits a plugin event before it denies, +// which is not decoration. Until it did, the denial event under test was +// the emptiest one this suite could build — no identity, no plugin +// payload, no asserted duration — so the comparison reduced to the fields +// every listener happened to agree on, and #936's three dropped fields +// (Plugins, Identity, Duration on the reverse proxy) compared nil to nil. +// A parity fixture only covers what it populates. func TestParity_DenyOnRequest(t *testing.T) { f := fixture{ name: "deny-on-request", @@ -43,9 +51,23 @@ func TestParity_DenyOnRequest(t *testing.T) { DenyStatus: 429, DenyReason: "spy.denied", DenyDetails: map[string]string{"cutoff_reason": "quota"}, + Subject: "alice@example.org", + ClientID: "weather-agent", + Scopes: []string{"openid", "weather.read"}, + EmitOnRequest: true, + RequestEvent: &spyEvent{Marker: "denied", Count: 1}, })}, method: "GET", path: "/parity/deny", + expectedPluginEvents: map[string]string{ + spyPluginA: jsonOf(spyEvent{Marker: "denied", Count: 1}), + }, + expectedIdentity: &identitySummary{ + Subject: "alice@example.org", + ClientID: "weather-agent", + Scopes: []string{"openid", "weather.read"}, + }, + expectDuration: true, } assertParity(t, f, pipeline.SessionDenied, inboundListeners) } @@ -81,9 +103,23 @@ func TestParity_OutboundDenyOnRequest(t *testing.T) { DenyStatus: 403, DenyReason: "spy.blocked", DenyDetails: map[string]string{"cutoff_reason": "egress-policy"}, + Subject: "alice@example.org", + ClientID: "weather-agent", + Scopes: []string{"openid", "llm.invoke"}, + EmitOnRequest: true, + RequestEvent: &spyEvent{Marker: "blocked", Count: 1}, })}, method: "GET", path: "/parity/egress", + expectedPluginEvents: map[string]string{ + spyPluginA: jsonOf(spyEvent{Marker: "blocked", Count: 1}), + }, + expectedIdentity: &identitySummary{ + Subject: "alice@example.org", + ClientID: "weather-agent", + Scopes: []string{"openid", "llm.invoke"}, + }, + expectDuration: true, } assertParity(t, f, pipeline.SessionDenied, outboundListeners) } @@ -389,6 +425,16 @@ func assertParity(t *testing.T, f fixture, wantPhase pipeline.SessionPhase, list t.Errorf("fixture %q listener %s: plugin event %q\n got: %s\n want: %s", f.name, g.listener, key, gotJSON, wantJSON) } } + if f.expectedIdentity != nil { + if g.observed.Identity == nil { + t.Errorf("fixture %q listener %s: event carried no Identity; the caller authenticated as %q and the event cannot be attributed to them", f.name, g.listener, f.expectedIdentity.Subject) + } else if !reflect.DeepEqual(g.observed.Identity, f.expectedIdentity) { + t.Errorf("fixture %q listener %s: Identity\n got: %s\n want: %s", f.name, g.listener, jsonPretty(g.observed.Identity), jsonPretty(f.expectedIdentity)) + } + } + if f.expectDuration && !g.observed.HasDuration { + t.Errorf("fixture %q listener %s: event carried no Duration; the request completed, so it has one", f.name, g.listener) + } } // Pairwise compare against the first listener. All observations must @@ -432,6 +478,15 @@ func observationDiff(a, b *observation) string { if !reflect.DeepEqual(a.PluginKeys, b.PluginKeys) { return "PluginKeys: " + jsonPretty(a.PluginKeys) + " vs " + jsonPretty(b.PluginKeys) } + // One listener attributing the request while another publishes an + // anonymous event for the same authenticated caller. Same shared-gap + // blind spot as Inference below, covered by fixture.expectedIdentity. + if !reflect.DeepEqual(a.Identity, b.Identity) { + return "Identity: " + jsonPretty(a.Identity) + " vs " + jsonPretty(b.Identity) + } + if a.HasDuration != b.HasDuration { + return fmt.Sprintf("HasDuration: %v vs %v", a.HasDuration, b.HasDuration) + } // One listener reporting token counts while another does not. Worth saying what // this check CANNOT do, because that is how the gap it was added for survived: two // listeners that both record nothing agree, so a gap shared by a direction's whole diff --git a/core/listener/parity/spy_test.go b/core/listener/parity/spy_test.go index da8d2d268..7e96283c6 100644 --- a/core/listener/parity/spy_test.go +++ b/core/listener/parity/spy_test.go @@ -28,6 +28,23 @@ type spyConfig struct { EmitOnResponse bool `json:"emit_on_response"` ResponseEvent *spyEvent `json:"response_event"` + // EmitOnRequest publishes RequestEvent at OnRequest, BEFORE any deny. + // Without it a deny fixture's Plugins map is empty on every listener, + // so the PluginKeys comparison below comes down to empty-vs-empty and + // a listener that never snapshots Plugins at all compares equal to one + // that does. That is how the drift in #936 survived this suite. + EmitOnRequest bool `json:"emit_on_request"` + RequestEvent *spyEvent `json:"request_event"` + + // Subject / ClientID / Scopes make the spy stand in for an auth + // plugin, which is the only thing that ever populates pctx.Identity + // (jwtvalidation/plugin.go:439 does exactly this assignment). A + // fixture that cannot authenticate cannot tell a listener that drops + // Identity from one that had none to record. + Subject string `json:"subject"` + ClientID string `json:"client_id"` + Scopes []string `json:"scopes"` + // RequiresLater names peer plugins that must appear at a HIGHER // index in the same pipeline. Populates PluginCapabilities so the // registry's dependency validator can reject wrong-order pipelines. @@ -80,6 +97,19 @@ func (s *spyPlugin) OnRequest(_ context.Context, pctx *pipeline.Context) pipelin if s.cfg.RecordRequestBody { s.publish(pctx, bodyReqStrippedSuffix+pipeline.PluginEventSuffix, bodyObservation{Body: string(pctx.Body)}) } + // Authenticate and emit before deciding to deny: both must land on the + // denial event, since a denial is exactly when an operator needs to + // know who was denied and which plugin said so. + if s.cfg.Subject != "" || s.cfg.ClientID != "" { + pctx.Identity = spyIdentity{ + subject: s.cfg.Subject, + clientID: s.cfg.ClientID, + scopes: s.cfg.Scopes, + } + } + if s.cfg.EmitOnRequest && s.cfg.RequestEvent != nil { + s.publish(pctx, pipeline.PluginEventSuffix, *s.cfg.RequestEvent) + } if !s.cfg.DenyOnRequest { return pipeline.Action{Type: pipeline.Continue} } @@ -108,6 +138,19 @@ func (s *spyPlugin) publish(pctx *pipeline.Context, suffix string, v any) { pctx.Extensions.Custom[s.name+suffix] = v } +// spyIdentity is the pipeline.Identity an authenticating plugin leaves on +// pctx. Kept local to this package because every test file that needs one +// defines its own; there is no shared stub to borrow. +type spyIdentity struct { + subject string + clientID string + scopes []string +} + +func (s spyIdentity) Subject() string { return s.subject } +func (s spyIdentity) ClientID() string { return s.clientID } +func (s spyIdentity) Scopes() []string { return s.scopes } + // spyStreamingPlugin wraps spyPlugin and adds OnResponseFrame, making it // a pipeline.StreamingResponder. Pipeline.RunResponse skips streaming // responders and dispatches only via OnResponseFrame; body-recording diff --git a/core/listener/reverseproxy/server.go b/core/listener/reverseproxy/server.go index 6a648b425..eb295aa2e 100644 --- a/core/listener/reverseproxy/server.go +++ b/core/listener/reverseproxy/server.go @@ -20,6 +20,7 @@ import ( "github.com/rossoctl/cortex/core/listener/httpx" "github.com/rossoctl/cortex/core/listener/internal/bodyread" + "github.com/rossoctl/cortex/core/listener/internal/sessionevent" "github.com/rossoctl/cortex/core/listener/internal/sseframe" "github.com/rossoctl/cortex/core/listener/internal/tlssniff" "github.com/rossoctl/cortex/core/listener/transparentproxy" @@ -303,14 +304,14 @@ func (s *Server) WrapListener(inner net.Listener) net.Listener { func (s *Server) MTLSEnabled() bool { return s.mtlsCfg != nil } // eventTLS builds a *pipeline.EventTLS from the pctx's connection -// state, extracting the peer SPIFFE ID via core/tlsconfig. Returns nil -// for plaintext or absent TLS state — sites that pass the result -// through to a SessionEvent get the right thing for any caller. +// state. Returns nil for plaintext or absent TLS state — sites that pass +// the result through to a SessionEvent get the right thing for any caller. +// +// Kept as a local alias because the accept paths below read better for it; +// the derivation itself is shared with the other listeners, which record +// the same field on a denial. func eventTLS(pctx *pipeline.Context) *pipeline.EventTLS { - if pctx == nil || pctx.TLS == nil { - return nil - } - return pipeline.NewEventTLS(pctx.TLS, authtls.PeerSPIFFEID(pctx.PeerCertificate())) + return sessionevent.TLS(pctx) } // Handler returns the HTTP handler for the reverse proxy. @@ -717,35 +718,7 @@ func (s *Server) recordInboundReject(pctx *pipeline.Context, action pipeline.Act // inboundSessionID helper, kept consistent so denial events land // in the same bucket the accepted request would have. sid := inboundSessionID(pctx) - var status int - var code, message string - if action.Violation != nil { - status = action.Violation.Status - if status == 0 { - status = pipeline.StatusFromCode(action.Violation.Code) - } - code = action.Violation.Code - message = action.Violation.Reason - } - ev := pipeline.SessionEvent{ - At: time.Now(), - Direction: pipeline.Inbound, - Phase: pipeline.SessionDenied, - RequestID: pctx.RequestID(), - Client: pctx.ClientInfo(), - Invocations: pipeline.SnapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest), - Host: pctx.Host, - HTTPMethod: pctx.Method, - HTTPPath: pctx.Path, - StatusCode: status, - Error: &pipeline.EventError{ - Kind: "policy", - Code: code, - Message: message, - }, - TLS: eventTLS(pctx), - } - s.Sessions.Append(sid, ev) + s.Sessions.Append(sid, sessionevent.Deny(pctx, action, pipeline.Inbound)) } // requestScheme derives the URL scheme for an incoming server-side diff --git a/core/pipeline/session.go b/core/pipeline/session.go index 664d559e8..369b8936e 100644 --- a/core/pipeline/session.go +++ b/core/pipeline/session.go @@ -158,7 +158,9 @@ type SessionEvent struct { // Duration is the wall-clock time from request entry into the listener // to response recording. Zero on request-phase events. On response - // events it's computed as now - matching-request.At. + // events it's computed as now - matching-request.At. On denied events + // it's DurationSince(pctx.StartedAt) — a denial ends the request, so + // it has a duration even though no response was ever recorded. Duration time.Duration // TLS, when non-nil, carries connection-level identity for events