fix(server): harden auth and namespace edges

Author: abhinav-galileo

docs/auth.md

@@ -51,9 +51,11 @@ Management auth is selected by `AGENT_CONTROL_AUTH_MODE`.
 | Mode | Meaning |
 | --- | --- |
 | `none` | No credentials required. Intended for local development only. |
-| `api_key` | Validate caller credentials locally with `AGENT_CONTROL_API_KEYS`. This is the default. `header` is accepted as a backwards-compatible alias. |
+| `api_key` | Validate caller credentials locally with `AGENT_CONTROL_API_KEYS` and/or `AGENT_CONTROL_ADMIN_API_KEYS`. Requires `AGENT_CONTROL_API_KEY_ENABLED=true`. `header` is accepted as a backwards-compatible alias. |
 | `http_upstream` | POST each management authorization decision to `AGENT_CONTROL_AUTH_UPSTREAM_URL`. |
 
+When `AGENT_CONTROL_AUTH_MODE` is unset, startup selects `api_key` if local API-key validation is enabled and `none` otherwise.
+
 Runtime auth is selected by `AGENT_CONTROL_RUNTIME_AUTH_MODE`.
 
 | Mode | Meaning |
@@ -68,7 +70,7 @@ Common combinations:
 | Management | Runtime | Use case |
 | --- | --- | --- |
 | `api_key` | unset | Existing standalone deployments. |
-| `api_key` | `jwt` | Local management keys with short-lived target-bound runtime tokens. |
+| `api_key` | `jwt` | Local management keys with short-lived target-bound runtime tokens. This does not perform per-target authorization; any valid local API key can exchange for any target in the local namespace. |
 | `http_upstream` | `jwt` | External identity or authorization service for management, local token verify for high-volume runtime calls. |
 | `none` | `none` | Single-process local development. Do not use in production. |
 
@@ -125,6 +127,7 @@ Status handling:
 | `429` | `503` with a rate-limit detail and `Retry-After` hint when present. |
 | Other statuses or upstream network errors | Fail closed with `503`. |
 | Malformed `200` principal response | Fail closed with `502`. |
+| `200` target grant that conflicts with request context | Fail closed with `403`. |
 
 ## Runtime JWT Claims
 

server/alembic/versions/b6f4c2d8e9a1_namespace_observability_events.py

@@ -0,0 +1,44 @@
+"""namespace observability events
+
+Revision ID: b6f4c2d8e9a1
+Revises: a7f3b1e0d9c5
+Create Date: 2026-05-14 12:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "b6f4c2d8e9a1"
+down_revision = "a7f3b1e0d9c5"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "control_execution_events",
+        sa.Column(
+            "namespace_key",
+            sa.String(length=255),
+            server_default=sa.text("'default'"),
+            nullable=False,
+        ),
+    )
+    op.create_index(
+        "ix_events_namespace_agent_time",
+        "control_execution_events",
+        ["namespace_key", "agent_name", sa.literal_column("timestamp DESC")],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_events_namespace_agent_time",
+        table_name="control_execution_events",
+    )
+    op.drop_column("control_execution_events", "namespace_key")

server/src/agent_control_server/auth_framework/config.py

@@ -28,6 +28,7 @@
 import os
 from dataclasses import dataclass
 
+from ..config import auth_settings
 from ..logging_utils import get_logger
 from .core import Operation, RequestAuthorizer, clear_authorizers, set_authorizer
 from .providers import (
@@ -88,8 +89,10 @@ def configure_auth_from_env() -> None:
     Default flow:
 
     - ``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=api_key``: :class:`HeaderAuthProvider`.
+      ``header`` remains accepted as a backwards-compatible alias. When the mode
+      is unset, startup selects ``api_key`` only if local API-key validation is
+      enabled; otherwise it selects ``none``.
     - ``AGENT_CONTROL_AUTH_MODE=http_upstream``: :class:`HttpUpstreamAuthProvider`
       pointed at ``AGENT_CONTROL_AUTH_UPSTREAM_URL``.
 
@@ -190,11 +193,17 @@ def set_runtime_auth_config(config: RuntimeAuthConfig | None) -> None:
 
 
 def _build_default_provider() -> RequestAuthorizer:
-    mode = os.environ.get(_MODE_ENV, "api_key").strip().lower()
+    raw_mode = os.environ.get(_MODE_ENV)
+    mode = (
+        raw_mode
+        if raw_mode is not None
+        else ("api_key" if auth_settings.api_key_enabled else "none")
+    ).strip().lower()
     if mode in {"none", "no_auth"}:
         _logger.info("Default auth provider: none")
         return NoAuthProvider()
     if mode in {"api_key", "header"}:
+        _validate_local_api_key_mode()
         _logger.info("Default auth provider: api_key (local credentials)")
         return HeaderAuthProvider()
     if mode == "http_upstream":
@@ -223,6 +232,20 @@ def _build_default_provider() -> RequestAuthorizer:
     )
 
 
+def _validate_local_api_key_mode() -> None:
+    """Fail startup when local API-key mode has no local key validator."""
+    if not auth_settings.api_key_enabled:
+        raise RuntimeError(
+            f"{_MODE_ENV}=api_key requires AGENT_CONTROL_API_KEY_ENABLED=true. "
+            f"Use {_MODE_ENV}=none for deployments without credential enforcement."
+        )
+    if not auth_settings.get_api_keys() and not auth_settings.get_admin_api_keys():
+        raise RuntimeError(
+            f"{_MODE_ENV}=api_key requires AGENT_CONTROL_API_KEYS or "
+            "AGENT_CONTROL_ADMIN_API_KEYS to be configured."
+        )
+
+
 def _parse_extra_forward_headers(raw: str | None) -> tuple[str, ...]:
     """Parse a comma-separated header list into a deduplicated tuple.
 

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

