-
Notifications
You must be signed in to change notification settings - Fork 40
Fix: Propagate every plugin header mutation in extproc and forwardproxy #760
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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,14 +161,15 @@ 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, | ||
| Shared: s.Shared, | ||
| 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,14 +187,15 @@ 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, | ||
| Shared: s.Shared, | ||
| 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 { | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. suggestion — With the Authorization special case retired, |
||
| 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, ","))}, | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nit — Two edges worth a line of comment or a follow-up:
|
||
| }) | ||
| } | ||
| 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 { | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nit — This comment says Host is what "the lineage plugin's |
||
| // 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) | ||
| } | ||
| }) | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
must-fix — The header-propagation fix is sound, but this hunk (and its twin at line 190) is not the neutral telemetry fix the description claims. On the inbound path
:authority/Hostis caller-controlled, andpctx.Hostis read by decision-making plugins, not just recorded:plugins/ibac/plugin.go:352—matchesAnyHost(p.bypassHosts, pctx.Host)→pctx.Skip("host_bypass"), with no direction guard.defaultBypassHostsincludes keycloak/spire/otel, plus whateveragent_llm_hostis set to. Today in ext_proc inbound this branch is inert becausepctx.Hostis""; after this change a caller who setsHost: keycloak...skips IBAC judging entirely.plugins/opa/plugin.go:525—"host": pctx.Hostbecomes caller-controlled policy input.plugins/jwtvalidation/plugin.go:393— withaudience_mode: per-host, the expected audience is derived from the caller-supplied authority.This repo already documents the hazard and guards for it:
plugins/cpex/plugin.go:306-310gatesmatchesAnyHostbehindpctx.Direction == pipeline.Outbound, with the comment "the Host header is attacker-controlled and identity has NOT been pre-validated."Two ways forward, either is fine: (a) drop the two inbound
authorityOfhunks and keep the outbound consolidation (lines 469/511, a pure no-op refactor); or (b) land them together with a direction guard on ibac's host-bypass check. Worth notingreverseproxyalready populates inbound Host, so ibac's exposure pre-dates this PR — but this widens it to the ext_proc sidecar path, and that shouldn't ride along in a PR framed as header propagation.