refactor(server): isolate Orbit identity URL selection

Author: josjeon

server/src/agent_control_server/auth_framework/config.py

@@ -32,6 +32,7 @@
 import os
 import ssl
 from dataclasses import dataclass
+from urllib.parse import urlsplit, urlunsplit
 
 from ..config import auth_settings
 from ..logging_utils import get_logger
@@ -64,6 +65,8 @@
 _UPSTREAM_MAX_KEEPALIVE_CONNECTIONS_ENV = (
     "AGENT_CONTROL_AUTH_UPSTREAM_MAX_KEEPALIVE_CONNECTIONS"
 )
+_ORBIT_MANAGEMENT_PATH = "/internal/auth/agent_control/check_management_access"
+_ORBIT_IDENTITY_PATH = "/internal/auth/resolve_tenant_context"
 
 # Runtime flow.
 _RUNTIME_MODE_ENV = "AGENT_CONTROL_RUNTIME_AUTH_MODE"
@@ -246,10 +249,14 @@ def _build_default_provider() -> RequestAuthorizer:
             max_keepalive_connections=max_keepalive_connections,
         )
         _logger.info("Default auth provider: http_upstream url=%s", url)
+        # Only the known Orbit management route has a predictable identity route.
+        # Other upstreams keep their existing namespace-wide authorization flow.
+        explicit_identity_url = (os.environ.get(_UPSTREAM_IDENTITY_URL_ENV) or "").strip()
+        identity_url = explicit_identity_url or _derive_orbit_identity_url(url)
         try:
             upstream_config = HttpUpstreamConfig(
                 url=url,
-                identity_url=(os.environ.get(_UPSTREAM_IDENTITY_URL_ENV) or "").strip() or None,
+                identity_url=identity_url,
                 timeout_seconds=timeout,
                 service_token=token,
                 service_token_header=token_header,
@@ -277,6 +284,17 @@ def _build_default_provider() -> RequestAuthorizer:
     )
 
 
+def _derive_orbit_identity_url(url: str) -> str | None:
+    """Resolve Orbit's existing identity route from its management route."""
+    parts = urlsplit(url)
+    if not parts.path.endswith(_ORBIT_MANAGEMENT_PATH):
+        return None
+    prefix = parts.path[: -len(_ORBIT_MANAGEMENT_PATH)]
+    return urlunsplit(
+        (parts.scheme, parts.netloc, prefix + _ORBIT_IDENTITY_PATH, parts.query, parts.fragment)
+    )
+
+
 def _validate_local_api_key_mode(mode_env: str = _MODE_ENV) -> None:
     """Fail startup when local API-key mode has no local key validator."""
     if not auth_settings.api_key_enabled:

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

@@ -46,7 +46,6 @@
 from datetime import datetime
 from time import perf_counter
 from typing import Any
-from urllib.parse import urlsplit, urlunsplit
 
 import httpx
 from agent_control_models.errors import ErrorCode, ErrorReason
@@ -68,9 +67,8 @@
 _logger = get_logger(__name__)
 
 _DEFAULT_FORWARDED_HEADERS = ("X-API-Key", "Authorization", "Cookie")
-_ORBIT_MANAGEMENT_PATH = "/internal/auth/agent_control/check_management_access"
-_ORBIT_IDENTITY_PATH = "/internal/auth/resolve_tenant_context"
-_IDENTITY_OPERATION = "identity.resolve"
+# Diagnostic label only; the identity POST has no operation payload.
+_IDENTITY_LOOKUP_LABEL = "identity.resolve"
 
 _AUTH_UPSTREAM_ATTEMPTS = Counter(
     "agent_control_server_auth_upstream_attempts_total",
@@ -185,8 +183,7 @@ class HttpUpstreamConfig:
     """Maximum idle connections retained for the auth upstream."""
 
     identity_url: str | None = None
-    """URL for credential and namespace resolution. If omitted, the Orbit
-    identity URL is derived from the known management authorization path."""
+    """Optional URL for credential and namespace resolution before target authorization."""
 
     def __post_init__(self) -> None:
         if self.keepalive_expiry_seconds < 0:
@@ -221,7 +218,7 @@ def __init__(
         client: httpx.AsyncClient | None = None,
     ) -> None:
         self._config = config
-        self._identity_url = config.identity_url or _derive_orbit_identity_url(config.url)
+        self._identity_url = config.identity_url
         self._owns_client = client is None
         if client is not None:
             self._client = client
@@ -274,7 +271,7 @@ async def resolve_identity(self, request: Request, operation: Operation) -> Prin
                 hint="Set AGENT_CONTROL_AUTH_UPSTREAM_IDENTITY_URL.",
             )
         response = await self._post_upstream(
-            _IDENTITY_OPERATION,
+            _IDENTITY_LOOKUP_LABEL,
             None,
             self._forward_headers(request),
             url=self._identity_url,
@@ -287,7 +284,7 @@ async def resolve_identity(self, request: Request, operation: Operation) -> Prin
                 detail="Authorization identity endpoint was not found.",
                 hint="Check the configured authorization identity URL.",
             )
