feat(server): operator-configurable extra forwarded headers on HttpUpstream

The default forward set (X-API-Key, Authorization, Cookie) only covers
credential headers Agent Control itself reads. Deployments whose
upstream authenticates against a different header name (e.g., a
deployer-specific API-key header) had no way to surface that
credential through HttpUpstreamAuthProvider — the inbound header
reached AC but never crossed the upstream call.

Add an extra_forward_headers config field on HttpUpstreamConfig
(defaulting to the empty tuple) that operators populate via the new
AGENT_CONTROL_AUTH_UPSTREAM_EXTRA_FORWARD_HEADERS env var (comma-
separated). The provider's _forward_headers iterates over the union
of the default set and the extras, deduplicating case-insensitively
so a duplicate name (cross-set or within extras) does not produce two
copies on the wire.

Tests:
- forwards a configured extra header alongside defaults
- default forward set unchanged when extras are empty
- extras dedupe against defaults case-insensitively
- _parse_extra_forward_headers parametric: None / empty / single /
  multiple / whitespace / empty-entries / case-folded duplicates
- configure_auth_from_env threads the parsed tuple onto the provider

Lint clean, typecheck clean, full server suite (747) green.

Author: abhinav-galileo

server/src/agent_control_server/auth_framework/config.py

@@ -46,6 +46,7 @@
 _UPSTREAM_TIMEOUT_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_TIMEOUT_SECONDS"
 _UPSTREAM_TOKEN_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_SERVICE_TOKEN"
 _UPSTREAM_TOKEN_HEADER_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_SERVICE_TOKEN_HEADER"
+_UPSTREAM_EXTRA_FORWARD_HEADERS_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_EXTRA_FORWARD_HEADERS"
 
 # Runtime flow.
 _RUNTIME_MODE_ENV = "AGENT_CONTROL_RUNTIME_AUTH_MODE"
@@ -196,20 +197,48 @@ def _build_default_provider() -> RequestAuthorizer:
         timeout = float(os.environ.get(_UPSTREAM_TIMEOUT_ENV, "5.0"))
         token = os.environ.get(_UPSTREAM_TOKEN_ENV)
         token_header = os.environ.get(_UPSTREAM_TOKEN_HEADER_ENV, "X-Agent-Control-Service-Token")
+        extra_forward_headers = _parse_extra_forward_headers(
+            os.environ.get(_UPSTREAM_EXTRA_FORWARD_HEADERS_ENV)
+        )
         _logger.info("Default auth provider: http_upstream url=%s", url)
         return HttpUpstreamAuthProvider(
             HttpUpstreamConfig(
                 url=url,
                 timeout_seconds=timeout,
                 service_token=token,
                 service_token_header=token_header,
+                extra_forward_headers=extra_forward_headers,
             )
         )
     raise RuntimeError(
         f"Unknown {_MODE_ENV}={mode!r}; expected 'none', 'api_key', or 'http_upstream'."
     )
 
 
+def _parse_extra_forward_headers(raw: str | None) -> tuple[str, ...]:
+    """Parse a comma-separated header list into a deduplicated tuple.
+
+    Empty / unset env var returns an empty tuple. Whitespace around each
+    name is stripped. Empty entries (e.g. ``"X-A,,X-B"``) are dropped.
+    Order is preserved; duplicates (case-insensitive) are dropped after
+    the first occurrence.
+    """
+    if not raw or not raw.strip():
+        return ()
+    seen: set[str] = set()
+    result: list[str] = []
+    for raw_name in raw.split(","):
+        name = raw_name.strip()
+        if not name:
+            continue
+        lower = name.lower()
+        if lower in seen:
+            continue
+        seen.add(lower)
+        result.append(name)
+    return tuple(result)
+
+
 def _resolve_runtime_mode() -> str:
     raw = os.environ.get(_RUNTIME_MODE_ENV)
     if raw is None or not raw.strip():

server/src/agent_control_server/auth_framework/providers/http_upstream.py

@@ -60,7 +60,7 @@
 
 _logger = get_logger(__name__)
 
-_FORWARDED_HEADERS = ("X-API-Key", "Authorization", "Cookie")
+_DEFAULT_FORWARDED_HEADERS = ("X-API-Key", "Authorization", "Cookie")
 
 
 class _UpstreamGrant(BaseModel):
@@ -136,6 +136,17 @@ class HttpUpstreamConfig:
 
     service_token_header: str = "X-Agent-Control-Service-Token"
 
+    extra_forward_headers: tuple[str, ...] = ()
+    """Additional inbound request headers to forward to the upstream
+    on top of the default ``(X-API-Key, Authorization, Cookie)`` set.
+
+    Use this when the upstream authenticates via a header the provider
+    does not forward by default (e.g., a deployer-specific API-key
+    header). Header lookups against the inbound request are
+    case-insensitive; an empty or absent inbound header is silently
+    dropped. Names duplicating the default set or each other (after
+    case-folding) are deduplicated."""
+
 
 class HttpUpstreamAuthProvider(RequestAuthorizer):
     """Delegates authorization to an upstream HTTP service."""
