"""Typed, secret-free machine-client lifecycle primitives for MCP onboarding.""" from __future__ import annotations from datetime import UTC, datetime, timedelta from enum import StrEnum from threading import RLock from uuid import UUID, uuid4 from pydantic import BaseModel, ConfigDict, Field class ClientKind(StrEnum): NATIVE_2026 = "native-2026" COMPATIBILITY_2025 = "compatibility-2025" STDIO_BRIDGE = "stdio-bridge" INSPECTOR = "inspector" class ClientStatus(StrEnum): PENDING = "pending" ENABLED = "enabled" DISABLED = "disabled" REVOKED = "revoked" class ClientRecord(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) id: UUID name: str = Field(min_length=1, max_length=128) project: str = Field(min_length=1, max_length=63, pattern=r"^[a-z0-9][a-z0-9-]*$") environment: str = Field(min_length=1, max_length=32, pattern=r"^[a-z0-9][a-z0-9-]*$") kind: ClientKind status: ClientStatus credential_reference: str = Field(min_length=1, max_length=256) created_at: datetime last_seen_at: datetime | None = None disabled_at: datetime | None = None revoked_at: datetime | None = None class ClientConfig(BaseModel): """Generated client configuration. It intentionally contains no token or secret.""" model_config = ConfigDict(extra="forbid", frozen=True, populate_by_name=True) client_id: UUID = Field(serialization_alias="clientId") kind: ClientKind protocol_version: str = Field(serialization_alias="protocolVersion") endpoint: str transport: str scopes: tuple[str, ...] credential_reference: str = Field(serialization_alias="credentialReference") environment_variable: str | None = Field( default=None, serialization_alias="environmentVariable" ) bridge_command: str | None = Field(default=None, serialization_alias="bridgeCommand") class BootstrapHandle(BaseModel): """A one-time Authentik hand-off reference, never a bearer credential.""" model_config = ConfigDict(extra="forbid", frozen=True, populate_by_name=True) client_id: UUID reference: str = Field(min_length=1, max_length=256) delivery: str = "authentik-service-account" expires_at: datetime = Field(serialization_alias="expiresAt") consumed: bool = False class ClientRegistry: """In-memory development registry; production persistence is supplied by the control plane.""" def __init__(self, *, revocation_cache_ttl_seconds: int = 60) -> None: if not 1 <= revocation_cache_ttl_seconds <= 60: raise ValueError("revocation cache ceiling is 60 seconds") self.revocation_cache_ttl_seconds = revocation_cache_ttl_seconds self._clients: dict[UUID, ClientRecord] = {} self._bootstrap: dict[str, BootstrapHandle] = {} self._lock = RLock() def create( self, *, name: str, project: str, environment: str, kind: ClientKind, now: datetime | None = None, ) -> ClientRecord: reference = _aware(now) client_id = uuid4() record = ClientRecord( id=client_id, name=name, project=project, environment=environment, kind=kind, status=ClientStatus.ENABLED, credential_reference=f"authentik://mcp-client/{client_id}", created_at=reference, ) with self._lock: self._clients[client_id] = record return record def list(self) -> tuple[ClientRecord, ...]: with self._lock: return tuple(self._clients.values()) def get(self, client_id: UUID) -> ClientRecord | None: with self._lock: return self._clients.get(client_id) def disable(self, client_id: UUID, *, now: datetime | None = None) -> ClientRecord: return self._transition(client_id, ClientStatus.DISABLED, _aware(now)) def revoke(self, client_id: UUID, *, now: datetime | None = None) -> ClientRecord: return self._transition(client_id, ClientStatus.REVOKED, _aware(now)) def _transition(self, client_id: UUID, status: ClientStatus, now: datetime) -> ClientRecord: with self._lock: current = self._clients.get(client_id) if current is None: raise KeyError(client_id) updated = current.model_copy( update={ "status": status, "disabled_at": now if status is ClientStatus.DISABLED else current.disabled_at, "revoked_at": now if status is ClientStatus.REVOKED else current.revoked_at, } ) self._clients[client_id] = updated return updated def touch(self, client_id: UUID, *, now: datetime | None = None) -> ClientRecord: reference = _aware(now) with self._lock: current = self._clients.get(client_id) if current is None: raise KeyError(client_id) if current.status is not ClientStatus.ENABLED: raise PermissionError("client is not enabled") updated = current.model_copy(update={"last_seen_at": reference}) self._clients[client_id] = updated return updated def authorize(self, client_id: UUID, *, project: str, environment: str) -> ClientRecord: record = self.get(client_id) if record is None or record.status is not ClientStatus.ENABLED: raise PermissionError("client is disabled or revoked") if record.project != project or record.environment != environment: raise PermissionError("client binding does not match endpoint scope") return record def issue_bootstrap( self, client_id: UUID, *, now: datetime | None = None, ttl_seconds: int = 300 ) -> BootstrapHandle: if not 1 <= ttl_seconds <= 900: raise ValueError("bootstrap TTL must be between 1 and 900 seconds") if self.get(client_id) is None: raise KeyError(client_id) reference = _aware(now) handle = BootstrapHandle( client_id=client_id, reference=f"authentik-bootstrap://{uuid4()}", expires_at=reference + timedelta(seconds=ttl_seconds), ) with self._lock: self._bootstrap[handle.reference] = handle return handle def consume_bootstrap(self, reference: str, *, now: datetime | None = None) -> BootstrapHandle: current = _aware(now) with self._lock: handle = self._bootstrap.get(reference) if handle is None or handle.consumed or handle.expires_at <= current: raise PermissionError("bootstrap reference is invalid or expired") consumed = handle.model_copy(update={"consumed": True}) self._bootstrap[reference] = consumed return consumed def _aware(value: datetime | None) -> datetime: result = value or datetime.now(UTC) if result.tzinfo is None or result.utcoffset() is None: raise ValueError("timestamps must be timezone-aware") return result.astimezone(UTC) def generate_config(record: ClientRecord, *, public_host: str = "mcp.example.com") -> ClientConfig: endpoint = f"https://{public_host}/mcp/{record.project}/{record.environment}" if record.kind is ClientKind.NATIVE_2026: return ClientConfig( client_id=record.id, kind=record.kind, protocol_version="2026-07-28", endpoint=endpoint, transport="streamable-http", scopes=("tools:read",), credential_reference=record.credential_reference, ) if record.kind is ClientKind.COMPATIBILITY_2025: return ClientConfig( client_id=record.id, kind=record.kind, protocol_version="2025-11-25", endpoint=endpoint, transport="streamable-http", scopes=("tools:read",), credential_reference=record.credential_reference, ) if record.kind is ClientKind.STDIO_BRIDGE: return ClientConfig( client_id=record.id, kind=record.kind, protocol_version="2026-07-28", endpoint=endpoint, transport="stdio", scopes=("tools:read",), credential_reference=record.credential_reference, environment_variable="ITWORX_MCP_ACCESS_TOKEN", bridge_command="itworx-mcp-bridge", ) return ClientConfig( client_id=record.id, kind=record.kind, protocol_version="2026-07-28", endpoint=f"http://127.0.0.1:18442/mcp/{record.project}/{record.environment}", transport="streamable-http", scopes=("tools:read",), credential_reference=record.credential_reference, )