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