-        principal = self._handle_response(response, _IDENTITY_OPERATION, None)
+        principal = self._handle_response(response, _IDENTITY_LOOKUP_LABEL, None)
         return Principal(namespace_key=principal.namespace_key, caller_id=principal.caller_id)
 
     async def _post_upstream(
@@ -493,17 +490,6 @@ def _authorization_service_unavailable_error() -> APIError:
     )
 
 
-def _derive_orbit_identity_url(url: str) -> str | None:
-    """Resolve Orbit's existing identity route from its management route."""
-    parts = urlsplit(url)
-    if not parts.path.endswith(_ORBIT_MANAGEMENT_PATH):
-        return None
-    prefix = parts.path[: -len(_ORBIT_MANAGEMENT_PATH)]
-    return urlunsplit(
-        (parts.scheme, parts.netloc, prefix + _ORBIT_IDENTITY_PATH, parts.query, parts.fragment)
-    )
-
-
 def _ensure_target_context_matches_grant(
     context: dict[str, Any] | None,
     principal: Principal,

server/tests/test_auth_framework.py

@@ -280,7 +280,7 @@ def factory(request: httpx.Request) -> httpx.Response:
 
 
 @pytest.mark.asyncio
-async def test_http_upstream_resolves_identity_at_derived_orbit_url():
+async def test_http_upstream_resolves_identity_at_configured_url():
     captured: dict[str, Any] = {}
 
     def factory(request: httpx.Request) -> httpx.Response:
@@ -300,7 +300,10 @@ def factory(request: httpx.Request) -> httpx.Response:
     provider = _build_upstream(
         factory,
         config_overrides={
-            "url": ("https://orbit.example/prefix/internal/auth/agent_control/check_management_access"),
+            "url": (
+                "https://orbit.example/prefix/internal/auth/agent_control/check_management_access"
+            ),
+            "identity_url": "https://orbit.example/prefix/internal/auth/resolve_tenant_context",
             "service_token": "service-secret",
             "extra_forward_headers": ("X-Deployer-Auth",),
         },
@@ -1671,13 +1674,51 @@ async def test_configure_http_upstream_extra_forward_headers_env(monkeypatch):
         await auth_config.teardown_auth()
 
 
+@pytest.mark.parametrize(
+    ("upstream_url", "expected_identity_url"),
+    [
+        (
+            "https://orbit.example/prefix/internal/auth/agent_control/check_management_access",
+            "https://orbit.example/prefix/internal/auth/resolve_tenant_context",
+        ),
+        ("https://auth.example.test/check", None),
+    ],
+)
 @pytest.mark.asyncio
-async def test_configure_http_upstream_identity_url_env(monkeypatch):
+async def test_configure_http_upstream_selects_identity_url(
+    monkeypatch, upstream_url: str, expected_identity_url: str | None
+):
     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_URL", upstream_url)
+    monkeypatch.delenv("AGENT_CONTROL_AUTH_UPSTREAM_IDENTITY_URL", raising=False)
+
+    try:
+        auth_config.configure_auth_from_env()
+        provider = get_authorizer(Operation.CONTROL_BINDINGS_READ)
+        assert isinstance(provider, HttpUpstreamAuthProvider)
+        assert provider._config.identity_url == expected_identity_url
+        assert provider.binding_target_authorization is (expected_identity_url is not None)
+    finally:
+        await auth_config.teardown_auth()
+
+
+@pytest.mark.parametrize(
+    "upstream_url",
+    [
+        "https://auth.example.test/check",
+        "https://orbit.example/internal/auth/agent_control/check_management_access",
+    ],
+)
+@pytest.mark.asyncio
+async def test_configure_http_upstream_identity_url_env(monkeypatch, upstream_url: str):
+    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", upstream_url)
     monkeypatch.setenv(
         "AGENT_CONTROL_AUTH_UPSTREAM_IDENTITY_URL",
         " https://identity.example.test/resolve ",
@@ -1688,6 +1729,7 @@ async def test_configure_http_upstream_identity_url_env(monkeypatch):
         provider = get_authorizer(Operation.CONTROL_BINDINGS_READ)
         assert isinstance(provider, HttpUpstreamAuthProvider)
         assert provider._config.identity_url == "https://identity.example.test/resolve"
+        assert provider.binding_target_authorization is True
     finally:
         await auth_config.teardown_auth()
 

server/tests/test_control_bindings_endpoints.py

@@ -473,7 +473,8 @@ def upstream_auth(request: httpx.Request) -> Response:
     set_authorizer(
         HttpUpstreamAuthProvider(
             HttpUpstreamConfig(
-                url="https://orbit.example/internal/auth/agent_control/check_management_access"
+                url="https://orbit.example/internal/auth/agent_control/check_management_access",
+                identity_url="https://orbit.example/internal/auth/resolve_tenant_context",
             ),
             client=upstream_client,
         )