diff --git a/sdks/python/src/agent_control/client.py b/sdks/python/src/agent_control/client.py index fcc0c398..ccf2c791 100644 --- a/sdks/python/src/agent_control/client.py +++ b/sdks/python/src/agent_control/client.py @@ -20,6 +20,8 @@ _logger = logging.getLogger(__name__) _RUNTIME_AUTH_MODE_ENV_VAR = "AGENT_CONTROL_RUNTIME_AUTH_MODE" +_RUNTIME_TOKEN_HEADER_ENV_VAR = "AGENT_CONTROL_RUNTIME_TOKEN_HEADER" +_DEFAULT_RUNTIME_TOKEN_HEADER = "Authorization" _DEFAULT_RUNTIME_TOKEN_REFRESH_MARGIN_SECONDS = 30 _AUTO_RUNTIME_TOKEN_FALLBACK_STATUSES = {404, 500, 502, 503, 504} _GLOBAL_RUNTIME_TOKEN_FALLBACK_STATUSES = {404} @@ -35,19 +37,39 @@ def _runtime_cache_identity(api_key: str | None, api_key_header: str) -> str: class _AgentControlAuth(httpx.Auth): - """Attach local API-key credentials unless a request already has Bearer auth.""" + """Attach local API-key credentials unless the request already carries a token. - def __init__(self, api_key: str | None, header_name: str = "X-API-Key") -> None: + The API key is suppressed when the request already presents a bearer + credential on ``Authorization`` or a runtime token on its dedicated + header, so a runtime-authenticated evaluation carries a single + credential regardless of which header the runtime token rides. + """ + + def __init__( + self, + api_key: str | None, + header_name: str = "X-API-Key", + runtime_token_header: str | None = None, + ) -> None: self._api_key = api_key self._header_name = header_name + self._runtime_token_header = runtime_token_header def auth_flow( self, request: httpx.Request, ) -> Generator[httpx.Request, httpx.Response, None]: - if self._api_key and "Authorization" not in request.headers: - if self._header_name not in request.headers: - request.headers[self._header_name] = self._api_key + runtime_token_on_dedicated_header = ( + self._runtime_token_header is not None + and self._runtime_token_header in request.headers + ) + if ( + self._api_key + and "Authorization" not in request.headers + and not runtime_token_on_dedicated_header + and self._header_name not in request.headers + ): + request.headers[self._header_name] = self._api_key yield request @@ -99,6 +121,7 @@ def __init__( api_key: str | None = None, api_key_header: str | None = None, runtime_auth_mode: RuntimeAuthMode | str | None = None, + runtime_token_header: str | None = None, runtime_token_cache: RuntimeTokenCache | None = None, runtime_token_refresh_margin_seconds: int = (_DEFAULT_RUNTIME_TOKEN_REFRESH_MARGIN_SECONDS), transport: httpx.AsyncBaseTransport | None = None, @@ -121,6 +144,13 @@ def __init__( request auth when the exchange endpoint is unavailable. ``jwt`` requires a successful exchange. ``api_key`` and ``none`` keep evaluation requests on the normal request-auth path. + runtime_token_header: HTTP header the runtime token is sent on. + Defaults to ``Authorization``; the + AGENT_CONTROL_RUNTIME_TOKEN_HEADER environment variable + overrides the default. Point this at a dedicated header (e.g. + ``X-Agent-Control-Runtime-Token``) when the server runs behind + a gateway that reserves ``Authorization`` for its own identity + JWT. The server must be configured to read the same header. runtime_token_cache: Optional cache shared across client instances. runtime_token_refresh_margin_seconds: Refresh cached runtime tokens before this many seconds of validity remain. @@ -140,6 +170,24 @@ def __init__( self._runtime_cache_identity = _runtime_cache_identity(self._api_key, self._api_key_header) configured_runtime_mode = runtime_auth_mode or os.environ.get(_RUNTIME_AUTH_MODE_ENV_VAR) self._runtime_auth_mode = normalize_runtime_auth_mode(configured_runtime_mode) + # Explicit blank param is a hard error; a blank env var falls back to + # the default (mirrors the server's _resolve_runtime_token_header). + if runtime_token_header is not None and not runtime_token_header.strip(): + raise ValueError("runtime_token_header must not be blank.") + env_runtime_token_header = os.environ.get(_RUNTIME_TOKEN_HEADER_ENV_VAR) + if env_runtime_token_header is not None and not env_runtime_token_header.strip(): + env_runtime_token_header = None + self._runtime_token_header = ( + runtime_token_header + or env_runtime_token_header + or _DEFAULT_RUNTIME_TOKEN_HEADER + ).strip() + # Bearer only on Authorization; a dedicated header carries the raw token + # so it can't collide with the gateway's Authorization JWT. Must stay in + # sync with LocalJwtVerifyProvider._require_bearer on the server. + self._runtime_token_use_bearer = ( + self._runtime_token_header.lower() == _DEFAULT_RUNTIME_TOKEN_HEADER.lower() + ) if runtime_token_refresh_margin_seconds < 0: raise ValueError("runtime_token_refresh_margin_seconds must be >= 0.") self._runtime_token_refresh_margin_seconds = runtime_token_refresh_margin_seconds @@ -198,7 +246,13 @@ async def __aenter__(self) -> "AgentControlClient": base_url=self.base_url, timeout=self.timeout, headers=self._get_headers(), - auth=_AgentControlAuth(self._api_key, self._api_key_header), + auth=_AgentControlAuth( + self._api_key, + self._api_key_header, + runtime_token_header=( + None if self._runtime_token_use_bearer else self._runtime_token_header + ), + ), transport=self._transport, event_hooks={"response": [self._check_server_version]}, ) @@ -295,18 +349,22 @@ async def post_runtime_evaluation( return response + def _format_runtime_token(self, token: str) -> str: + """Sole place the Bearer prefix is applied (see __init__ for the rule).""" + return f"Bearer {token}" if self._runtime_token_use_bearer else token + def _merge_runtime_headers( self, headers: dict[str, str] | None, runtime_authorization: str | None, ) -> dict[str, str] | None: - """Merge caller headers with an optional Bearer token.""" + """Merge caller headers with an optional runtime token header.""" if headers is None and runtime_authorization is None: return None merged = dict(headers or {}) if runtime_authorization is not None: - merged["Authorization"] = runtime_authorization + merged[self._runtime_token_header] = runtime_authorization return merged async def _runtime_authorization( @@ -350,7 +408,7 @@ async def _runtime_authorization( refresh_margin_seconds=self._runtime_token_refresh_margin_seconds, ) if cached is not None: - return f"Bearer {cached.token}" + return self._format_runtime_token(cached.token) exchange_lock = self._runtime_token_cache.exchange_lock( self.base_url, @@ -379,7 +437,7 @@ async def _runtime_authorization( refresh_margin_seconds=self._runtime_token_refresh_margin_seconds, ) if cached is not None: - return f"Bearer {cached.token}" + return self._format_runtime_token(cached.token) token = await self._exchange_runtime_token( target_type=target_type, @@ -388,7 +446,7 @@ async def _runtime_authorization( ) if token is None: return None - return f"Bearer {token}" + return self._format_runtime_token(token) async def _exchange_runtime_token( self, diff --git a/sdks/python/tests/test_client.py b/sdks/python/tests/test_client.py index 02c9c174..1a479398 100644 --- a/sdks/python/tests/test_client.py +++ b/sdks/python/tests/test_client.py @@ -1081,3 +1081,169 @@ async def test_check_server_version_ignores_missing_header() -> None: # Then: no warning is emitted mock_warning.assert_not_called() + + +# --------------------------------------------------------------------------- +# HYBIM-741: configurable runtime-token header (gateway Authorization collision) +# --------------------------------------------------------------------------- + + +def test_runtime_token_header_defaults_to_authorization() -> None: + client = AgentControlClient(base_url="https://agent-control.test") + assert client._runtime_token_header == "Authorization" + assert client._runtime_token_use_bearer is True + + +def test_runtime_token_header_param_selects_dedicated_header() -> None: + client = AgentControlClient( + base_url="https://agent-control.test", + runtime_token_header="X-Agent-Control-Runtime-Token", + ) + assert client._runtime_token_header == "X-Agent-Control-Runtime-Token" + assert client._runtime_token_use_bearer is False + + +def test_runtime_token_header_env_var_overrides_default( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv( + "AGENT_CONTROL_RUNTIME_TOKEN_HEADER", "X-Agent-Control-Runtime-Token" + ) + client = AgentControlClient(base_url="https://agent-control.test") + assert client._runtime_token_header == "X-Agent-Control-Runtime-Token" + assert client._runtime_token_use_bearer is False + + +def test_runtime_token_header_rejects_blank_param() -> None: + with pytest.raises(ValueError, match="runtime_token_header"): + AgentControlClient( + base_url="https://agent-control.test", runtime_token_header=" " + ) + + +def test_runtime_token_header_whitespace_env_falls_back( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("AGENT_CONTROL_RUNTIME_TOKEN_HEADER", " ") + client = AgentControlClient(base_url="https://agent-control.test") + assert client._runtime_token_header == "Authorization" + assert client._runtime_token_use_bearer is True + + +@pytest.mark.asyncio +async def test_runtime_evaluation_sends_raw_token_on_custom_header() -> None: + """Custom header carries the raw token; Authorization stays free for the + gateway JWT and the API key does not ride along.""" + seen: dict[str, str | None] = {} + expires_at = (datetime.now(UTC) + timedelta(minutes=5)).isoformat() + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/runtime-token-exchange"): + assert request.headers.get("X-API-Key") == "test-key" + return httpx.Response( + 200, + json={ + "token": "runtime-token", + "expires_at": expires_at, + "target_type": "log_stream", + "target_id": "ls-1", + "scopes": ["runtime.use"], + }, + ) + seen["runtime"] = request.headers.get("X-Agent-Control-Runtime-Token") + seen["authorization"] = request.headers.get("Authorization") + seen["api_key"] = request.headers.get("X-API-Key") + return httpx.Response(200, json={"is_safe": True, "confidence": 1.0}) + + transport = httpx.MockTransport(handler) + async with AgentControlClient( + base_url="https://agent-control.test", + api_key="test-key", + runtime_auth_mode="jwt", + runtime_token_header="X-Agent-Control-Runtime-Token", + transport=transport, + ) as client: + response = await client.post_runtime_evaluation( + json={"target_type": "log_stream", "target_id": "ls-1"}, + target_type="log_stream", + target_id="ls-1", + ) + + assert response.status_code == 200 + assert seen["runtime"] == "runtime-token" + assert seen["authorization"] is None + assert seen["api_key"] is None + + +@pytest.mark.asyncio +async def test_runtime_evaluation_default_sends_bearer_on_authorization() -> None: + """Default (unset) behavior is unchanged: Bearer token on Authorization.""" + seen: dict[str, str | None] = {} + expires_at = (datetime.now(UTC) + timedelta(minutes=5)).isoformat() + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/runtime-token-exchange"): + return httpx.Response( + 200, + json={ + "token": "runtime-token", + "expires_at": expires_at, + "target_type": "log_stream", + "target_id": "ls-1", + "scopes": ["runtime.use"], + }, + ) + seen["authorization"] = request.headers.get("Authorization") + return httpx.Response(200, json={"is_safe": True, "confidence": 1.0}) + + transport = httpx.MockTransport(handler) + async with AgentControlClient( + base_url="https://agent-control.test", + api_key="test-key", + runtime_auth_mode="jwt", + transport=transport, + ) as client: + await client.post_runtime_evaluation( + json={"target_type": "log_stream", "target_id": "ls-1"}, + target_type="log_stream", + target_id="ls-1", + ) + + assert seen["authorization"] == "Bearer runtime-token" + + +@pytest.mark.asyncio +async def test_runtime_evaluation_custom_header_fallback_keeps_api_key() -> None: + """Custom-header mode must still fall back to the API key when the exchange + is unavailable: with no runtime token minted, the dedicated header is absent + and X-API-Key must authenticate the evaluation request.""" + exchange_calls = 0 + seen: dict[str, str | None] = {} + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal exchange_calls + if request.url.path.endswith("/runtime-token-exchange"): + exchange_calls += 1 + return httpx.Response(503, json={"detail": "runtime auth disabled"}) + seen["runtime"] = request.headers.get("X-Agent-Control-Runtime-Token") + seen["api_key"] = request.headers.get("X-API-Key") + return httpx.Response(200, json={"is_safe": True, "confidence": 1.0}) + + transport = httpx.MockTransport(handler) + async with AgentControlClient( + base_url="https://agent-control.test", + api_key="test-key", + runtime_auth_mode="auto", + runtime_token_header="X-Agent-Control-Runtime-Token", + transport=transport, + ) as client: + response = await client.post_runtime_evaluation( + json={"target_type": "log_stream", "target_id": "ls-1"}, + target_type="log_stream", + target_id="ls-1", + ) + assert response.status_code == 200 + + assert exchange_calls == 1 + assert seen["runtime"] is None + assert seen["api_key"] == "test-key" diff --git a/server/src/agent_control_server/auth_framework/config.py b/server/src/agent_control_server/auth_framework/config.py index 06246a46..16d0a566 100644 --- a/server/src/agent_control_server/auth_framework/config.py +++ b/server/src/agent_control_server/auth_framework/config.py @@ -21,6 +21,10 @@ The ``runtime.token_exchange`` operation continues to flow through the default authorizer because the exchange itself is shaped like a management call (forward credential, get grant). + ``AGENT_CONTROL_RUNTIME_TOKEN_HEADER`` (default ``Authorization``) + selects which request header the ``jwt`` verifier reads the runtime + token from, so the server can run behind a gateway that reserves + ``Authorization`` for its own downstream identity JWT. """ from __future__ import annotations @@ -39,6 +43,7 @@ NoAuthProvider, ) from .providers.http_upstream import HttpUpstreamConfig +from .providers.local_jwt import DEFAULT_RUNTIME_TOKEN_HEADER _logger = get_logger(__name__) @@ -60,6 +65,7 @@ _RUNTIME_MODE_ENV = "AGENT_CONTROL_RUNTIME_AUTH_MODE" _RUNTIME_TOKEN_SECRET_ENV = "AGENT_CONTROL_RUNTIME_TOKEN_SECRET" _RUNTIME_TOKEN_TTL_ENV = "AGENT_CONTROL_RUNTIME_TOKEN_TTL_SECONDS" +_RUNTIME_TOKEN_HEADER_ENV = "AGENT_CONTROL_RUNTIME_TOKEN_HEADER" _DEFAULT_RUNTIME_TOKEN_TTL_SECONDS = 300 # HS256 needs at least 256 bits (32 bytes) of secret material to be safe # against brute force; reject anything shorter so production deployments @@ -378,12 +384,28 @@ def _build_runtime_provider( if mode == "jwt": if config is None: raise RuntimeError(f"{_RUNTIME_MODE_ENV}=jwt but runtime auth config is missing.") - return LocalJwtVerifyProvider(secret=config.secret) + return LocalJwtVerifyProvider( + secret=config.secret, + header_name=_resolve_runtime_token_header(), + ) raise RuntimeError( f"Unknown runtime auth mode {mode!r}; expected 'none', 'api_key', or 'jwt'." ) +def _resolve_runtime_token_header() -> str: + """Header the runtime JWT verifier reads the token from (default ``Authorization``). + + Behind a gateway that overwrites ``Authorization`` with its own identity + JWT, set a dedicated header so the two tokens don't collide. Blank falls + back to the default. + """ + raw = os.environ.get(_RUNTIME_TOKEN_HEADER_ENV) + if raw is None or not raw.strip(): + return DEFAULT_RUNTIME_TOKEN_HEADER + return raw.strip() + + def _load_runtime_auth_config(*, require_secret: bool = False) -> RuntimeAuthConfig | None: """Parse, validate, and return the runtime-auth config from env. diff --git a/server/src/agent_control_server/auth_framework/providers/local_jwt.py b/server/src/agent_control_server/auth_framework/providers/local_jwt.py index 3f39e6fd..28a5f647 100644 --- a/server/src/agent_control_server/auth_framework/providers/local_jwt.py +++ b/server/src/agent_control_server/auth_framework/providers/local_jwt.py @@ -1,11 +1,15 @@ """Authorizer that verifies a locally-minted runtime token. -Wired to the runtime resolution path. Reads a Bearer token from the -``Authorization`` header, verifies the signature against the runtime -secret, checks the token's scope covers the requested operation, and -returns a :class:`Principal` carrying the bound target. When a -``context_builder`` on the dependency must surface matching -``target_type`` / ``target_id`` values for target-bound tokens. +Wired to the runtime resolution path. Reads the runtime token from a +configurable header (``Authorization`` by default), verifies the +signature against the runtime secret, checks the token's scope covers +the requested operation, and returns a :class:`Principal` carrying the +bound target. When a ``context_builder`` on the dependency must surface +matching ``target_type`` / ``target_id`` values for target-bound tokens. + +The header is configurable so the server can sit behind a gateway that +reserves ``Authorization`` for its own identity JWT (point the verifier at +a dedicated header to avoid the collision). """ from __future__ import annotations @@ -20,13 +24,28 @@ from ..runtime_token import RuntimeTokenError, verify_runtime_token +DEFAULT_RUNTIME_TOKEN_HEADER = "Authorization" + + class LocalJwtVerifyProvider(RequestAuthorizer): """Verifies a runtime Bearer token and emits a target-bound :class:`Principal`.""" - def __init__(self, *, secret: str) -> None: + def __init__( + self, + *, + secret: str, + header_name: str = DEFAULT_RUNTIME_TOKEN_HEADER, + ) -> None: if not secret: raise ValueError("LocalJwtVerifyProvider requires a non-empty secret.") + if not header_name or not header_name.strip(): + raise ValueError("LocalJwtVerifyProvider requires a non-empty header_name.") self._secret = secret + self._header_name = header_name.strip() + # Bearer only on Authorization; a dedicated header carries the raw token + # so it can't collide with the gateway's Authorization JWT. Must stay in + # sync with AgentControlClient._runtime_token_use_bearer in the SDK. + self._require_bearer = self._header_name.lower() == "authorization" async def authorize( self, @@ -79,18 +98,29 @@ async def authorize( ) def _extract_bearer_token(self, request: Request) -> str: - header = request.headers.get("Authorization") + header = request.headers.get(self._header_name) if not header: raise AuthenticationError( error_code=ErrorCode.AUTH_MISSING_KEY, - detail="Missing Authorization header.", + detail=f"Missing {self._header_name} header.", hint="Present a Bearer runtime token.", ) - scheme, _, value = header.partition(" ") - if scheme.lower() != "bearer" or not value: + scheme, sep, value = header.partition(" ") + if sep and scheme.lower() == "bearer": + token = value.strip() + elif self._require_bearer: raise AuthenticationError( error_code=ErrorCode.AUTH_MISSING_KEY, - detail="Authorization header must be a Bearer token.", - hint="Format: ``Authorization: Bearer ``.", + detail=f"{self._header_name} header must be a Bearer token.", + hint=f"Format: ``{self._header_name}: Bearer ``.", + ) + else: + # Dedicated runtime-token header: accept the raw token value. + token = header.strip() + if not token: + raise AuthenticationError( + error_code=ErrorCode.AUTH_MISSING_KEY, + detail=f"{self._header_name} header is empty.", + hint="Present a Bearer runtime token.", ) - return value.strip() + return token diff --git a/server/tests/test_auth_framework.py b/server/tests/test_auth_framework.py index c3514fba..b05cce57 100644 --- a/server/tests/test_auth_framework.py +++ b/server/tests/test_auth_framework.py @@ -1764,3 +1764,134 @@ async def test_teardown_auth_clears_registry(): assert not _operation_authorizers with pytest.raises(RuntimeError, match="No RequestAuthorizer"): get_authorizer(Operation.CONTROL_BINDINGS_WRITE) + + +# --------------------------------------------------------------------------- +# HYBIM-741: configurable runtime-token header (gateway Authorization collision) +# --------------------------------------------------------------------------- + + +def _mint_runtime_use_token(): + from agent_control_server.auth_framework.runtime_token import mint_runtime_token + + token, _ = mint_runtime_token( + namespace_key="default", + actor_id="actor-1", + target_type="log_stream", + target_id="ls-1", + scopes=("runtime.use",), + secret=_TEST_SECRET, + ttl_seconds=60, + ) + return token + + +@pytest.mark.asyncio +async def test_local_jwt_default_reads_authorization_bearer(): + """Default behavior unchanged: token read as Bearer from Authorization.""" + provider = LocalJwtVerifyProvider(secret=_TEST_SECRET) + token = _mint_runtime_use_token() + principal = await provider.authorize( + _build_request(headers={"Authorization": f"Bearer {token}"}), + Operation.RUNTIME_USE, + context={"target_type": "log_stream", "target_id": "ls-1"}, + ) + assert principal.target_type == "log_stream" + assert principal.target_id == "ls-1" + + +@pytest.mark.asyncio +async def test_local_jwt_default_rejects_raw_token_on_authorization(): + """On Authorization the Bearer scheme stays mandatory (back-compat).""" + provider = LocalJwtVerifyProvider(secret=_TEST_SECRET) + token = _mint_runtime_use_token() + with pytest.raises(AuthenticationError): + await provider.authorize( + _build_request(headers={"Authorization": token}), + Operation.RUNTIME_USE, + context={"target_type": "log_stream", "target_id": "ls-1"}, + ) + + +@pytest.mark.asyncio +async def test_local_jwt_custom_header_reads_raw_token(): + """On a dedicated header the raw token is accepted and Authorization is + left free for the gateway's own identity JWT (no collision).""" + provider = LocalJwtVerifyProvider( + secret=_TEST_SECRET, header_name="X-Agent-Control-Runtime-Token" + ) + token = _mint_runtime_use_token() + principal = await provider.authorize( + _build_request( + headers={ + "X-Agent-Control-Runtime-Token": token, + "Authorization": "Bearer gateway-identity-jwt", + } + ), + Operation.RUNTIME_USE, + context={"target_type": "log_stream", "target_id": "ls-1"}, + ) + assert principal.target_id == "ls-1" + + +@pytest.mark.asyncio +async def test_local_jwt_custom_header_also_accepts_bearer_prefix(): + provider = LocalJwtVerifyProvider( + secret=_TEST_SECRET, header_name="X-Agent-Control-Runtime-Token" + ) + token = _mint_runtime_use_token() + principal = await provider.authorize( + _build_request(headers={"X-Agent-Control-Runtime-Token": f"Bearer {token}"}), + Operation.RUNTIME_USE, + context={"target_type": "log_stream", "target_id": "ls-1"}, + ) + assert principal.target_id == "ls-1" + + +@pytest.mark.asyncio +async def test_local_jwt_custom_header_missing_reports_that_header(): + provider = LocalJwtVerifyProvider( + secret=_TEST_SECRET, header_name="X-Agent-Control-Runtime-Token" + ) + with pytest.raises(AuthenticationError, match="X-Agent-Control-Runtime-Token"): + await provider.authorize( + _build_request(headers={"Authorization": "Bearer gateway-jwt"}), + Operation.RUNTIME_USE, + context={"target_type": "log_stream", "target_id": "ls-1"}, + ) + + +def test_local_jwt_rejects_blank_header_name(): + with pytest.raises(ValueError, match="header_name"): + LocalJwtVerifyProvider(secret=_TEST_SECRET, header_name=" ") + + +# --------------------------------------------------------------------------- +# HYBIM-741: AGENT_CONTROL_RUNTIME_TOKEN_HEADER env resolution (config wiring) +# --------------------------------------------------------------------------- + + +def test_resolve_runtime_token_header_defaults_to_authorization(monkeypatch): + from agent_control_server.auth_framework import config as auth_config + + monkeypatch.delenv("AGENT_CONTROL_RUNTIME_TOKEN_HEADER", raising=False) + assert auth_config._resolve_runtime_token_header() == "Authorization" + + +def test_resolve_runtime_token_header_reads_env(monkeypatch): + from agent_control_server.auth_framework import config as auth_config + + monkeypatch.setenv( + "AGENT_CONTROL_RUNTIME_TOKEN_HEADER", " X-Agent-Control-Runtime-Token " + ) + # Value is honored and trimmed. + assert ( + auth_config._resolve_runtime_token_header() == "X-Agent-Control-Runtime-Token" + ) + + +def test_resolve_runtime_token_header_blank_env_falls_back(monkeypatch): + from agent_control_server.auth_framework import config as auth_config + + monkeypatch.setenv("AGENT_CONTROL_RUNTIME_TOKEN_HEADER", " ") + assert auth_config._resolve_runtime_token_header() == "Authorization"