feat(broker): per-user credential injection + per-(user,server) connection keying (spec 074 T7)

docs/features/auth-broker.md

@@ -90,6 +90,21 @@ On each proxied request the broker resolves the per-user credential to inject, i
 
 Concurrent requests for the same `(user, upstream)` are coalesced (single-flight) so a burst does not trigger duplicate upstream token flows. A policy-decision hook is evaluated per call immediately before the credential is returned; no policy engine ships yet, so it permits every injection by default.
 
+## Header injection
+
+The resolved per-user credential is injected into the configured outbound header (`header`, default `Authorization`) using the value template (`header_format`, default `Bearer {token}`), then the request is forwarded to the upstream.
+
+Injection is a **replacement**, not a merge:
+
+- Any header on the upstream config whose name matches `header` (case-insensitively) is **removed** before the resolved credential is set, so a brokered upstream presents exactly one value for that header.
+- The inbound gateway/IdP token is **never forwarded** to the upstream. Brokering exists precisely so the upstream sees a credential minted *for it*, scoped to the calling user — not the token the user presented to the gateway.
+
+Injection applies only to **HTTP-family** upstreams (`http`, `sse`, `streamable-http`). Brokering on a `stdio` upstream is rejected — at config validation, and again as a runtime guard at the injection boundary — with a clear "unsupported in this phase" message.
+
+## Per-(user, server) connection keying
+
+A **shared** upstream that is brokered per-user must carry **each user's own** credential. Brokered upstream connections are therefore keyed by `(user, server)`, never by server alone: one user's connection (and the credential injected on it) is never reused for another user. The server-component of the key reuses the same `name + URL` scheme as the credential store, so a connection and its cached credential stay in lockstep.
+
 ## See also
 
 - [OAuth Authentication](./oauth-authentication.md) — upstream OAuth for the personal edition.

internal/serveredition/broker/injector.go

@@ -0,0 +1,108 @@
+//go:build server
+
+package broker
+
+import (
+	"context"
+	"errors"
+
+	"github.com/smart-mcp-proxy/mcpproxy-go/internal/config"
+	"github.com/smart-mcp-proxy/mcpproxy-go/internal/oauth"
+	"github.com/smart-mcp-proxy/mcpproxy-go/internal/transport"
+)
+
+// ErrBrokerStdioUnsupported is returned when brokering is requested for a
+// non-HTTP-family (stdio) upstream. Credential injection only works over
+// HTTP/SSE/streamable-HTTP transports in this phase (spec 074 FR-002). This is a
+// runtime defense-in-depth check; config validation already rejects such blocks
+// at load time.
+var ErrBrokerStdioUnsupported = errors.New("auth broker: credential injection is only supported on HTTP-family upstreams (http, sse, streamable-http); stdio brokering is unsupported in this phase")
+
+// Fallback header/format used when a brokered server's config has not had its
+// defaults applied. These mirror config.AuthBrokerConfig.ApplyDefaults (FR-016);
+// config validation normally applies them at load time.
+const (
+	fallbackBrokerHeader       = "Authorization"
+	fallbackBrokerHeaderFormat = "Bearer {token}"
+)
+
+// resolver is the subset of *CredentialResolver the injector depends on. It is
+// an interface so tests can substitute a fake without a real store/exchanger.
+type resolver interface {
+	Resolve(ctx context.Context, userID string, server *config.ServerConfig) (*UpstreamCredential, error)
+}
+
+// HeaderInjector turns a per-user resolved upstream credential into the
+// transport-layer BrokeredAuth injected on a proxied request. It is the bridge
+// between the credential broker (server edition) and the edition-neutral
+// transport layer: the transport never imports the broker, it only receives the
+// resolved credential as plain data.
+//
+// The injector enforces the spec-074 brokering invariants at the injection
+// boundary:
+//   - per-user only: an empty userID is rejected (FR-014);
+//   - HTTP-family only: stdio brokering is rejected (FR-002);
+//   - replacement, not forwarding: the produced BrokeredAuth replaces any
+//     configured/inbound auth header (FR-016/FR-017, enforced in transport).
+type HeaderInjector struct {
+	resolver resolver
+}
+
+// NewHeaderInjector builds an injector over a credential resolver. *CredentialResolver
+// satisfies the resolver interface.
+func NewHeaderInjector(r resolver) *HeaderInjector {
+	return &HeaderInjector{resolver: r}
+}
+
+// InjectFor resolves the per-user credential for (userID, server) and returns
+// the transport.BrokeredAuth to inject. It returns:
+//   - ErrUnauthenticated if userID is empty (FR-014);
+//   - ErrBrokerNotConfigured if the server has no auth_broker block;
+//   - ErrBrokerStdioUnsupported if the server is not HTTP-family (FR-002);
+//   - any resolver error (e.g. *NotConnectedError carrying a connect URL).
+func (h *HeaderInjector) InjectFor(ctx context.Context, userID string, server *config.ServerConfig) (*transport.BrokeredAuth, error) {
+	if userID == "" {
+		return nil, ErrUnauthenticated
+	}
+	if server == nil || server.AuthBroker == nil {
+		return nil, ErrBrokerNotConfigured
+	}
+	// Defense-in-depth: reject brokering on stdio/non-HTTP upstreams (FR-002).
+	if transport.DetermineTransportType(server) == transport.TransportStdio {
+		return nil, ErrBrokerStdioUnsupported
+	}
+
+	cred, err := h.resolver.Resolve(ctx, userID, server)
+	if err != nil {
+		return nil, err
+	}
+	if cred == nil || cred.AccessToken == "" {
+		return nil, ErrNoCredential
+	}
+
+	header := server.AuthBroker.Header
+	if header == "" {
+		header = fallbackBrokerHeader
+	}
+	format := server.AuthBroker.HeaderFormat
+	if format == "" {
+		format = fallbackBrokerHeaderFormat
+	}
+	return &transport.BrokeredAuth{
+		Header: header,
+		Format: format,
+		Token:  cred.AccessToken,
+	}, nil
+}
+
+// ConnectionKey derives the pooling key for a brokered upstream connection. It
+// binds the connection to a single (user, server) pair so a shared upstream
+// brokered per-user never reuses one user's credential/connection for another
+// (FR-018). The server component reuses the existing oauth.GenerateServerKey
+// scheme (name + URL) so it matches the credential store's keying.
+func ConnectionKey(userID string, server *config.ServerConfig) string {
+	if server == nil {
+		return userID + "\x00"
+	}
+	return userID + "\x00" + oauth.GenerateServerKey(server.Name, server.URL)
+}

