From 4440ef96c746a219eed88097bf1c90d61cbc22a4 Mon Sep 17 00:00:00 2001 From: YehoshuaSagron Date: Fri, 14 Aug 2026 19:48:56 +0300 Subject: [PATCH 1/3] Fix: Propagate every plugin header mutation in extproc and forwardproxy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reverseproxy already syncs the pipeline's whole header set onto the forwarded request, and its comment states the bug it fixed: Only Authorization used to be forwarded, silently dropping any other injected header (e.g. static-inject's x-api-key). extproc and forwardproxy still behave the way that comment describes. This brings them to parity. extproc gains a generic withHeaderMutation: diff pctx.Headers against a clone taken before the pipeline ran, emit the difference as SetHeaders/RemoveHeaders. It skips ':'-prefixed pseudo-headers, which govern routing, and Content-Length/Content-Encoding, which the body-rewrite path and the transport manage — the same exclusions reverseproxy makes. forwardproxy takes the equivalent block. Both drop the Authorization special case. Every writer in-tree emits "Bearer "+token, so extract-and-re-prefix was the identity function on all real inputs, and it mangled non-Bearer schemes because ExtractBearer returns empty for them. Removing it takes three lines out of each of the four ext_proc handlers and drops the auth import from the file. No header is special in any listener now. Affected today, with no telemetry involved: static-inject writes a configurable header name (plugin.go:221) and deletes Authorization (plugin.go:229) — neither reached the wire, the deletion because the old path only ever set that header. cpex writes arbitrary pairs (manager_cpex.go:492). Also here, separable in review: a 4-line authorityOf helper used at five sites. The inbound ext_proc handlers never set pctx.Host while the outbound ones did, though pipeline.SessionEvent documents Host for both directions and reverseproxy always populated it. A 107-line table test covers every handler site and both header forms. Six listener-level regression tests come with this, asserting at the ProcessingResponse layer that a plugin-level test cannot observe. Three use ordinary header names to pin the general behaviour: an arbitrary header reaches the wire, a deleted header is removed, pseudo-headers are never emitted. Out of scope: extauthz (waypoint mode) has the same Authorization-only pattern at server.go:86-92 and is untouched here. Signed-off-by: YehoshuaSagron --- authbridge/authlib/listener/extproc/server.go | 120 ++++++-- .../listener/extproc/server_authority_test.go | 107 +++++++ .../extproc/server_headerdiff_test.go | 288 ++++++++++++++++++ .../authlib/listener/forwardproxy/server.go | 27 +- 4 files changed, 506 insertions(+), 36 deletions(-) create mode 100644 authbridge/authlib/listener/extproc/server_authority_test.go create mode 100644 authbridge/authlib/listener/extproc/server_headerdiff_test.go diff --git a/authbridge/authlib/listener/extproc/server.go b/authbridge/authlib/listener/extproc/server.go index 82621c0f6..bf3a011c9 100644 --- a/authbridge/authlib/listener/extproc/server.go +++ b/authbridge/authlib/listener/extproc/server.go @@ -10,6 +10,7 @@ import ( "io" "log/slog" "net/http" + "slices" "strconv" "strings" "time" @@ -21,7 +22,6 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/status" - "github.com/rossoctl/cortex/authbridge/authlib/auth" "github.com/rossoctl/cortex/authbridge/authlib/listener/httpx" "github.com/rossoctl/cortex/authbridge/authlib/listener/internal/sseframe" "github.com/rossoctl/cortex/authbridge/authlib/listener/skiphost" @@ -161,6 +161,7 @@ func (s *Server) handleInbound(stream extprocv3.ExternalProcessor_ProcessServer, Direction: pipeline.Inbound, Method: getHeader(headers, ":method"), Scheme: getHeader(headers, ":scheme"), + Host: authorityOf(headers), Path: getHeader(headers, ":path"), Headers: headerMapToHTTP(headers), Body: body, @@ -168,7 +169,7 @@ func (s *Server) handleInbound(stream extprocv3.ExternalProcessor_ProcessServer, StartedAt: time.Now(), } - originalAuth := pctx.Headers.Get("Authorization") + originalHeaders := pctx.Headers.Clone() action := s.InboundPipeline.Run(ctx, pctx) if action.Type == pipeline.Reject { s.recordInboundReject(pctx, action) @@ -177,10 +178,7 @@ func (s *Server) handleInbound(stream extprocv3.ExternalProcessor_ProcessServer, } s.recordInboundSession(pctx) - if newAuth := pctx.Headers.Get("Authorization"); newAuth != originalAuth { - return replaceTokenResponse(auth.ExtractBearer(newAuth)), pctx - } - return allowResponse(), pctx + return withHeaderMutation(allowResponse(), pctx, originalHeaders), pctx } func (s *Server) handleInboundBody(stream extprocv3.ExternalProcessor_ProcessServer, headers *corev3.HeaderMap, body []byte) (*extprocv3.ProcessingResponse, *pipeline.Context) { @@ -189,6 +187,7 @@ func (s *Server) handleInboundBody(stream extprocv3.ExternalProcessor_ProcessSer Direction: pipeline.Inbound, Method: getHeader(headers, ":method"), Scheme: getHeader(headers, ":scheme"), + Host: authorityOf(headers), Path: getHeader(headers, ":path"), Headers: headerMapToHTTP(headers), Body: body, @@ -196,7 +195,7 @@ func (s *Server) handleInboundBody(stream extprocv3.ExternalProcessor_ProcessSer StartedAt: time.Now(), } - originalAuth := pctx.Headers.Get("Authorization") + originalHeaders := pctx.Headers.Clone() action := s.InboundPipeline.Run(ctx, pctx) if action.Type == pipeline.Reject { s.recordInboundReject(pctx, action) @@ -205,10 +204,8 @@ func (s *Server) handleInboundBody(stream extprocv3.ExternalProcessor_ProcessSer } s.recordInboundSession(pctx) - if newAuth := pctx.Headers.Get("Authorization"); newAuth != originalAuth { - return withBodyMutation(replaceTokenBodyResponse(auth.ExtractBearer(newAuth)), pctx), pctx - } - return withBodyMutation(allowBodyResponse(), pctx), pctx + resp := withHeaderMutation(allowBodyResponse(), pctx, originalHeaders) + return withBodyMutation(resp, pctx), pctx } // inboundSessionID returns the bucket ID for an inbound event. Trusts the @@ -469,16 +466,13 @@ func (s *Server) handleOutbound(stream extprocv3.ExternalProcessor_ProcessServer Direction: pipeline.Outbound, Method: getHeader(headers, ":method"), Scheme: getHeader(headers, ":scheme"), - Host: getHeader(headers, ":authority"), + Host: authorityOf(headers), Path: getHeader(headers, ":path"), Headers: headerMapToHTTP(headers), Body: body, Shared: s.Shared, StartedAt: time.Now(), } - if pctx.Host == "" { - pctx.Host = getHeader(headers, "host") - } // SkipHosts short-circuit: forward the request as a transparent // proxy without running the pipeline or recording a session event. @@ -495,7 +489,7 @@ func (s *Server) handleOutbound(stream extprocv3.ExternalProcessor_ProcessServer } } - originalAuth := pctx.Headers.Get("Authorization") + originalHeaders := pctx.Headers.Clone() action := s.OutboundPipeline.Run(ctx, pctx) if action.Type == pipeline.Reject { s.recordOutboundReject(pctx, action) @@ -505,11 +499,7 @@ func (s *Server) handleOutbound(stream extprocv3.ExternalProcessor_ProcessServer s.recordOutboundSession(pctx) - newAuth := pctx.Headers.Get("Authorization") - if newAuth != originalAuth { - return replaceTokenResponse(auth.ExtractBearer(newAuth)), pctx - } - return passResponse(), pctx + return withHeaderMutation(passResponse(), pctx, originalHeaders), pctx } func (s *Server) handleOutboundBody(stream extprocv3.ExternalProcessor_ProcessServer, headers *corev3.HeaderMap, body []byte) (*extprocv3.ProcessingResponse, *pipeline.Context) { @@ -518,16 +508,13 @@ func (s *Server) handleOutboundBody(stream extprocv3.ExternalProcessor_ProcessSe Direction: pipeline.Outbound, Method: getHeader(headers, ":method"), Scheme: getHeader(headers, ":scheme"), - Host: getHeader(headers, ":authority"), + Host: authorityOf(headers), Path: getHeader(headers, ":path"), Headers: headerMapToHTTP(headers), Body: body, Shared: s.Shared, StartedAt: time.Now(), } - if pctx.Host == "" { - pctx.Host = getHeader(headers, "host") - } // SkipHosts short-circuit: see handleOutbound for rationale. The // body-phase entry point needs the same gate because Envoy may @@ -547,7 +534,7 @@ func (s *Server) handleOutboundBody(stream extprocv3.ExternalProcessor_ProcessSe } } - originalAuth := pctx.Headers.Get("Authorization") + originalHeaders := pctx.Headers.Clone() action := s.OutboundPipeline.Run(ctx, pctx) if action.Type == pipeline.Reject { s.recordOutboundReject(pctx, action) @@ -557,11 +544,8 @@ func (s *Server) handleOutboundBody(stream extprocv3.ExternalProcessor_ProcessSe s.recordOutboundSession(pctx) - newAuth := pctx.Headers.Get("Authorization") - if newAuth != originalAuth { - return withBodyMutation(replaceTokenBodyResponse(auth.ExtractBearer(newAuth)), pctx), pctx - } - return withBodyMutation(passBodyResponse(), pctx), pctx + resp := withHeaderMutation(passBodyResponse(), pctx, originalHeaders) + return withBodyMutation(resp, pctx), pctx } func (s *Server) handleResponseHeaders(ctx context.Context, headers *corev3.HeaderMap, pctx *pipeline.Context, direction string) *extprocv3.ProcessingResponse { @@ -705,6 +689,80 @@ func (s *Server) handleResponseBody(ctx context.Context, body []byte, pctx *pipe } } +// withHeaderMutation emits every header mutation the request pipeline made to +// pctx.Headers — including the Authorization replacement. ext_proc forwards no +// header change it does not explicitly emit, so only Authorization used to be +// propagated, silently dropping any other injected header (e.g. static-inject's +// x-api-key). Symmetric to withBodyMutation, and to reverseproxy's +// forwarded-request header sync. Skipped: HTTP/2 pseudo-headers, which +// headerMapToHTTP copies into pctx.Headers and whose :authority governs routing; +// and Content-Length / Content-Encoding, managed by withBodyMutation and the +// transport. +func withHeaderMutation(resp *extprocv3.ProcessingResponse, pctx *pipeline.Context, orig http.Header) *extprocv3.ProcessingResponse { + skip := func(k string) bool { + return strings.HasPrefix(k, ":") || + k == "Content-Length" || k == "Content-Encoding" + } + var set []*corev3.HeaderValueOption + var del []string + for k, vv := range pctx.Headers { + if skip(k) || slices.Equal(orig[k], vv) { + continue + } + // Wire header names are lowercase; pctx.Headers keys were + // canonicalised by http.Header.Set in headerMapToHTTP. + // Multi-value join uses ",": correct per RFC 9110 for every header a + // plugin realistically rewrites, and known-wrong only for Cookie + // (whose separator is "; ") — no plugin rewrites Cookie today, and + // one that does must split this out rather than discover it here. + set = append(set, &corev3.HeaderValueOption{ + Header: &corev3.HeaderValue{Key: strings.ToLower(k), RawValue: []byte(strings.Join(vv, ","))}, + }) + } + for k := range orig { + if _, ok := pctx.Headers[k]; !ok && !skip(k) { + del = append(del, strings.ToLower(k)) // plugin removed it + } + } + if len(set) == 0 && len(del) == 0 { + return resp + } + var cr *extprocv3.CommonResponse + switch r := resp.Response.(type) { + case *extprocv3.ProcessingResponse_RequestHeaders: + if r.RequestHeaders.Response == nil { + r.RequestHeaders.Response = &extprocv3.CommonResponse{} + } + cr = r.RequestHeaders.Response + case *extprocv3.ProcessingResponse_RequestBody: + if r.RequestBody.Response == nil { + r.RequestBody.Response = &extprocv3.CommonResponse{} + } + cr = r.RequestBody.Response + default: + return resp // ImmediateResponse or response-phase; nothing to forward. + } + if cr.HeaderMutation == nil { + cr.HeaderMutation = &extprocv3.HeaderMutation{} + } + // Append, never assign: composes with allowResponse's + // x-authbridge-direction removal. + cr.HeaderMutation.SetHeaders = append(cr.HeaderMutation.SetHeaders, set...) + cr.HeaderMutation.RemoveHeaders = append(cr.HeaderMutation.RemoveHeaders, del...) + return resp +} + +// authorityOf returns the request's authority: the HTTP/2 :authority +// pseudo-header, falling back to the HTTP/1 Host header. Both directions +// need it — outbound it names the service being called, inbound the address +// this workload was reached on (see pipeline.SessionEvent.Host). +func authorityOf(headers *corev3.HeaderMap) string { + if a := getHeader(headers, ":authority"); a != "" { + return a + } + return getHeader(headers, "host") +} + func headerMapToHTTP(headers *corev3.HeaderMap) http.Header { h := make(http.Header) if headers != nil { diff --git a/authbridge/authlib/listener/extproc/server_authority_test.go b/authbridge/authlib/listener/extproc/server_authority_test.go new file mode 100644 index 000000000..e48a655cc --- /dev/null +++ b/authbridge/authlib/listener/extproc/server_authority_test.go @@ -0,0 +1,107 @@ +package extproc + +import ( + "context" + "testing" + + extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" + + "github.com/rossoctl/cortex/authbridge/authlib/pipeline" + "github.com/rossoctl/cortex/authbridge/authlib/plugins/plugintesting" +) + +// hostCapture records the pctx.Host the listener built, so a test can assert +// what plugins actually see (Host is what SessionEvent.Host and the lineage +// plugin's lineage.peer.host fact are derived from). +type hostCapture struct { + host string +} + +func (p *hostCapture) Name() string { return "host-capture" } +func (p *hostCapture) Capabilities() pipeline.PluginCapabilities { + return pipeline.PluginCapabilities{} +} +func (p *hostCapture) OnResponse(context.Context, *pipeline.Context) pipeline.Action { + return pipeline.Action{Type: pipeline.Continue} +} + +func (p *hostCapture) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { + p.host = pctx.Host + return pipeline.Action{Type: pipeline.Continue} +} + +func newHostCaptureServer(t *testing.T) (*Server, *hostCapture, *hostCapture) { + t.Helper() + in, out := &hostCapture{}, &hostCapture{} + inbound, err := plugintesting.BuildPipeline([]pipeline.Plugin{in}) + if err != nil { + t.Fatalf("building inbound pipeline: %v", err) + } + outbound, err := plugintesting.BuildPipeline([]pipeline.Plugin{out}) + if err != nil { + t.Fatalf("building outbound pipeline: %v", err) + } + return &Server{ + InboundPipeline: pipeline.NewHolder(inbound), + OutboundPipeline: pipeline.NewHolder(outbound), + }, in, out +} + +func runOne(t *testing.T, srv *Server, req *extprocv3.ProcessingRequest) { + t.Helper() + _ = srv.Process(&mockStream{ctx: context.Background(), requests: []*extprocv3.ProcessingRequest{req}}) +} + +// TestExtProc_Authority asserts both directions carry the request authority on +// pctx.Host, from either the HTTP/2 pseudo-header or the HTTP/1 Host header. +// Inbound used to be left empty, which cost every inbound observation the +// address the workload was reached on. +func TestExtProc_Authority(t *testing.T) { + cases := []struct { + name string + inbound bool + headers []string + wantHost string + }{ + { + name: "inbound from :authority", + inbound: true, + headers: []string{"x-authbridge-direction", "inbound", ":authority", "weather-service.team1.svc.cluster.local:8000", ":path", "/"}, + wantHost: "weather-service.team1.svc.cluster.local:8000", + }, + { + name: "inbound falls back to the host header", + inbound: true, + headers: []string{"x-authbridge-direction", "inbound", "host", "weather-service:8000", ":path", "/"}, + wantHost: "weather-service:8000", + }, + { + name: "outbound from :authority", + headers: []string{":authority", "weather-tool-mcp.team1.svc.cluster.local:8000", ":path", "/mcp"}, + wantHost: "weather-tool-mcp.team1.svc.cluster.local:8000", + }, + { + name: "outbound falls back to the host header", + headers: []string{"host", "weather-tool-mcp:8000", ":path", "/mcp"}, + wantHost: "weather-tool-mcp:8000", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + srv, in, out := newHostCaptureServer(t) + headers := makeHeaders(tc.headers...) + if tc.inbound { + runOne(t, srv, inboundRequest(headers)) + if in.host != tc.wantHost { + t.Errorf("inbound pctx.Host = %q; want %q", in.host, tc.wantHost) + } + return + } + runOne(t, srv, outboundRequest(headers)) + if out.host != tc.wantHost { + t.Errorf("outbound pctx.Host = %q; want %q", out.host, tc.wantHost) + } + }) + } +} diff --git a/authbridge/authlib/listener/extproc/server_headerdiff_test.go b/authbridge/authlib/listener/extproc/server_headerdiff_test.go new file mode 100644 index 000000000..f69fec30b --- /dev/null +++ b/authbridge/authlib/listener/extproc/server_headerdiff_test.go @@ -0,0 +1,288 @@ +package extproc + +import ( + "context" + "fmt" + "testing" + + extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" + + "github.com/rossoctl/cortex/authbridge/authlib/auth" + "github.com/rossoctl/cortex/authbridge/authlib/pipeline" + "github.com/rossoctl/cortex/authbridge/authlib/plugins/plugintesting" +) + +// traceRewriterPlugin mimics a pipeline plugin's header writes — the lineage +// plugin's tracestate stamp today (wire contract v1.5), a traceparent rewrite +// to prove the mechanism is not stamp-specific, and arbitrary set/delete to +// prove it is not trace-specific either (static-inject's x-api-key is the +// upstream case). Used to assert the listener forwards plugin header writes +// as mutations: before withHeaderMutation everything but Authorization died +// in pctx.Headers (inert on the wire — the phantom-root forests). +type traceRewriterPlugin struct { + traceparent string + tracestate string + set map[string]string + del []string + readsBody bool +} + +func (p *traceRewriterPlugin) Name() string { return "trace-rewriter" } +func (p *traceRewriterPlugin) Capabilities() pipeline.PluginCapabilities { + return pipeline.PluginCapabilities{ReadsBody: p.readsBody} +} +func (p *traceRewriterPlugin) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { + if p.traceparent != "" { + pctx.Headers.Set("traceparent", p.traceparent) + } + if p.tracestate != "" { + pctx.Headers.Set("tracestate", p.tracestate) + } + for k, v := range p.set { + pctx.Headers.Set(k, v) + } + for _, k := range p.del { + pctx.Headers.Del(k) + } + return pipeline.Action{Type: pipeline.Continue} +} +func (p *traceRewriterPlugin) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action { + return pipeline.Action{Type: pipeline.Continue} +} + +func traceRewriterServer(t *testing.T, plugin pipeline.Plugin) *Server { + t.Helper() + outbound, err := pipeline.New([]pipeline.Plugin{plugin}) + if err != nil { + t.Fatal(err) + } + inbound, err := plugintesting.BuildPipeline([]pipeline.Plugin{plugintesting.NewJWTValidation(auth.New(auth.Config{}), false)}) + if err != nil { + t.Fatal(err) + } + return &Server{InboundPipeline: pipeline.NewHolder(inbound), OutboundPipeline: pipeline.NewHolder(outbound)} +} + +// mutationHeaderValue returns the mutation value for key, or "" when absent. +// (setHeaderValue in placeholder_test.go unwraps a full RequestHeaders +// response; this one takes the bare mutation so body-phase responses can +// share it.) +func mutationHeaderValue(hm *extprocv3.HeaderMutation, key string) string { + if hm == nil { + return "" + } + for _, sh := range hm.SetHeaders { + if sh.Header != nil && sh.Header.Key == key { + return string(sh.Header.RawValue) + } + } + return "" +} + +// TestExtProc_Outbound_TraceRewriteReachesWire: a plugin rewrite of the outbound +// traceparent/tracestate must be emitted as SetHeaders on the headers-phase +// response — this is what puts the lineage stamp on the wire. +func TestExtProc_Outbound_TraceRewriteReachesWire(t *testing.T) { + const newTP = "00-4bf92f3577b34da6a3ce929d0e0e4736-aaaaaaaaaaaaaaaa-01" + const newTS = "dg-parent=aaaaaaaaaaaaaaaa" + srv := traceRewriterServer(t, &traceRewriterPlugin{traceparent: newTP, tracestate: newTS}) + + stream := &mockStream{ + ctx: context.Background(), + requests: []*extprocv3.ProcessingRequest{ + outboundRequest(makeHeaders( + ":authority", "fanin-echo", + "traceparent", "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + "tracestate", "dg-parent=00f067aa0ba902b7", + )), + }, + } + _ = srv.Process(stream) + + if len(stream.responses) != 1 { + t.Fatalf("expected 1 response, got %d", len(stream.responses)) + } + rh := stream.responses[0].GetRequestHeaders() + if rh == nil || rh.Response == nil || rh.Response.HeaderMutation == nil { + t.Fatalf("expected HeadersResponse with header mutation, got %+v", stream.responses[0]) + } + if got := mutationHeaderValue(rh.Response.HeaderMutation, "traceparent"); got != newTP { + t.Errorf("traceparent mutation = %q, want %q (trace rewrite lost on the wire)", got, newTP) + } + if got := mutationHeaderValue(rh.Response.HeaderMutation, "tracestate"); got != newTS { + t.Errorf("tracestate mutation = %q, want %q", got, newTS) + } +} + +// TestExtProc_OutboundBody_TraceRewriteReachesWire: same guarantee on the +// body-phase path (a ReadsBody plugin defers the pipeline to the body +// message; the trace-header diff must ride that response instead). +func TestExtProc_OutboundBody_TraceRewriteReachesWire(t *testing.T) { + const newTP = "00-4bf92f3577b34da6a3ce929d0e0e4736-bbbbbbbbbbbbbbbb-01" + srv := traceRewriterServer(t, &traceRewriterPlugin{traceparent: newTP, readsBody: true}) + + body := []byte(`{"jsonrpc":"2.0"}`) + stream := &mockStream{ + ctx: context.Background(), + requests: []*extprocv3.ProcessingRequest{ + outboundRequest(makeHeaders( + ":authority", "fanin-echo", + "traceparent", "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + "content-length", fmt.Sprintf("%d", len(body)), + )), + {Request: &extprocv3.ProcessingRequest_RequestBody{ + RequestBody: &extprocv3.HttpBody{Body: body}, + }}, + }, + } + _ = srv.Process(stream) + + if len(stream.responses) != 2 { + t.Fatalf("expected 2 responses, got %d", len(stream.responses)) + } + rb := stream.responses[1].GetRequestBody() + if rb == nil || rb.Response == nil || rb.Response.HeaderMutation == nil { + t.Fatalf("expected RequestBody response with header mutation, got %+v", stream.responses[1]) + } + if got := mutationHeaderValue(rb.Response.HeaderMutation, "traceparent"); got != newTP { + t.Errorf("traceparent mutation = %q, want %q (trace rewrite lost on the body path)", got, newTP) + } +} + +// TestExtProc_Outbound_UnchangedTraceHeadersEmitNothing: when no plugin +// touches the trace headers, the listener must not emit mutations for them +// (echoing an unchanged header back would be a silent no-op today but +// masks diff regressions). +func TestExtProc_Outbound_UnchangedTraceHeadersEmitNothing(t *testing.T) { + srv := traceRewriterServer(t, &traceRewriterPlugin{}) + + stream := &mockStream{ + ctx: context.Background(), + requests: []*extprocv3.ProcessingRequest{ + outboundRequest(makeHeaders( + ":authority", "fanin-echo", + "traceparent", "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + "tracestate", "dg-parent=00f067aa0ba902b7", + )), + }, + } + _ = srv.Process(stream) + + if len(stream.responses) != 1 { + t.Fatalf("expected 1 response, got %d", len(stream.responses)) + } + rh := stream.responses[0].GetRequestHeaders() + if rh == nil { + t.Fatal("expected HeadersResponse") + } + if rh.Response != nil && rh.Response.HeaderMutation != nil { + hm := rh.Response.HeaderMutation + if v := mutationHeaderValue(hm, "traceparent"); v != "" { + t.Errorf("unexpected traceparent mutation %q on unchanged header", v) + } + if v := mutationHeaderValue(hm, "tracestate"); v != "" { + t.Errorf("unexpected tracestate mutation %q on unchanged header", v) + } + } +} + +// mutationRemovesHeader reports whether hm removes key. +func mutationRemovesHeader(hm *extprocv3.HeaderMutation, key string) bool { + if hm == nil { + return false + } + for _, rh := range hm.RemoveHeaders { + if rh == key { + return true + } + } + return false +} + +// TestExtProc_Outbound_ArbitraryHeaderReachesWire: the sync is generic — a +// plugin injecting any header (static-inject's x-api-key is the upstream +// case) must reach the wire, not just the trace headers. +func TestExtProc_Outbound_ArbitraryHeaderReachesWire(t *testing.T) { + srv := traceRewriterServer(t, &traceRewriterPlugin{ + set: map[string]string{"x-api-key": "secret-value"}, + }) + + stream := &mockStream{ + ctx: context.Background(), + requests: []*extprocv3.ProcessingRequest{ + outboundRequest(makeHeaders(":authority", "fanin-echo")), + }, + } + _ = srv.Process(stream) + + if len(stream.responses) != 1 { + t.Fatalf("expected 1 response, got %d", len(stream.responses)) + } + rh := stream.responses[0].GetRequestHeaders() + if rh == nil || rh.Response == nil || rh.Response.HeaderMutation == nil { + t.Fatalf("expected HeadersResponse with header mutation, got %+v", stream.responses[0]) + } + if got := mutationHeaderValue(rh.Response.HeaderMutation, "x-api-key"); got != "secret-value" { + t.Errorf("x-api-key mutation = %q, want %q (injected header dropped)", got, "secret-value") + } +} + +// TestExtProc_Outbound_DeletedHeaderIsRemoved: a plugin deleting a header +// must emit RemoveHeaders — the narrow two-name diff could not express this. +func TestExtProc_Outbound_DeletedHeaderIsRemoved(t *testing.T) { + srv := traceRewriterServer(t, &traceRewriterPlugin{del: []string{"x-drop-me"}}) + + stream := &mockStream{ + ctx: context.Background(), + requests: []*extprocv3.ProcessingRequest{ + outboundRequest(makeHeaders(":authority", "fanin-echo", "x-drop-me", "present")), + }, + } + _ = srv.Process(stream) + + if len(stream.responses) != 1 { + t.Fatalf("expected 1 response, got %d", len(stream.responses)) + } + rh := stream.responses[0].GetRequestHeaders() + if rh == nil || rh.Response == nil || rh.Response.HeaderMutation == nil { + t.Fatalf("expected HeadersResponse with header mutation, got %+v", stream.responses[0]) + } + if !mutationRemovesHeader(rh.Response.HeaderMutation, "x-drop-me") { + t.Errorf("expected x-drop-me in RemoveHeaders, got %+v", rh.Response.HeaderMutation) + } +} + +// TestExtProc_Outbound_PseudoHeadersNeverEmitted: headerMapToHTTP copies the +// HTTP/2 pseudo-headers into pctx.Headers (Go's canonicaliser leaves +// ":"-prefixed keys alone). Emitting a mutation for :authority would rewrite +// routing, so the sync must skip them even while emitting a real change. +func TestExtProc_Outbound_PseudoHeadersNeverEmitted(t *testing.T) { + srv := traceRewriterServer(t, &traceRewriterPlugin{ + set: map[string]string{"x-api-key": "secret-value"}, + }) + + stream := &mockStream{ + ctx: context.Background(), + requests: []*extprocv3.ProcessingRequest{ + outboundRequest(makeHeaders( + ":authority", "fanin-echo", + ":method", "POST", + ":path", "/rpc", + )), + }, + } + _ = srv.Process(stream) + + rh := stream.responses[0].GetRequestHeaders() + if rh == nil || rh.Response == nil || rh.Response.HeaderMutation == nil { + t.Fatalf("expected HeadersResponse with header mutation, got %+v", stream.responses[0]) + } + for _, pseudo := range []string{":authority", ":method", ":path", ":scheme"} { + if got := mutationHeaderValue(rh.Response.HeaderMutation, pseudo); got != "" { + t.Errorf("pseudo-header %s emitted as %q — would rewrite routing", pseudo, got) + } + if mutationRemovesHeader(rh.Response.HeaderMutation, pseudo) { + t.Errorf("pseudo-header %s emitted in RemoveHeaders", pseudo) + } + } +} diff --git a/authbridge/authlib/listener/forwardproxy/server.go b/authbridge/authlib/listener/forwardproxy/server.go index e3d33cc79..0e289becc 100644 --- a/authbridge/authlib/listener/forwardproxy/server.go +++ b/authbridge/authlib/listener/forwardproxy/server.go @@ -19,7 +19,6 @@ import ( "sync" "time" - "github.com/rossoctl/cortex/authbridge/authlib/auth" "github.com/rossoctl/cortex/authbridge/authlib/listener/httpx" "github.com/rossoctl/cortex/authbridge/authlib/listener/internal/sseframe" "github.com/rossoctl/cortex/authbridge/authlib/listener/skiphost" @@ -272,7 +271,6 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, isBridge } } - originalAuth := pctx.Headers.Get("Authorization") if !skipped { action := s.OutboundPipeline.Run(r.Context(), pctx) @@ -324,9 +322,28 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, isBridge s.Sessions.Append(sid, ev) } - newAuth := pctx.Headers.Get("Authorization") - if newAuth != originalAuth { - r.Header.Set("Authorization", "Bearer "+auth.ExtractBearer(newAuth)) + // Propagate every header mutation the outbound pipeline made to the + // forwarded request. pctx.Headers started as a clone of r.Header, so + // plugins' set / replace / delete operations on it are the intended + // upstream-facing header set. Only Authorization used to be forwarded, + // silently dropping any other injected header (e.g. static-inject's + // x-api-key). Content-Length / Content-Encoding are managed by the + // body-rewrite block below and the transport, so leave them untouched. + // Mirrors reverseproxy's forwarded-request header sync. + skip := func(k string) bool { return k == "Content-Length" || k == "Content-Encoding" } + for k := range r.Header { + if skip(k) { + continue + } + if _, ok := pctx.Headers[k]; !ok { + r.Header.Del(k) // plugin removed it + } + } + for k, vv := range pctx.Headers { + if skip(k) { + continue + } + r.Header[k] = append([]string(nil), vv...) // set / overwrite } // If a WritesBody plugin rewrote pctx.Body, ship the new bytes From 4ddbc1ec58205b064083efb5e4b451e213baafa7 Mon Sep 17 00:00:00 2001 From: Igor Gokhman Date: Thu, 20 Aug 2026 12:03:59 +0300 Subject: [PATCH 2/3] Test: Cover forwardproxy plugin header mutation on the wire PR #760 gave forwardproxy the same generic header-sync as extproc/ reverseproxy but added no forwardproxy tests; the set/replace/delete behaviour was covered only indirectly via Authorization. Add four listener-level regression tests that assert on the headers the upstream backend actually receives: TestForwardProxy_ArbitraryHeaderReachesWire (plugin-Set x-api-key) TestForwardProxy_OverwrittenHeaderReachesWire (Set replaces client value) TestForwardProxy_DeletedHeaderIsRemoved (plugin Del strips it) TestForwardProxy_UnchangedHeaderPreserved (untouched header survives) All four fail against the pre-PR Authorization-only path and pass on the generic sync, mirroring the extproc server_headerdiff_test.go suite. Assisted-By: Claude (Anthropic AI) Signed-off-by: Igor Gokhman --- .../forwardproxy/server_headerdiff_test.go | 186 ++++++++++++++++++ 1 file changed, 186 insertions(+) create mode 100644 authbridge/authlib/listener/forwardproxy/server_headerdiff_test.go diff --git a/authbridge/authlib/listener/forwardproxy/server_headerdiff_test.go b/authbridge/authlib/listener/forwardproxy/server_headerdiff_test.go new file mode 100644 index 000000000..d9b61924d --- /dev/null +++ b/authbridge/authlib/listener/forwardproxy/server_headerdiff_test.go @@ -0,0 +1,186 @@ +package forwardproxy + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/rossoctl/cortex/authbridge/authlib/pipeline" + "github.com/rossoctl/cortex/authbridge/authlib/plugins/plugintesting" +) + +// headerMutatorPlugin performs an arbitrary set / overwrite / delete on +// pctx.Headers during OnRequest. It is the forwardproxy analog of the +// ext_proc traceRewriterPlugin: the header names are deliberately +// ordinary (x-api-key, x-drop-me) rather than Authorization, because the +// point of PR #760 is that EVERY plugin header mutation — not just the +// old Authorization special case — must reach the upstream request. The +// plugin declares no capabilities: a header write does not need +// ReadsBody/WritesBody, mirroring how staticinject/cpex mutate headers. +type headerMutatorPlugin struct { + set map[string]string // header -> value to Set (set or overwrite) + del []string // headers to Del +} + +func (p *headerMutatorPlugin) Name() string { return "header-mutator" } +func (p *headerMutatorPlugin) Capabilities() pipeline.PluginCapabilities { + return pipeline.PluginCapabilities{} +} +func (p *headerMutatorPlugin) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { + for k, v := range p.set { + pctx.Headers.Set(k, v) + } + for _, k := range p.del { + pctx.Headers.Del(k) + } + return pipeline.Action{Type: pipeline.Continue} +} +func (p *headerMutatorPlugin) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action { + return pipeline.Action{Type: pipeline.Continue} +} + +// headerMutatorFixture wires a forward proxy whose outbound pipeline runs +// the given mutator, plus a backend that captures the headers it actually +// received on the wire. +type headerMutatorFixture struct { + client *http.Client + backendURL string + headers func() http.Header +} + +// newHeaderMutatorFixture returns a fixture and a cleanup func. The proxy +// dials the httptest backend because the request URL (backendURL) is the +// backend's own URL — the forward-proxy contract. The captured headers +// are what reached the backend AFTER the outbound pipeline + the sync +// block PR #760 added, so asserting on them proves the sync fired. +func newHeaderMutatorFixture(t *testing.T, mut *headerMutatorPlugin) (*headerMutatorFixture, func()) { + t.Helper() + + gotHeaders := make(chan http.Header, 1) + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotHeaders <- r.Header.Clone() + w.WriteHeader(http.StatusOK) + })) + + p, err := plugintesting.BuildPipeline([]pipeline.Plugin{mut}) + if err != nil { + backend.Close() + t.Fatalf("BuildPipeline: %v", err) + } + srv := &Server{OutboundPipeline: pipeline.NewHolder(p), Client: http.DefaultClient} + proxy := httptest.NewServer(srv.Handler()) + + fx := &headerMutatorFixture{ + client: &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(mustParseURL(proxy.URL))}}, + backendURL: backend.URL, + headers: func() http.Header { + select { + case h := <-gotHeaders: + return h + default: + t.Fatal("backend was never reached") + return nil + } + }, + } + cleanup := func() { + proxy.Close() + backend.Close() + } + return fx, cleanup +} + +// do sends a GET through the proxy to the backend, applying setup to the +// outgoing request (e.g. seeding client headers), and returns the headers +// the backend saw. It fails the test on transport error or non-200. +func (fx *headerMutatorFixture) do(t *testing.T, setup func(*http.Request)) http.Header { + t.Helper() + req, err := http.NewRequest(http.MethodGet, fx.backendURL+"/x", nil) + if err != nil { + t.Fatalf("NewRequest: %v", err) + } + if setup != nil { + setup(req) + } + resp, err := fx.client.Do(req) + if err != nil { + t.Fatalf("request failed: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + return fx.headers() +} + +// TestForwardProxy_ArbitraryHeaderReachesWire is the forwardproxy analog +// of the ext_proc TestExtProc_Outbound_ArbitraryHeaderReachesWire: a +// plugin that Sets a non-Authorization header must have that header reach +// the upstream. Before PR #760, forwardproxy forwarded only Authorization, +// silently dropping this injected header (e.g. static-inject's x-api-key). +func TestForwardProxy_ArbitraryHeaderReachesWire(t *testing.T) { + fx, cleanup := newHeaderMutatorFixture(t, &headerMutatorPlugin{ + set: map[string]string{"X-Api-Key": "secret-123"}, + }) + defer cleanup() + + // The request carries no X-Api-Key; only the plugin injects it. + h := fx.do(t, nil) + if got := h.Get("X-Api-Key"); got != "secret-123" { + t.Errorf("backend X-Api-Key = %q, want secret-123 (plugin-injected header did not reach the wire)", got) + } +} + +// TestForwardProxy_OverwrittenHeaderReachesWire asserts a plugin that +// Sets an already-present header overwrites the client's value on the +// forwarded request (set/replace, not append). +func TestForwardProxy_OverwrittenHeaderReachesWire(t *testing.T) { + fx, cleanup := newHeaderMutatorFixture(t, &headerMutatorPlugin{ + set: map[string]string{"X-Tenant": "server-chosen"}, + }) + defer cleanup() + + h := fx.do(t, func(r *http.Request) { r.Header.Set("X-Tenant", "client-supplied") }) + if got := h.Values("X-Tenant"); len(got) != 1 || got[0] != "server-chosen" { + t.Errorf("backend X-Tenant = %v, want [server-chosen] (plugin overwrite did not replace client value)", got) + } +} + +// TestForwardProxy_DeletedHeaderIsRemoved is the forwardproxy analog of +// the ext_proc TestExtProc_Outbound_DeletedHeaderIsRemoved: a plugin that +// Dels a header the client sent must strip it from the forwarded request. +// Before PR #760 the Authorization-only path had no way to express a +// deletion, so a plugin asking to remove a header was ignored. +func TestForwardProxy_DeletedHeaderIsRemoved(t *testing.T) { + fx, cleanup := newHeaderMutatorFixture(t, &headerMutatorPlugin{ + del: []string{"X-Drop-Me"}, + }) + defer cleanup() + + h := fx.do(t, func(r *http.Request) { r.Header.Set("X-Drop-Me", "please-remove") }) + if got := h.Get("X-Drop-Me"); got != "" { + t.Errorf("backend X-Drop-Me = %q, want empty (plugin deletion did not reach the wire)", got) + } +} + +// TestForwardProxy_UnchangedHeaderPreserved guards the other direction: +// a header the client sent that NO plugin touches must still reach the +// upstream unchanged. This pins that the delete loop only strips headers +// the plugin actually removed, never a spurious drop of untouched ones. +func TestForwardProxy_UnchangedHeaderPreserved(t *testing.T) { + // Plugin mutates an unrelated header so the pipeline runs, but leaves + // X-Keep alone. + fx, cleanup := newHeaderMutatorFixture(t, &headerMutatorPlugin{ + set: map[string]string{"X-Added": "1"}, + }) + defer cleanup() + + h := fx.do(t, func(r *http.Request) { r.Header.Set("X-Keep", "keep-me") }) + if got := h.Get("X-Keep"); got != "keep-me" { + t.Errorf("backend X-Keep = %q, want keep-me (untouched client header was dropped)", got) + } + if got := h.Get("X-Added"); got != "1" { + t.Errorf("backend X-Added = %q, want 1", got) + } +} From d0cb475d4e5bcc060ca2a12d0385b5d02c85c6d1 Mon Sep 17 00:00:00 2001 From: Igor Gokhman Date: Thu, 20 Aug 2026 12:13:23 +0300 Subject: [PATCH 3/3] Refactor: Remove dead replaceToken* helpers in extproc The PR retired the Authorization special case in the four ext_proc handlers in favour of the generic withHeaderMutation diff, leaving replaceTokenResponse and replaceTokenBodyResponse with no callers. They compiled only because Go does not flag unused package-level functions and CI runs no unused-code linter; the only remaining mentions were stale comments in placeholder_test.go describing the removed mechanism. Delete both functions and refresh the placeholder_test.go comments to name the withHeaderMutation path the tests actually exercise. No behaviour change: TestExtProc_Inbound{,Body}_AuthorizationMutation still assert the same SetHeaders mutation and pass. Assisted-By: Claude (Anthropic AI) Signed-off-by: Igor Gokhman --- .../listener/extproc/placeholder_test.go | 19 ++++--- authbridge/authlib/listener/extproc/server.go | 50 ------------------- 2 files changed, 9 insertions(+), 60 deletions(-) diff --git a/authbridge/authlib/listener/extproc/placeholder_test.go b/authbridge/authlib/listener/extproc/placeholder_test.go index f5a1a84b8..c877e7347 100644 --- a/authbridge/authlib/listener/extproc/placeholder_test.go +++ b/authbridge/authlib/listener/extproc/placeholder_test.go @@ -11,7 +11,7 @@ import ( // mintPlugin rewrites the inbound Authorization header to a minted // credential. Used to assert handleInbound emits a SetHeaders mutation -// (via replaceTokenResponse) carrying the new value so Envoy rewrites the +// (via withHeaderMutation) carrying the new value so Envoy rewrites the // request to the agent. type mintPlugin struct{} @@ -28,7 +28,7 @@ func (mintPlugin) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Ac } // setHeaderValue extracts the value for the named SetHeaders key from a -// RequestHeaders ProcessingResponse. replaceTokenResponse stores the value +// RequestHeaders ProcessingResponse. withHeaderMutation stores the value // in RawValue; fall back to Value for robustness. Returns ("", false) when // the key is absent. func setHeaderValue(resp *extprocv3.ProcessingResponse, key string) (string, bool) { @@ -100,12 +100,11 @@ func headerRemoved(cr *extprocv3.CommonResponse, key string) bool { } // bodyHeaderValue extracts the value for the named SetHeaders key from a -// RequestBody ProcessingResponse. The body path (replaceTokenBodyResponse -// wrapped by withBodyMutation) nests the SetHeaders mutation inside the -// RequestBody's CommonResponse rather than the RequestHeaders response that -// setHeaderValue reads, so it needs its own accessor. replaceTokenBodyResponse -// stores the value in RawValue; fall back to Value for robustness. Returns -// ("", false) when the key is absent. +// RequestBody ProcessingResponse. On the body path withHeaderMutation nests +// the SetHeaders mutation inside the RequestBody's CommonResponse rather than +// the RequestHeaders response that setHeaderValue reads, so it needs its own +// accessor. withHeaderMutation stores the value in RawValue; fall back to +// Value for robustness. Returns ("", false) when the key is absent. func bodyHeaderValue(resp *extprocv3.ProcessingResponse, key string) (string, bool) { rb := resp.GetRequestBody() if rb == nil || rb.GetResponse() == nil || rb.GetResponse().GetHeaderMutation() == nil { @@ -129,8 +128,8 @@ func bodyHeaderValue(resp *extprocv3.ProcessingResponse, key string) (string, bo // (handleInboundBody) instead of the header path. A plugin that rewrites the // inbound Authorization header must cause handleInboundBody to emit a // SetHeaders mutation carrying the new value — nested in the RequestBody -// response via replaceTokenBodyResponse/withBodyMutation — so Envoy rewrites -// the request to the agent on the body phase too. +// response via withHeaderMutation — so Envoy rewrites the request to the +// agent on the body phase too. func TestExtProc_InboundBody_AuthorizationMutation(t *testing.T) { p, err := pipeline.New([]pipeline.Plugin{mintPlugin{}}) if err != nil { diff --git a/authbridge/authlib/listener/extproc/server.go b/authbridge/authlib/listener/extproc/server.go index bf3a011c9..afecfc87b 100644 --- a/authbridge/authlib/listener/extproc/server.go +++ b/authbridge/authlib/listener/extproc/server.go @@ -860,56 +860,6 @@ func allowBodyResponse() *extprocv3.ProcessingResponse { } } -func replaceTokenBodyResponse(token string) *extprocv3.ProcessingResponse { - return &extprocv3.ProcessingResponse{ - Response: &extprocv3.ProcessingResponse_RequestBody{ - RequestBody: &extprocv3.BodyResponse{ - Response: &extprocv3.CommonResponse{ - HeaderMutation: &extprocv3.HeaderMutation{ - SetHeaders: []*corev3.HeaderValueOption{ - { - Header: &corev3.HeaderValue{ - Key: "authorization", - RawValue: []byte("Bearer " + token), - }, - }, - }, - // Strip the internal direction header before forwarding, - // matching allowResponse/allowBodyResponse — otherwise - // Envoy leaks x-authbridge-direction to the agent/target. - RemoveHeaders: []string{"x-authbridge-direction"}, - }, - }, - }, - }, - } -} - -func replaceTokenResponse(token string) *extprocv3.ProcessingResponse { - return &extprocv3.ProcessingResponse{ - Response: &extprocv3.ProcessingResponse_RequestHeaders{ - RequestHeaders: &extprocv3.HeadersResponse{ - Response: &extprocv3.CommonResponse{ - HeaderMutation: &extprocv3.HeaderMutation{ - SetHeaders: []*corev3.HeaderValueOption{ - { - Header: &corev3.HeaderValue{ - Key: "authorization", - RawValue: []byte("Bearer " + token), - }, - }, - }, - // Strip the internal direction header before forwarding, - // matching allowResponse/allowBodyResponse — otherwise - // Envoy leaks x-authbridge-direction to the agent/target. - RemoveHeaders: []string{"x-authbridge-direction"}, - }, - }, - }, - }, - } -} - // rejectFromActionForRequest is the MCP-aware sibling of rejectFromAction. // When pctx carries an MCP JSON-RPC request shape (Method + non-nil RPCID), // the response is an HTTP 200 carrying a JSON-RPC 2.0 error frame so the