Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 3 additions & 65 deletions core/listener/extproc/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down
30 changes: 2 additions & 28 deletions core/listener/forwardproxy/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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.
Expand Down
86 changes: 86 additions & 0 deletions core/listener/internal/sessionevent/sessionevent.go
Original file line number Diff line number Diff line change
@@ -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{

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: pipeline/snapshot.go:104-109 lists the SnapshotInference exemptions as "the four SessionDenied recorders and forwardproxy.recordTunnelOpened", then says "Anything else that builds a SessionEvent and skips this call is the bug above, not a sixth exemption." After this PR the four recorders don't build an event; this function does, and it skips SnapshotInference. Read literally, that comment now flags sessionevent.Deny as the bug. Suggest naming sessionevent.Deny there, as the one builder the four denial recorders share.

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),

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: in proxy-sidecar mode this line also changes a figure users see. cost/usage/usage.go:1348 counts every event with a Duration of 1ms or more toward latency, denied events included, so reverse- and forward-proxy denials now feed the /v1/usage latency mean in the laptop's Usage pane. A token-exchange deny after a slow IdP failure is one example. This matches extproc, and arguably it's the right number, since the client did wait that long. But it moves a visible metric, so it deserves a line in the PR body.

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()))
}
Loading
Loading