internal/serveredition/broker/injector_test.go

@@ -0,0 +1,163 @@
+//go:build server
+
+package broker
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/smart-mcp-proxy/mcpproxy-go/internal/config"
+	"github.com/smart-mcp-proxy/mcpproxy-go/internal/transport"
+)
+
+// fakeResolver returns a per-user token so the injector can be exercised without
+// a real store/exchanger. It records the per-user resolution so tests can assert
+// one user's credential is never produced for another (FR-018).
+type fakeResolver struct {
+	tokens map[string]string // userID -> access token
+	err    error
+}
+
+func (f *fakeResolver) Resolve(_ context.Context, userID string, _ *config.ServerConfig) (*UpstreamCredential, error) {
+	if f.err != nil {
+		return nil, f.err
+	}
+	tok, ok := f.tokens[userID]
+	if !ok {
+		return nil, ErrNoCredential
+	}
+	return &UpstreamCredential{AccessToken: tok, TokenType: "Bearer"}, nil
+}
+
+func httpBrokerServer() *config.ServerConfig {
+	s := &config.ServerConfig{
+		Name:     "ghe",
+		URL:      "https://ghe.example/mcp",
+		Protocol: "streamable-http",
+		AuthBroker: &config.AuthBrokerConfig{
+			Mode:          config.AuthBrokerModeTokenExchange,
+			TokenEndpoint: "https://idp.example/token",
+		},
+	}
+	s.AuthBroker.ApplyDefaults()
+	return s
+}
+
+// FR-016: the resolved per-user credential is rendered into the configured
+// header/format (default Authorization: Bearer {token}).
+func TestInjector_InjectFor_ProducesBrokeredAuth(t *testing.T) {
+	inj := NewHeaderInjector(&fakeResolver{tokens: map[string]string{"alice": "alice-tok"}})
+
+	ba, err := inj.InjectFor(context.Background(), "alice", httpBrokerServer())
+	if err != nil {
+		t.Fatalf("InjectFor: %v", err)
+	}
+	if ba.Header != "Authorization" {
+		t.Fatalf("header = %q, want Authorization", ba.Header)
+	}
+	if ba.HeaderValue() != "Bearer alice-tok" {
+		t.Fatalf("header value = %q, want %q", ba.HeaderValue(), "Bearer alice-tok")
+	}
+}
+
+// FR-018 + SC-002/003: two users brokered against the SAME shared upstream get
+// two distinct outbound tokens; one user's credential is never reused for the
+// other.
+func TestInjector_TwoUsers_TwoTokens(t *testing.T) {
+	inj := NewHeaderInjector(&fakeResolver{tokens: map[string]string{
+		"alice": "alice-tok",
+		"bob":   "bob-tok",
+	}})
+	server := httpBrokerServer()
+
+	aliceBA, err := inj.InjectFor(context.Background(), "alice", server)
+	if err != nil {
+		t.Fatalf("alice: %v", err)
+	}
+	bobBA, err := inj.InjectFor(context.Background(), "bob", server)
+	if err != nil {
+		t.Fatalf("bob: %v", err)
+	}
+
+	if aliceBA.HeaderValue() == bobBA.HeaderValue() {
+		t.Fatalf("two users produced the same outbound token %q (FR-018 violation)", aliceBA.HeaderValue())
+	}
+
+	// And the effective outbound headers must carry each user's own token.
+	aliceHdr := transport.EffectiveHeaders(nil, aliceBA)
+	bobHdr := transport.EffectiveHeaders(nil, bobBA)
+	if aliceHdr["Authorization"] != "Bearer alice-tok" || bobHdr["Authorization"] != "Bearer bob-tok" {
+		t.Fatalf("cross-user token leak: alice=%q bob=%q", aliceHdr["Authorization"], bobHdr["Authorization"])
+	}
+
+	// Per-(user,server) connection keys must differ so connections are never
+	// pooled across users (FR-018).
+	if ConnectionKey("alice", server) == ConnectionKey("bob", server) {
+		t.Fatalf("connection key collided across users (FR-018 violation)")
+	}
+}
+
+// FR-018: ConnectionKey is stable for the same (user, server) and distinct per
+// user and per server so a brokered connection is never reused across either.
+func TestConnectionKey_StableAndDistinct(t *testing.T) {
+	s1 := httpBrokerServer()
+	s2 := httpBrokerServer()
+	s2.Name = "other"
+	s2.URL = "https://other.example/mcp"
+
+	k1 := ConnectionKey("alice", s1)
+	k1again := ConnectionKey("alice", s1)
+	if k1 != k1again {
+		t.Fatal("ConnectionKey must be stable for the same (user, server)")
+	}
+	if ConnectionKey("alice", s1) == ConnectionKey("alice", s2) {
+		t.Fatal("ConnectionKey must differ per server")
+	}
+	if ConnectionKey("alice", s1) == ConnectionKey("bob", s1) {
+		t.Fatal("ConnectionKey must differ per user")
+	}
+}
+
+// FR-002: brokering on a stdio upstream is rejected with a clear, actionable
+// message — never silently injected.
+func TestInjector_RejectsStdio(t *testing.T) {
+	inj := NewHeaderInjector(&fakeResolver{tokens: map[string]string{"alice": "x"}})
+	stdio := &config.ServerConfig{
+		Name:     "local",
+		Protocol: "stdio",
+		Command:  "my-mcp",
+		AuthBroker: &config.AuthBrokerConfig{
+			Mode:          config.AuthBrokerModeTokenExchange,
+			TokenEndpoint: "https://idp.example/token",
+		},
+	}
+	stdio.AuthBroker.ApplyDefaults()
+
+	_, err := inj.InjectFor(context.Background(), "alice", stdio)
+	if !errors.Is(err, ErrBrokerStdioUnsupported) {
+		t.Fatalf("stdio brokering must be rejected with ErrBrokerStdioUnsupported, got %v", err)
+	}
+}
+
+// A server with no auth_broker block is not brokered: the injector returns
+// ErrBrokerNotConfigured and the caller proceeds with today's behaviour.
+func TestInjector_NotConfigured(t *testing.T) {
+	inj := NewHeaderInjector(&fakeResolver{})
+	plain := &config.ServerConfig{Name: "plain", URL: "https://x/mcp", Protocol: "streamable-http"}
+
+	_, err := inj.InjectFor(context.Background(), "alice", plain)
+	if !errors.Is(err, ErrBrokerNotConfigured) {
+		t.Fatalf("non-brokered server must return ErrBrokerNotConfigured, got %v", err)
+	}
+}
+
+// An empty userID is rejected before any resolution — brokering is strictly
+// per-user (FR-014).
+func TestInjector_RejectsAnonymous(t *testing.T) {
+	inj := NewHeaderInjector(&fakeResolver{tokens: map[string]string{"alice": "x"}})
+	_, err := inj.InjectFor(context.Background(), "", httpBrokerServer())
+	if !errors.Is(err, ErrUnauthenticated) {
+		t.Fatalf("anonymous caller must be rejected with ErrUnauthenticated, got %v", err)
+	}
+}

