feat(pipeline): Add Scheme to Context and plumb through listeners

authbridge/authlib/pipeline/context.go

@@ -82,10 +82,22 @@ func (d *Direction) UnmarshalJSON(data []byte) error {
 type Context struct {
 	Direction Direction
 	Method    string
-	Host      string
-	Path      string
-	Headers   http.Header
-	Body      []byte // nil unless at least one plugin declares BodyAccess: true
+	// Scheme is the URL scheme of the request, typically "http" or
+	// "https" — future transports ("ws" / "wss" / gRPC-specific
+	// schemes) pass through unchanged as free-form strings. Populated
+	// by the listener at pctx construction from the transport-native
+	// field: the :scheme pseudo-header in ext_proc and ext_authz,
+	// r.URL.Scheme in the forward and reverse proxies.
+	//
+	// Empty when the listener can't determine scheme (legacy test
+	// fixtures, an unrecognized transport, etc.). Plugins that need a
+	// concrete scheme should pick a default explicitly — treating ""
+	// as "assume http" would silently mask missing listener plumbing.
+	Scheme  string
+	Host    string
+	Path    string
+	Headers http.Header
+	Body    []byte // nil unless at least one plugin declares BodyAccess: true
 
 	// StartedAt is the wall-clock time this context was constructed by the
 	// listener at the start of a request. Used on the response path to

authbridge/authlib/plugins/tokenbroker/plugin.go

@@ -224,8 +224,16 @@ func (p *TokenBroker) OnRequest(ctx context.Context, pctx *pipeline.Context) pip
 			"broker route requires authorization token")
 	}
 
-	// Derive server URL from host
-	serverURL := "http://" + host
+	// Derive server URL from scheme + host. pctx.Scheme is populated
+	// by the listener from :scheme (ext_proc / ext_authz) or
+	// r.URL.Scheme / r.TLS (forward / reverse proxy). Defaults to
+	// "http" when the listener couldn't determine one — matches the
+	// previous hardcoded behavior.
+	scheme := pctx.Scheme
+	if scheme == "" {
+		scheme = "http"
+	}
+	serverURL := scheme + "://" + host
 
 	// Use the plugin's configured broker URL
 	brokerURL := p.cfg.BrokerURL

authbridge/authlib/plugins/tokenbroker/plugin_request_test.go

@@ -475,3 +475,52 @@ func TestTokenBroker_OnResponse(t *testing.T) {
 		t.Errorf("OnResponse() action.Type = %v, want %v", action.Type, pipeline.Continue)
 	}
 }
+
+// TestTokenBroker_ServerURL_SchemeAware verifies the serverURL sent
+// to the broker (via X-Server-Url) uses pctx.Scheme when populated.
+// Defaults to "http" when pctx.Scheme is empty — matches the
+// previous hardcoded behavior so existing callers upgrade without
+// surprise. See issue #397.
+func TestTokenBroker_ServerURL_SchemeAware(t *testing.T) {
+	tests := []struct {
+		name   string
+		scheme string
+		want   string
+	}{
+		{name: "scheme_https", scheme: "https", want: "https://api.example.com"},
+		{name: "scheme_http_explicit", scheme: "http", want: "http://api.example.com"},
+		{name: "scheme_empty_defaults_http", scheme: "", want: "http://api.example.com"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			var gotServerURL string
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				gotServerURL = r.Header.Get("X-Server-Url")
+				w.Header().Set("Content-Type", "application/json")
+				w.WriteHeader(http.StatusOK)
+				_ = json.NewEncoder(w).Encode(map[string]string{"token": "x"})
+			}))
+			defer srv.Close()
+
+			p := NewTokenBroker()
+			cfg := `{"broker_url": "` + srv.URL + `", "default_policy": "broker"}`
+			if err := p.Configure(json.RawMessage(cfg)); err != nil {
+				t.Fatalf("Configure: %v", err)
+			}
+
+			pctx := &pipeline.Context{
+				Scheme: tc.scheme,
+				Host:   "api.example.com",
+				Headers: http.Header{
+					"Authorization": []string{"Bearer user-token"},
+				},
+			}
+			if action := p.OnRequest(context.Background(), pctx); action.Type != pipeline.Continue {
+				t.Fatalf("OnRequest action = %v, want Continue", action.Type)
+			}
+			if gotServerURL != tc.want {
+				t.Errorf("X-Server-Url = %q, want %q", gotServerURL, tc.want)
+			}
+		})
+	}
+}

