diff --git a/CLAUDE.md b/CLAUDE.md index fc9c079d0..33f6863bb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -618,11 +618,42 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries: disagree about the status on purpose. **A 426 with `error.kind: "upgrade_declined"` is not one of these:** it is the forward proxy refusing a WebSocket on OpenAI's Responses API (a path ending `/responses` on `chatgpt.com` or - `api.openai.com`; other hosts keep the strip), because the origin - would only answer the stripped GET with a 405, which Codex retries for ~6s, while a - 426 sends it straight to HTTPS. Nothing is forwarded, the Finisher sees an allow with + `api.openai.com`) — a deliberate steer, since a 426 sends Codex straight to HTTPS + where its traffic stays parseable, while a forwarded WebSocket's frames are opaque + to every plugin. Nothing is forwarded, the Finisher sees an allow with status 426, and `/v1/usage` does not count it as an error. On by default; - `listener.decline_responses_websockets: false` forwards the upgrade instead. + `listener.decline_responses_websockets: false` forwards the upgrade for real + instead. **Every other WebSocket is forwarded** (#1488): the upgrade crosses the + proxy (normalized to `Connection: Upgrade` / `Upgrade: websocket`, which net/http + pins to HTTP/1.1 even on the bridge's h2-negotiating client), the origin's 101 is + hijacked and spliced, and the session records the request row plus a paired 101 + response row carrying both directions' byte counts — the frames themselves are + never inspected or buffered. **An upgrade is both headers, on both sides**, and + since the proxy WRITES the normalized pair a loose predicate forges one rather + than missing one: a request needs `Connection: upgrade` as well as `Upgrade: + websocket` (RFC 9110 §7.8, and `Request.requiresHTTP1`'s own test) or it is + stripped and takes the ordinary path, and a 101 that switches to some other + protocol than the one requested is refused 502 before the hijack instead of + spliced. Either check missing turns a plain request into an opaque pipe no + plugin inspects. **The forward proxy's 101 row is kept out of `/v1/usage` for the + same reasons a tunnel row is** — it is recorded after both splice directions end, + so its + `durationMs` is the WebSocket SESSION's lifetime, and folding it as a request + latency moved one host's mean from 500ms to 6.4s in testing; its bytes are opaque + frames with no model, token or cost to attribute; and counting it would add one + Request whose latency is its whole lifetime. The `durationMs` stays on the + session timeline, where the lifetime is the feature — only the aggregation + misread it. **The exclusion is scoped to outbound, and a 101 is not excluded + everywhere.** The reverse proxy records one too, but *at handshake time* — + `ReverseProxy` never copies an upgraded body, so a deferred record would never + fire — which makes that row's `durationMs` the handshake latency rather than a + session lifetime. An inbound 101 is therefore an ordinary request and still + counts as one; excluding it on the status alone would change what `Requests` + means on every inbound listener. **A relay that broke mid-session carries `error.kind: + "proxy_error"`** on that row, so a WebSocket that died is distinguishable from + one that closed cleanly; an ordinary close (`net.ErrClosed`, from the direction + that lost the close race) is deliberately not marked, or every healthy session + would read as broken. **A 403 with `error.kind: "destination_private"` is the egress guard's answer, not the upstream's** (`listener.egress_guard`, off by default): the destination resolved into loopback, link-local, private or this host's own @@ -675,7 +706,7 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries: `timing:` line. - `requestedHost` — present only when a plugin redirected the request (`pctx.Redirect`): the host the client asked for. `host` is where the request actually went, and usage, the cost ledger and pricing all follow `host`. agentop's detail pane shows both on a `redirected:` line. - `inference.requestedModel` — present only when a plugin changed the model (`pctx.SetRequestModel`): the model the client asked for. `inference.model` is the model the request was sent for, which settlement prices; agentop's detail pane shows both on a `model:` line. -- `bytesUp`, `bytesDown` — **body** bytes, one direction per row: a request row carries `bytesUp`, the body **as forwarded** (so a `tool-prune`-style rewrite is reflected, since the count is taken after the pipeline ran), and a response row carries `bytesDown`, the body **as it came back from the destination** — not as delivered, because extproc records the row before it emits a response mutator's replacement. Headers are not counted. A tunnel close is the one row carrying both. **Zero means the listener counted nothing, not that the body was empty**, and there is no third value for "unknown": every listener buffers bodies only when a plugin asks, so an unbuffered request reports zero. A response relayed straight through is counted, not zero — both proxies record the row after the copy (#1251's fix), so the bytes they relayed through their own process are on it. The one arm that still reports ↓0 is extproc with a body-free pipeline: it asked Envoy for body mode NONE, so the bytes it never saw are bytes it asked not to see — `core/listener/parity`'s `TestDivergence_InboundResponseBodyCountWhenNoPluginReadsIt` pins that disagreement with its reason. agentop renders a zero side blank rather than as `0B`, and that is why its BYTES column can be trusted as a size when it shows one. Everywhere else the three listeners count the same thing for the same response, which took some doing: extproc sums the chunks Envoy hands it, while the proxies count off the upstream body at one choke point each, so every arm — buffered, block-by-block SSE, or chunk copy — reports the same figure. `core/listener/internal/bodycount` and `core/listener/parity`'s bytes fixtures hold that line. +- `bytesUp`, `bytesDown` — **body** bytes, one direction per row: a request row carries `bytesUp`, the body **as forwarded** (so a `tool-prune`-style rewrite is reflected, since the count is taken after the pipeline ran), and a response row carries `bytesDown`, the body **as it came back from the destination** — not as delivered, because extproc records the row before it emits a response mutator's replacement. Headers are not counted. A tunnel close and a forwarded WebSocket's 101 response row are the rows carrying both. **Zero means the listener counted nothing, not that the body was empty**, and there is no third value for "unknown": every listener buffers bodies only when a plugin asks, so an unbuffered request reports zero. A response relayed straight through is counted, not zero — both proxies record the row after the copy (#1251's fix), so the bytes they relayed through their own process are on it. The one arm that still reports ↓0 is extproc with a body-free pipeline: it asked Envoy for body mode NONE, so the bytes it never saw are bytes it asked not to see — `core/listener/parity`'s `TestDivergence_InboundResponseBodyCountWhenNoPluginReadsIt` pins that disagreement with its reason. agentop renders a zero side blank rather than as `0B`, and that is why its BYTES column can be trusted as a size when it shows one. Everywhere else the three listeners count the same thing for the same response, which took some doing: extproc sums the chunks Envoy hands it, while the proxies count off the upstream body at one choke point each, so every arm — buffered, block-by-block SSE, or chunk copy — reports the same figure. `core/listener/internal/bodycount` and `core/listener/parity`'s bytes fixtures hold that line. - `tunnel`, `tunnelReason` — an opaque CONNECT (or transparent-redirect) tunnel records two rows sharing a `requestId`: the open (`phase: "request"`, `tunnelReason` saying why the bytes stayed opaque) and, when the tunnel ends, the close (`phase: "response"`). The close carries the CONNECT's own `statusCode` — 200, or 502 with the dial error in `error` when the destination could not be reached and no TLS bridge could take the connection (`tunnelReason: "dial-failed"`; where the bridge is enabled for the port the CONNECT is answered 200 first, because a redirect can still apply, and a bridge that then declines closes with that 200 and the dial error) — plus `durationMs` for how long the tunnel stayed open and, in `bytesUp`/`bytesDown`, the bytes it carried each way (up = client to destination; opaque bytes have no body/header split, so these are everything that crossed). It is not the destination's status: that travels inside the client's end-to-end TLS. A bridged tunnel's open is recorded with its first decrypted request — in that request's session, directly before it, stamped with its time — and records no close, because the request carries its own response; a bridged tunnel that recorded no request gets its open and a close when it ends. Tunnel rows are kept out of `/v1/usage`: a tunnel's lifetime is not a request latency. `tunnelReason: "dest-private"` with a 403 close and `error.kind: "destination_private"` is the egress guard refusing the destination before any dial — never confused with `dial-failed`, which attempted. - `dialAddr` — the address the egress guard checked and the proxy dialed, `"ip:port"`, on a tunnel's open row when `listener.egress_guard` is on. With `host` it is the audit trail against DNS rebinding: under the guard the proxy resolves once, checks that answer, and dials exactly it, so the two fields together say "asked for the name, reached this address". Absent when the guard is off and on rows that dialed nothing. - `httpMethod`, `httpPath` — the HTTP verb and path, so a request no parser recognized is still identifiable rather than showing only a host. Distinct from the `method` inside `a2a` / `mcp`, which is a protocol method name. On an opaque tunnel `httpMethod` is `CONNECT` and `httpPath` is absent — opaque bytes carry no request line. The path is query-stripped and percent-decoded, so query-borne credentials never reach the timeline, but a secret in a path *segment* (a bot token, a webhook path) does survive on this unauthenticated surface — worth knowing before exporting events off-box. diff --git a/core/config/config.go b/core/config/config.go index b5ffe3919..1f11cfb46 100644 --- a/core/config/config.go +++ b/core/config/config.go @@ -1045,8 +1045,8 @@ type ListenerConfig struct { // DeclineResponsesWebSockets: whether the forward proxy answers a WebSocket // upgrade on an OpenAI Responses API path (one ending /responses, on chatgpt.com - // or api.openai.com) with 426 itself, instead of forwarding the request with its Upgrade stripped. nil means - // unset, and the default is on; explicit false restores the forwarding. See + // or api.openai.com) with 426 itself, instead of forwarding the upgrade. nil + // means unset, and the default is on; explicit false forwards it. See // DeclinesResponsesWebSockets for why on is the default. // // A pointer for the reason SessionConfig.Enabled is one: with a plain bool, @@ -1100,13 +1100,14 @@ type EgressGuardConfig struct { // DeclinesResponsesWebSockets reports whether the forward proxy answers a WebSocket // upgrade on a Responses API path with 426 rather than forwarding it. // -// On unless set false, in every deployment. The proxy must strip Upgrade (RFC 9110 -// makes it hop-by-hop), so a forwarded upgrade reaches the origin as a bare GET, and -// OpenAI's Responses endpoint answers that 405. Codex reads a 405 as transient and -// retries for about six seconds before falling back to HTTPS; it reads a 426 as "use -// HTTP" and falls back at once. The forwarded request can never become a WebSocket, so -// nothing is lost by answering it here. False is the escape hatch for a client that -// reads 426 some other way. +// On unless set false, in every deployment. The proxy CAN forward a WebSocket (it +// does for every other host, #1488), but a Responses session on one is opaque to +// every parser — no token counts, no cost — so the 426 keeps Codex on HTTPS where +// its traffic stays observable. That, and history: before forwarding existed the +// stripped handshake drew a 405 that Codex retried for about six seconds before +// falling back to HTTPS, while it reads a 426 as "use HTTP" and falls back at once. +// False is the escape hatch for a client that reads 426 some other way — or wants +// the WebSocket for real. func (l ListenerConfig) DeclinesResponsesWebSockets() bool { return l.DeclineResponsesWebSockets == nil || *l.DeclineResponsesWebSockets } diff --git a/core/cost/usage/replay_test.go b/core/cost/usage/replay_test.go index 9133d5b06..7722d79b6 100644 --- a/core/cost/usage/replay_test.go +++ b/core/cost/usage/replay_test.go @@ -2,6 +2,7 @@ package usage import ( "encoding/json" + "net/http" "reflect" "strings" "testing" @@ -154,3 +155,45 @@ func TestAbandonReplay_StopsLogging(t *testing.T) { t.Fatalf("after AbandonReplay: replaying=%v, %d ops logged", live.replaying, len(live.replayOps)) } } + +// A 101 row is excluded on the REPLAY path too, which only holds because +// ReplayCopy is a whole-struct copy (`c := *e`) and so carries every field the +// exclusion reads. There are THREE: StatusCode and Tunnel, and — since the +// exclusion was scoped to outbound — Direction. If ReplayCopy ever became a +// field-by-field copy that dropped or mangled any of them, the exclusion would +// silently stop applying to archived WebSocket sessions and their lifetimes +// would reappear in the latency mean on every restart — see +// TestRecord_WebSocket101IsNotARequestLatency for what that costs. +// +// The first two are asserted directly below; Direction is covered by the +// end-to-end half, which is why that half is not redundant: mangling Direction +// to Inbound moves the mean from 500ms to 300s and fails there. +func TestReplayCopy_KeepsWhatExcludesA101Row(t *testing.T) { + now := time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC) + ws := respEvent(now, http.StatusSwitchingProtocols, 10*time.Minute, "", 0) + if c := ReplayCopy(ws); c.StatusCode != http.StatusSwitchingProtocols { + t.Fatalf("ReplayCopy dropped StatusCode (%d); the 101 exclusion in Record keys on it", + c.StatusCode) + } + tun := respEvent(now, 200, 4*time.Minute, "", 0) + tun.Tunnel = true + if c := ReplayCopy(tun); !c.Tunnel { + t.Fatal("ReplayCopy dropped Tunnel; the tunnel exclusion in Record keys on it") + } + + // And end to end through the ring: a replayed 101 changes nothing. + a := New(WithClock(func() time.Time { return now })) + a.Record("s1", respEvent(now, 200, 500*time.Millisecond, "", 0)) + before := a.Snapshot(time.Minute, BucketWidth, "", GroupNone) + c := ReplayCopy(ws) + a.Record("s1", &c) + after := a.Snapshot(time.Minute, BucketWidth, "", GroupNone) + if after.Totals.Requests != before.Totals.Requests { + t.Errorf("requests = %d, want %d — a replayed 101 must not count either", + after.Totals.Requests, before.Totals.Requests) + } + if weightedLatMean(after) != weightedLatMean(before) { + t.Errorf("latency mean moved %.1f -> %.1f on replay", + weightedLatMean(before), weightedLatMean(after)) + } +} diff --git a/core/cost/usage/usage.go b/core/cost/usage/usage.go index ec2dcc265..168b01476 100644 --- a/core/cost/usage/usage.go +++ b/core/cost/usage/usage.go @@ -16,6 +16,7 @@ package usage import ( "math" "net" + "net/http" "sort" "strconv" "strings" @@ -1056,7 +1057,37 @@ func (a *Aggregator) Record(sessionID string, e *pipeline.SessionEvent) { // were opaque, so there is no model, token or cost to attribute. Counting it would // also quietly change what Requests means for every reader of /v1/usage. A bridged // tunnel's decrypted requests are ordinary events and are counted as usual. - if e.Tunnel { + // + // A FORWARDED WEBSOCKET'S 101 ROW IS THE SAME SHAPE, and is excluded for the + // same three reasons (#1488, #1549 review). It is recorded after both splice + // directions end, so its Duration is the WebSocket SESSION's lifetime, not a + // request latency: measured, one 10-minute WS session against a host with a + // hundred 500ms requests moved that host's latency mean from 500ms to 6.4s + // (TestRecord_WebSocket101IsNotARequestLatency). Its bytes are opaque frames + // with no model, token or cost to attribute. And counting it would make + // Requests mean something new, inconsistently: the upgrade GET's own request + // row is already not counted (Invocations nil), so one WebSocket would add + // exactly one Request whose latency is its whole lifetime. + // + // Keyed on the status rather than a new event field because 101 IS the + // condition on the forward proxy — it records that status for no other + // reason, and a 101 that reached a client always means a spliced upgrade + // (the relay refuses any other switched protocol before the hijack). The row + // keeps its durationMs on the session timeline, where the WebSocket's + // lifetime is the feature; only the aggregation misread it. + // + // SCOPED TO OUTBOUND, because the status alone does not identify the + // condition across listeners. The reverse proxy records a 101 too + // (ModifyResponse's StatusSwitchingProtocols arm, Direction: Inbound) and + // records it AT HANDSHAKE TIME — ReverseProxy never copies an upgraded body, + // so a deferred record would never fire — which makes that row's Duration + // the handshake latency, not a session lifetime. It is an ordinary request + // with an ordinary latency, and excluding it would change what Requests + // means on every inbound listener: the same silent redefinition this comment + // warns about, in the other direction. Pinned both ways by + // TestRecord_WebSocket101IsNotARequestLatency and + // TestRecord_InboundWebSocket101IsStillARequest (#1549 review). + if e.Tunnel || (e.Direction == pipeline.Outbound && e.StatusCode == http.StatusSwitchingProtocols) { return } diff --git a/core/cost/usage/usage_test.go b/core/cost/usage/usage_test.go index 59f94b714..01596d8c4 100644 --- a/core/cost/usage/usage_test.go +++ b/core/cost/usage/usage_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "math" + "net/http" "reflect" "strings" "testing" @@ -615,6 +616,91 @@ func TestRecord_IgnoresTunnelRows(t *testing.T) { } } +// A forwarded WebSocket's 101 row is the same shape as a tunnel close and gets the +// same treatment (#1488, #1549 review). The row is recorded after both splice +// directions end, so its Duration is the WebSocket SESSION's lifetime — minutes or +// hours — and folding it as a request latency is exactly the error the tunnel +// exclusion above exists to prevent. +// +// The numbers are the point: a hundred ordinary 500ms requests against one host, +// plus ONE ten-minute WebSocket, used to report a 6.4-SECOND mean. That is the +// figure an operator reads off /v1/usage and agentop's per-server latency. +func TestRecord_WebSocket101IsNotARequestLatency(t *testing.T) { + now := time.Date(2026, 9, 4, 23, 30, 30, 0, time.UTC) + a := New(WithClock(fixedClock(now))) + for i := 0; i < 100; i++ { + a.Record("s1", respEvent(now, 200, 500*time.Millisecond, "", 0)) + } + before := a.Snapshot(time.Minute, BucketWidth, "", GroupNone) + + // The 101 a ten-minute WebSocket session leaves behind, byte counts and all. + ws := respEvent(now, http.StatusSwitchingProtocols, 10*time.Minute, "", 0) + ws.BytesUp, ws.BytesDown = 1234, 5678 + a.Record("s1", ws) + after := a.Snapshot(time.Minute, BucketWidth, "", GroupNone) + + if got := after.Totals.Requests; got != before.Totals.Requests { + t.Errorf("requests = %d, want %d — a WebSocket session is not one more request", + got, before.Totals.Requests) + } + bl, al := weightedLatMean(before), weightedLatMean(after) + if al != bl { + t.Errorf("latency mean moved %.1fms -> %.1fms; a 101 row carries the session's "+ + "lifetime, not a request latency", bl, al) + } + if bl == 0 { + t.Fatal("the ordinary requests recorded no latency; the test would pass vacuously") + } +} + +// The exclusion above is scoped to OUTBOUND, and this is why. The reverse proxy +// records a 101 row too (its ModifyResponse's StatusSwitchingProtocols arm, +// Direction: Inbound) — but it records it AT HANDSHAKE TIME, because +// ReverseProxy never copies an upgraded body, so that row's Duration is the +// handshake latency and not the session's lifetime. The lifetime is the only +// thing the exclusion exists to keep out of the mean. +// +// So an inbound 101 is an ordinary request with a sensible latency, and dropping +// it would silently change what Requests MEANS on every inbound listener — the +// very thing the exclusion's own comment warns against, in the other direction +// (#1549 review). +func TestRecord_InboundWebSocket101IsStillARequest(t *testing.T) { + now := time.Date(2026, 9, 4, 23, 30, 30, 0, time.UTC) + a := New(WithClock(fixedClock(now))) + before := a.Snapshot(time.Minute, BucketWidth, "", GroupNone) + + // The reverse proxy's 101: inbound, and its duration is the HANDSHAKE, so it + // is a normal-looking latency rather than a session lifetime. + ws := respEvent(now, http.StatusSwitchingProtocols, 12*time.Millisecond, "", 0) + ws.Direction = pipeline.Inbound + a.Record("s1", ws) + after := a.Snapshot(time.Minute, BucketWidth, "", GroupNone) + + if got, want := after.Totals.Requests, before.Totals.Requests+1; got != want { + t.Errorf("requests = %d, want %d — an inbound 101 is a handshake, and was "+ + "counted as a request before the outbound exclusion was added", got, want) + } + if m := weightedLatMean(after); m == 0 { + t.Error("inbound 101 contributed no latency; its handshake time is a real request latency") + } +} + +// weightedLatMean is the sample-weighted latency mean across a snapshot's buckets, +// which is how a client showing one window-wide figure has to compute it (see +// Bucket.LatSamples). +func weightedLatMean(s Snapshot) float64 { + var sum float64 + var n int64 + for _, b := range s.Buckets { + sum += b.LatMeanMs * float64(b.LatSamples) + n += b.LatSamples + } + if n == 0 { + return 0 + } + return sum / float64(n) +} + // All three groupings accumulate simultaneously, so an operator cycling the // group parameter sees the same history from each angle rather than each // grouping starting empty when first selected. diff --git a/core/listener/forwardproxy/server.go b/core/listener/forwardproxy/server.go index 59bce0e2e..28ab012c0 100644 --- a/core/listener/forwardproxy/server.go +++ b/core/listener/forwardproxy/server.go @@ -602,13 +602,25 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne } } + // A WebSocket upgrade is the one hop-by-hop request this proxy forwards + // (#1488): the CONNECT-host check above has already held a bridged request to + // its tunnel's destination, the pipeline has run, and what is left is to hand + // the header across and splice the 101. Every other protocol named in Upgrade + // keeps the strip, and so does the one WebSocket this proxy declines — the + // Responses-API case below, which answers 426 itself. + forwardUpgrade := upgradesToWebSocket(r.Header) && + !(s.DeclineResponsesWebSockets && wantsResponsesWebSocket(r)) + // Before the request row is recorded, 20-odd lines below, and NOT down beside // stripForUpstream where the strip itself happens — the row is appended first, // so a field set after it is set on nothing. (It was, and three unit tests // driving this method directly all passed while the feature recorded nothing - // end to end.) Reading it here is equivalent: the strip is unconditional, so an - // Upgrade present on the inbound request is one that will be removed. - pctx.RecordStrippedUpgrade(r.Header) + // end to end.) Reading it here is equivalent: an Upgrade present on the + // inbound request is one the strip below will remove, because the one case it + // is kept is exactly the case excluded by forwardUpgrade above. + if !forwardUpgrade { + pctx.RecordStrippedUpgrade(r.Header) + } // After the pipeline, so a plugin's reject still wins; before any row, since this // records its own pair. See declineResponsesWebSocket. @@ -664,7 +676,7 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne pinned := s.OutboundPipeline.ForPctx(pctx) ownEncoding := !skipped && (pinned.NeedsResponseBody() || pinned.HasStreamingResponders()) - stripForUpstream(r, ownEncoding) + stripForUpstream(r, ownEncoding, forwardUpgrade) // Clear RequestURI — set by the server but must be empty for client requests r.RequestURI = "" @@ -697,10 +709,30 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne client := s.Client if isBridge && s.TLSBridge != nil { + // No upgrade-specific client, though this one negotiates h2: net/http + // pins a WebSocket upgrade to HTTP/1.1 on its own (Request.requiresHTTP1 + // matches the Connection/Upgrade pair stripForUpstream just normalized + // to), so the exchange rides this client as HTTP/1.1. client = s.TLSBridge.Upstream } resp, err := client.Do(r) - if err == nil && !skipped { + // A Resender reads refused answers, and a 101 is not one — it is the upgrade + // succeeding. Peeking at its (empty) body and sending the GET again would tear + // down the connection the origin just switched, so a forwarded upgrade never + // resends. + // + // THE RULE IS KEYED ON THE REQUEST, NOT THE ANSWER, so it also covers an + // upgrade the origin DECLINED with a non-101 (401/403/429 — those take the + // ordinary response path, and a Resender does not see them here). That is + // deliberate, and narrowing this to `resp.StatusCode != 101` would break it: + // the resend path strips the Upgrade pair (`stripForUpstream(again, + // ownEncoding, false)`), so the second attempt would reach the origin as a + // bare GET — not the handshake the client sent, and on an endpoint that only + // answers an upgrade, not a request worth making. A re-auth that cannot carry + // the upgrade cannot rescue the upgrade; the client retries the handshake + // itself, with whatever credential it has now. Left as a request-keyed rule + // rather than made cleverer (#1549 review). + if err == nil && !skipped && !forwardUpgrade { resp, err = s.resendIfAsked(r, pctx, client, resp, tl, ownEncoding) } if err != nil { @@ -755,6 +787,18 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne } return } + + // The 101: the origin took the upgrade. Hand both connections to a splice and + // get out of the way (#1488). The caller's resp.Body.Close() below must NOT + // run — the body IS the origin's connection now, and the splice owns closing + // it. Anything else the origin answered (a 200 serving the bare GET, a 4xx + // from an endpoint that does not upgrade) falls through to the ordinary + // response path, headers and body relayed as usual. + if forwardUpgrade && resp.StatusCode == http.StatusSwitchingProtocols { + pctx.ResponseHeadersAt = time.Now() + s.relayUpgradedWebSocket(w, r, pctx, resp, !skipped) + return + } defer resp.Body.Close() // Response phase: populate pctx and run plugins in reverse order. @@ -1351,8 +1395,8 @@ const declinedUpgradeBody = `{"detail":"WebSocket transport is not available thr // Scoped twice: to OpenAI's own hosts (isOpenAIResponsesHost), and to the Responses // suffix inference-parser reads (/v1/responses, and Codex's // /backend-api/codex/responses). A 426 is an answer only Codex is known to read as -// "use HTTPS", and only against OpenAI is that known: every other WebSocket keeps the -// strip and whatever its origin makes of the bare GET. +// "use HTTPS", and only against OpenAI is that known: every other WebSocket is +// forwarded (#1488), whatever its origin makes of the upgrade. func wantsResponsesWebSocket(r *http.Request) bool { return r.Method == http.MethodGet && upgradesToWebSocket(r.Header) && @@ -1382,11 +1426,30 @@ func isOpenAIResponsesHost(authority string) bool { return false } -// upgradesToWebSocket reports whether h's Upgrade names websocket. RFC 9110 §7.8 makes -// Upgrade a comma-separated list, matched case-insensitively and in any repeat of the -// header, so "websocket, h2c" asks for a WebSocket as surely as "websocket" does. A -// token may carry a version ("websocket/13"); only the name before it is compared. +// upgradesToWebSocket reports whether h is a WebSocket upgrade REQUEST. RFC 9110 +// §7.8 makes Upgrade a comma-separated list, matched case-insensitively and in any +// repeat of the header, so "websocket, h2c" asks for a WebSocket as surely as +// "websocket" does. A token may carry a version ("websocket/13"); only the name +// before it is compared. +// +// CONNECTION IS CHECKED FIRST, and it is not a formality. RFC 9110 §7.8 is +// explicit that a sender MUST send Connection: upgrade with the Upgrade header, +// and that a recipient ignores Upgrade when Connection does not list it — which is +// exactly what net/http's own httpguts-based upgradeType does before +// ReverseProxy honours an upgrade at all. +// +// Keying on Upgrade alone made this proxy SYNTHESIZE an upgrade the client had not +// requested: stripForUpstream sets the normalized Connection/Upgrade pair whenever +// keepUpgrade is true, so a request carrying a bare "Upgrade: websocket" and no +// Connection token reached the origin as a well-formed handshake, and an origin +// that answered 101 got its connection hijacked and spliced — turning a request +// the client expected an ordinary HTTP response to into an opaque byte pipe whose +// frames no plugin inspects. Measured end to end before this check existed +// (TestForwardProxy_UpgradeWithoutConnectionTokenIsStripped) (#1549 review). func upgradesToWebSocket(h http.Header) bool { + if !headerListContainsToken(h, "Connection", "upgrade") { + return false + } for _, line := range h.Values("Upgrade") { for _, tok := range strings.Split(line, ",") { name, _, _ := strings.Cut(strings.TrimSpace(tok), "/") @@ -1398,19 +1461,40 @@ func upgradesToWebSocket(h http.Header) bool { return false } +// headerListContainsToken reports whether the named header — in any repeat, as a +// comma-separated list — carries token, compared case-insensitively. The shape +// RFC 9110 gives Connection, and the check net/http makes with +// httpguts.HeaderValuesContainsToken, which is not importable from here. +func headerListContainsToken(h http.Header, name, token string) bool { + for _, line := range h.Values(name) { + for _, tok := range strings.Split(line, ",") { + if strings.EqualFold(strings.TrimSpace(tok), token) { + return true + } + } + } + return false +} + // declineResponsesWebSocket answers a WebSocket upgrade on a Responses API path with // 426 Upgrade Required, without contacting the origin. // -// THE ORIGIN'S ANSWER IS ALREADY KNOWN, AND IT IS THE EXPENSIVE ONE. stripForUpstream -// must remove Upgrade, so the origin would get a bare GET on an endpoint that takes a -// POST or an upgrade and answer 405. Codex reads a 405 as transient: it retries five -// times on a 200ms-doubling backoff, about six seconds, printing `Reconnecting… n/5`, -// before falling back to HTTPS. A 426 is the one status it reads as "this endpoint wants +// THE EXPENSIVE ANSWER IS THE ONE A STRIPPED HANDSHAKE DREW. Before upgrades were +// forwarded (#1488) the strip turned this handshake into a bare GET on an endpoint +// that takes a POST or an upgrade, and the origin answered 405. Codex reads a 405 as +// transient: it retries five times on a 200ms-doubling backoff, about six seconds, +// printing `Reconnecting… n/5`, before falling back to HTTPS. A 426 is the one status +// it reads as "this endpoint wants // HTTP": it switches at once, for the rest of the session, with no retry and no warning // (codex-rs/core/src/client.rs, the UPGRADE_REQUIRED arms). Measured on Codex 0.163.0, // eight runs each of a one-word `codex exec` turn: 10.2–15.1s (median 11.0s) through the // 405, 2.4–6.4s (median 3.0s) through this, with one handshake attempt instead of seven. // +// TODAY THE DECLINE IS A STEER RATHER THAN A LIMIT: a forwarded Responses WebSocket +// works (relayUpgradedWebSocket), but its frames are opaque to every plugin, so a +// Codex session moved onto one stops being parseable — no token counts, no cost. The +// 426 keeps it on HTTPS, where its traffic stays observable. +// // RFC 9110 gives 426 the opposite sense — the server requires the client to upgrade. // OpenAI's backend uses it this way and Codex reads it this way, and a client that reads // neither treats it as an ordinary failed handshake, which is what the 405 already was. @@ -1443,6 +1527,255 @@ func (s *Server) declineResponsesWebSocket(w http.ResponseWriter, r *http.Reques }) } +// relayUpgradedWebSocket splices a 101 Switching Protocols answer onto the +// client's connection and blocks until either side closes (#1488). It mirrors +// httputil.ReverseProxy's handleUpgradeResponse — hijack the client connection, +// write the origin's 101 through it, then copy both ways — with two things that +// proxy does not need and this one does: per-direction byte counts for the +// session row, and the row itself. +// +// resp.Body is the origin's connection once the origin answered 101 (the +// transport hands back an io.ReadWriteCloser or it never sent a 101), and this +// function owns it: serveOutbound returns without its resp.Body.Close() defer +// running, because closing the body here would close the WebSocket. +// +// THE ROWS: the request row was recorded before client.Do, as for any request. +// Its pair is recorded here, after the splice ends, carrying the 101 and both +// directions' byte counts — the one response row with bytesUp on it, like a +// tunnel close, because after the 101 the bytes are protocol frames the proxy +// does not parse and there is no body/header split to report. Duration covers +// the whole WebSocket session, not the handshake; ttfb is the handshake, set +// where client.Do returned. That is one audit event per upgrade, the request row +// plus its paired 101 (MOCA MI1 §8.5 item 5). +// +// Response plugins never see the 101 — pctx.ResponseHeaders is not set from it, +// so no RunResponse runs on this answer: the frames after a successful upgrade +// belong to the client and the origin ("streamed, never inspected or buffered", +// the same contract), and a response mutator would have nothing to act on. A +// non-101 answer never reaches here; it takes the ordinary response path. +// +// record=false is skip_hosts: the splice still happens — a skipped request is +// forwarded as a transparent proxy or not at all — but no row is appended, which +// is what skip_hosts promises. +func (s *Server) relayUpgradedWebSocket(w http.ResponseWriter, r *http.Request, pctx *pipeline.Context, resp *http.Response, record bool) { + // The host as this function logs it. r.Host is client-controlled, so it goes + // through SanitizeControlRunes — every control rune replaced — and then + // through a strings.ReplaceAll that is a NO-OP by construction, because the + // sanitizer already took every control rune, "\n" included. The ReplaceAll + // is not ornament: CodeQL's go/log-injection (this PR's finding) cannot see + // through a custom function however thorough, and recognizes exactly two + // sanitizers — this ReplaceAll shape and %q formatting — so the one it knows + // has to sit in THIS function, on the value's last step before a log call + // (the query's own test suite proves the shape only in the sink's function; + // fencing inside SanitizeControlRunes was tried at 6cae5d14 and changed + // nothing). A helper would put it back inside a callee — do not lift it. + host := strings.ReplaceAll(pipeline.SanitizeControlRunes(r.Host), "\n", "") + // BOTH early returns below must close the origin themselves. This function + // owns resp.Body (serveOutbound skipped its Close defer, see the doc above), + // and the goroutine that closes it on context-done is only armed further + // down — so returning before that point without closing here leaves the + // origin holding a connection it believes is an established WebSocket, until + // its own idle timeout reaps it. The client meanwhile has a 502, so nothing + // will ever come back for it (#1549 review). + // THE ORIGIN MUST BE SWITCHING TO THE PROTOCOL THE CLIENT ASKED FOR. A 101 is + // not self-describing: it says "switching", and its own Upgrade header says to + // what. net/http's ReverseProxy compares the two (handleUpgradeResponse refuses + // when resUpType != reqUpType), and this function is modelled on it, but the + // comparison was the one part left out — so an origin answering "101 Upgrade: + // h2c" to a WebSocket handshake got spliced anyway, handing the client an + // opaque pipe speaking a protocol it never negotiated and cannot parse. The + // client is the one that would have to detect that, after the proxy already + // committed its connection. Refuse before the hijack, while a 502 is still + // writable (#1549 review). + if !upgradesToWebSocket(resp.Header) { + got := strings.ReplaceAll(pipeline.SanitizeControlRunes(resp.Header.Get("Upgrade")), "\n", "") + slog.Error("forward-proxy: origin switched to another protocol than the WebSocket requested — not relaying", + "host", host, "upgrade", got) + _ = resp.Body.Close() + s.failUpgradedWebSocket(w, pctx, record, + fmt.Sprintf("origin switched to %q, not the websocket requested", got)) + return + } + backConn, ok := resp.Body.(io.ReadWriteCloser) + if !ok { + // Not reachable through net/http's own transport, which gives a 101 a + // writable body; a custom RoundTripper could violate it. + slog.Error("forward-proxy: 101 switching protocols with a non-writable body — cannot relay the upgrade", "host", host) + // Not an io.ReadWriteCloser, but still an io.Closer: the Body contract + // guarantees that much, so the origin's end can be released. + _ = resp.Body.Close() + s.failUpgradedWebSocket(w, pctx, record, "101 switching protocols with a non-writable body") + return + } + conn, brw, err := http.NewResponseController(w).Hijack() + if err != nil { + // The host is fenced above; the hijack error is the runtime's own text. + slog.Error("forward-proxy: hijack failed relaying a WebSocket upgrade", "host", host, "error", err) + _ = backConn.Close() + s.failUpgradedWebSocket(w, pctx, record, fmt.Sprintf("hijack failed relaying the upgrade: %v", err)) + return + } + // The origin's end closes when the client's request context is done — which + // covers a client hang-up the copies have not noticed yet (golang.org/issue + // 35559, as in ReverseProxy) — or when the splice below ends, whichever is + // first; the deferred close of conn takes care of the client's end. + backConnCloseCh := make(chan struct{}) + go func() { + select { + case <-r.Context().Done(): + case <-backConnCloseCh: + } + backConn.Close() + }() + defer close(backConnCloseCh) + defer conn.Close() + + // The origin's 101, as it wrote it, onto the hijacked connection. Moving + // resp.Header into w's still-empty header map first is ReverseProxy's trick: + // resp.Write then serializes status line and headers as one response, and + // nil'ing the body stops it at the blank line. + for k, vv := range resp.Header { + w.Header()[k] = vv + } + resp.Header = w.Header() + resp.Body = nil + if err := resp.Write(brw); err != nil { + slog.Error("forward-proxy: writing the 101 failed", "host", host, "error", err) + s.recordSpliceFailure(pctx, record, fmt.Sprintf("writing the 101 failed: %v", err)) + return + } + if err := brw.Flush(); err != nil { + slog.Error("forward-proxy: flushing the 101 failed", "host", host, "error", err) + s.recordSpliceFailure(pctx, record, fmt.Sprintf("flushing the 101 failed: %v", err)) + return + } + + // A delivered answer from here: the deferred RunFinish reads any non-zero + // status as OutcomeAllow, and 101 is what the client got. + pctx.StatusCode = http.StatusSwitchingProtocols + + // Both directions, each one io.Copy so its return count needs no second + // counter. Up is client → origin, read from brw rather than conn so bytes + // the hijack buffered are not lost; down is origin → client, written + // straight to conn, which has no write buffer to strand. An EOF propagates + // as CloseWrite where the connection has it (ReverseProxy's + // switchProtocolCopier), so a half-closed WebSocket shuts down instead of + // hanging on the open half. + var up, down int64 + errc := make(chan error, 2) + go func() { + var err error + if up, err = io.Copy(backConn, brw); err != nil { + errc <- err + return + } + if wc, ok := backConn.(interface{ CloseWrite() error }); ok { + errc <- wc.CloseWrite() + return + } + errc <- nil + }() + go func() { + var err error + if down, err = io.Copy(conn, backConn); err != nil { + errc <- err + return + } + if wc, ok := conn.(interface{ CloseWrite() error }); ok { + errc <- wc.CloseWrite() + return + } + errc <- nil + }() + // Both, not one: the counts are read below, and a goroutine's send on errc + // is what publishes its write. + // + // But a FAILED direction has to tear both ends down before we wait for the + // second, or that wait never returns. A clean EOF propagates as CloseWrite + // above, which is what lets a half-closed WebSocket finish; an error has no + // such signal, so the peer of the broken direction stays blocked in Read + // with nothing coming. Measured: an origin that RSTs mid-session ends the + // origin→client copy with an error while client→origin stays blocked on a + // silent client, and the 101 row — recorded after both — is never written at + // all. That is the one-event-per-upgrade promise lost on exactly the + // connections worth auditing. + // + // ReverseProxy avoids this by returning after the FIRST result and letting + // its deferred closes unblock the other; this function waits for both + // because it needs their byte counts, so it closes explicitly instead. The + // counts stay correct: io.Copy returns what it copied before the error, and + // the closes only release a direction that is already finished or broken + // (#1549 review). + spliceErr := <-errc + if spliceErr != nil { + _ = conn.Close() + _ = backConn.Close() + } + if err := <-errc; spliceErr == nil { + spliceErr = err + } + + if !record || s.Sessions == nil { + return + } + // THE ROW SAYS WHETHER THE SESSION ENDED CLEANLY. Both directions have + // finished either way and the status is still 101 — the upgrade did succeed, + // and the client got its 101 — but a relay that ended in an error is not the + // same event as one that closed cleanly, and without this the only symptom + // was truncated byte counts: an operator auditing a WebSocket that broke + // mid-session read a successful upgrade. One event per upgrade is the MOCA + // contract; one event that cannot distinguish the two outcomes only half + // keeps it. + // + // NOT EVERY ERROR HERE IS A FAILURE, and net.ErrClosed is the one that is not. + // A WebSocket normally ends with one side closing; the deferred conn.Close() + // and the context goroutine's backConn.Close() then race the direction still + // blocked in Read, which returns "use of closed network connection". That is + // the ordinary shutdown of a healthy session — marking it would flag EVERY + // WebSocket as broken, which is worse than the gap this closes. Caught by + // assertUpgradedRows' own "no error on a clean session" assertion, which is + // why that one exists alongside the broken-direction test: a marker that is + // always set distinguishes nothing. + // + // proxy_error otherwise, the kind the two pre-splice failure paths already + // use, because a relay that broke is on our side. Sanitized: a network + // error's text can carry a client-supplied address (#1549 review). + var fail *pipeline.EventError + if spliceErr != nil && !errors.Is(spliceErr, net.ErrClosed) { + fail = &pipeline.EventError{ + Kind: "proxy_error", + Message: "websocket relay ended in error: " + pipeline.SanitizeControlRunes(spliceErr.Error()), + } + } + ev := outboundResponseEvent(pctx, http.StatusSwitchingProtocols, fail) + ev.BytesUp, ev.BytesDown = up, down + s.Sessions.Append(s.responseSessionID(pctx), ev) +} + +// recordSpliceFailure records the response row for a 101 whose relay broke after +// the hijack — the connection is the client's problem now, so there is no status +// to write it, but the request row still exists and a row with no pair reads as a +// request still in flight (#1045). 502 with proxy_error, as failUpgradedWebSocket: +// this failure is ours. +func (s *Server) recordSpliceFailure(pctx *pipeline.Context, record bool, cause string) { + if !record { + return + } + s.recordOutboundResponseEvent(pctx, http.StatusBadGateway, &pipeline.EventError{ + Kind: "proxy_error", + Message: cause, + }) +} + +// failUpgradedWebSocket answers a 101 this proxy could not relay and records the +// failure row. The status stays off pctx (zero reads as OutcomeError to the +// deferred RunFinish) for the same reason the client.Do error path in +// serveOutbound keeps its 502 off it: nothing succeeded here. +func (s *Server) failUpgradedWebSocket(w http.ResponseWriter, pctx *pipeline.Context, record bool, cause string) { + http.Error(w, `{"error":"bad gateway"}`, http.StatusBadGateway) + s.recordSpliceFailure(pctx, record, cause) +} + // tunnelAuthorityMatches reports whether host, a bridged request's Host, names the same // authority as tunnel, the CONNECT target or transparent destination it was decrypted // from. Hosts compare case-insensitively without a trailing dot, IP literals by value, @@ -1591,7 +1924,17 @@ func syncHeaders(r *http.Request, h http.Header) { // stripForUpstream removes from r what is the proxy's and not the upstream's: the // hop-by-hop headers, and, when ownEncoding, the client's Accept-Encoding. It runs on // every request the proxy sends upstream for a client, a resend included. -func stripForUpstream(r *http.Request, ownEncoding bool) { +// +// keepUpgrade is the WebSocket exception (#1488). RFC 9110 makes Upgrade +// hop-by-hop, and the strip is right everywhere the proxy is a reader between two +// endpoints — but an upgrade exchange makes it an endpoint of the hop instead: +// everything after the 101 is opaque to it, so the header must cross or the +// origin never sees an upgrade request at all. The pair is normalized rather than +// forwarded raw (mirroring httputil.ReverseProxy): Connection becomes "Upgrade" +// and Upgrade "websocket", the one protocol this proxy forwards — a client's +// "Connection: keep-alive, Upgrade" is not the proxy's to pass on, and any other +// token sharing the client's Upgrade list was never going to be forwarded anyway. +func stripForUpstream(r *http.Request, ownEncoding, keepUpgrade bool) { // Remove hop-by-hop headers r.Header.Del("Connection") r.Header.Del("Keep-Alive") @@ -1602,6 +1945,10 @@ func stripForUpstream(r *http.Request, ownEncoding bool) { r.Header.Del("Trailer") r.Header.Del("Transfer-Encoding") r.Header.Del("Upgrade") + if keepUpgrade { + r.Header.Set("Connection", "Upgrade") + r.Header.Set("Upgrade", "websocket") + } // Strip the client's Accept-Encoding so Go's transport negotiates content // coding on its own behalf. @@ -1686,7 +2033,10 @@ func (s *Server) resendIfAsked(r *http.Request, pctx *pipeline.Context, client * again := r.Clone(r.Context()) syncHeaders(again, pctx.Headers) setBody(again, pctx.Body) - stripForUpstream(again, ownEncoding) + // No keepUpgrade: a resent request is never a forwarded upgrade — the + // 101 branch in serveOutbound skips resendIfAsked entirely, and only that + // branch leaves the headers unstripped. + stripForUpstream(again, ownEncoding, false) var err error if resp, err = client.Do(again); err != nil { return nil, err @@ -1984,8 +2334,19 @@ func (s *Server) recordOutboundResponseEvent(pctx *pipeline.Context, statusCode // when nothing was pinned (defensive — a response-only path), then to // the default bucket. See Context.OutboundSessionID. sid := s.responseSessionID(pctx) + // Always record — see the request-phase comment. This is what surfaces + // responses no plugin acted on (e.g. a generic 404), carrying StatusCode + // + Error even with empty invocations. + s.Sessions.Append(sid, outboundResponseEvent(pctx, statusCode, fail)) +} + +// outboundResponseEvent builds the SessionResponse row every outbound answer is +// recorded through — the shared shape, so the WebSocket splice +// (relayUpgradedWebSocket) can adjust its byte fields without forking the +// snapshotting logic. +func outboundResponseEvent(pctx *pipeline.Context, statusCode int, fail *pipeline.EventError) pipeline.SessionEvent { plugins := pipeline.SnapshotPlugins(pctx.Extensions.Custom) - ev := pipeline.SessionEvent{ + return pipeline.SessionEvent{ At: time.Now(), Direction: pipeline.Outbound, Phase: pipeline.SessionResponse, @@ -2009,10 +2370,6 @@ func (s *Server) recordOutboundResponseEvent(pctx *pipeline.Context, statusCode ParentSubagent: pctx.ParentSubagent(), Proc: pctx.AttributedProc(), } - // Always record — see the request-phase comment. This is what surfaces - // responses no plugin acted on (e.g. a generic 404), carrying StatusCode - // + Error even with empty invocations. - s.Sessions.Append(sid, ev) } // bufferResponseBody reads resp's body whole for the response phase. When it diff --git a/core/listener/forwardproxy/upgradedeclined_test.go b/core/listener/forwardproxy/upgradedeclined_test.go index fd60edeef..f004afba3 100644 --- a/core/listener/forwardproxy/upgradedeclined_test.go +++ b/core/listener/forwardproxy/upgradedeclined_test.go @@ -280,7 +280,9 @@ func TestForwardProxy_WebSocketTokenMatchesCaseInsensitively(t *testing.T) { } // Upgrade is a comma-separated list (RFC 9110 §7.8), possibly repeated, and websocket -// anywhere in it is a WebSocket request. +// anywhere in it is a WebSocket request. Every case here carries the Connection: +// upgrade token the same section requires alongside it — see the Connection table +// below for what happens without it. func TestUpgradesToWebSocket_ReadsTheTokenList(t *testing.T) { cases := []struct { name string @@ -299,7 +301,7 @@ func TestUpgradesToWebSocket_ReadsTheTokenList(t *testing.T) { } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - h := http.Header{} + h := http.Header{"Connection": []string{"Upgrade"}} for _, v := range tc.values { h.Add("Upgrade", v) } @@ -310,6 +312,43 @@ func TestUpgradesToWebSocket_ReadsTheTokenList(t *testing.T) { } } +// RFC 9110 §7.8: Upgrade is only an upgrade request when Connection lists the +// upgrade token, and a recipient ignores Upgrade otherwise — the check net/http's +// own upgradeType makes before ReverseProxy honours one. +// +// Not a formality here: stripForUpstream SETS the normalized Connection/Upgrade +// pair whenever the request is treated as an upgrade, so honouring a bare Upgrade +// would have this proxy synthesize a handshake the client never sent, and splice +// the origin's 101 onto a connection whose client is waiting for an ordinary +// response (#1549 review). +func TestUpgradesToWebSocket_RequiresTheConnectionToken(t *testing.T) { + cases := []struct { + name string + connection []string + want bool + }{ + {"absent", nil, false}, + {"keep-alive only", []string{"keep-alive"}, false}, + {"close", []string{"close"}, false}, + {"upgrade", []string{"upgrade"}, true}, + {"capitalised", []string{"Upgrade"}, true}, + {"in a list", []string{"keep-alive, Upgrade"}, true}, + {"in a repeated header", []string{"keep-alive", "upgrade"}, true}, + {"a longer token", []string{"upgrades"}, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + h := http.Header{"Upgrade": []string{"websocket"}} + for _, v := range tc.connection { + h.Add("Connection", v) + } + if got := upgradesToWebSocket(h); got != tc.want { + t.Errorf("upgradesToWebSocket(Connection=%q) = %v, want %v", tc.connection, got, tc.want) + } + }) + } +} + // A plugin's reject still wins: the decline runs after the pipeline, so a policy that // refuses the request is not overridden by a 426 the client would then route around. func TestForwardProxy_PluginRejectWinsOverTheDecline(t *testing.T) { diff --git a/core/listener/forwardproxy/upgradestripped_test.go b/core/listener/forwardproxy/upgradestripped_test.go index bd15ec43a..a1fe9056e 100644 --- a/core/listener/forwardproxy/upgradestripped_test.go +++ b/core/listener/forwardproxy/upgradestripped_test.go @@ -77,19 +77,20 @@ func TestRecordOutboundRequest_OrdinaryRequestRecordsNoUpgrade(t *testing.T) { } } -// stripForUpstream must still remove the header — the marker is a record of the -// strip, not a replacement for it. Forwarding Upgrade would violate RFC 9110's -// hop-by-hop rule, so a change that recorded the value and then left it in place -// would pass the two tests above while breaking the behaviour they describe. +// stripForUpstream must still remove the header on a request that is not having its +// WebSocket forwarded — the marker is a record of the strip, not a replacement for it. +// Forwarding Upgrade on an ordinary request would violate RFC 9110's hop-by-hop rule, +// so a change that recorded the value and then left it in place would pass the two +// tests above while breaking the behaviour they describe. func TestStripForUpstream_StillRemovesTheUpgrade(t *testing.T) { r, err := http.NewRequest("GET", "https://chatgpt.com/backend-api/codex/responses", nil) if err != nil { t.Fatal(err) } - r.Header.Set("Upgrade", "websocket") + r.Header.Set("Upgrade", "h2c") r.Header.Set("Connection", "Upgrade") - stripForUpstream(r, false) + stripForUpstream(r, false, false) if got := r.Header.Get("Upgrade"); got != "" { t.Errorf("Upgrade = %q, want it removed — forwarding it breaks the hop-by-hop rule", got) @@ -99,20 +100,48 @@ func TestStripForUpstream_StillRemovesTheUpgrade(t *testing.T) { } } -// END TO END, and the only test here that would notice the call site being deleted: -// the three above drive pctx.RecordStrippedUpgrade themselves, so removing -// handleRequest's call leaves them green. This one sends a real request with a real -// Upgrade header through a real proxy and reads the recorded event. +// The forwarded case normalizes rather than passes through (#1488): whatever else +// shared the client's Connection and Upgrade lists, the origin sees exactly +// "Connection: Upgrade" and "Upgrade: websocket" — the one protocol this proxy +// forwards, in the one form RFC 9110 §7.8 asks for. The two tests around this one +// hold the other side: a non-websocket token, and the no-upgrade request, strip. +func TestStripForUpstream_ForwardedWebSocketIsNormalized(t *testing.T) { + r, err := http.NewRequest("GET", "https://ws.example/chat", nil) + if err != nil { + t.Fatal(err) + } + r.Header.Set("Upgrade", "websocket, h2c") + r.Header.Set("Connection", "keep-alive, Upgrade") + + stripForUpstream(r, false, true) + + if got := r.Header.Get("Upgrade"); got != "websocket" { + t.Errorf("Upgrade = %q, want exactly websocket", got) + } + if got := r.Header.Get("Connection"); got != "Upgrade" { + t.Errorf("Connection = %q, want exactly Upgrade", got) + } +} + +// END TO END, and the only test here that would notice the call sites being deleted: +// the ones above drive pctx.RecordStrippedUpgrade and stripForUpstream themselves, so +// removing serveOutbound's calls leaves them green. This one sends a real request with +// a real Upgrade header through a real proxy and reads both the wire and the recorded +// events. // -// It also asserts the OTHER half in the same pass — that the origin received no -// Upgrade — so a change that recorded the marker by keeping the header fails here -// rather than looking like an improvement. +// SINCE #1488 a WebSocket upgrade is FORWARDED: the origin receives Connection: +// Upgrade and Upgrade: websocket, and the request row records no marker — the +// marker is the record of a strip, and nothing was stripped. The origin here +// answers 200 rather than 101, which is an origin declining the upgrade per RFC +// 9110 §7.8; the 200 is relayed the ordinary way, and that ordinary path is what +// leaves the row this test reads. The 101 itself has its own end-to-end test in +// websocketupgrade_test.go. // -// WITH THE DECLINE OFF. On a Responses path a WebSocket is otherwise answered 426 -// before anything is forwarded (upgradedeclined_test.go), so the origin would see -// nothing and the "received no Upgrade" check below would pass without testing the -// strip at all. That is why it also asserts the origin was reached. -func TestForwardProxy_EndToEnd_RecordsAndStripsTheUpgrade(t *testing.T) { +// WITH THE DECLINE OFF, as before: on a Responses path a WebSocket is otherwise +// answered 426 before anything is forwarded (upgradedeclined_test.go), so the +// origin would see nothing and the forwarding checks below would pass without +// testing them at all. That is why it also asserts the origin was reached. +func TestForwardProxy_EndToEnd_ForwardsTheWebSocketUpgrade(t *testing.T) { store := session.New(5*time.Minute, 100, 0) defer store.Close() @@ -155,6 +184,7 @@ func TestForwardProxy_EndToEnd_RecordsAndStripsTheUpgrade(t *testing.T) { t.Fatal(err) } req.Header.Set("Upgrade", "websocket") + req.Header.Set("Connection", "keep-alive, Upgrade") resp, err := client.Do(req) if err != nil { t.Fatal(err) @@ -165,29 +195,29 @@ func TestForwardProxy_EndToEnd_RecordsAndStripsTheUpgrade(t *testing.T) { gotReached, gotUpgrade, gotConnection := reached, sawUpgrade, sawConnection mu.Unlock() if !gotReached { - t.Fatal("the origin was never reached, so the strip below is untested") + t.Fatal("the origin was never reached, so the forwarding below is untested") } - if gotUpgrade != "" { - t.Errorf("the origin received Upgrade: %q — it must be stripped", gotUpgrade) + if gotUpgrade != "websocket" { + t.Errorf("the origin received Upgrade: %q — want websocket, forwarded and normalized", gotUpgrade) } - if gotConnection != "" { - t.Errorf("the origin received Connection: %q — it must be stripped", gotConnection) + if gotConnection != "Upgrade" { + t.Errorf("the origin received Connection: %q — want exactly Upgrade", gotConnection) } v := store.View(session.DefaultSessionID) if v == nil || len(v.Events) == 0 { t.Fatalf("no events recorded: %+v", v) } - var found bool + var stripped int for _, ev := range v.Events { - if ev.Phase == pipeline.SessionRequest && ev.UpgradeStripped == "websocket" { - found = true + if ev.Phase == pipeline.SessionRequest && ev.UpgradeStripped != "" { + stripped++ } } - if !found { + if stripped != 0 { for _, ev := range v.Events { t.Logf(" phase=%v upgradeStripped=%q path=%q", ev.Phase, ev.UpgradeStripped, ev.HTTPPath) } - t.Error("no request event carried UpgradeStripped=websocket — is handleRequest still calling RecordStrippedUpgrade before stripForUpstream?") + t.Errorf("%d request events carried UpgradeStripped — a forwarded upgrade must not be marked stripped", stripped) } } diff --git a/core/listener/forwardproxy/websocketclose_test.go b/core/listener/forwardproxy/websocketclose_test.go new file mode 100644 index 000000000..296d12f00 --- /dev/null +++ b/core/listener/forwardproxy/websocketclose_test.go @@ -0,0 +1,254 @@ +package forwardproxy + +import ( + "bufio" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/session" + "github.com/rossoctl/cortex/core/tlsbridge" +) + +// wsClosingHandler writes the 101, waits, then closes — an origin that hangs up +// FIRST. That is RFC 6455 §7.1.1's server-initiated close (FIN after the close +// frame), and also what an origin restart or an LB idle timeout looks like. +// +// Every other upgrade test has the CLIENT close first, which is exactly why the +// bridged half-close gap survived: a client-initiated close is driven from the +// agent side and never needs the origin's FIN to travel back. +func wsClosingHandler(delay time.Duration) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || r.Header.Get("Upgrade") != "websocket" { + http.Error(w, "expected a websocket upgrade", http.StatusBadRequest) + return + } + conn, brw, err := http.NewResponseController(w).Hijack() + if err != nil { + return + } + fmt.Fprintf(brw, "HTTP/1.1 101 Switching Protocols\r\n"+ + "Upgrade: websocket\r\nConnection: Upgrade\r\n"+ + "Sec-WebSocket-Accept: %s\r\n\r\n", wsAccept(r.Header.Get("Sec-WebSocket-Key"))) + if err := brw.Flush(); err != nil { + return + } + time.Sleep(delay) + _ = conn.Close() // FIN toward the proxy; the splice must forward it + } +} + +// readAfterOriginClose reads until the read fails and reports the error. A +// forwarded close surfaces as EOF (or a reset); a close that was NOT forwarded +// leaves the client blocked until its own deadline, so the error is a timeout. +func readAfterOriginClose(t *testing.T, r io.Reader) error { + t.Helper() + buf := make([]byte, 64) + for { + if _, err := r.Read(buf); err != nil { + return err + } + } +} + +func isTimeout(err error) bool { + var ne net.Error + return errors.As(err, &ne) && ne.Timeout() +} + +// THE REGRESSION. On the TLS-bridged path the origin's half-close must reach the +// agent. It did not: tlsbridge.ServeConn serves the terminated conn through +// newOneConnListener, which hands Hijack a *notifyConn — and notifyConn embeds +// the net.Conn INTERFACE, so CloseWrite is not in its method set whatever the +// concrete type underneath is. The splice's +// +// if wc, ok := conn.(interface{ CloseWrite() error }); ok +// +// therefore failed on the bridge and the goroutine sent nil, forwarding nothing. +// The agent got neither close_notify nor FIN and waited on a WebSocket whose +// other end was gone, while the other copy stayed blocked on the silent agent — +// so the 101 row, recorded after BOTH directions, never landed either. Same +// one-event-per-upgrade loss as the RST case, on the clean-close path. +// +// The plain path was always fine: there conn is a *net.TCPConn, which has +// CloseWrite. +// +// Both paths are asserted here, in one test, because the bug WAS the asymmetry — +// splitting them invites fixing one and leaving the other (#1549 review). +func TestWebSocketSplice_OriginClosesFirst(t *testing.T) { + const originDelay = 100 * time.Millisecond + + t.Run("plain", func(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + t.Cleanup(store.Close) + origin := httptest.NewServer(wsClosingHandler(originDelay)) + t.Cleanup(origin.Close) + + p, err := pipeline.New(nil) + if err != nil { + t.Fatalf("pipeline.New: %v", err) + } + srv := &Server{ + OutboundPipeline: pipeline.NewHolder(p), + Sessions: store, + Client: http.DefaultClient, + } + proxy := httptest.NewServer(srv.Handler()) + t.Cleanup(proxy.Close) + + originURL, err := url.Parse(origin.URL) + if err != nil { + t.Fatalf("parse origin URL: %v", err) + } + proxyURL, err := url.Parse(proxy.URL) + if err != nil { + t.Fatalf("parse proxy URL: %v", err) + } + conn, err := net.Dial("tcp", proxyURL.Host) + if err != nil { + t.Fatalf("dial proxy: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + // Generous relative to the origin's 100ms, so a timeout here means the + // close was never forwarded rather than that the test was impatient. + _ = conn.SetDeadline(time.Now().Add(3 * time.Second)) + + writeUpgradeRequest(conn, origin.URL+"/ws", originURL.Host) + br := bufio.NewReader(conn) + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("read upgrade response: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusSwitchingProtocols { + t.Fatalf("upgrade status = %d, want 101", resp.StatusCode) + } + + if err := readAfterOriginClose(t, br); isTimeout(err) { + t.Fatalf("plain: client never saw the origin's close (%v) — the splice dropped the half-close", err) + } + }) + + t.Run("bridged", func(t *testing.T) { + st := newClosingBridgeStack(t, originDelay) + + tconn := st.connect(t) + writeUpgradeRequest(tconn, "/ws", st.originHostPort) + br := bufio.NewReader(tconn) + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("read upgrade response over the bridge: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusSwitchingProtocols { + t.Fatalf("upgrade status = %d, want 101 through the bridge", resp.StatusCode) + } + + // The headline assertion: the origin hung up, so this read must END. + // Before the fix it blocked for the full deadline and reported a + // timeout, with the agent still holding a dead WebSocket. + if err := readAfterOriginClose(t, br); isTimeout(err) { + t.Fatalf("bridged: client never saw the origin's close (%v) — notifyConn has no CloseWrite, so the splice forwarded nothing", err) + } + + // And the 101 row lands. The client closes on seeing EOF, as a real + // WebSocket client does once the peer has gone — the row is recorded + // after BOTH directions, and the client→origin copy is still blocked + // on this (silent) client until it does. The point of the assertion + // is that the close reaching the client is what UNBLOCKS that: before + // the fix the client never learned to close, so the row never landed + // at all until the client's own deadline expired. + // + // assertUpgradedRows is not reused here: it requires non-zero byte + // counts both ways, and this session deliberately exchanges no frames + // (the origin closes straight after the 101). What matters on this row + // is that it EXISTS and reports a clean close, so the fields are + // checked directly. + _ = tconn.Close() + events := waitForUpgradedRows(t, st.store) + var row *pipeline.SessionEvent + for i := range events { + if events[i].Phase == pipeline.SessionResponse && events[i].StatusCode == http.StatusSwitchingProtocols { + row = &events[i] + break + } + } + if row == nil { + t.Fatalf("no 101 response row among %d events — one event per upgrade is the contract", len(events)) + } + // An origin closing first is an ORDINARY end to a WebSocket, not a + // broken relay, so the row must not be marked. proxy_error here would + // make every server-initiated close read as a failure. + if row.Error != nil { + t.Errorf("101 row carries error %+v — an origin-initiated close is a clean end", row.Error) + } + }) +} + +// newClosingBridgeStack is newBridgeStack with an origin that closes first. Kept +// separate rather than parameterising newBridgeStack, so the existing tests' +// setup is untouched. +func newClosingBridgeStack(t *testing.T, delay time.Duration) *bridgeStack { + t.Helper() + st := &bridgeStack{} + origin := httptest.NewUnstartedServer(wsClosingHandler(delay)) + origin.EnableHTTP2 = true + origin.StartTLS() + + originURL, err := url.Parse(origin.URL) + if err != nil { + origin.Close() + t.Fatalf("parse origin URL: %v", err) + } + originCAPEM := certPEM(origin) + + src, err := tlsbridge.NewEphemeralSource() + if err != nil { + origin.Close() + t.Fatalf("NewEphemeralSource: %v", err) + } + up, err := tlsbridge.NewUpstreamClient(originCAPEM, false) + if err != nil { + origin.Close() + t.Fatalf("NewUpstreamClient: %v", err) + } + engine := &tlsbridge.Engine{ + Decision: mustBridgeDecision(t, portOf(originURL.Host)), + Term: tlsbridge.NewTerminator(tlsbridge.NewMinter(src, tlsbridge.MinterOpts{})), + Skip: tlsbridge.NewSkipSet(), + Upstream: up, + CAPEM: src.CACertPEM(), + } + + store := session.New(5*time.Minute, 100, 0) + p, err := pipeline.New(nil) + if err != nil { + origin.Close() + store.Close() + t.Fatalf("pipeline.New: %v", err) + } + srv := &Server{ + OutboundPipeline: pipeline.NewHolder(p), + Sessions: store, + Client: http.DefaultClient, + TLSBridge: engine, + } + proxy := httptest.NewServer(srv.Handler()) + t.Cleanup(func() { + proxy.Close() + store.Close() + origin.Close() + }) + st.store = store + st.proxy = proxy + st.originHostPort = originURL.Host + st.engine = engine + return st +} diff --git a/core/listener/forwardproxy/websocketupgrade_test.go b/core/listener/forwardproxy/websocketupgrade_test.go new file mode 100644 index 000000000..3c5e62ae6 --- /dev/null +++ b/core/listener/forwardproxy/websocketupgrade_test.go @@ -0,0 +1,811 @@ +package forwardproxy + +// #1488 end to end: a WebSocket upgrade crossing the forward proxy — plain and +// TLS-bridged — is forwarded, the 101 is spliced, frames flow both ways, and the +// audit shows one event: the request row plus its paired 101 response row. + +import ( + "bufio" + "bytes" + "crypto/sha1" //nolint:gosec // RFC 6455 §1.3 mandates SHA-1 over key+GUID for the accept token; a sanity check, not a secret + "crypto/tls" + "crypto/x509" + "encoding/base64" + "encoding/pem" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "sync" + "testing" + "time" + + "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/session" + "github.com/rossoctl/cortex/core/tlsbridge" +) + +// wsGUID is the fixed GUID RFC 6455 §1.3 folds into the accept computation. +const wsGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + +// The RFC 6455 §1.3 sample handshake pair. Using the spec's own key means the +// origin's accept is assertable against a published constant, which is what +// proves the 101 the client read is the origin's own answer relayed unmodified +// rather than something the proxy synthesized. +const ( + rfc6455Key = "dGhlIHNhbXBsZSBub25jZQ==" + rfc6455Accept = "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=" +) + +func wsAccept(key string) string { + sum := sha1.Sum([]byte(key + wsGUID)) + return base64.StdEncoding.EncodeToString(sum[:]) +} + +// wsEchoHandler is a minimal WebSocket echo endpoint: a real 101 (accept +// computed per RFC 6455) followed by a byte-level echo of everything after it. +// No frame parsing — the proxy under test must not parse them either, and a +// byte-level echo is what proves the splice is transparent in both directions. +func wsEchoHandler(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || r.Header.Get("Upgrade") != "websocket" || r.Header.Get("Sec-WebSocket-Version") != "13" { + http.Error(w, "expected a version-13 websocket upgrade", http.StatusBadRequest) + return + } + conn, brw, err := http.NewResponseController(w).Hijack() + if err != nil { + return + } + defer conn.Close() + fmt.Fprintf(brw, "HTTP/1.1 101 Switching Protocols\r\n"+ + "Upgrade: websocket\r\nConnection: Upgrade\r\n"+ + "Sec-WebSocket-Accept: %s\r\n\r\n", wsAccept(r.Header.Get("Sec-WebSocket-Key"))) + if err := brw.Flush(); err != nil { + return + } + // The echo: client → origin → client, one copy. Reads from brw so bytes + // buffered past the request line are not lost — the same rule the proxy's + // own splice follows. + _, _ = io.Copy(conn, brw) +} + +// maskedTextFrame builds one masked RFC 6455 text frame — masking is mandatory +// client-to-server — so the exchange after the 101 has the shape of a real +// WebSocket's. A byte-level echo returns the frame unchanged, so the test +// asserts the exact bytes back. +func maskedTextFrame(payload string) []byte { + p := []byte(payload) + frame := []byte{0x81, 0x80 | byte(len(p))} // FIN + text opcode; MASK set, 7-bit length + mask := [4]byte{0x37, 0xfa, 0x21, 0x3d} + frame = append(frame, mask[:]...) + for i, c := range p { + frame = append(frame, c^mask[i%4]) + } + return frame +} + +// writeUpgradeRequest sends the upgrade GET by hand: no stock Go client will +// both speak to a raw TCP peer and surrender its connection after a 101, which +// is exactly the behaviour under test. target is the request target as the +// receiving server must see it — absolute-form through the plain forward proxy, +// origin-form on the bridged connection — and host the Host header. +func writeUpgradeRequest(w io.Writer, target, host string) { + fmt.Fprintf(w, "GET %s HTTP/1.1\r\nHost: %s\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"+ + "Sec-WebSocket-Key: %s\r\nSec-WebSocket-Version: 13\r\n\r\n", target, host, rfc6455Key) +} + +// echoRound sends one masked frame over conn and reads what comes back on br, +// asserting the echo is byte-identical. Used by both the plain and bridged +// tests, after each has read the 101 its own way. +func echoRound(t *testing.T, conn io.Writer, br io.Reader) { + t.Helper() + frame := maskedTextFrame("hello through the proxy") + if _, err := conn.Write(frame); err != nil { + t.Fatalf("write frame: %v", err) + } + echo := make([]byte, len(frame)) + if _, err := io.ReadFull(br, echo); err != nil { + t.Fatalf("read echo: %v", err) + } + if !bytes.Equal(echo, frame) { + t.Errorf("echo = %x, want the frame back unchanged: %x", echo, frame) + } +} + +// waitForUpgradedRows polls the store until the 101's response row arrives. +// That row is recorded only after BOTH splice directions end — i.e. after the +// client closes — so waiting for event COUNT is not enough: the tunnel-open and +// request rows land before client.Do even returns, and a count-based poll would +// race the splice's teardown. +func waitForUpgradedRows(t *testing.T, store *session.Store) []pipeline.SessionEvent { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for { + v := store.View(session.DefaultSessionID) + if v != nil { + for _, ev := range v.Events { + if ev.Phase == pipeline.SessionResponse && ev.StatusCode == http.StatusSwitchingProtocols { + return v.Events + } + } + } + if time.Now().After(deadline) { + if v == nil { + t.Fatal("no events recorded within 5s") + } + for _, ev := range v.Events { + t.Logf(" phase=%v tunnel=%v path=%q status=%d", ev.Phase, ev.Tunnel, ev.HTTPPath, ev.StatusCode) + } + t.Fatalf("no 101 response row within 5s; %d events recorded", len(v.Events)) + } + time.Sleep(10 * time.Millisecond) + } +} + +// assertUpgradedRows checks the audit shape #1488 promises: a request row with +// no strip marker, paired by RequestID with a 101 response row carrying both +// directions' byte counts — the one response row that does, like a tunnel +// close, because after the 101 the proxy does not parse what crosses. +func assertUpgradedRows(t *testing.T, events []pipeline.SessionEvent, path string) { + t.Helper() + var req *pipeline.SessionEvent + var resp *pipeline.SessionEvent + for i := range events { + ev := &events[i] + switch { + case ev.Phase == pipeline.SessionRequest && !ev.Tunnel && ev.HTTPPath == path: + if req == nil { + req = ev + } + case ev.Phase == pipeline.SessionResponse && ev.StatusCode == http.StatusSwitchingProtocols: + if resp == nil { + resp = ev + } + } + } + if req == nil { + for i := range events { + t.Logf(" phase=%v tunnel=%v path=%q status=%d", events[i].Phase, events[i].Tunnel, events[i].HTTPPath, events[i].StatusCode) + } + t.Fatalf("no request row for %q", path) + } + if resp == nil { + t.Fatalf("no 101 response row among %d events", len(events)) + } + if req.UpgradeStripped != "" { + t.Errorf("request row carries UpgradeStripped=%q — a forwarded upgrade must not be marked stripped", req.UpgradeStripped) + } + if req.RequestID != resp.RequestID { + t.Errorf("request row id %q does not pair with the 101 row id %q", req.RequestID, resp.RequestID) + } + if resp.BytesUp == 0 || resp.BytesDown == 0 { + t.Errorf("101 row bytes: up=%d down=%d — both directions carried the frame, both must be counted", resp.BytesUp, resp.BytesDown) + } + // A cleanly closed session carries NO error. The counterpart assertion lives + // in TestForwardProxy_WebSocketBrokenDirectionDoesNotStrandTheOther, which + // requires one; both halves are needed, because a marker that is always set + // distinguishes nothing and would make every WebSocket read as broken. + if resp.Error != nil { + t.Errorf("101 row carries error %+v on a clean session — the marker must mean a relay that failed", resp.Error) + } +} + +// THE PLAIN PATH. A ws:// upgrade arrives as an absolute-form GET on the +// forward proxy's ordinary listener; the 101 comes back over the same +// connection, hijacked and spliced, and the echo crosses both ways through the +// proxy's own process. The done-when of #1488, on the non-intercepted side. +func TestForwardProxy_WebSocketEcho(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + + origin := httptest.NewServer(http.HandlerFunc(wsEchoHandler)) + defer origin.Close() + originURL, err := url.Parse(origin.URL) + if err != nil { + t.Fatal(err) + } + + p, err := pipeline.New(nil) + if err != nil { + t.Fatal(err) + } + srv := &Server{ + OutboundPipeline: pipeline.NewHolder(p), + Sessions: store, + Client: http.DefaultClient, + } + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + proxyURL, err := url.Parse(proxy.URL) + if err != nil { + t.Fatal(err) + } + + conn, err := net.Dial("tcp", proxyURL.Host) + if err != nil { + t.Fatalf("dial proxy: %v", err) + } + _ = conn.SetDeadline(time.Now().Add(10 * time.Second)) + + writeUpgradeRequest(conn, origin.URL+"/ws", originURL.Host) + br := bufio.NewReader(conn) + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("read upgrade response: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusSwitchingProtocols { + t.Fatalf("upgrade status = %d, want 101", resp.StatusCode) + } + if got := resp.Header.Get("Sec-WebSocket-Accept"); got != rfc6455Accept { + t.Errorf("Sec-WebSocket-Accept = %q, want the RFC 6455 sample %q — the 101 must be the origin's own, relayed unmodified", got, rfc6455Accept) + } + + echoRound(t, conn, br) + + // Closing is what ends the splice's two copies, and the 101 row records + // after they both do. + _ = conn.Close() + assertUpgradedRows(t, waitForUpgradedRows(t, store), "/ws") +} + +// THE BRIDGED PATH, the one #1488 is about: the agent CONNECTs, the bridge +// terminates the TLS, and the upgrade GET arrives decrypted. The origin is +// h2-capable on purpose, and the test asserts it still served the upgrade over +// HTTP/1.1 — the load-bearing stdlib behavior the whole forwarding rests on +// (net/http Request.requiresHTTP1 pins a WebSocket upgrade to HTTP/1.1, so the +// bridge's h2-negotiating upstream client needs no upgrade-specific twin). The +// echo crosses two splices (agent↔bridge and bridge↔origin) in each direction. +func TestConnectBridge_WebSocketEcho(t *testing.T) { + st := newBridgeStack(t) + + tconn := st.connect(t) + writeUpgradeRequest(tconn, "/ws", st.originHostPort) + br := bufio.NewReader(tconn) + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("read upgrade response over the bridge: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusSwitchingProtocols { + t.Fatalf("upgrade status = %d, want 101 through the bridge", resp.StatusCode) + } + if got := resp.Header.Get("Sec-WebSocket-Accept"); got != rfc6455Accept { + t.Errorf("Sec-WebSocket-Accept = %q, want the RFC 6455 sample %q — the 101 must be the origin's own, relayed unmodified", got, rfc6455Accept) + } + + echoRound(t, tconn, br) + + _ = tconn.Close() + if proto := st.upgradeProto(); proto != "HTTP/1.1" { + t.Errorf("the origin served the upgrade over %s — net/http should have pinned the exchange to HTTP/1.1 (Request.requiresHTTP1)", proto) + } + events := waitForUpgradedRows(t, st.store) + // The bridged tunnel's open rides along with the request row, so three + // rows: tunnel open, request, 101. + var tunnelOpen bool + for _, ev := range events { + if ev.Tunnel && ev.Phase == pipeline.SessionRequest { + tunnelOpen = true + } + } + if !tunnelOpen { + t.Errorf("no tunnel-open row among %d events — a bridged upgrade's tunnel must still be audited", len(events)) + } + assertUpgradedRows(t, events, "/ws") +} + +// The CONNECT-host check applies to the upgrade request too (MOCA MI1 §8.5 +// item 5): an upgrade GET whose Host names another authority than its tunnel's +// CONNECT target is refused 421 before anything is forwarded. +func TestConnectBridge_WebSocketMisdirectedHost(t *testing.T) { + st := newBridgeStack(t) + + tconn := st.connect(t) + writeUpgradeRequest(tconn, "/ws", "other.example:443") + resp, err := http.ReadResponse(bufio.NewReader(tconn), nil) + if err != nil { + t.Fatalf("read response over the bridge: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusMisdirectedRequest { + t.Fatalf("misdirected upgrade status = %d, want 421", resp.StatusCode) + } +} + +// bridgeStack is the CONNECT-bridge scaffolding the WebSocket tests share: an +// h2-capable TLS origin running the echo handler, a bridge engine, a forward +// proxy wired to it, and the session store the rows land in. +type bridgeStack struct { + store *session.Store + proxy *httptest.Server + originHostPort string + engine *tlsbridge.Engine + + // protoMu guards originProto, written by the origin's handler goroutine + // and read by the test after the response — a response arriving over a + // socket is not a happens-before edge Go's memory model recognises. + protoMu sync.Mutex + originProto string +} + +// upgradeProto reports the protocol the origin served the upgrade over, +// recorded by the handler below. Empty when the origin never saw an upgrade +// request. +func (st *bridgeStack) upgradeProto() string { + st.protoMu.Lock() + defer st.protoMu.Unlock() + return st.originProto +} + +func newBridgeStack(t *testing.T) *bridgeStack { + t.Helper() + + st := &bridgeStack{} + origin := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Upgrade") != "" { + st.protoMu.Lock() + st.originProto = r.Proto + st.protoMu.Unlock() + } + wsEchoHandler(w, r) + })) + origin.EnableHTTP2 = true + origin.StartTLS() + + originURL, err := url.Parse(origin.URL) + if err != nil { + origin.Close() + t.Fatalf("parse origin URL: %v", err) + } + + originCAPEM := pem.EncodeToMemory(&pem.Block{ + Type: "CERTIFICATE", + Bytes: origin.Certificate().Raw, + }) + + src, err := tlsbridge.NewEphemeralSource() + if err != nil { + origin.Close() + t.Fatalf("NewEphemeralSource: %v", err) + } + minter := tlsbridge.NewMinter(src, tlsbridge.MinterOpts{}) + up, err := tlsbridge.NewUpstreamClient(originCAPEM, false) + if err != nil { + origin.Close() + t.Fatalf("NewUpstreamClient: %v", err) + } + engine := &tlsbridge.Engine{ + Decision: mustBridgeDecision(t, portOf(originURL.Host)), + Term: tlsbridge.NewTerminator(minter), + Skip: tlsbridge.NewSkipSet(), + Upstream: up, + CAPEM: src.CACertPEM(), + } + + store := session.New(5*time.Minute, 100, 0) + p, err := pipeline.New(nil) + if err != nil { + origin.Close() + store.Close() + t.Fatalf("pipeline.New: %v", err) + } + srv := &Server{ + OutboundPipeline: pipeline.NewHolder(p), + Sessions: store, + Client: http.DefaultClient, + TLSBridge: engine, + } + proxy := httptest.NewServer(srv.Handler()) + + t.Cleanup(func() { + proxy.Close() + store.Close() + origin.Close() + }) + st.store = store + st.proxy = proxy + st.originHostPort = originURL.Host + st.engine = engine + return st +} + +// connect does the CONNECT and the agent-side TLS handshake, returning the +// decrypted connection the test speaks HTTP/1.1 on — the same sequence +// TestConnectBridge walks, kept here so each WebSocket test starts at the +// upgrade GET. +func (st *bridgeStack) connect(t *testing.T) *tls.Conn { + t.Helper() + proxyURL, err := url.Parse(st.proxy.URL) + if err != nil { + t.Fatalf("parse proxy URL: %v", err) + } + rawConn, err := net.Dial("tcp", proxyURL.Host) + if err != nil { + t.Fatalf("dial proxy: %v", err) + } + _ = rawConn.SetDeadline(time.Now().Add(10 * time.Second)) + + connectReq, err := http.NewRequest(http.MethodConnect, "//"+st.originHostPort, nil) + if err != nil { + t.Fatalf("new CONNECT request: %v", err) + } + connectReq.Host = st.originHostPort + connectReq.URL.Host = st.originHostPort + if err := connectReq.Write(rawConn); err != nil { + t.Fatalf("write CONNECT: %v", err) + } + br := bufio.NewReader(rawConn) + connectResp, err := http.ReadResponse(br, connectReq) + if err != nil { + t.Fatalf("read CONNECT response: %v", err) + } + _ = connectResp.Body.Close() + if connectResp.StatusCode != http.StatusOK { + t.Fatalf("CONNECT status = %d, want 200", connectResp.StatusCode) + } + + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(st.engine.CAPEM) { + t.Fatalf("failed to append bridge CA PEM to pool") + } + tconn := tls.Client(&bufferedConn{Conn: rawConn, r: br}, &tls.Config{ + ServerName: hostOnly(st.originHostPort), + RootCAs: pool, + NextProtos: []string{"http/1.1"}, + }) + _ = tconn.SetDeadline(time.Now().Add(10 * time.Second)) + if err := tconn.Handshake(); err != nil { + t.Fatalf("agent-side TLS handshake through the bridge failed: %v", err) + } + return tconn +} + +// A FAILED SPLICE MUST NOT LEAK THE ORIGIN'S CONNECTION. relayUpgradedWebSocket +// owns resp.Body once the origin answered 101 — serveOutbound deliberately skips +// its Close defer, because closing the body would close the WebSocket — and the +// goroutine that closes the origin on context-done is armed only after the +// hijack succeeds. So the two early returns before it have to close the origin +// themselves, or the origin is left holding a connection it believes is an +// established WebSocket while the client already has a 502. Nothing would ever +// come back for it; it sits until the origin's own idle timeout reaps it. +// +// Driven through the hijack-failure path, which is reachable with a plain +// ResponseWriter: httptest.NewRecorder does not implement http.Hijacker. The +// non-writable-body path cannot be reached through net/http's own transport at +// all (it hands a 101 an io.ReadWriteCloser), so it is covered by inspection and +// the same close, not by a test. +// +// The assertion is the ORIGIN's view, not ours: its handler hijacks and blocks +// in a copy, which returns only when our end of that connection closes. So "the +// copy returned" is the only honest evidence that we released it (#1549 review). +func TestForwardProxy_WebSocketSpliceFailureClosesTheOrigin(t *testing.T) { + copyReturned := make(chan struct{}) + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, brw, err := http.NewResponseController(w).Hijack() + if err != nil { + t.Errorf("origin could not hijack: %v", err) + return + } + defer func() { _ = conn.Close() }() + fmt.Fprintf(brw, "HTTP/1.1 101 Switching Protocols\r\n"+ + "Upgrade: websocket\r\nConnection: Upgrade\r\n"+ + "Sec-WebSocket-Accept: %s\r\n\r\n", wsAccept(r.Header.Get("Sec-WebSocket-Key"))) + if err := brw.Flush(); err != nil { + return + } + // Blocks until the proxy's end of this connection is closed. + _, _ = io.Copy(io.Discard, brw) + close(copyReturned) + })) + defer origin.Close() + + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + p, err := pipeline.New(nil) + if err != nil { + t.Fatal(err) + } + srv := &Server{ + OutboundPipeline: pipeline.NewHolder(p), + Sessions: store, + Client: origin.Client(), + } + + // httptest.NewRecorder is not an http.Hijacker, so the hijack fails after + // the origin has already answered 101 — the state this test is about. + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, origin.URL+"/ws", nil) + req.Host = req.URL.Host + req.Header.Set("Connection", "Upgrade") + req.Header.Set("Upgrade", "websocket") + req.Header.Set("Sec-WebSocket-Key", rfc6455Key) + req.Header.Set("Sec-WebSocket-Version", "13") + + srv.serveOutbound(rec, req, nil) + + if rec.Code != http.StatusBadGateway { + t.Errorf("client status = %d, want 502 — a failed splice is a bad gateway", rec.Code) + } + select { + case <-copyReturned: + case <-time.After(5 * time.Second): + t.Fatal("the origin's copy never returned: its connection was left open after the splice failed") + } +} + +// A BROKEN DIRECTION MUST NOT STRAND THE OTHER. The splice waits for both +// directions so it can report both byte counts, which means a direction that +// ends in an ERROR has to tear both ends down before that wait — a clean EOF +// propagates as CloseWrite and releases its peer, but an error carries no such +// signal, so the peer sits in Read with nothing coming. +// +// The cost is the audit, not just a goroutine: the 101 row is recorded after +// both directions end, so a stranded direction means the upgrade is never +// evented at all. Measured before the fix — an origin that RSTs mid-session +// against a client that has gone quiet leaves the row unwritten past any +// deadline; after it, the row lands in ~0.1s (#1549 review). +// +// The origin here is a raw listener rather than an httptest server because the +// test needs SetLinger(0) — a Close that sends RST — to make one direction fail +// rather than EOF. The client is deliberately silent after reading the 101: that +// is what keeps client→origin blocked, so the test observes the strand and not +// merely a slow teardown. +func TestForwardProxy_WebSocketBrokenDirectionDoesNotStrandTheOther(t *testing.T) { + origin, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer func() { _ = origin.Close() }() + go func() { + c, err := origin.Accept() + if err != nil { + return + } + req, err := http.ReadRequest(bufio.NewReader(c)) + if err != nil { + return + } + fmt.Fprintf(c, "HTTP/1.1 101 Switching Protocols\r\n"+ + "Upgrade: websocket\r\nConnection: Upgrade\r\n"+ + "Sec-WebSocket-Accept: %s\r\n\r\n", wsAccept(req.Header.Get("Sec-WebSocket-Key"))) + // RST rather than FIN, so the proxy's origin→client copy ends with an + // error instead of the clean EOF that would CloseWrite its peer. + if tc, ok := c.(*net.TCPConn); ok { + _ = tc.SetLinger(0) + } + time.Sleep(100 * time.Millisecond) + _ = c.Close() + }() + + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + p, err := pipeline.New(nil) + if err != nil { + t.Fatal(err) + } + srv := &Server{ + OutboundPipeline: pipeline.NewHolder(p), + Sessions: store, + Client: http.DefaultClient, + } + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + cc, err := net.Dial("tcp", proxy.Listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = cc.Close() }() + writeUpgradeRequest(cc, "http://"+origin.Addr().String()+"/ws", origin.Addr().String()) + resp, err := http.ReadResponse(bufio.NewReader(cc), nil) + if err != nil { + t.Fatalf("reading the 101: %v", err) + } + if resp.StatusCode != http.StatusSwitchingProtocols { + t.Fatalf("status = %d, want 101", resp.StatusCode) + } + // Then say nothing at all: client→origin stays blocked on this read. + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if v := store.View(session.DefaultSessionID); v != nil { + for _, e := range v.Events { + if e.Phase == pipeline.SessionResponse && e.StatusCode == http.StatusSwitchingProtocols { + // And the row says the session ended badly. Without this the + // only symptom of a WebSocket that broke mid-session was + // truncated byte counts, so an audit read it as a clean + // upgrade — one event per upgrade that cannot tell the two + // outcomes apart only half keeps the contract (#1549 review). + if e.Error == nil { + t.Fatal("the 101 row carries no error after a broken splice; a session that died " + + "reads as one that closed cleanly") + } + if e.Error.Kind != "proxy_error" { + t.Errorf("error.kind = %q, want proxy_error — the relay's failure is ours", e.Error.Kind) + } + return + } + } + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("the 101 response row was never recorded: a broken splice direction stranded the other, so the upgrade went unaudited") +} + +// TestForwardProxy_UpgradeWithoutConnectionTokenIsStripped: an "Upgrade: +// websocket" with no Connection: upgrade token is NOT an upgrade request (RFC +// 9110 §7.8 — a recipient ignores Upgrade unless Connection lists it), so it +// must take the ordinary request path: the pair is stripped, the origin sees a +// plain GET, and the client gets that GET's response. +// +// The regression this pins is an escalation, not a missed feature. +// stripForUpstream SETS the normalized Connection/Upgrade pair for anything it +// treats as an upgrade, so keying on Upgrade alone had the proxy synthesize a +// handshake the client never sent — and when the origin answered 101, hijack +// the client's connection and splice it. A client waiting for an ordinary HTTP +// response instead got an opaque byte pipe whose frames no plugin inspects. +// Measured end to end: the client read the origin's post-101 bytes (#1549 +// review). +func TestForwardProxy_UpgradeWithoutConnectionTokenIsStripped(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + + seen := make(chan http.Header, 1) + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + seen <- r.Header.Clone() + // An origin that WOULD upgrade if asked, so the only thing keeping this + // request on the ordinary path is the proxy's own refusal to forge the pair. + if r.Header.Get("Upgrade") != "" { + hj, ok := w.(http.Hijacker) + if !ok { + t.Error("origin: no hijacker") + return + } + c, _, err := hj.Hijack() + if err != nil { + t.Errorf("origin hijack: %v", err) + return + } + defer func() { _ = c.Close() }() + _, _ = c.Write([]byte("HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\n" + + "Upgrade: websocket\r\nSec-WebSocket-Accept: " + wsAccept(rfc6455Key) + "\r\n\r\nORIGINBYTES")) + return + } + w.Header().Set("Content-Type", "text/plain") + _, _ = io.WriteString(w, "ordinary") + })) + defer origin.Close() + originURL, err := url.Parse(origin.URL) + if err != nil { + t.Fatal(err) + } + + p, err := pipeline.New(nil) + if err != nil { + t.Fatal(err) + } + srv := &Server{OutboundPipeline: pipeline.NewHolder(p), Sessions: store, Client: http.DefaultClient} + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + proxyURL, err := url.Parse(proxy.URL) + if err != nil { + t.Fatal(err) + } + + conn, err := net.Dial("tcp", proxyURL.Host) + if err != nil { + t.Fatalf("dial proxy: %v", err) + } + defer func() { _ = conn.Close() }() + _ = conn.SetDeadline(time.Now().Add(10 * time.Second)) + // Note the absent Connection header — that is the whole point. + fmt.Fprintf(conn, "GET %s/ws HTTP/1.1\r\nHost: %s\r\nUpgrade: websocket\r\n"+ + "Sec-WebSocket-Key: %s\r\nSec-WebSocket-Version: 13\r\n\r\n", + origin.URL, originURL.Host, rfc6455Key) + + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200 — a bare Upgrade must take the ordinary path, not be spliced", resp.StatusCode) + } + body, _ := io.ReadAll(resp.Body) + if string(body) != "ordinary" { + t.Errorf("body = %q, want %q", body, "ordinary") + } + select { + case h := <-seen: + if got := h.Get("Upgrade"); got != "" { + t.Errorf("origin saw Upgrade = %q, want it stripped — the proxy must not forge a handshake", got) + } + if got := h.Get("Connection"); got != "" { + t.Errorf("origin saw Connection = %q, want it stripped", got) + } + case <-time.After(5 * time.Second): + t.Fatal("origin never received the request") + } +} + +// TestForwardProxy_RefusesA101ToAnotherProtocol: a 101 is not self-describing — +// it says "switching", and its Upgrade header says to what. An origin that +// answers a WebSocket handshake by switching to something else (h2c here) must +// not be spliced: net/http's ReverseProxy compares the response's upgrade type +// against the request's and refuses a mismatch, and this relay is modelled on +// it, but the comparison was the part left out. +// +// Refused BEFORE the hijack, so a 502 is still writable and the client learns +// the upgrade failed rather than being handed a pipe speaking a protocol it +// never negotiated (#1549 review). +func TestForwardProxy_RefusesA101ToAnotherProtocol(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hj, ok := w.(http.Hijacker) + if !ok { + t.Error("origin: no hijacker") + return + } + c, _, err := hj.Hijack() + if err != nil { + t.Errorf("origin hijack: %v", err) + return + } + defer func() { _ = c.Close() }() + _, _ = c.Write([]byte("HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\n" + + "Upgrade: h2c\r\n\r\nNOTWEBSOCKET")) + })) + defer origin.Close() + originURL, err := url.Parse(origin.URL) + if err != nil { + t.Fatal(err) + } + + p, err := pipeline.New(nil) + if err != nil { + t.Fatal(err) + } + srv := &Server{OutboundPipeline: pipeline.NewHolder(p), Sessions: store, Client: http.DefaultClient} + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + proxyURL, err := url.Parse(proxy.URL) + if err != nil { + t.Fatal(err) + } + + conn, err := net.Dial("tcp", proxyURL.Host) + if err != nil { + t.Fatalf("dial proxy: %v", err) + } + defer func() { _ = conn.Close() }() + _ = conn.SetDeadline(time.Now().Add(10 * time.Second)) + writeUpgradeRequest(conn, origin.URL+"/ws", originURL.Host) + + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusBadGateway { + t.Fatalf("status = %d, want 502 — a 101 to another protocol must not be spliced", resp.StatusCode) + } + + // And the failure is audited: a request row with no pair reads as still in + // flight, which is the thing recordSpliceFailure exists to prevent. + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if v := store.View(session.DefaultSessionID); v != nil { + for _, e := range v.Events { + if e.Phase == pipeline.SessionResponse && e.StatusCode == http.StatusBadGateway { + return + } + } + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("no 502 response row recorded for the refused upgrade") +} diff --git a/core/listener/reverseproxy/bytescount_test.go b/core/listener/reverseproxy/bytescount_test.go index 6050d0652..bbf578e5a 100644 --- a/core/listener/reverseproxy/bytescount_test.go +++ b/core/listener/reverseproxy/bytescount_test.go @@ -60,6 +60,11 @@ func bytesOf(t *testing.T, plugins []pipeline.Plugin, method, path, reqBody stri } resp.Body.Close() + // Waited rather than read: the unbuffered relay defers its response row to + // the end of ReverseProxy's copy (recordOnEOFBody), which can finish after + // the body was drained here. + waitForResponseEvent(t, store, session.DefaultSessionID) + v := store.View(session.DefaultSessionID) if v == nil || len(v.Events) != 2 { t.Fatalf("want a request row and a response row, got %+v", v) diff --git a/core/listener/reverseproxy/finalize_context_test.go b/core/listener/reverseproxy/finalize_context_test.go index f4e351fcd..64b0e23f5 100644 --- a/core/listener/reverseproxy/finalize_context_test.go +++ b/core/listener/reverseproxy/finalize_context_test.go @@ -233,6 +233,10 @@ func TestModifyResponse_BufferedArmSurvivesAHangup(t *testing.T) { if calls, ctxErr := plain.seen(); calls != 1 || ctxErr != nil { t.Errorf("response phases = %d, ctx.Err = %v, want 1 and nil: the phase was skipped or ran on a done context", calls, ctxErr) } + // Waited rather than read: the response row is deferred to the end of + // ReverseProxy's copy (recordOnEOFBody), which can finish after the body + // was drained above. + waitForResponseEvent(t, store, session.DefaultSessionID) v := store.View(session.DefaultSessionID) if v == nil || len(v.Events) != 1 || v.Events[0].Phase != pipeline.SessionResponse { t.Errorf("session events = %+v, want one response row: the early return took the telemetry with it", v) diff --git a/core/listener/reverseproxy/httppath_test.go b/core/listener/reverseproxy/httppath_test.go index a32f0d4aa..a4e582387 100644 --- a/core/listener/reverseproxy/httppath_test.go +++ b/core/listener/reverseproxy/httppath_test.go @@ -50,7 +50,8 @@ func TestReverseProxyRecordsMethodAndPath(t *testing.T) { // The response row lands when ReverseProxy finishes relaying the body, which can be // after the client has its response: wait for it rather than race it. - waitForResponseEvent(t, store) + waitForResponseEvent(t, store, session.DefaultSessionID) + v := store.View(session.DefaultSessionID) // Both the request and the response event must carry the pair; a response diff --git a/core/listener/reverseproxy/recording_test.go b/core/listener/reverseproxy/recording_test.go index 70ab4a927..c38e84a3d 100644 --- a/core/listener/reverseproxy/recording_test.go +++ b/core/listener/reverseproxy/recording_test.go @@ -75,7 +75,7 @@ func TestReverseProxy_RecordsInboundAllowWithoutA2A(t *testing.T) { // for the row rather than racing the recording (waitForResponseEvent, // upstreamfailure_test.go); the request row is appended before the // exchange, so it is on the view once the response row is. - respEvent := waitForResponseEvent(t, store) + respEvent := waitForResponseEvent(t, store, session.DefaultSessionID) // No A2A means no contextId, so the event lands in the default bucket. v := store.View(session.DefaultSessionID) diff --git a/core/listener/reverseproxy/server_test.go b/core/listener/reverseproxy/server_test.go index a15f894cd..9cc0ba753 100644 --- a/core/listener/reverseproxy/server_test.go +++ b/core/listener/reverseproxy/server_test.go @@ -856,7 +856,7 @@ func TestReverseProxy_ResponseEventDurationIncludesUnbufferedBody(t *testing.T) // after the client has read the body and returned — wait for it rather // than racing the recording (see recording_test.go for the flake this // shape produced in CI). - waitForResponseEvent(t, store) + waitForResponseEvent(t, store, session.DefaultSessionID) v := store.View(session.DefaultSessionID) if v == nil || len(v.Events) != 2 { diff --git a/core/listener/reverseproxy/upstreamfailure_test.go b/core/listener/reverseproxy/upstreamfailure_test.go index facffd2a9..dc4142454 100644 --- a/core/listener/reverseproxy/upstreamfailure_test.go +++ b/core/listener/reverseproxy/upstreamfailure_test.go @@ -50,19 +50,9 @@ func TestReverseProxy_UnreachableBackend_RecordsResponseEvent(t *testing.T) { t.Errorf("client status = %d, want 502", resp.StatusCode) } - v := store.View(session.DefaultSessionID) - if v == nil { - t.Fatal("no session recorded") - } - var respEvent *pipeline.SessionEvent - for i := range v.Events { - if v.Events[i].Phase == pipeline.SessionResponse { - respEvent = &v.Events[i] - } - } - if respEvent == nil { - t.Fatalf("no response event recorded for a failed backend call; events = %+v", v.Events) - } + // Waited rather than read: the failure row is appended from errorHandler + // after the 502 went out, so the client can be back first. + respEvent := waitForResponseEvent(t, store, session.DefaultSessionID) if respEvent.Direction != pipeline.Inbound { t.Errorf("Direction = %q, want inbound", respEvent.Direction) } @@ -265,7 +255,7 @@ func TestReverseProxy_ClientGivesUp_RecordsClientCanceled(t *testing.T) { t.Fatal("request succeeded; wanted the client to give up on a hung backend") } - ev := waitForResponseEvent(t, store) + ev := waitForResponseEvent(t, store, session.DefaultSessionID) if ev.StatusCode != pipeline.StatusClientClosedRequest { t.Errorf("StatusCode = %d, want %d — a 502 here is a response nobody received", ev.StatusCode, pipeline.StatusClientClosedRequest) } @@ -312,7 +302,7 @@ func TestReverseProxy_FailedUpgrade_RecordsOneResponse(t *testing.T) { } resp.Body.Close() - ev := waitForResponseEvent(t, store) + ev := waitForResponseEvent(t, store, session.DefaultSessionID) // Give a second row the time to land, if it were going to. time.Sleep(100 * time.Millisecond) v := store.View(session.DefaultSessionID) @@ -360,7 +350,7 @@ func failedExchange(t *testing.T, target string, timeout time.Duration, plugins t.Fatalf("client.Do: %v", err) } resp.Body.Close() - return waitForResponseEvent(t, store), resp.StatusCode + return waitForResponseEvent(t, store, session.DefaultSessionID), resp.StatusCode } // assertBufferingFailure checks the shape every buffering failure records: the @@ -394,14 +384,17 @@ func assertOutcomeError(t *testing.T, f *finisherStub) { } } -// waitForResponseEvent polls the default bucket for a response event. A failure -// is recorded after the client may already have returned — immediately, on a -// hangup — so a bare read races the recording. -func waitForResponseEvent(t *testing.T, store *session.Store) pipeline.SessionEvent { +// waitForResponseEvent polls the session's events for a response row. A +// response is recorded after the client may already have returned — the +// unbuffered relay defers its row to the end of ReverseProxy's body copy +// (recordOnEOFBody), and a failure row is appended from errorHandler after the +// 502 went out — so a bare read races the recording. The session id is a +// parameter because a rekeyed turn's rows land under the contextId bucket. +func waitForResponseEvent(t *testing.T, store *session.Store, sid string) pipeline.SessionEvent { t.Helper() deadline := time.Now().Add(5 * time.Second) for time.Now().Before(deadline) { - if v := store.View(session.DefaultSessionID); v != nil { + if v := store.View(sid); v != nil { for _, ev := range v.Events { if ev.Phase == pipeline.SessionResponse { return ev @@ -410,7 +403,7 @@ func waitForResponseEvent(t *testing.T, store *session.Store) pipeline.SessionEv } time.Sleep(10 * time.Millisecond) } - t.Fatal("no response event recorded") + t.Fatalf("no response event recorded in session %q", sid) return pipeline.SessionEvent{} } diff --git a/core/pipeline/context.go b/core/pipeline/context.go index 813be7754..f8b1852d8 100644 --- a/core/pipeline/context.go +++ b/core/pipeline/context.go @@ -126,21 +126,24 @@ type Context struct { // UpgradeStripped names the protocol the client asked to switch to — // "websocket" in every case seen so far — on a request whose Upgrade header // the forward proxy then removed. Empty on every other request, which is - // almost all of them. + // almost all of them: a WebSocket the proxy FORWARDS (#1488) carries no + // marker, because nothing was stripped. // // Set by the listener, not a plugin, and purely diagnostic: nothing reads it // to change behaviour. It exists because the stripping is CORRECT and its - // consequence is baffling. RFC 9110 makes Upgrade hop-by-hop, so a proxy must - // not forward it; but the request then reaches the origin as a bare GET, and - // an origin whose endpoint takes a POST or an upgrade and nothing else answers - // 405. Codex does exactly this against chatgpt.com, retries about five times + // consequence was baffling. RFC 9110 makes Upgrade hop-by-hop, so a proxy + // must not forward it on an ordinary request; but a stripped upgrade request + // reaches the origin as a bare GET, and an origin whose endpoint takes a POST + // or an upgrade and nothing else answers 405. Codex did exactly this against + // chatgpt.com, retried about five times // over roughly six seconds, falls back to HTTPS and works — leaving the user // looking at `405 Method Not Allowed` in their agent's output with every // reason to think Cortex broke, and an event saying only `backend_error/405` // with nothing connecting the two. (Codex no longer reaches that 405 by default: // the forward proxy answers a Responses API WebSocket with 426 itself, which it // reads as "use HTTPS" at once — see listener.decline_responses_websockets. The - // marker is still set there, since the Upgrade is still not forwarded.) + // marker is still set there, since the Upgrade is still not forwarded. Every + // other WebSocket is forwarded, and those rows carry no marker.) // // Recorded on the REQUEST row, where the fact is true, and paired with the // response row's status by RequestID — the same split Tunnel/TunnelReason use. diff --git a/core/tlsbridge/serve.go b/core/tlsbridge/serve.go index 425328855..273a8c02a 100644 --- a/core/tlsbridge/serve.go +++ b/core/tlsbridge/serve.go @@ -55,6 +55,31 @@ func (c *notifyConn) Close() error { return c.Conn.Close() } +// CloseWrite forwards a half-close to the terminated TLS conn — a close_notify +// on the wire — so a WebSocket spliced over the bridge sees the origin's close. +// +// It has to be written out rather than inherited: notifyConn embeds the net.Conn +// INTERFACE, so only net.Conn's method set is promoted and CloseWrite is not in +// it, whatever the concrete type underneath happens to be. The forward proxy's +// splice probes its conn for `interface{ CloseWrite() error }`, and on the +// bridged path that probe is this wrapper — it failed, the goroutine forwarded +// nothing, and an origin that hung up first left the agent waiting on a dead +// WebSocket while the 101 row (recorded after BOTH directions) never landed. +// The plain path was unaffected: there the conn is a *net.TCPConn (#1549 review). +// +// nil when the wrapped conn cannot half-close, which is the right answer rather +// than an error: the splice treats a non-nil result as a broken relay and marks +// the row `proxy_error`, so reporting "could not half-close" here would flag +// every clean close on such a conn as a failed session. A *tls.Conn — what +// ServeConn is always given — does implement CloseWrite, so in practice this +// forwards. +func (c *notifyConn) CloseWrite() error { + if cw, ok := c.Conn.(interface{ CloseWrite() error }); ok { + return cw.CloseWrite() + } + return nil +} + type dummyAddr struct{} func (dummyAddr) Network() string { return "tcp" } diff --git a/core/tlsbridge/upstream.go b/core/tlsbridge/upstream.go index 37f1a237a..880a67c8c 100644 --- a/core/tlsbridge/upstream.go +++ b/core/tlsbridge/upstream.go @@ -33,6 +33,13 @@ func WithDialContext(dial func(ctx context.Context, network, addr string) (net.C // re-origination does NOT verify the origin cert — for internal/self-signed // upstreams (e.g. a private LiteLLM endpoint). Prefer extraRootsPEM // (upstream_ca_bundle) over insecure whenever the origin CA is available. +// +// WebSocket upgrades need no client of their own, tempting as one pinned to +// HTTP/1.1 looks: net/http already pins a request carrying "Connection: +// upgrade" and "Upgrade: websocket" to HTTP/1.1 (Request.requiresHTTP1) — the +// exact pair forwardproxy's stripForUpstream normalizes a forwarded upgrade to +// — so this client carries one as HTTP/1.1 even against an origin it would +// otherwise negotiate h2 with. func NewUpstreamClient(extraRootsPEM []byte, insecure bool, opts ...UpstreamOption) (*http.Client, error) { var o upstreamOptions for _, opt := range opts { diff --git a/docs/agents/codex.md b/docs/agents/codex.md index 2f61c30ea..d9e1be4e8 100644 --- a/docs/agents/codex.md +++ b/docs/agents/codex.md @@ -260,8 +260,10 @@ is one — so a consumer that parses `ToolCalls[].Arguments` as JSON must tolera for this dialect. **HTTPS, not WebSocket.** Codex prefers a WebSocket to the same `/responses` path, and -Cortex cannot carry one: `Upgrade` is a hop-by-hop header a proxy must not forward, and -the parser reads HTTP requests, not frames. So the forward proxy answers that handshake +Cortex declines one on purpose: the proxy can carry a WebSocket (it forwards the +upgrade and splices the 101, #1488), but the frames after it are opaque — the parsers +read HTTP requests, not frames — so a Codex session moved onto a WebSocket stops +being parseable: no token counts, no cost. The forward proxy answers that handshake itself with **426 Upgrade Required**, which Codex reads as "this endpoint wants HTTP" and switches to HTTPS straight away, for the rest of its session. That covers both of Codex's endpoints — `chatgpt.com` on a ChatGPT login and `api.openai.com` on an API key @@ -285,7 +287,8 @@ hardcodes `supports_websockets = true` and Codex refuses a config that overrides built-in provider. The 426 is Cortex's setting, not Codex's, and it is on by default, so there is nothing to -configure. To turn it off — which brings back the 405 and its retries — add the key to the +configure. To turn it off — which forwards the upgrade for real, putting the session on +a WebSocket the parsers cannot see — add the key to the `listener:` block that is already in `~/.cortex/config.yaml`: ```yaml diff --git a/docs/framework-architecture.md b/docs/framework-architecture.md index d408781f4..224ee4771 100644 --- a/docs/framework-architecture.md +++ b/docs/framework-architecture.md @@ -907,6 +907,52 @@ or token-exchange policy applied — bypass means bypass. It exists for traffic whose recording floods the session store and crowds out the user intent a guardrail needs (see the intent-pin note in [`laptop-service.md`](laptop-service.md)). +#### Forwarding WebSocket upgrades + +Every WebSocket upgrade the forward proxy does not decline is forwarded (#1488), plain +and TLS-bridged alike. `Upgrade` is hop-by-hop, which is why it is stripped from +ordinary requests — but an upgrade exchange makes the proxy an endpoint of the hop +instead of a reader between two: everything after the 101 is opaque to it, so the +header must cross or the origin never sees an upgrade request at all. The forwarded +pair is normalized to `Connection: Upgrade` / `Upgrade: websocket` (the one protocol +forwarded, mirroring `httputil.ReverseProxy`), which also pins the exchange to +HTTP/1.1 — net/http does that by itself (`Request.requiresHTTP1`), so the bridge's +h2-negotiating upstream client needs no HTTP/1.1-only twin for upgrades. An upgrade +GET on a bridged tunnel is held to the CONNECT-host check like any bridged request +(421 otherwise); on a bridged tunnel the splice is doubled, agent↔bridge and +bridge↔origin. + +**What counts as an upgrade is both headers, on both sides, and neither check is a +formality** — because the proxy *writes* the normalized pair, a predicate that is too +loose does not merely miss a forward, it manufactures one: + +- **The request** needs `Connection: upgrade` as well as `Upgrade: websocket` (RFC + 9110 §7.8: a recipient ignores `Upgrade` unless `Connection` lists it — the same + test `Request.requiresHTTP1` and ReverseProxy's `upgradeType` make). A bare + `Upgrade: websocket` with no `Connection` token takes the ordinary path and is + stripped. Honouring it would have the proxy forge a handshake the client never + sent and splice the origin's 101 onto a connection whose client is waiting for an + ordinary HTTP response — an opaque byte pipe no plugin inspects, from a request + that was not an upgrade. +- **The response** must switch to the protocol that was asked for. A 101 only says + "switching"; its own `Upgrade` header says to what. An origin answering `101 + Upgrade: h2c` to a WebSocket handshake is refused with 502 *before* the hijack — + while a status is still writable — rather than spliced, matching + `handleUpgradeResponse`, which rejects a `resUpType`/`reqUpType` mismatch. + +The 101 is hijacked and spliced onto the client's connection +(`relayUpgradedWebSocket`, modelled on ReverseProxy's `handleUpgradeResponse`), and +the session records the request row plus a paired 101 response row carrying both +directions' byte counts, recorded after both splice directions end — `durationMs` +covers the whole WebSocket session, `ttfbMs` the handshake. The frames after the 101 +are relayed byte for byte and never inspected or buffered, so no response plugin runs +on a 101 and a body parser sees only the upgrade GET. A non-101 answer (an origin +serving the bare GET, refusing the upgrade) takes the ordinary response path. + +Out of scope: WebSocket over HTTP/2 downstream (RFC 8441 Extended CONNECT) — a client +that wants h2 websockets through the proxy gets its fallback path, same as before +forwarding existed. + #### Declining Responses API WebSockets (`listener.decline_responses_websockets`) The forward proxy answers a WebSocket upgrade on OpenAI's Responses API — a `GET` with @@ -916,18 +962,18 @@ deployment. ```yaml listener: - decline_responses_websockets: false # forward the upgrade instead, as before + decline_responses_websockets: false # forward the upgrade for real ``` -The proxy cannot carry the WebSocket either way: `Upgrade` is hop-by-hop, so a -forwarded handshake reaches the origin as a bare `GET`, which OpenAI's endpoint answers -405. Codex — the one client known to use this transport — retries a 405 five times over -about six seconds before falling back to HTTPS, and reads a 426 as "use HTTP" and falls -back at once. `false` restores the forwarding byte for byte, for a client that reads a -426 some other way. Every other WebSocket is forwarded whatever this says, including -one to another host serving the same API (a LiteLLM gateway, OpenCode Zen, a local -server): what a 426 means is only known for Codex against OpenAI, and another client -might stop where today it falls back. Restart to +The decline is a steer, not a capability gap: a forwarded Responses WebSocket works +(the section above), but its frames are opaque to every plugin, so a Codex session +moved onto one stops being parseable — no token counts, no cost. A 426 sends Codex +straight to HTTPS, where its traffic stays observable; before forwarding existed the +calculation was harsher still, because the stripped handshake drew a 405 that Codex +retried for ~6s. `false` forwards the upgrade for real instead. The scope is exact — +OpenAI's own hosts only — because what a 426 means is only known for Codex against +OpenAI, and another client (or another host serving the same API: a LiteLLM gateway, +OpenCode Zen, a local server) is forwarded whatever this says. Restart to apply, like every `listener.*` field. The key goes in the runtime config: on a laptop, `~/.cortex/config.yaml`, then `agentop service restart`; in a cluster, the `authbridge-runtime-config` ConfigMap, then a pod restart. See diff --git a/docs/session-api.md b/docs/session-api.md index f3185f2c9..8f95624d6 100644 --- a/docs/session-api.md +++ b/docs/session-api.md @@ -281,8 +281,29 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries: "upgrade_declined"` is the forward proxy's own answer to a WebSocket upgrade on OpenAI's Responses API (`chatgpt.com` or `api.openai.com`): nothing is forwarded, and a client such as Codex reads it as "use HTTPS" and retries at once over - HTTPS. It is not counted as an error in `/v1/usage`. A **403** with - `error.kind: "destination_private"` is the egress guard's refusal + HTTPS. It is not counted as an error in `/v1/usage`. Every other WebSocket + upgrade is forwarded (#1488): the 101 the origin answers with is spliced onto + the client's connection, the frames after it are relayed uninspected, and the + session records the request row plus a **101 response row** paired to it by + `requestId`, carrying `bytesUp` and `bytesDown` for the whole WebSocket + session — the one response row besides a tunnel close that carries both — + with `durationMs` covering the session and `ttfbMs` the handshake. Because + that `durationMs` is a session lifetime rather than a request latency, the + 101 row is **excluded from `/v1/usage`** exactly as a tunnel row is: folding + it would drag the latency mean toward WebSocket lifetimes (one 10-minute + session moved a host's mean from 500ms to 6.4s in testing), its frames are + opaque so there is no model, token or cost to attribute, and counting it + would add one Request whose latency is its whole lifetime. On the timeline + the lifetime is the point, so the field stays. **That exclusion is outbound + only.** The reverse proxy also records a 101, but at *handshake* time + (`ReverseProxy` hands off the connection instead of copying a body, so a + deferred record would never fire), so its `durationMs` is the handshake + latency — an ordinary request figure. An inbound WebSocket handshake is + therefore still counted as a Request. A relay that ended in an + error carries `error.kind: "proxy_error"` on the 101 row, so a session that + broke mid-stream is distinguishable from one that closed cleanly; an + ordinary close is not marked. A **403** + with `error.kind: "destination_private"` is the egress guard's refusal (`listener.egress_guard`): the destination resolved into loopback, link-local, private or this host's own address space and no permit covered it. Also the proxy's own answer rather than the upstream's, and also not @@ -310,8 +331,9 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries: - `bytesUp`, `bytesDown` — **body** bytes, one direction per row: a request row carries `bytesUp`, the body as forwarded (so a rewrite is reflected); a response row carries `bytesDown`, the body as it came back from the - destination. Headers are not counted. A tunnel close is the one row carrying - both. **Zero means the listener counted nothing, not that the body was + destination. Headers are not counted. A tunnel close and a forwarded + WebSocket's 101 response row are the rows carrying both. **Zero means the + listener counted nothing, not that the body was empty** — every listener buffers bodies only when a plugin asks, so an unbuffered request reports zero. A response relayed straight through is counted, not zero: both proxies record the row after they finish relaying