internal/serveredition/multiuser/router.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/smart-mcp-proxy/mcpproxy-go/internal/auth"
 	"github.com/smart-mcp-proxy/mcpproxy-go/internal/config"
+	"github.com/smart-mcp-proxy/mcpproxy-go/internal/serveredition/broker"
 	"github.com/smart-mcp-proxy/mcpproxy-go/internal/serveredition/workspace"
 )
 
@@ -149,6 +150,29 @@ func (r *Router) GetServerForUser(ctx context.Context, serverName string) (*Serv
 	return nil, fmt.Errorf("server %q not found or not accessible", serverName)
 }
 
+// BrokeredConnectionKey returns the per-(user, server) pooling key for a
+// brokered upstream connection (spec 074 FR-018). It resolves the calling user
+// and the named server (shared or personal), then keys the connection by both
+// so a shared upstream brokered per-user never reuses one user's
+// credential/connection for another. The connection pool MUST use this key when
+// caching brokered upstream clients.
+//
+// It errors if there is no auth context or the server is not accessible to the
+// user; it does not require the server to actually declare an auth_broker block,
+// so callers can key uniformly and decide per server whether to broker.
+func (r *Router) BrokeredConnectionKey(ctx context.Context, serverName string) (string, error) {
+	ac := auth.AuthContextFromContext(ctx)
+	if ac == nil {
+		return "", fmt.Errorf("no authentication context")
+	}
+
+	info, err := r.GetServerForUser(ctx, serverName)
+	if err != nil {
+		return "", err
+	}
+	return broker.ConnectionKey(ac.GetUserID(), info.Config), nil
+}
+
 // IsServerAccessible returns true if the user from the context can access
 // the named server. This is a convenience method that does not return
 // detailed error information.