@@ -147,6 +147,18 @@ class HttpUpstreamConfig:
     dropped. Names duplicating the default set or each other (after
     case-folding) are deduplicated."""
 
+    def __post_init__(self) -> None:
+        if self.service_token is None:
+            return
+        forwarded = {
+            name.lower()
+            for name in (*_DEFAULT_FORWARDED_HEADERS, *self.extra_forward_headers)
+        }
+        if self.service_token_header.lower() in forwarded:
+            raise ValueError(
+                "service_token_header must not match a forwarded caller credential header"
+            )
+
 
 class HttpUpstreamAuthProvider(RequestAuthorizer):
     """Delegates authorization to an upstream HTTP service."""
@@ -197,7 +209,7 @@ async def authorize(
                 hint="Retry the request; if the failure persists, contact the operator.",
             ) from exc
 
-        return self._handle_response(response, operation)
+        return self._handle_response(response, operation, context)
 
     def _forward_headers(self, request: Request) -> dict[str, str]:
         headers: dict[str, str] = {}
@@ -215,11 +227,16 @@ def _forward_headers(self, request: Request) -> dict[str, str]:
         return headers
 
     def _handle_response(
-        self, response: httpx.Response, operation: Operation
+        self,
+        response: httpx.Response,
+        operation: Operation,
+        context: dict[str, Any] | None,
     ) -> Principal:
         status = response.status_code
         if status == 200:
-            return self._parse_principal(response)
+            principal = self._parse_principal(response)
+            _ensure_target_context_matches_grant(context, principal)
+            return principal
         if status == 401:
             raise AuthenticationError(
                 error_code=ErrorCode.AUTH_INVALID_KEY,
@@ -309,3 +326,27 @@ def _parse_principal(self, response: httpx.Response) -> Principal:
             scopes=grant.scopes,
             grant_expires_at=grant.expires_at,
         )
+
+
+def _ensure_target_context_matches_grant(
+    context: dict[str, Any] | None,
+    principal: Principal,
+) -> None:
+    """Reject target-bound grants that do not match the requested target."""
+    if principal.target_type is None and principal.target_id is None:
+        return
+    if context is None:
+        return
+
+    expected_type = context.get("target_type")
+    expected_id = context.get("target_id")
+    if not isinstance(expected_type, str) or not isinstance(expected_id, str):
+        return
+    if principal.target_type == expected_type and principal.target_id == expected_id:
+        return
+
+    raise ForbiddenError(
+        error_code=ErrorCode.AUTH_INSUFFICIENT_PRIVILEGES,
+        detail="Authorization grant target does not match the requested target.",
+        hint="Retry with credentials authorized for the requested target.",
+    )

server/src/agent_control_server/auth_framework/runtime_token.py

@@ -92,6 +92,12 @@ def mint_runtime_token(
         )
     if not namespace_key:
         raise RuntimeTokenError("namespace_key is required to mint a runtime token")
+    if not actor_id:
+        raise RuntimeTokenError("actor_id is required to mint a runtime token")
+    if not target_type:
+        raise RuntimeTokenError("target_type is required to mint a runtime token")
+    if not target_id:
+        raise RuntimeTokenError("target_id is required to mint a runtime token")
     if ttl_seconds <= 0:
         raise RuntimeTokenError("ttl_seconds must be positive")
     if upstream_expires_at is not None and (

server/src/agent_control_server/endpoints/agents.py

@@ -174,6 +174,23 @@ def _ensure_target_principal_matches_namespace(
     )
 
 
+async def _authorize_existing_agent_overwrite(
+    request: Request,
+    principal: Principal,
+) -> None:
+    update_principal = await get_authorizer(Operation.AGENTS_UPDATE).authorize(
+        request,
+        Operation.AGENTS_UPDATE,
+    )
+    if update_principal.namespace_key == principal.namespace_key:
+        return
+    raise ForbiddenError(
+        error_code=ErrorCode.AUTH_INSUFFICIENT_PRIVILEGES,
+        detail="Update authorization resolved to a different namespace.",
+        hint="Ensure the credential is scoped to the requested agent namespace.",
+    )
+
+
 # =============================================================================
 # List Agents Models
 # =============================================================================
@@ -532,6 +549,7 @@ async def list_agents(
 )
 async def init_agent(
     request: InitAgentRequest,
+    http_request: Request,
     db: AsyncSession = Depends(get_async_db),
     principal: Principal = Depends(require_operation(Operation.AGENTS_CREATE)),
     target_principal: Principal | None = Depends(_init_agent_target_principal),
@@ -664,6 +682,9 @@ async def init_agent(
         )
         return InitAgentResponse(created=created, controls=controls)
 
+    if request.force_replace or request.conflict_mode == ConflictMode.OVERWRITE:
+        await _authorize_existing_agent_overwrite(http_request, principal)
+
     # Parse existing data via AgentData Pydantic model
     try:
         data_model = AgentData.model_validate(existing.data)

server/src/agent_control_server/endpoints/auth.py

@@ -13,6 +13,7 @@
 
 from __future__ import annotations
 
+import hashlib
 from datetime import datetime
 from typing import Any
 
@@ -34,6 +35,10 @@
 _logger = get_logger(__name__)
 
 
+def _log_hash(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]
+
+
 class RuntimeTokenExchangeRequest(BaseModel):
     """Body for the runtime token exchange endpoint."""
 
@@ -181,7 +186,7 @@ async def runtime_token_exchange(
         "Runtime token exchanged",
         extra={
             "namespace_key": claims.namespace_key,
-            "actor_id": claims.actor_id,
+            "actor_id_hash": _log_hash(claims.actor_id),
             "target_type": claims.target_type,
             "target_id": claims.target_id,
             "scopes": list(claims.scopes),

server/src/agent_control_server/endpoints/evaluation.py

@@ -127,10 +127,13 @@ async def _evaluation_context(request: Request) -> dict[str, object]:
         return {}
     if not isinstance(body, dict):
         return {}
-    return {
-        "target_type": body.get("target_type"),
-        "target_id": body.get("target_id"),
-    }
+    target_type = body.get("target_type")
+    target_id = body.get("target_id")
+    if not isinstance(target_type, str) or not isinstance(target_id, str):
+        return {}
+    if not target_type or not target_id:
+        return {}
+    return {"target_type": target_type, "target_id": target_id}
 
 
 @router.post(