feat(server): add runtime auth namespace cutover

Add explicit none, api_key, and jwt runtime auth modes, including a generic no-auth provider.

Move controls, bindings, policies, agents, and evaluation storage lookups onto principal namespace scoping.

Cover auth mode selection and principal namespace isolation with server tests.

Author: abhinav-galileo

server/src/agent_control_server/auth_framework/__init__.py

@@ -2,10 +2,9 @@
 
 Endpoints declare an :class:`Operation` they need; an installed
 :class:`RequestAuthorizer` decides whether the request is allowed and
-returns the resulting :class:`Principal`. Two providers ship in-tree:
-:class:`HeaderAuthProvider` (uses local credential checks) and
-:class:`HttpUpstreamAuthProvider` (delegates to a configurable
-upstream HTTP service).
+returns the resulting :class:`Principal`. Providers ship in-tree for
+disabled auth, local credential checks, upstream HTTP authorization,
+and local runtime-JWT verification.
 """
 
 from .core import (

server/src/agent_control_server/auth_framework/config.py

@@ -8,15 +8,19 @@
 
 - **Default flow** (everything except runtime). One authorizer handles
   every operation that does not have a specific override:
-  :class:`HeaderAuthProvider` (local credentials) or
+  :class:`NoAuthProvider` (no credentials),
+  :class:`HeaderAuthProvider` (local API keys), or
   :class:`HttpUpstreamAuthProvider` (forwards to a configurable URL).
-- **Runtime flow.** When ``AGENT_CONTROL_RUNTIME_TOKEN_SECRET`` is
-  configured, :class:`LocalJwtVerifyProvider` is registered as the
-  override for :data:`Operation.RUNTIME_USE`; 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). Without the secret,
-  no runtime override is installed.
+- **Runtime flow.** ``AGENT_CONTROL_RUNTIME_AUTH_MODE`` selects the
+  override for :data:`Operation.RUNTIME_USE`: ``none`` uses
+  :class:`NoAuthProvider`, ``api_key`` uses
+  :class:`HeaderAuthProvider`, and ``jwt`` uses
+  :class:`LocalJwtVerifyProvider`. When the mode is unset, startup
+  preserves historical behavior by selecting ``jwt`` if
+  ``AGENT_CONTROL_RUNTIME_TOKEN_SECRET`` is set, otherwise ``api_key``.
+  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).
 """
 
 from __future__ import annotations
@@ -30,6 +34,7 @@
     HeaderAuthProvider,
     HttpUpstreamAuthProvider,
     LocalJwtVerifyProvider,
+    NoAuthProvider,
 )
 from .providers.http_upstream import HttpUpstreamConfig
 
@@ -43,6 +48,7 @@
 _UPSTREAM_TOKEN_HEADER_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_SERVICE_TOKEN_HEADER"
 
 # Runtime flow.
+_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"
 _DEFAULT_RUNTIME_TOKEN_TTL_SECONDS = 300
@@ -80,15 +86,19 @@ def configure_auth_from_env() -> None:
 
     Default flow:
 
-    - ``AGENT_CONTROL_AUTH_MODE=header`` (default): :class:`HeaderAuthProvider`.
+    - ``AGENT_CONTROL_AUTH_MODE=none``: :class:`NoAuthProvider`.
+    - ``AGENT_CONTROL_AUTH_MODE=api_key`` (default): :class:`HeaderAuthProvider`.
+      ``header`` remains accepted as a backwards-compatible alias.
     - ``AGENT_CONTROL_AUTH_MODE=http_upstream``: :class:`HttpUpstreamAuthProvider`
       pointed at ``AGENT_CONTROL_AUTH_UPSTREAM_URL``.
 
     Runtime flow:
 
-    - When ``AGENT_CONTROL_RUNTIME_TOKEN_SECRET`` is set, register
-      :class:`LocalJwtVerifyProvider` as an override for
-      :data:`Operation.RUNTIME_USE`.
+    - ``AGENT_CONTROL_RUNTIME_AUTH_MODE=none``: :class:`NoAuthProvider`.
+    - ``AGENT_CONTROL_RUNTIME_AUTH_MODE=api_key`` (default when no runtime
+      token secret is configured): :class:`HeaderAuthProvider`.
+    - ``AGENT_CONTROL_RUNTIME_AUTH_MODE=jwt`` (default when a runtime token
+      secret is configured): :class:`LocalJwtVerifyProvider`.
 
     Clears any previously-installed default and operation overrides
     before installing fresh ones, so reconfiguration cannot leave
@@ -101,27 +111,27 @@ def configure_auth_from_env() -> None:
     global _runtime_auth_config
     clear_authorizers()
     _active_providers.clear()