internal/serveredition/multiuser/router_broker_test.go

@@ -0,0 +1,45 @@
+//go:build server
+
+package multiuser
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Spec 074 T7 / FR-018: a shared upstream brokered per-user must key its
+// connection by (user, server) so one user's credential/connection is never
+// reused for another. The router exposes the per-(user,server) connection key
+// for the connection pool to use.
+
+func TestRouter_BrokeredConnectionKey_DistinctPerUser(t *testing.T) {
+	router, _ := setupRouter(t, []string{"shared-ghe"}, nil)
+
+	aliceKey, err := router.BrokeredConnectionKey(userCtx("alice"), "shared-ghe")
+	require.NoError(t, err)
+	bobKey, err := router.BrokeredConnectionKey(userCtx("bob"), "shared-ghe")
+	require.NoError(t, err)
+
+	assert.NotEmpty(t, aliceKey)
+	assert.NotEqual(t, aliceKey, bobKey,
+		"the same shared brokered upstream must key a distinct connection per user (FR-018)")
+
+	// Stable for the same (user, server).
+	aliceKey2, err := router.BrokeredConnectionKey(userCtx("alice"), "shared-ghe")
+	require.NoError(t, err)
+	assert.Equal(t, aliceKey, aliceKey2, "connection key must be stable for the same (user, server)")
+}
+
+func TestRouter_BrokeredConnectionKey_RequiresAuth(t *testing.T) {
+	router, _ := setupRouter(t, []string{"shared-ghe"}, nil)
+	_, err := router.BrokeredConnectionKey(noAuthCtx(), "shared-ghe")
+	assert.Error(t, err, "no auth context must be rejected")
+}
+
+func TestRouter_BrokeredConnectionKey_UnknownServer(t *testing.T) {
+	router, _ := setupRouter(t, []string{"shared-ghe"}, nil)
+	_, err := router.BrokeredConnectionKey(userCtx("alice"), "nope")
+	assert.Error(t, err, "unknown/inaccessible server must error")
+}