@@ -190,7 +201,12 @@ async def authorize(
 
     def _forward_headers(self, request: Request) -> dict[str, str]:
         headers: dict[str, str] = {}
-        for name in _FORWARDED_HEADERS:
+        seen: set[str] = set()
+        for name in (*_DEFAULT_FORWARDED_HEADERS, *self._config.extra_forward_headers):
+            lower = name.lower()
+            if lower in seen:
+                continue
+            seen.add(lower)
             value = request.headers.get(name)
             if value is not None:
                 headers[name] = value

server/tests/test_auth_framework.py

@@ -261,6 +261,75 @@ def factory(request: httpx.Request) -> httpx.Response:
     assert captured["headers"]["x-custom-token"] == "shh"
 
 
+@pytest.mark.asyncio
+async def test_http_upstream_forwards_extra_headers():
+    # Given: a provider configured with an extra header in its forward list
+    captured: dict[str, Any] = {}
+
+    def factory(request: httpx.Request) -> httpx.Response:
+        captured["headers"] = dict(request.headers)
+        return httpx.Response(200, json={"namespace_key": "ns"})
+
+    provider = _build_upstream(
+        factory,
+        config_overrides={"extra_forward_headers": ("X-Deployer-Auth",)},
+    )
+
+    # When: the inbound request carries the extra header
+    inbound = _build_request(headers={"X-Deployer-Auth": "k_abc", "X-API-Key": "k1"})
+    await provider.authorize(inbound, Operation.CONTROL_BINDINGS_READ)
+
+    # Then: both the default and the extra header reach the upstream
+    assert captured["headers"]["x-deployer-auth"] == "k_abc"
+    assert captured["headers"]["x-api-key"] == "k1"
+
+
+@pytest.mark.asyncio
+async def test_http_upstream_default_forward_set_unchanged():
+    # Given: a provider with no extra_forward_headers
+    captured: dict[str, Any] = {}
+
+    def factory(request: httpx.Request) -> httpx.Response:
+        captured["headers"] = dict(request.headers)
+        return httpx.Response(200, json={"namespace_key": "ns"})
+
+    provider = _build_upstream(factory)
+
+    # When: the inbound carries an unlisted header alongside a default one
+    inbound = _build_request(
+        headers={"X-API-Key": "k1", "X-Deployer-Auth": "should-not-forward"}
+    )
+    await provider.authorize(inbound, Operation.CONTROL_BINDINGS_READ)
+
+    # Then: only the default-set header reaches the upstream
+    assert captured["headers"].get("x-api-key") == "k1"
+    assert "x-deployer-auth" not in captured["headers"]
+
+
+@pytest.mark.asyncio
+async def test_http_upstream_extra_forward_dedupes_against_defaults():
+    # Given: extra list duplicates a default header (different case)
+    captured: dict[str, Any] = {}
+
+    def factory(request: httpx.Request) -> httpx.Response:
+        captured["headers"] = dict(request.headers)
+        return httpx.Response(200, json={"namespace_key": "ns"})
+
+    provider = _build_upstream(
+        factory,
+        config_overrides={"extra_forward_headers": ("x-api-key", "Authorization")},
+    )
+
+    # When: inbound has both
+    inbound = _build_request(headers={"X-API-Key": "k1", "Authorization": "Bearer t"})
+    await provider.authorize(inbound, Operation.CONTROL_BINDINGS_READ)
+
+    # Then: each header appears exactly once on the upstream request
+    forwarded = captured["headers"]
+    assert sum(1 for k in forwarded if k.lower() == "x-api-key") == 1
+    assert sum(1 for k in forwarded if k.lower() == "authorization") == 1
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "status, expected",
@@ -1053,6 +1122,52 @@ async def test_configure_http_upstream_management_with_jwt_runtime(monkeypatch):
         await auth_config.teardown_auth()
 
 
+@pytest.mark.parametrize(
+    "raw, expected",
+    [
+        (None, ()),
+        ("", ()),
+        ("   ", ()),
+        ("X-One", ("X-One",)),
+        ("X-One,X-Two", ("X-One", "X-Two")),
+        ("  X-One  ,  X-Two  ", ("X-One", "X-Two")),
+        ("X-One,,X-Two", ("X-One", "X-Two")),
+        ("X-One,x-one,X-One", ("X-One",)),
+        ("X-A,X-B,x-a,X-C,X-b", ("X-A", "X-B", "X-C")),
+    ],
+)
+def test_parse_extra_forward_headers(raw, expected):
+    from agent_control_server.auth_framework.config import _parse_extra_forward_headers
+
+    assert _parse_extra_forward_headers(raw) == expected
+
+
+@pytest.mark.asyncio
+async def test_configure_http_upstream_extra_forward_headers_env(monkeypatch):
+    """Setting the env var threads extra_forward_headers into the provider."""
+    from agent_control_server.auth_framework import config as auth_config
+
+    clear_authorizers()
+
+    monkeypatch.setenv("AGENT_CONTROL_AUTH_MODE", "http_upstream")
+    monkeypatch.setenv("AGENT_CONTROL_AUTH_UPSTREAM_URL", "https://auth.example.test/check")
+    monkeypatch.setenv(
+        "AGENT_CONTROL_AUTH_UPSTREAM_EXTRA_FORWARD_HEADERS",
+        "X-Deployer-Auth, X-Deployer-Trace",
+    )
+
+    try:
+        auth_config.configure_auth_from_env()
+        provider = get_authorizer(Operation.CONTROLS_READ)
+        assert isinstance(provider, HttpUpstreamAuthProvider)
+        assert provider._config.extra_forward_headers == (
+            "X-Deployer-Auth",
+            "X-Deployer-Trace",
+        )
+    finally:
+        await auth_config.teardown_auth()
+
+
 def test_configure_runtime_jwt_requires_secret(monkeypatch):
     from agent_control_server.auth_framework import config as auth_config