authbridge/cmd/authbridge/listener/extauthz/server.go

@@ -44,10 +44,12 @@ func (s *Server) Check(ctx context.Context, req *authv3.CheckRequest) (*authv3.C
 		host = headers["host"]
 	}
 	path := httpReq.GetPath()
+	scheme := httpReq.GetScheme()
 
 	// Inbound validation via pipeline
 	inPctx := &pipeline.Context{
 		Direction: pipeline.Inbound,
+		Scheme:    scheme,
 		Host:      host,
 		Path:      path,
 		Headers:   mapToHTTPHeader(headers),
@@ -60,6 +62,7 @@ func (s *Server) Check(ctx context.Context, req *authv3.CheckRequest) (*authv3.C
 	// Outbound exchange via pipeline
 	outPctx := &pipeline.Context{
 		Direction: pipeline.Outbound,
+		Scheme:    scheme,
 		Host:      host,
 		Path:      path,
 		Headers:   mapToHTTPHeader(headers),

authbridge/cmd/authbridge/listener/extauthz/server_test.go

@@ -12,12 +12,12 @@ import (
 	"google.golang.org/grpc/codes"
 
 	authpkg "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth"
-	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/cache"
-	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/exchange"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/jwtvalidation/validation"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/plugintesting"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/cache"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/exchange"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/routing"
-	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/jwtvalidation/validation"
 )
 
 type mockVerifier struct {
@@ -151,3 +151,76 @@ func TestCheck_MissingHTTPAttributes(t *testing.T) {
 		t.Fatal("expected DeniedResponse for missing attributes")
 	}
 }
+
+// schemeCapturePlugin records pctx.Scheme. Duplicated in each listener
+// test package because these are the smallest-unit Plugin types that
+// don't belong in authlib.
+type schemeCapturePlugin struct {
+	got string
+}
+
+func (p *schemeCapturePlugin) Name() string { return "scheme-capture" }
+func (p *schemeCapturePlugin) Capabilities() pipeline.PluginCapabilities {
+	return pipeline.PluginCapabilities{}
+}
+func (p *schemeCapturePlugin) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action {
+	p.got = pctx.Scheme
+	return pipeline.Action{Type: pipeline.Continue}
+}
+func (p *schemeCapturePlugin) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action {
+	return pipeline.Action{Type: pipeline.Continue}
+}
+
+// TestCheck_PopulatesSchemeFromAttribute verifies that both the
+// inbound and outbound pctx constructed by Check() read Scheme from
+// the CheckRequest's HTTP attribute, so plugins composing target
+// URLs or branching on transport see the right value in waypoint
+// mode.
+func TestCheck_PopulatesSchemeFromAttribute(t *testing.T) {
+	tests := []struct {
+		name   string
+		scheme string
+		want   string
+	}{
+		{name: "http", scheme: "http", want: "http"},
+		{name: "https", scheme: "https", want: "https"},
+		{name: "empty_passes_through", scheme: "", want: ""},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			inCap := &schemeCapturePlugin{}
+			outCap := &schemeCapturePlugin{}
+			in, err := plugintesting.BuildPipeline([]pipeline.Plugin{inCap})
+			if err != nil {
+				t.Fatalf("BuildPipeline inbound: %v", err)
+			}
+			out, err := plugintesting.BuildPipeline([]pipeline.Plugin{outCap})
+			if err != nil {
+				t.Fatalf("BuildPipeline outbound: %v", err)
+			}
+			srv := &Server{
+				InboundPipeline:  pipeline.NewHolder(in),
+				OutboundPipeline: pipeline.NewHolder(out),
+			}
+
+			req := &authv3.CheckRequest{
+				Attributes: &authv3.AttributeContext{
+					Request: &authv3.AttributeContext_Request{
+						Http: &authv3.AttributeContext_HttpRequest{
+							Scheme:  tc.scheme,
+							Headers: map[string]string{":authority": "agent.example"},
+							Path:    "/x",
+						},
+					},
+				},
+			}
+			_, _ = srv.Check(context.Background(), req)
+			if inCap.got != tc.want {
+				t.Errorf("inbound pctx.Scheme = %q, want %q", inCap.got, tc.want)
+			}
+			if outCap.got != tc.want {
+				t.Errorf("outbound pctx.Scheme = %q, want %q", outCap.got, tc.want)
+			}
+		})
+	}
+}

authbridge/cmd/authbridge/listener/extproc/server.go

@@ -127,6 +127,7 @@ func (s *Server) handleInbound(stream extprocv3.ExternalProcessor_ProcessServer,
 	ctx := stream.Context()
 	pctx := &pipeline.Context{
 		Direction: pipeline.Inbound,
+		Scheme:    getHeader(headers, ":scheme"),
 		Path:      getHeader(headers, ":path"),
 		Headers:   headerMapToHTTP(headers),
 		Body:      body,
@@ -147,6 +148,7 @@ func (s *Server) handleInboundBody(stream extprocv3.ExternalProcessor_ProcessSer
 	ctx := stream.Context()
 	pctx := &pipeline.Context{
 		Direction: pipeline.Inbound,
+		Scheme:    getHeader(headers, ":scheme"),
 		Path:      getHeader(headers, ":path"),
 		Headers:   headerMapToHTTP(headers),
 		Body:      body,
@@ -559,6 +561,7 @@ func (s *Server) handleOutbound(stream extprocv3.ExternalProcessor_ProcessServer
 	ctx := stream.Context()
 	pctx := &pipeline.Context{
 		Direction: pipeline.Outbound,
+		Scheme:    getHeader(headers, ":scheme"),
 		Host:      getHeader(headers, ":authority"),
 		Path:      getHeader(headers, ":path"),
 		Headers:   headerMapToHTTP(headers),
@@ -595,6 +598,7 @@ func (s *Server) handleOutboundBody(stream extprocv3.ExternalProcessor_ProcessSe
 	ctx := stream.Context()
 	pctx := &pipeline.Context{
 		Direction: pipeline.Outbound,
+		Scheme:    getHeader(headers, ":scheme"),
 		Host:      getHeader(headers, ":authority"),
 		Path:      getHeader(headers, ":path"),
 		Headers:   headerMapToHTTP(headers),

authbridge/cmd/authbridge/listener/extproc/server_test.go

@@ -16,13 +16,13 @@ import (
 
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/auth"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/bypass"
-	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/cache"
-	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/exchange"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/jwtvalidation/validation"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/plugintesting"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/cache"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/exchange"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/routing"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/session"
-	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/jwtvalidation/validation"
 )
 
 // mockStream implements ExternalProcessor_ProcessServer for testing.
@@ -1331,7 +1331,7 @@ func TestRecordOutboundReject_EmitsDeniedPhase(t *testing.T) {
 	// Seed an active session so ActiveSession() returns it instead of
 	// empty — mirrors the real request flow (inbound recorded first).
 	store.Append("sess-active", pipeline.SessionEvent{
-		At: time.Now().Add(-50 * time.Millisecond),
+		At:        time.Now().Add(-50 * time.Millisecond),
 		Direction: pipeline.Inbound, Phase: pipeline.SessionRequest,
 		A2A: &pipeline.A2AExtension{Method: "message/send", SessionID: "sess-active"},
 	})
@@ -1486,3 +1486,88 @@ func keysOf(m map[string]json.RawMessage) []string {
 	}
 	return out
 }
+
+// schemeCapturePlugin captures the pctx.Scheme value it sees on
+// OnRequest. Used by the per-listener scheme-wiring tests to verify
+// each listener populates the field from its transport-native source.
+//
+// Duplicated byte-for-byte across the four listener test packages
+// (extauthz, extproc, forwardproxy, reverseproxy) rather than
+// promoted to authlib/plugintesting — the plugin shape is only
+// useful in listener tests and not worth exporting. Each package
+// has a copy with this same godoc.
+type schemeCapturePlugin struct {
+	got string
+}
+
+func (p *schemeCapturePlugin) Name() string { return "scheme-capture" }
+func (p *schemeCapturePlugin) Capabilities() pipeline.PluginCapabilities {
+	return pipeline.PluginCapabilities{}
+}
+func (p *schemeCapturePlugin) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action {
+	p.got = pctx.Scheme
+	return pipeline.Action{Type: pipeline.Continue}
+}
+func (p *schemeCapturePlugin) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action {
+	return pipeline.Action{Type: pipeline.Continue}
+}
+
+// TestExtProc_PopulatesSchemeFromPseudoHeader verifies that each pctx
+// construction path in the ext_proc listener (inbound + outbound,
+// headers-only + body) reads the :scheme pseudo-header and surfaces
+// it on pctx.Scheme. Plugins composing target URLs or branching on
+// transport depend on this being populated unconditionally.
+func TestExtProc_PopulatesSchemeFromPseudoHeader(t *testing.T) {
+	tests := []struct {
+		name      string
+		direction string
+		scheme    string
+		want      string
+	}{
+		{name: "inbound_http", direction: "inbound", scheme: "http", want: "http"},
+		{name: "inbound_https", direction: "inbound", scheme: "https", want: "https"},
+		{name: "outbound_http", direction: "outbound", scheme: "http", want: "http"},
+		{name: "outbound_https", direction: "outbound", scheme: "https", want: "https"},
+		{name: "missing_pseudo_header_is_empty", direction: "inbound", scheme: "", want: ""},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			capturer := &schemeCapturePlugin{}
+			p, err := plugintesting.BuildPipeline([]pipeline.Plugin{capturer})
+			if err != nil {
+				t.Fatalf("BuildPipeline: %v", err)
+			}
+			srv := &Server{
+				InboundPipeline:  pipeline.NewHolder(p),
+				OutboundPipeline: pipeline.NewHolder(p),
+			}
+
+			hdrs := []string{
+				"x-authbridge-direction", tc.direction,
+				":path", "/x",
+				":authority", "agent.example",
+			}
+			if tc.scheme != "" {
+				hdrs = append(hdrs, ":scheme", tc.scheme)
+			}
+			// Use the direction-matching helper even though
+			// inboundRequest and outboundRequest share the same
+			// shape today — the test reads more clearly and guards
+			// against future drift if the helpers diverge.
+			reqFn := inboundRequest
+			if tc.direction == "outbound" {
+				reqFn = outboundRequest
+			}
+			stream := &mockStream{
+				ctx: context.Background(),
+				requests: []*extprocv3.ProcessingRequest{
+					reqFn(makeHeaders(hdrs...)),
+				},
+			}
+			_ = srv.Process(stream)
+			if capturer.got != tc.want {
+				t.Errorf("pctx.Scheme = %q, want %q", capturer.got, tc.want)
+			}
+		})
+	}
+}

authbridge/cmd/authbridge/listener/forwardproxy/server.go

@@ -56,6 +56,7 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
 
 	pctx := &pipeline.Context{
 		Direction: pipeline.Outbound,
+		Scheme:    r.URL.Scheme,
 		Host:      r.Host,
 		Path:      r.URL.Path,
 		Headers:   r.Header.Clone(),