Skip to content

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.

Sign up for GitHub

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

Open
JoshSag wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Fix: Propagate every plugin header mutation in extproc and forwardproxy #760

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4440ef9
Fix: Propagate every plugin header mutation in extproc and forwardproxy
JoshSag Aug 14, 2026
4ddbc1e
Test: Cover forwardproxy plugin header mutation on the wire
faraway7 Aug 20, 2026
d0cb475
Refactor: Remove dead replaceToken* helpers in extproc
faraway7 Aug 20, 2026
a898234
Fix: Address review — drop inbound Host population, nil-value delete,…
JoshSag Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 9 additions & 10 deletions authbridge/authlib/listener/extproc/placeholder_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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{}

Expand All @@ -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) {
Expand Down Expand Up @@ -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 {
Expand All @@ -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 {
Expand Down
182 changes: 101 additions & 81 deletions authbridge/authlib/listener/extproc/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import (
"io"
"log/slog"
"net/http"
"slices"
"strconv"
"strings"
"time"
Expand All @@ -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"
Expand Down Expand Up @@ -168,7 +168,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)
Expand All @@ -177,10 +177,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) {
Expand All @@ -196,7 +193,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)
Expand All @@ -205,10 +202,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
Expand Down Expand Up @@ -469,16 +464,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.
Expand All @@ -495,7 +487,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)
Expand All @@ -505,11 +497,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) {
Expand All @@ -518,16 +506,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
Expand All @@ -547,7 +532,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)
Expand All @@ -557,11 +542,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 {
Expand Down Expand Up @@ -705,6 +687,94 @@ 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 {

@huang195 huang195 Aug 17, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

suggestion — With the Authorization special case retired, replaceTokenResponse (line 888) and replaceTokenBodyResponse (line 863) have no callers left. The five references in placeholder_test.go (lines 14, 31, 103, 106, 132) are comments, not calls, and now describe a path production no longer takes. Deleting both helpers and rewording those comments to name withHeaderMutation keeps the next reader from tracing a dead path.

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
}
if len(vv) == 0 {
// pctx.Headers[k] = nil is a delete, same as Del(k). Emitting
// an empty SetHeaders value instead would leave the outcome to
// Envoy's keep_empty_value setting.
del = append(del, strings.ToLower(k))
continue
}
// Wire header names are lowercase; pctx.Headers keys were
// canonicalised by http.Header.Set in headerMapToHTTP — which also
// collapses duplicate wire entries to their last value, so a header
// a plugin mutates is emitted as one value even if it arrived as
// several. 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, ","))},

@huang195 huang195 Aug 17, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

nit — Two edges worth a line of comment or a follow-up:

  • headerMapToHTTP (line 766) uses h.Set, so a header that arrived on the wire with duplicate entries is already collapsed to its last value in pctx.Headers. Unchanged headers emit nothing so nothing regresses, but for a header a plugin does mutate, the emitted SetHeaders replaces all wire values with the collapsed one.
  • A plugin doing pctx.Headers[k] = nil instead of Del(k) lands here rather than in the remove loop, emitting an empty RawValue — Envoy drops empty values without keep_empty_value, so the effect is right by accident. Treating a zero-length slice as a delete makes it right by construction.

@JoshSag JoshSag Aug 20, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

Both taken, with one correction worth making.

In extproc a zero-length value now lands in RemoveHeaders, so the outcome no longer depends on keep_empty_value — pinned by TestExtProc_Outbound_NilValueHeaderIsRemoved, which fails against the previous code with set_headers:{header:{key:"x-drop-me"}}.

In forwardproxy the same change alters no bytes: net/http already omits a header whose value slice is empty, so the previous line was correct on the wire and only the in-memory map differs. It is a consistency change, and no wire-level test can distinguish the two spellings. Happy to drop that hunk if you'd rather the PR carry only behavioural changes.

The duplicate-header collapse is now stated in the withHeaderMutation comment.

})
}
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. Outbound only —
// there it names the service being called (pipeline.SessionEvent.Host).
// The inbound handlers deliberately leave pctx.Host empty: the inbound
// authority is caller-controlled and pctx.Host feeds enforcement decisions
// (ibac's host-bypass skip, opa's policy input, per-host JWT audiences),
// so a spoofed Host header must not reach them. See cpex's outbound-only
// host-bypass guard for the same rule stated plugin-side.
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 {
Expand Down Expand Up @@ -802,56 +872,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
Expand Down
Loading