-    _runtime_auth_config = _load_runtime_auth_config()
+    runtime_mode = _resolve_runtime_mode()
+    _runtime_auth_config = (
+        _load_runtime_auth_config(require_secret=True) if runtime_mode == "jwt" else None
+    )
 
     default = _build_default_provider()
     set_authorizer(default)
     _active_providers.append(default)
 
-    if _runtime_auth_config is not None:
-        runtime_provider = LocalJwtVerifyProvider(secret=_runtime_auth_config.secret)
-        set_authorizer(runtime_provider, operation=Operation.RUNTIME_USE)
-        _active_providers.append(runtime_provider)
+    runtime_provider = _build_runtime_provider(runtime_mode, _runtime_auth_config)
+    set_authorizer(runtime_provider, operation=Operation.RUNTIME_USE)
+    _active_providers.append(runtime_provider)
+    if runtime_mode == "jwt":
         _logger.info(
-            "Runtime auth enabled: LocalJwtVerifyProvider override installed for %s",
+            "Runtime auth provider: jwt override installed for %s",
             Operation.RUNTIME_USE.value,
         )
     else:
-        _logger.warning(
-            "Runtime auth disabled (%s not set); %s falls through to the "
-            "default authorizer, which may grant any authenticated credential. "
-            "Set the runtime token secret to bind runtime calls to a "
-            "short-lived target-scoped JWT.",
-            _RUNTIME_TOKEN_SECRET_ENV,
+        _logger.info(
+            "Runtime auth provider: %s override installed for %s",
+            runtime_mode,
             Operation.RUNTIME_USE.value,
         )
 
@@ -172,9 +182,12 @@ def set_runtime_auth_config(config: RuntimeAuthConfig | None) -> None:
 
 
 def _build_default_provider() -> RequestAuthorizer:
-    mode = os.environ.get(_MODE_ENV, "header").strip().lower()
-    if mode == "header":
-        _logger.info("Default auth provider: header (local credentials)")
+    mode = os.environ.get(_MODE_ENV, "api_key").strip().lower()
+    if mode in {"none", "no_auth"}:
+        _logger.info("Default auth provider: none")
+        return NoAuthProvider()
+    if mode in {"api_key", "header"}:
+        _logger.info("Default auth provider: api_key (local credentials)")
         return HeaderAuthProvider()
     if mode == "http_upstream":
         url = os.environ.get(_UPSTREAM_URL_ENV)
@@ -192,19 +205,60 @@ def _build_default_provider() -> RequestAuthorizer:
                 service_token_header=token_header,
             )
         )
-    raise RuntimeError(f"Unknown {_MODE_ENV}={mode!r}; expected 'header' or 'http_upstream'.")
+    raise RuntimeError(
+        f"Unknown {_MODE_ENV}={mode!r}; expected 'none', 'api_key', or 'http_upstream'."
+    )
+
+
+def _resolve_runtime_mode() -> str:
+    raw = os.environ.get(_RUNTIME_MODE_ENV)
+    if raw is None or not raw.strip():
+        return "jwt" if os.environ.get(_RUNTIME_TOKEN_SECRET_ENV) else "api_key"
+
+    mode = raw.strip().lower()
+    if mode in {"none", "no_auth"}:
+        return "none"
+    if mode in {"api_key", "header"}:
+        return "api_key"
+    if mode == "jwt":
+        return mode
+    raise RuntimeError(
+        f"Unknown {_RUNTIME_MODE_ENV}={mode!r}; expected 'none', 'api_key', or 'jwt'."
+    )
+
+
+def _build_runtime_provider(
+    mode: str,
+    config: RuntimeAuthConfig | None,
+) -> RequestAuthorizer:
+    if mode == "none":
+        return NoAuthProvider()
+    if mode == "api_key":
+        return HeaderAuthProvider()
+    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)
+    raise RuntimeError(
+        f"Unknown runtime auth mode {mode!r}; expected 'none', 'api_key', or 'jwt'."
+    )
 
 
-def _load_runtime_auth_config() -> RuntimeAuthConfig | None:
+def _load_runtime_auth_config(*, require_secret: bool = False) -> RuntimeAuthConfig | None:
     """Parse, validate, and return the runtime-auth config from env.
 
-    Returns ``None`` when no runtime secret is configured. Raises
-    ``RuntimeError`` when the secret is too short or the TTL is invalid
-    so misconfiguration surfaces at startup, not on the first
-    request-time mint.
+    Returns ``None`` when no runtime secret is configured and
+    ``require_secret`` is false. Raises ``RuntimeError`` when the
+    secret is required, too short, or the TTL is invalid so
+    misconfiguration surfaces at startup, not on the first request-time
+    mint.
     """
     secret = os.environ.get(_RUNTIME_TOKEN_SECRET_ENV)
     if not secret:
+        if require_secret:
+            raise RuntimeError(
+                f"{_RUNTIME_MODE_ENV}=jwt requires {_RUNTIME_TOKEN_SECRET_ENV} to be set."
+            )
         return None
     if len(secret.encode("utf-8")) < _RUNTIME_TOKEN_SECRET_MIN_BYTES:
         raise RuntimeError(

server/src/agent_control_server/auth_framework/core.py

@@ -42,14 +42,21 @@ class Operation(StrEnum):
     CONTROL_BINDINGS_READ = "control_bindings.read"
     CONTROL_BINDINGS_WRITE = "control_bindings.write"
 
-    # Runtime token exchange — wired on the exchange endpoint.
+    # Runtime token exchange - wired on the exchange endpoint.
     RUNTIME_TOKEN_EXCHANGE = "runtime.token_exchange"
 
-    # Reserved for follow-up migrations; not yet wired on endpoints.
     CONTROLS_READ = "controls.read"
     CONTROLS_CREATE = "controls.create"
     CONTROLS_UPDATE = "controls.update"
     CONTROLS_DELETE = "controls.delete"
+    POLICIES_READ = "policies.read"
+    POLICIES_CREATE = "policies.create"
+    POLICIES_UPDATE = "policies.update"
+    POLICIES_DELETE = "policies.delete"
+    AGENTS_READ = "agents.read"
+    AGENTS_CREATE = "agents.create"
+    AGENTS_UPDATE = "agents.update"
+    AGENTS_DELETE = "agents.delete"
     RUNTIME_USE = "runtime.use"
 
 
@@ -61,8 +68,7 @@ class Principal:
         namespace_key: The namespace the request runs in. Endpoints use
             this to scope every read and write.
         is_admin: Whether the caller has admin privileges in the
-            current namespace. Mostly informational for endpoints that
-            still gate on the legacy admin-key contract.
+            current namespace.
         caller_id: Opaque, provider-supplied identifier for the caller
             (e.g., a key fingerprint or user id). Useful for audit
             logging; never echo back to clients.
@@ -122,7 +128,7 @@ def set_authorizer(
 
     Without ``operation``, this becomes the default authorizer used by
     every operation that does not have a specific override. With
-    ``operation``, it overrides the default for that operation only —
+    ``operation``, it overrides the default for that operation only -
     used to route a different family (e.g., runtime) through a
     different provider.
 

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

@@ -3,10 +3,12 @@
 from .header import AccessLevel, HeaderAuthProvider
 from .http_upstream import HttpUpstreamAuthProvider
 from .local_jwt import LocalJwtVerifyProvider
+from .no_auth import NoAuthProvider
 
 __all__ = [
     "AccessLevel",
     "HeaderAuthProvider",
     "HttpUpstreamAuthProvider",
     "LocalJwtVerifyProvider",
+    "NoAuthProvider",
 ]

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

@@ -1,23 +1,14 @@
 """Default :class:`RequestAuthorizer` that uses local credentials only.
 
-Resolves the namespace from a header (or falls back to
-``DEFAULT_NAMESPACE_KEY``) and enforces a per-operation access level
-using the legacy API-key + session-cookie credential check from
-:mod:`agent_control_server.auth`. Behavior matches the pre-framework
-local auth path verbatim:
+Returns ``DEFAULT_NAMESPACE_KEY`` and enforces a per-operation access
+level using the local API-key + session-cookie credential check from
+:mod:`agent_control_server.auth`:
 
 - ``ADMIN`` operations require an admin key (or admin session).
 - ``AUTHENTICATED`` operations require any valid credential.
 - ``PUBLIC`` operations are open.
-- When ``api_key_enabled`` is ``False`` (no-auth mode), every
-  operation succeeds with a non-admin :class:`Principal` — preserved
-  by the underlying credential check.
-
-The header lookup is wired but currently inert: the provider always
-returns the default namespace because non-binding write endpoints
-still hardcode it. The header is kept here so a follow-up that
-threads namespace resolution through the rest of the API can flip it
-on without changing the provider contract.
+- When the underlying local credential layer is disabled, every
+  operation succeeds with a non-admin :class:`Principal`.
 """
 
 from __future__ import annotations
@@ -51,6 +42,14 @@ class AccessLevel(Enum):
     Operation.CONTROLS_CREATE: AccessLevel.ADMIN,
     Operation.CONTROLS_UPDATE: AccessLevel.ADMIN,
     Operation.CONTROLS_DELETE: AccessLevel.ADMIN,
+    Operation.POLICIES_READ: AccessLevel.AUTHENTICATED,
+    Operation.POLICIES_CREATE: AccessLevel.ADMIN,
+    Operation.POLICIES_UPDATE: AccessLevel.ADMIN,
+    Operation.POLICIES_DELETE: AccessLevel.ADMIN,
+    Operation.AGENTS_READ: AccessLevel.AUTHENTICATED,
+    Operation.AGENTS_CREATE: AccessLevel.AUTHENTICATED,
+    Operation.AGENTS_UPDATE: AccessLevel.ADMIN,
+    Operation.AGENTS_DELETE: AccessLevel.ADMIN,
     Operation.RUNTIME_TOKEN_EXCHANGE: AccessLevel.AUTHENTICATED,
     Operation.RUNTIME_USE: AccessLevel.AUTHENTICATED,
 }
@@ -60,7 +59,7 @@ class HeaderAuthProvider(RequestAuthorizer):
     """Default authorizer.
 
     For each operation's configured access level, validates the
-    request's credentials via the legacy local check; on success,
+    request's credentials via the local credential check; on success,
     returns a :class:`Principal` scoped to the resolved namespace.
     """
 
@@ -100,8 +99,7 @@ async def authorize(
         )
         # Runtime token exchange returns a normalized scope grant so the
         # exchange endpoint can require ``runtime.use`` uniformly across
-        # providers; an upstream that explicitly grants no scopes ends
-        # up with an empty tuple and is rejected.
+        # providers.
         scopes: tuple[str, ...] = (
             (Operation.RUNTIME_USE.value,) if operation is Operation.RUNTIME_TOKEN_EXCHANGE else ()
         )
@@ -113,10 +111,7 @@ async def authorize(
         )
 
     def _resolve_namespace_key(self, request: Request) -> str:
-        # The provider always returns the default namespace because
-        # non-binding write endpoints still hardcode it; serving
-        # anything else here would create rows the rest of the API
-        # cannot find. The branch is preserved so a future change can
-        # lift the lock without touching the provider contract.
+        # Local credentials do not carry namespace metadata. Providers
+        # that resolve a namespace can return a different principal.
         del request
         return self._default_namespace_key

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

@@ -67,8 +67,8 @@ class _UpstreamGrant(BaseModel):
     """Strict schema for the upstream authorization-service response.
 
     Unknown fields are tolerated (so the upstream can evolve), but every
-    *known* field is type-checked. A wrong type on any field — or a
-    half-supplied target binding — causes the provider to fail closed
+    *known* field is type-checked. A wrong type on any field - or a
+    half-supplied target binding - causes the provider to fail closed
     with a 502.
     """
 
@@ -108,7 +108,7 @@ def _target_must_be_paired(self) -> _UpstreamGrant:
         A target is meaningful only as a ``(target_type, target_id)``
         pair; allowing one side without the other would let a malformed
         grant pass and the exchange endpoint mint a token for the
-        request's value of the missing half — outside the upstream's
+        request's value of the missing half - outside the upstream's
         intended authorization.
         """
         if (self.target_type is None) != (self.target_id is None):

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

@@ -6,7 +6,7 @@
 returns a :class:`Principal` carrying the bound target. When a
 ``context_builder`` on the dependency surfaces ``target_type`` /
 ``target_id``, the provider also enforces that they match the token's
-binding — runtime endpoints get the request-target check for free.
+binding - runtime endpoints get the request-target check for free.
 """
 
 from __future__ import annotations

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

@@ -0,0 +1,29 @@
+"""Authorizer for deployments that intentionally disable authentication."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import Request
+
+from ...models import DEFAULT_NAMESPACE_KEY
+from ..core import Operation, Principal, RequestAuthorizer
+
+
+class NoAuthProvider(RequestAuthorizer):
+    """Allows every operation and returns the default namespace."""
+
+    def __init__(self, *, default_namespace_key: str = DEFAULT_NAMESPACE_KEY) -> None:
+        self._default_namespace_key = default_namespace_key
+
+    async def authorize(
+        self,
+        request: Request,
+        operation: Operation,
+        context: dict[str, Any] | None = None,
+    ) -> Principal:
+        del request, context
+        scopes: tuple[str, ...] = (
+            (Operation.RUNTIME_USE.value,) if operation is Operation.RUNTIME_TOKEN_EXCHANGE else ()
+        )
+        return Principal(namespace_key=self._default_namespace_key, scopes=scopes)