diff --git a/fastmcp_slim/fastmcp/cli/deploy/__init__.py b/fastmcp_slim/fastmcp/cli/deploy/__init__.py new file mode 100644 index 000000000..94b708080 --- /dev/null +++ b/fastmcp_slim/fastmcp/cli/deploy/__init__.py @@ -0,0 +1 @@ +"""Horizon deployment support for the FastMCP CLI.""" diff --git a/fastmcp_slim/fastmcp/cli/deploy/authentication.py b/fastmcp_slim/fastmcp/cli/deploy/authentication.py new file mode 100644 index 000000000..6e71e5935 --- /dev/null +++ b/fastmcp_slim/fastmcp/cli/deploy/authentication.py @@ -0,0 +1,101 @@ +"""Horizon device authorization workflow.""" + +from __future__ import annotations + +import asyncio +import time +import webbrowser +from collections.abc import Awaitable, Callable +from contextlib import suppress + +from pydantic import SecretStr + +from fastmcp.cli.deploy.horizon_client import ( + DeviceAuthorization, + DeviceMetadata, + HorizonClient, +) + + +class DeviceAuthorizationError(RuntimeError): + """Device authorization did not complete.""" + + +class DeviceAuthorizationDeniedError(DeviceAuthorizationError): + """The user denied the device authorization request.""" + + +class DeviceAuthorizationExpiredError(DeviceAuthorizationError): + """The device authorization request expired.""" + + +async def poll_device_authorization( + client: HorizonClient, + authorization: DeviceAuthorization, + *, + sleep: Callable[[float], Awaitable[None]] | None = None, + monotonic: Callable[[], float] = time.monotonic, +) -> SecretStr: + """Poll at the server interval until the device request completes.""" + sleep = asyncio.sleep if sleep is None else sleep + deadline = monotonic() + authorization.expires_in + interval = float(authorization.interval) + + while True: + remaining = deadline - monotonic() + if remaining <= 0: + raise DeviceAuthorizationExpiredError( + "The device authorization request expired" + ) + + await sleep(min(interval, remaining)) + if monotonic() >= deadline: + raise DeviceAuthorizationExpiredError( + "The device authorization request expired" + ) + + result = await client.exchange_device_authorization(authorization.device_code) + if result.access_token is not None: + return result.access_token + if result.error == "authorization_pending": + continue + if result.error == "slow_down": + interval += 5 + continue + if result.error == "access_denied": + raise DeviceAuthorizationDeniedError( + "The device authorization request was denied" + ) + if result.error == "expired_token": + raise DeviceAuthorizationExpiredError( + "The device authorization request expired" + ) + + raise DeviceAuthorizationError("Device authorization failed") + + +async def authorize_device( + client: HorizonClient, + *, + metadata: DeviceMetadata | None = None, + on_challenge: Callable[[DeviceAuthorization], None] | None = None, + open_browser: bool = False, + browser_opener: Callable[[str], object] = webbrowser.open, + sleep: Callable[[float], Awaitable[None]] | None = None, + monotonic: Callable[[], float] = time.monotonic, +) -> SecretStr: + """Create, present, and complete a Horizon device authorization.""" + authorization = await client.create_device_authorization(metadata) + if on_challenge is not None: + on_challenge(authorization) + + if open_browser: + with suppress(OSError, webbrowser.Error): + browser_opener(authorization.verification_uri_complete) + + return await poll_device_authorization( + client, + authorization, + sleep=sleep, + monotonic=monotonic, + ) diff --git a/fastmcp_slim/fastmcp/cli/deploy/configuration.py b/fastmcp_slim/fastmcp/cli/deploy/configuration.py new file mode 100644 index 000000000..750bb8e83 --- /dev/null +++ b/fastmcp_slim/fastmcp/cli/deploy/configuration.py @@ -0,0 +1,68 @@ +"""Global non-secret configuration for the FastMCP CLI.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from fastmcp.cli.deploy.credentials import CredentialStore +from fastmcp.cli.deploy.horizon_client import ( + DEFAULT_HORIZON_API_ORIGIN, + normalize_api_origin, +) +from fastmcp.cli.deploy.state import read_state, state_lock, write_state + + +class HorizonConfiguration(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, populate_by_name=True) + + schema_version: Literal[1] = Field(alias="schemaVersion") + api_origin: str = Field(alias="apiOrigin") + + @field_validator("api_origin") + @classmethod + def validate_api_origin(cls, value: str) -> str: + return normalize_api_origin(value) + + +class ConfigurationStore: + """Persist the Horizon API origin without organization state.""" + + def __init__(self, state_directory: Path | None = None) -> None: + if state_directory is None: + import fastmcp + + state_directory = fastmcp.settings.home / "cli" + self.path = state_directory / "config.json" + + def load(self) -> HorizonConfiguration: + state = read_state(self.path, HorizonConfiguration) + if state is not None: + return state + return HorizonConfiguration( + schemaVersion=1, + apiOrigin=DEFAULT_HORIZON_API_ORIGIN, + ) + + def save(self, configuration: HorizonConfiguration) -> None: + write_state( + self.path, + configuration.model_dump(mode="json", by_alias=True), + ) + + def set_api_origin( + self, + api_origin: str, + *, + credentials: CredentialStore, + ) -> HorizonConfiguration: + """Set the origin and clear credentials before an origin change.""" + with state_lock(self.path.parent): + current = self.load() + updated = HorizonConfiguration(schemaVersion=1, apiOrigin=api_origin) + if updated.api_origin != current.api_origin: + credentials.clear() + self.save(updated) + return updated diff --git a/fastmcp_slim/fastmcp/cli/deploy/credentials.py b/fastmcp_slim/fastmcp/cli/deploy/credentials.py new file mode 100644 index 000000000..bd129abee --- /dev/null +++ b/fastmcp_slim/fastmcp/cli/deploy/credentials.py @@ -0,0 +1,142 @@ +"""Restricted Horizon credential storage and resolution.""" + +from __future__ import annotations + +import os +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import Literal + +from pydantic import ( + BaseModel, + ConfigDict, + Field, + SecretStr, + ValidationError, + field_validator, +) + +from fastmcp.cli.deploy.horizon_client import HorizonClient, normalize_api_origin +from fastmcp.cli.deploy.state import ( + StateFileError, + read_state, + remove_state, + state_lock, + write_state, +) + +CredentialSource = Literal["environment", "stored", "interactive"] + + +class AuthenticationRequiredError(RuntimeError): + """No Horizon credential is available without interactive authorization.""" + + +class AuthState(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, populate_by_name=True) + + schema_version: Literal[1] = Field(alias="schemaVersion") + api_key: SecretStr = Field(alias="apiKey") + + @field_validator("api_key") + @classmethod + def require_nonempty_api_key(cls, value: SecretStr) -> SecretStr: + if not value.get_secret_value().strip(): + raise ValueError("The API key is empty") + return value + + +@dataclass(frozen=True) +class ResolvedCredential: + api_key: SecretStr + source: CredentialSource + + +class CredentialStore: + """Persist the active personal Horizon API key.""" + + def __init__(self, state_directory: Path | None = None) -> None: + if state_directory is None: + import fastmcp + + state_directory = fastmcp.settings.home / "cli" + self.path = state_directory / "auth.json" + + def load(self) -> SecretStr | None: + state = read_state(self.path, AuthState, secret=True) + return state.api_key if state is not None else None + + def save(self, api_key: SecretStr | str) -> None: + try: + state = AuthState(schemaVersion=1, apiKey=api_key) + except ValidationError: + raise StateFileError("The Horizon API key is invalid") from None + write_state( + self.path, + { + "schemaVersion": state.schema_version, + "apiKey": state.api_key.get_secret_value(), + }, + ) + + def save_for_origin( + self, + api_key: SecretStr | str, + *, + expected_api_origin: str, + ) -> None: + """Save a key only while its issuing Horizon origin is active.""" + from fastmcp.cli.deploy.configuration import ConfigurationStore + + expected_api_origin = normalize_api_origin(expected_api_origin) + with state_lock(self.path.parent): + active_api_origin = ConfigurationStore(self.path.parent).load().api_origin + if active_api_origin != expected_api_origin: + raise StateFileError("The Horizon host changed during login") + self.save(api_key) + + def clear(self) -> None: + remove_state(self.path) + + +async def resolve_credential( + store: CredentialStore, + *, + environ: Mapping[str, str] | None = None, + authorize: Callable[[], Awaitable[SecretStr]] | None = None, + expected_api_origin: str | None = None, +) -> ResolvedCredential: + """Resolve environment, stored, then interactive credentials.""" + environ = os.environ if environ is None else environ + environment_key = environ.get("HORIZON_API_KEY") + if environment_key: + return ResolvedCredential( + api_key=SecretStr(environment_key), + source="environment", + ) + + stored_key = store.load() + if stored_key is not None: + return ResolvedCredential(api_key=stored_key, source="stored") + + if authorize is None: + raise AuthenticationRequiredError("Horizon authentication is required") + + api_key = await authorize() + if expected_api_origin is None: + store.save(api_key) + else: + store.save_for_origin(api_key, expected_api_origin=expected_api_origin) + return ResolvedCredential(api_key=api_key, source="interactive") + + +async def revoke_and_clear_credential( + client: HorizonClient, + store: CredentialStore, +) -> None: + """Attempt remote revocation and always remove the stored credential.""" + try: + await client.revoke_current_api_key() + finally: + store.clear() diff --git a/fastmcp_slim/fastmcp/cli/deploy/horizon_client.py b/fastmcp_slim/fastmcp/cli/deploy/horizon_client.py new file mode 100644 index 000000000..26a66a10f --- /dev/null +++ b/fastmcp_slim/fastmcp/cli/deploy/horizon_client.py @@ -0,0 +1,332 @@ +"""Typed HTTP client for the Horizon control plane.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from types import TracebackType +from typing import Annotated, Literal, TypeVar +from urllib.parse import urlsplit, urlunsplit + +import httpx2 +from pydantic import ( + BaseModel, + ConfigDict, + Field, + SecretStr, + ValidationError, + field_validator, +) + +DEVICE_AUTH_CLIENT_ID = "fastmcp-cli" +DEVICE_AUTH_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:device_code" +DEFAULT_HORIZON_API_ORIGIN = "https://horizon.prefect.io" + +DeviceTokenError = Literal[ + "authorization_pending", + "slow_down", + "access_denied", + "expired_token", +] + + +class HorizonError(RuntimeError): + """A safe Horizon client error.""" + + +class HorizonUnavailableError(HorizonError): + """The Horizon API could not be reached.""" + + +class HorizonUnauthorizedError(HorizonError): + """The Horizon credential was rejected.""" + + +class HorizonResponseError(HorizonError): + """Horizon returned an unexpected response.""" + + def __init__(self, message: str, *, status_code: int | None = None) -> None: + super().__init__(message) + self.status_code = status_code + + +class _ResponseModel(BaseModel): + model_config = ConfigDict(extra="ignore", frozen=True) + + +ResponseModelT = TypeVar("ResponseModelT", bound=_ResponseModel) + + +class DeviceAuthorization(_ResponseModel): + device_code: Annotated[str, Field(min_length=1)] + user_code: Annotated[str, Field(min_length=1)] + verification_uri: Annotated[str, Field(pattern=r"^https?://")] + verification_uri_complete: Annotated[str, Field(pattern=r"^https?://")] + expires_in: Annotated[int, Field(gt=0)] + interval: Annotated[int, Field(gt=0)] + + +class DeviceAccessToken(_ResponseModel): + access_token: SecretStr + token_type: Literal["Bearer"] + + @field_validator("access_token") + @classmethod + def require_nonempty_access_token(cls, value: SecretStr) -> SecretStr: + if not value.get_secret_value().strip(): + raise ValueError("The access token is empty") + return value + + +class _DeviceTokenErrorResponse(_ResponseModel): + error: DeviceTokenError + + +class HorizonUser(_ResponseModel): + id: str + email: str + name: str | None + + +class _CurrentUserResponse(_ResponseModel): + user: HorizonUser + + +class HorizonOrganization(_ResponseModel): + id: str + name: str + slug: str + + +class _PaginationMeta(_ResponseModel): + nextCursor: str | None + limit: int + + +class _OrganizationsResponse(_ResponseModel): + items: tuple[HorizonOrganization, ...] + meta: _PaginationMeta + + +@dataclass(frozen=True) +class DeviceMetadata: + device_name: str | None = None + platform: str | None = None + architecture: str | None = None + client_version: str | None = None + + +@dataclass(frozen=True) +class DeviceTokenPoll: + access_token: SecretStr | None = None + error: DeviceTokenError | None = None + + def __post_init__(self) -> None: + if (self.access_token is None) == (self.error is None): + raise ValueError("A device token poll must contain one result") + + +def normalize_api_origin(value: str) -> str: + """Validate and normalize a Horizon API origin.""" + parts = urlsplit(value) + try: + _ = parts.port + except ValueError: + raise ValueError("The Horizon API origin must be an HTTP origin") from None + if ( + parts.scheme not in {"http", "https"} + or not parts.hostname + or parts.username is not None + or parts.password is not None + or parts.query + or parts.fragment + or parts.path not in {"", "/"} + ): + raise ValueError("The Horizon API origin must be an HTTP origin") + + return urlunsplit((parts.scheme, parts.netloc, "", "", "")) + + +class HorizonClient: + """Call the Horizon routes used by FastMCP CLI authentication.""" + + def __init__( + self, + api_origin: str = DEFAULT_HORIZON_API_ORIGIN, + *, + api_key: SecretStr | str | None = None, + transport: httpx2.AsyncBaseTransport | None = None, + timeout: float = 30.0, + ) -> None: + self.api_origin = normalize_api_origin(api_origin) + self._api_key = ( + api_key + if isinstance(api_key, SecretStr) + else SecretStr(api_key) + if api_key is not None + else None + ) + self._client = httpx2.AsyncClient( + base_url=self.api_origin, + follow_redirects=False, + timeout=timeout, + transport=transport, + ) + + async def __aenter__(self) -> HorizonClient: + await self._client.__aenter__() + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + await self._client.__aexit__(exc_type, exc_value, traceback) + + async def aclose(self) -> None: + await self._client.aclose() + + async def _request( + self, + method: str, + path: str, + *, + authenticated: bool = False, + data: Mapping[str, str] | None = None, + params: Mapping[str, str | int] | None = None, + ) -> httpx2.Response: + headers: dict[str, str] = {} + if authenticated: + if self._api_key is None: + raise HorizonUnauthorizedError("Horizon authentication is required") + headers["Authorization"] = f"Bearer {self._api_key.get_secret_value()}" + + try: + response = await self._client.request( + method, + path, + headers=headers, + data=data, + params=params, + ) + except httpx2.RequestError as exc: + raise HorizonUnavailableError("The Horizon API is unavailable") from exc + + if authenticated and response.status_code == 401: + raise HorizonUnauthorizedError("The Horizon credential is not valid") + return response + + @staticmethod + def _validate_response( + response: httpx2.Response, + model: type[ResponseModelT], + ) -> ResponseModelT: + try: + return model.model_validate_json(response.content) + except (ValidationError, ValueError): + raise HorizonResponseError( + "Horizon returned an invalid response", + status_code=response.status_code, + ) from None + + @staticmethod + def _require_status(response: httpx2.Response, expected: int) -> None: + if response.status_code != expected: + raise HorizonResponseError( + "Horizon returned an unexpected status", + status_code=response.status_code, + ) + + async def create_device_authorization( + self, + metadata: DeviceMetadata | None = None, + ) -> DeviceAuthorization: + metadata = metadata or DeviceMetadata() + form = { + "client_id": DEVICE_AUTH_CLIENT_ID, + "device_name": metadata.device_name, + "platform": metadata.platform, + "architecture": metadata.architecture, + "client_version": metadata.client_version, + } + response = await self._request( + "POST", + "/api/v0/oauth/device/authorization", + data={key: value for key, value in form.items() if value is not None}, + ) + self._require_status(response, 200) + return self._validate_response(response, DeviceAuthorization) + + async def exchange_device_authorization( + self, + device_code: str, + ) -> DeviceTokenPoll: + response = await self._request( + "POST", + "/api/v0/oauth/device/token", + data={ + "grant_type": DEVICE_AUTH_GRANT_TYPE, + "client_id": DEVICE_AUTH_CLIENT_ID, + "device_code": device_code, + }, + ) + + if response.status_code == 200: + result = self._validate_response(response, DeviceAccessToken) + return DeviceTokenPoll(access_token=result.access_token) + + if response.status_code == 400: + result = self._validate_response(response, _DeviceTokenErrorResponse) + return DeviceTokenPoll(error=result.error) + + self._require_status(response, 200) + raise AssertionError("unreachable") + + async def get_current_user(self) -> HorizonUser: + response = await self._request( + "GET", + "/api/v0/me", + authenticated=True, + ) + self._require_status(response, 200) + result = self._validate_response(response, _CurrentUserResponse) + return result.user + + async def list_organizations(self) -> tuple[HorizonOrganization, ...]: + organizations: list[HorizonOrganization] = [] + cursor: str | None = None + seen_cursors: set[str] = set() + + while True: + params = {"limit": 100} + if cursor is not None: + params["cursor"] = cursor + response = await self._request( + "GET", + "/api/v0/me/organizations", + authenticated=True, + params=params, + ) + self._require_status(response, 200) + result = self._validate_response(response, _OrganizationsResponse) + organizations.extend(result.items) + + cursor = result.meta.nextCursor + if cursor is None: + return tuple(organizations) + if cursor in seen_cursors: + raise HorizonResponseError( + "Horizon returned an invalid organization cursor", + status_code=response.status_code, + ) + seen_cursors.add(cursor) + + async def revoke_current_api_key(self) -> None: + response = await self._request( + "DELETE", + "/api/v0/me/api-key", + authenticated=True, + ) + self._require_status(response, 204) diff --git a/fastmcp_slim/fastmcp/cli/deploy/state.py b/fastmcp_slim/fastmcp/cli/deploy/state.py new file mode 100644 index 000000000..55cad4d74 --- /dev/null +++ b/fastmcp_slim/fastmcp/cli/deploy/state.py @@ -0,0 +1,226 @@ +"""Versioned JSON state helpers for the FastMCP CLI.""" + +from __future__ import annotations + +import errno +import json +import os +import subprocess +import tempfile +from collections.abc import Iterator +from contextlib import contextmanager, suppress +from pathlib import Path +from typing import Any, TypeVar + +from pydantic import BaseModel, ValidationError + +ModelT = TypeVar("ModelT", bound=BaseModel) + + +class StateFileError(RuntimeError): + """A CLI state file could not be read or written safely.""" + + +_WINDOWS_ACL_SCRIPT = r""" +$ErrorActionPreference = "Stop" +$path = $env:FASTMCP_STATE_PATH +$sid = [System.Security.Principal.WindowsIdentity]::GetCurrent().User +$acl = Get-Acl -LiteralPath $path +$acl.SetAccessRuleProtection($true, $false) +foreach ($existingRule in @($acl.Access)) { + $acl.RemoveAccessRuleSpecific($existingRule) +} + +if ([System.IO.Directory]::Exists($path)) { + $inheritance = [System.Security.AccessControl.InheritanceFlags]::ContainerInherit ` + -bor [System.Security.AccessControl.InheritanceFlags]::ObjectInherit + $rule = [System.Security.AccessControl.FileSystemAccessRule]::new( + $sid, + [System.Security.AccessControl.FileSystemRights]::FullControl, + $inheritance, + [System.Security.AccessControl.PropagationFlags]::None, + [System.Security.AccessControl.AccessControlType]::Allow + ) +} else { + $rule = [System.Security.AccessControl.FileSystemAccessRule]::new( + $sid, + [System.Security.AccessControl.FileSystemRights]::FullControl, + [System.Security.AccessControl.AccessControlType]::Allow + ) +} + +$acl.AddAccessRule($rule) +Set-Acl -LiteralPath $path -AclObject $acl +""" + + +def _restrict_windows_access(path: Path) -> None: + try: + subprocess.run( + [ + "powershell.exe", + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + _WINDOWS_ACL_SCRIPT, + ], + check=True, + capture_output=True, + text=True, + env={**os.environ, "FASTMCP_STATE_PATH": str(path)}, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise StateFileError("Could not restrict access to CLI state") from exc + + +def _restrict_access(path: Path, *, directory: bool = False) -> None: + try: + if os.name == "nt": + _restrict_windows_access(path) + else: + path.chmod(0o700 if directory else 0o600) + except OSError as exc: + raise StateFileError("Could not restrict access to CLI state") from exc + + +def _prepare_directory(path: Path) -> None: + try: + path.mkdir(parents=True, exist_ok=True) + except OSError as exc: + raise StateFileError("Could not create the CLI state directory") from exc + _restrict_access(path, directory=True) + + +@contextmanager +def state_lock(directory: Path) -> Iterator[None]: + """Lock related CLI state changes across processes.""" + _prepare_directory(directory) + lock_path = directory / ".state.lock" + if lock_path.is_symlink(): + raise StateFileError("The CLI state lock must not be a symbolic link") + + lock_file = None + try: + lock_file = lock_path.open("a+b") + _restrict_access(lock_path) + if os.name == "nt": + import msvcrt + + if lock_path.stat().st_size == 0: + lock_file.write(b"\0") + lock_file.flush() + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1) + else: + import fcntl + + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + except (OSError, StateFileError) as exc: + if lock_file is not None: + with suppress(OSError): + lock_file.close() + if isinstance(exc, StateFileError): + raise + raise StateFileError("Could not lock CLI state") from exc + + try: + yield + finally: + if os.name == "nt": + import msvcrt + + with suppress(OSError): + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + + with suppress(OSError): + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + with suppress(OSError): + lock_file.close() + + +def read_state( + path: Path, + model: type[ModelT], + *, + secret: bool = False, +) -> ModelT | None: + """Read and validate a versioned JSON state file.""" + if not path.exists(): + return None + if path.is_symlink(): + raise StateFileError(f"CLI state must not be a symbolic link: {path.name}") + + if secret: + _restrict_access(path.parent, directory=True) + _restrict_access(path) + + try: + return model.model_validate_json(path.read_text(encoding="utf-8")) + except (ValidationError, ValueError): + raise StateFileError(f"CLI state is invalid: {path.name}") from None + except OSError as exc: + raise StateFileError(f"Could not read CLI state: {path.name}") from exc + + +def write_state(path: Path, data: dict[str, Any]) -> None: + """Write JSON through a restricted temporary file and atomic replacement.""" + _prepare_directory(path.parent) + payload = (json.dumps(data, indent=2, sort_keys=True) + "\n").encode() + descriptor: int | None = None + temporary_path: Path | None = None + + try: + descriptor, temporary_name = tempfile.mkstemp( + dir=path.parent, + prefix=f".{path.name}.", + suffix=".tmp", + ) + temporary_path = Path(temporary_name) + if os.name != "nt": + os.fchmod(descriptor, 0o600) + + temporary_file = os.fdopen(descriptor, "wb") + descriptor = None + with temporary_file: + temporary_file.write(payload) + temporary_file.flush() + os.fsync(temporary_file.fileno()) + + _restrict_access(temporary_path) + os.replace(temporary_path, path) + temporary_path = None + + if os.name != "nt": + directory_descriptor = os.open(path.parent, os.O_RDONLY) + try: + try: + os.fsync(directory_descriptor) + except OSError as exc: + unsupported = {errno.EINVAL, errno.ENOTSUP} + if exc.errno not in unsupported: + raise + finally: + os.close(directory_descriptor) + except StateFileError: + raise + except OSError as exc: + raise StateFileError(f"Could not write CLI state: {path.name}") from exc + finally: + if descriptor is not None: + with suppress(OSError): + os.close(descriptor) + if temporary_path is not None: + with suppress(OSError): + temporary_path.unlink(missing_ok=True) + + +def remove_state(path: Path) -> None: + """Remove a state file when it exists.""" + try: + path.unlink(missing_ok=True) + except OSError as exc: + raise StateFileError(f"Could not remove CLI state: {path.name}") from exc diff --git a/tests/cli/deploy/test_authentication.py b/tests/cli/deploy/test_authentication.py new file mode 100644 index 000000000..4b9b215cc --- /dev/null +++ b/tests/cli/deploy/test_authentication.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +import httpx2 +import pytest + +from fastmcp.cli.deploy.authentication import ( + DeviceAuthorizationDeniedError, + DeviceAuthorizationExpiredError, + authorize_device, + poll_device_authorization, +) +from fastmcp.cli.deploy.horizon_client import DeviceAuthorization, HorizonClient + + +class Clock: + def __init__(self) -> None: + self.now = 0.0 + self.sleeps: list[float] = [] + + def monotonic(self) -> float: + return self.now + + async def sleep(self, delay: float) -> None: + self.sleeps.append(delay) + self.now += delay + + +def authorization(*, expires_in: int = 600, interval: int = 5) -> DeviceAuthorization: + return DeviceAuthorization( + device_code="device-secret", + user_code="BCDF-GHJK", + verification_uri="https://horizon.prefect.io/oauth/device", + verification_uri_complete=( + "https://horizon.prefect.io/oauth/device?user_code=BCDF-GHJK" + ), + expires_in=expires_in, + interval=interval, + ) + + +def sequenced_transport( + responses: list[httpx2.Response], +) -> httpx2.MockTransport: + def handler(request: httpx2.Request) -> httpx2.Response: + return responses.pop(0) + + return httpx2.MockTransport(handler) + + +async def test_polling_handles_pending_slow_down_and_approval() -> None: + clock = Clock() + async with HorizonClient( + transport=sequenced_transport( + [ + httpx2.Response(400, json={"error": "authorization_pending"}), + httpx2.Response(400, json={"error": "slow_down"}), + httpx2.Response( + 200, + json={"access_token": "fmcp_secret", "token_type": "Bearer"}, + ), + ] + ) + ) as client: + api_key = await poll_device_authorization( + client, + authorization(), + sleep=clock.sleep, + monotonic=clock.monotonic, + ) + + assert api_key.get_secret_value() == "fmcp_secret" + assert clock.sleeps == [5, 5, 10] + + +@pytest.mark.parametrize( + ("error", "exception"), + [ + ("access_denied", DeviceAuthorizationDeniedError), + ("expired_token", DeviceAuthorizationExpiredError), + ], +) +async def test_polling_handles_terminal_errors( + error: str, + exception: type[Exception], +) -> None: + clock = Clock() + async with HorizonClient( + transport=sequenced_transport([httpx2.Response(400, json={"error": error})]) + ) as client: + with pytest.raises(exception): + await poll_device_authorization( + client, + authorization(), + sleep=clock.sleep, + monotonic=clock.monotonic, + ) + + +async def test_polling_stops_at_the_local_expiry_deadline() -> None: + clock = Clock() + requests: list[httpx2.Request] = [] + + def handler(request: httpx2.Request) -> httpx2.Response: + requests.append(request) + return httpx2.Response(400, json={"error": "authorization_pending"}) + + async with HorizonClient(transport=httpx2.MockTransport(handler)) as client: + with pytest.raises(DeviceAuthorizationExpiredError): + await poll_device_authorization( + client, + authorization(expires_in=5, interval=5), + sleep=clock.sleep, + monotonic=clock.monotonic, + ) + + assert requests == [] + + +async def test_authorize_device_presents_challenge_before_opening_browser() -> None: + events: list[str] = [] + clock = Clock() + + def handler(request: httpx2.Request) -> httpx2.Response: + if request.url.path.endswith("/authorization"): + return httpx2.Response(200, json=authorization().model_dump()) + return httpx2.Response( + 200, + json={"access_token": "fmcp_secret", "token_type": "Bearer"}, + ) + + def present(challenge: DeviceAuthorization) -> None: + events.append(f"present:{challenge.user_code}") + + def open_browser(url: str) -> None: + events.append(f"browser:{url}") + + async with HorizonClient(transport=httpx2.MockTransport(handler)) as client: + await authorize_device( + client, + on_challenge=present, + open_browser=True, + browser_opener=open_browser, + sleep=clock.sleep, + monotonic=clock.monotonic, + ) + + assert events == [ + "present:BCDF-GHJK", + "browser:https://horizon.prefect.io/oauth/device?user_code=BCDF-GHJK", + ] + + +async def test_browser_failure_does_not_stop_remote_login() -> None: + clock = Clock() + + def handler(request: httpx2.Request) -> httpx2.Response: + if request.url.path.endswith("/authorization"): + return httpx2.Response(200, json=authorization().model_dump()) + return httpx2.Response( + 200, + json={"access_token": "fmcp_secret", "token_type": "Bearer"}, + ) + + def fail_to_open(url: str) -> None: + raise OSError("no browser") + + async with HorizonClient(transport=httpx2.MockTransport(handler)) as client: + api_key = await authorize_device( + client, + open_browser=True, + browser_opener=fail_to_open, + sleep=clock.sleep, + monotonic=clock.monotonic, + ) + + assert api_key.get_secret_value() == "fmcp_secret" diff --git a/tests/cli/deploy/test_configuration.py b/tests/cli/deploy/test_configuration.py new file mode 100644 index 000000000..476e68b89 --- /dev/null +++ b/tests/cli/deploy/test_configuration.py @@ -0,0 +1,131 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from fastmcp.cli.deploy.configuration import ( + ConfigurationStore, + HorizonConfiguration, +) +from fastmcp.cli.deploy.credentials import CredentialStore +from fastmcp.cli.deploy.horizon_client import DEFAULT_HORIZON_API_ORIGIN +from fastmcp.cli.deploy.state import StateFileError + + +def test_configuration_defaults_to_the_production_origin(tmp_path: Path) -> None: + store = ConfigurationStore(tmp_path) + + configuration = store.load() + + assert configuration.api_origin == DEFAULT_HORIZON_API_ORIGIN + assert store.path.exists() is False + + +def test_configuration_stores_only_schema_and_api_origin(tmp_path: Path) -> None: + store = ConfigurationStore(tmp_path) + configuration = HorizonConfiguration( + schemaVersion=1, + apiOrigin="https://example.com/", + ) + + store.save(configuration) + + assert json.loads(store.path.read_text()) == { + "schemaVersion": 1, + "apiOrigin": "https://example.com", + } + assert store.load() == configuration + + +def test_configuration_rejects_organization_state(tmp_path: Path) -> None: + store = ConfigurationStore(tmp_path) + store.path.write_text( + json.dumps( + { + "schemaVersion": 1, + "apiOrigin": DEFAULT_HORIZON_API_ORIGIN, + "currentOrganizationId": "org-id", + } + ) + ) + + with pytest.raises(StateFileError): + store.load() + + +def test_origin_change_clears_credentials_before_writing_configuration( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + credentials = CredentialStore(tmp_path) + credentials.save("fmcp_secret") + configuration = ConfigurationStore(tmp_path) + configuration.save( + HorizonConfiguration( + schemaVersion=1, + apiOrigin=DEFAULT_HORIZON_API_ORIGIN, + ) + ) + events: list[str] = [] + original_clear = credentials.clear + original_save = configuration.save + + def clear() -> None: + events.append("clear") + original_clear() + + def save(value: HorizonConfiguration) -> None: + events.append("save") + original_save(value) + + monkeypatch.setattr(credentials, "clear", clear) + monkeypatch.setattr(configuration, "save", save) + + result = configuration.set_api_origin( + "https://dev.horizon.prefect.io", + credentials=credentials, + ) + + assert events == ["clear", "save"] + assert credentials.load() is None + assert result.api_origin == "https://dev.horizon.prefect.io" + + +def test_same_origin_does_not_clear_credentials(tmp_path: Path) -> None: + credentials = CredentialStore(tmp_path) + credentials.save("fmcp_secret") + configuration = ConfigurationStore(tmp_path) + + configuration.set_api_origin( + DEFAULT_HORIZON_API_ORIGIN, + credentials=credentials, + ) + + stored = credentials.load() + assert stored is not None + assert stored.get_secret_value() == "fmcp_secret" + + +def test_failed_origin_write_leaves_no_cross_origin_credential( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + credentials = CredentialStore(tmp_path) + credentials.save("fmcp_secret") + configuration = ConfigurationStore(tmp_path) + + def fail_save(value: HorizonConfiguration) -> None: + raise StateFileError("write failed") + + monkeypatch.setattr(configuration, "save", fail_save) + + with pytest.raises(StateFileError): + configuration.set_api_origin( + "https://dev.horizon.prefect.io", + credentials=credentials, + ) + + assert credentials.load() is None + assert configuration.load().api_origin == DEFAULT_HORIZON_API_ORIGIN diff --git a/tests/cli/deploy/test_credentials.py b/tests/cli/deploy/test_credentials.py new file mode 100644 index 000000000..23e98ca62 --- /dev/null +++ b/tests/cli/deploy/test_credentials.py @@ -0,0 +1,309 @@ +from __future__ import annotations + +import errno +import json +import os +import stat +import subprocess +import traceback +from pathlib import Path +from typing import cast + +import httpx2 +import pytest +from pydantic import SecretStr + +from fastmcp.cli.deploy.configuration import ( + ConfigurationStore, + HorizonConfiguration, +) +from fastmcp.cli.deploy.credentials import ( + AuthenticationRequiredError, + CredentialStore, + resolve_credential, + revoke_and_clear_credential, +) +from fastmcp.cli.deploy.horizon_client import HorizonClient, HorizonUnavailableError +from fastmcp.cli.deploy.state import ( + StateFileError, + _restrict_windows_access, +) + + +def load_secret(store: CredentialStore) -> SecretStr: + secret = store.load() + assert secret is not None + return secret + + +def test_credential_store_writes_only_the_approved_contract(tmp_path: Path) -> None: + store = CredentialStore(tmp_path) + store.save("fmcp_secret") + + assert json.loads(store.path.read_text()) == { + "schemaVersion": 1, + "apiKey": "fmcp_secret", + } + assert load_secret(store).get_secret_value() == "fmcp_secret" + assert "user" not in store.path.read_text() + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") +def test_credential_store_restricts_file_and_directory_modes(tmp_path: Path) -> None: + state_directory = tmp_path / "cli" + store = CredentialStore(state_directory) + store.save("fmcp_secret") + + assert store.path.stat().st_mode & 0o777 == 0o600 + assert state_directory.stat().st_mode & 0o777 == 0o700 + + +def test_credential_store_restricts_an_existing_secret_file(tmp_path: Path) -> None: + store = CredentialStore(tmp_path) + store.path.write_text('{"schemaVersion": 1, "apiKey": "fmcp_secret"}') + if os.name != "nt": + store.path.chmod(0o644) + + assert load_secret(store).get_secret_value() == "fmcp_secret" + if os.name != "nt": + assert store.path.stat().st_mode & 0o777 == 0o600 + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX directory fsync") +def test_atomic_write_ignores_unsupported_directory_fsync( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + store = CredentialStore(tmp_path) + original_fsync = os.fsync + + def fsync(descriptor: int) -> None: + if stat.S_ISDIR(os.fstat(descriptor).st_mode): + raise OSError(errno.EINVAL, "directory sync is not supported") + original_fsync(descriptor) + + monkeypatch.setattr("fastmcp.cli.deploy.state.os.fsync", fsync) + + store.save("fmcp_secret") + + assert load_secret(store).get_secret_value() == "fmcp_secret" + + +def test_atomic_write_preserves_previous_state_on_replace_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + store = CredentialStore(tmp_path) + store.save("fmcp_original") + + def fail_replace(source: Path, destination: Path) -> None: + raise OSError("replace failed") + + monkeypatch.setattr("fastmcp.cli.deploy.state.os.replace", fail_replace) + with pytest.raises(StateFileError): + store.save("fmcp_new") + + assert json.loads(store.path.read_text())["apiKey"] == "fmcp_original" + assert list(tmp_path.glob(".*.tmp")) == [] + + +async def test_environment_credential_takes_precedence_and_is_not_stored( + tmp_path: Path, +) -> None: + store = CredentialStore(tmp_path) + store.save("fmcp_stored") + authorize_called = False + + async def authorize() -> SecretStr: + nonlocal authorize_called + authorize_called = True + return SecretStr("fmcp_interactive") + + result = await resolve_credential( + store, + environ={"HORIZON_API_KEY": "fmcp_environment"}, + authorize=authorize, + ) + + assert result.source == "environment" + assert result.api_key.get_secret_value() == "fmcp_environment" + assert load_secret(store).get_secret_value() == "fmcp_stored" + assert authorize_called is False + + +async def test_stored_credential_precedes_interactive_authorization( + tmp_path: Path, +) -> None: + store = CredentialStore(tmp_path) + store.save("fmcp_stored") + + async def authorize() -> SecretStr: + raise AssertionError("interactive authorization must not run") + + result = await resolve_credential(store, environ={}, authorize=authorize) + + assert result.source == "stored" + assert result.api_key.get_secret_value() == "fmcp_stored" + + +async def test_interactive_credential_is_persisted(tmp_path: Path) -> None: + store = CredentialStore(tmp_path) + + async def authorize() -> SecretStr: + return SecretStr("fmcp_interactive") + + result = await resolve_credential(store, environ={}, authorize=authorize) + + assert result.source == "interactive" + assert load_secret(store).get_secret_value() == "fmcp_interactive" + + +async def test_interactive_credential_rejects_an_origin_change( + tmp_path: Path, +) -> None: + store = CredentialStore(tmp_path) + ConfigurationStore(tmp_path).save( + HorizonConfiguration( + schemaVersion=1, + apiOrigin="https://dev.horizon.prefect.io", + ) + ) + + async def authorize() -> SecretStr: + return SecretStr("fmcp_old_origin") + + with pytest.raises(StateFileError, match="host changed"): + await resolve_credential( + store, + environ={}, + authorize=authorize, + expected_api_origin="https://horizon.prefect.io", + ) + + assert store.load() is None + + +async def test_missing_noninteractive_credential_is_explicit(tmp_path: Path) -> None: + with pytest.raises(AuthenticationRequiredError): + await resolve_credential(CredentialStore(tmp_path), environ={}) + + +async def test_remote_revoke_always_removes_the_local_credential( + tmp_path: Path, +) -> None: + store = CredentialStore(tmp_path) + store.save("fmcp_stored") + + def unavailable(request: httpx2.Request) -> httpx2.Response: + raise httpx2.ConnectError("offline", request=request) + + async with HorizonClient( + api_key="fmcp_stored", + transport=httpx2.MockTransport(unavailable), + ) as client: + with pytest.raises(HorizonUnavailableError): + await revoke_and_clear_credential(client, store) + + assert store.load() is None + + +def test_windows_acl_replaces_the_existing_access_list( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + path = tmp_path / "auth.json" + path.write_text("{}") + calls: list[list[str]] = [] + state_paths: list[str] = [] + + def run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: + calls.append(command) + environment = cast(dict[str, str], kwargs["env"]) + state_paths.append(environment["FASTMCP_STATE_PATH"]) + return subprocess.CompletedProcess(command, 0, "", "") + + monkeypatch.setattr("fastmcp.cli.deploy.state.subprocess.run", run) + _restrict_windows_access(path) + + assert calls == [ + [ + "powershell.exe", + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + calls[0][5], + ] + ] + assert state_paths == [str(path)] + assert "$path = $env:FASTMCP_STATE_PATH" in calls[0][5] + assert "Get-Acl -LiteralPath $path" in calls[0][5] + assert "SetAccessRuleProtection($true, $false)" in calls[0][5] + assert "RemoveAccessRuleSpecific($existingRule)" in calls[0][5] + + +@pytest.mark.skipif(os.name != "nt", reason="Windows ACL inspection") +def test_windows_credential_state_allows_only_the_current_user(tmp_path: Path) -> None: + state_directory = tmp_path / "cli" + store = CredentialStore(state_directory) + store.save("fmcp_secret") + inspect_acl = r""" +$acl = Get-Acl -LiteralPath $env:FASTMCP_STATE_PATH +$current = [System.Security.Principal.WindowsIdentity]::GetCurrent().User.Value +$access = @($acl.Access | ForEach-Object { + $_.IdentityReference.Translate( + [System.Security.Principal.SecurityIdentifier] + ).Value +}) +[pscustomobject]@{ + current = $current + access = $access + protected = $acl.AreAccessRulesProtected + inherited = @($acl.Access | ForEach-Object { $_.IsInherited }) +} | ConvertTo-Json -Compress +""" + + for path in (state_directory, store.path): + result = subprocess.run( + [ + "powershell.exe", + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + inspect_acl, + ], + check=True, + capture_output=True, + text=True, + env={**os.environ, "FASTMCP_STATE_PATH": str(path)}, + ) + acl = json.loads(result.stdout) + assert set(acl["access"]) == {acl["current"]} + assert acl["protected"] is True + assert not any(acl["inherited"]) + + +def test_credential_store_rejects_empty_api_keys(tmp_path: Path) -> None: + store = CredentialStore(tmp_path) + + for api_key in ("", " "): + with pytest.raises(StateFileError): + store.save(api_key) + + assert store.path.exists() is False + + +def test_malformed_credential_state_has_a_safe_error(tmp_path: Path) -> None: + store = CredentialStore(tmp_path) + store.path.write_text( + '{"schemaVersion": 1, "apiKey": "fmcp_valid", "metadata": "fmcp_secret"}' + ) + + with pytest.raises(StateFileError) as exc_info: + store.load() + + formatted_exception = "".join(traceback.format_exception(exc_info.value)) + assert "fmcp_secret" not in formatted_exception + assert exc_info.value.__cause__ is None + assert exc_info.value.__suppress_context__ is True diff --git a/tests/cli/deploy/test_horizon_client.py b/tests/cli/deploy/test_horizon_client.py new file mode 100644 index 000000000..aa0e4dee0 --- /dev/null +++ b/tests/cli/deploy/test_horizon_client.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +from collections.abc import Callable +from urllib.parse import parse_qs + +import httpx2 +import pytest +from pydantic import SecretStr + +from fastmcp.cli.deploy.horizon_client import ( + DEVICE_AUTH_CLIENT_ID, + DEVICE_AUTH_GRANT_TYPE, + DeviceMetadata, + HorizonClient, + HorizonResponseError, + HorizonUnauthorizedError, + normalize_api_origin, +) + + +def mock_transport( + handler: Callable[[httpx2.Request], httpx2.Response], +) -> httpx2.MockTransport: + return httpx2.MockTransport(handler) + + +async def test_device_authorization_uses_the_oauth_form_contract() -> None: + def handler(request: httpx2.Request) -> httpx2.Response: + assert request.url.path == "/api/v0/oauth/device/authorization" + assert request.headers["content-type"].startswith( + "application/x-www-form-urlencoded" + ) + assert "authorization" not in request.headers + assert parse_qs(request.content.decode()) == { + "client_id": [DEVICE_AUTH_CLIENT_ID], + "device_name": ["Avery's laptop"], + "platform": ["darwin"], + "architecture": ["arm64"], + "client_version": ["4.0.0"], + } + return httpx2.Response( + 200, + json={ + "device_code": "device-secret", + "user_code": "BCDF-GHJK", + "verification_uri": "https://horizon.prefect.io/oauth/device", + "verification_uri_complete": "https://horizon.prefect.io/oauth/device?user_code=BCDF-GHJK", + "expires_in": 600, + "interval": 5, + }, + ) + + async with HorizonClient( + transport=mock_transport(handler), + ) as client: + result = await client.create_device_authorization( + DeviceMetadata( + device_name="Avery's laptop", + platform="darwin", + architecture="arm64", + client_version="4.0.0", + ) + ) + + assert result.user_code == "BCDF-GHJK" + assert result.interval == 5 + + +@pytest.mark.parametrize( + "error", + ["authorization_pending", "slow_down", "access_denied", "expired_token"], +) +async def test_device_token_exchange_returns_expected_poll_errors(error: str) -> None: + def handler(request: httpx2.Request) -> httpx2.Response: + assert parse_qs(request.content.decode()) == { + "grant_type": [DEVICE_AUTH_GRANT_TYPE], + "client_id": [DEVICE_AUTH_CLIENT_ID], + "device_code": ["device-secret"], + } + return httpx2.Response(400, json={"error": error}) + + async with HorizonClient(transport=mock_transport(handler)) as client: + result = await client.exchange_device_authorization("device-secret") + + assert result.error == error + assert result.access_token is None + + +@pytest.mark.parametrize("access_token", ["", " "]) +async def test_device_token_exchange_rejects_empty_access_tokens( + access_token: str, +) -> None: + async with HorizonClient( + transport=mock_transport( + lambda request: httpx2.Response( + 200, + json={"access_token": access_token, "token_type": "Bearer"}, + ) + ) + ) as client: + with pytest.raises(HorizonResponseError): + await client.exchange_device_authorization("device-secret") + + +async def test_device_token_exchange_keeps_the_api_key_secret() -> None: + async with HorizonClient( + transport=mock_transport( + lambda request: httpx2.Response( + 200, + json={"access_token": "fmcp_secret", "token_type": "Bearer"}, + ) + ) + ) as client: + result = await client.exchange_device_authorization("device-secret") + + assert isinstance(result.access_token, SecretStr) + assert result.access_token.get_secret_value() == "fmcp_secret" + assert "fmcp_secret" not in repr(result) + + +async def test_authenticated_routes_use_the_current_key_and_paginate() -> None: + cursors: list[str | None] = [] + + def handler(request: httpx2.Request) -> httpx2.Response: + assert request.headers["authorization"] == "Bearer fmcp_secret" + if request.url.path == "/api/v0/me": + return httpx2.Response( + 200, + json={ + "user": { + "id": "user-id", + "email": "avery@example.com", + "name": "Avery", + "workosUserId": "workos-id", + "createdAt": "2026-08-08T00:00:00Z", + } + }, + ) + + assert request.url.path == "/api/v0/me/organizations" + cursor = request.url.params.get("cursor") + cursors.append(cursor) + if cursor is None: + return httpx2.Response( + 200, + json={ + "items": [{"id": "org-1", "name": "First", "slug": "first"}], + "meta": {"nextCursor": "next-page", "limit": 100}, + }, + ) + return httpx2.Response( + 200, + json={ + "items": [{"id": "org-2", "name": "Second", "slug": "second"}], + "meta": {"nextCursor": None, "limit": 100}, + }, + ) + + async with HorizonClient( + api_key="fmcp_secret", + transport=mock_transport(handler), + ) as client: + user = await client.get_current_user() + organizations = await client.list_organizations() + + assert user.email == "avery@example.com" + assert [organization.slug for organization in organizations] == ["first", "second"] + assert cursors == [None, "next-page"] + + +@pytest.mark.parametrize("count", [0, 1, 3]) +async def test_organization_memberships_preserve_zero_one_and_many(count: int) -> None: + organizations = [ + {"id": f"org-{index}", "name": f"Org {index}", "slug": f"org-{index}"} + for index in range(count) + ] + async with HorizonClient( + api_key="fmcp_secret", + transport=mock_transport( + lambda request: httpx2.Response( + 200, + json={ + "items": organizations, + "meta": {"nextCursor": None, "limit": 100}, + }, + ) + ), + ) as client: + result = await client.list_organizations() + + assert len(result) == count + + +async def test_revoke_uses_the_current_authenticated_key() -> None: + def handler(request: httpx2.Request) -> httpx2.Response: + assert request.method == "DELETE" + assert request.url.path == "/api/v0/me/api-key" + assert request.headers["authorization"] == "Bearer fmcp_current" + return httpx2.Response(204) + + async with HorizonClient( + api_key="fmcp_current", + transport=mock_transport(handler), + ) as client: + await client.revoke_current_api_key() + + +async def test_protected_routes_require_a_credential() -> None: + async with HorizonClient( + transport=mock_transport(lambda request: httpx2.Response(200)) + ) as client: + with pytest.raises(HorizonUnauthorizedError): + await client.get_current_user() + + +async def test_protected_routes_report_a_rejected_credential() -> None: + async with HorizonClient( + api_key="fmcp_invalid", + transport=mock_transport(lambda request: httpx2.Response(401)), + ) as client: + with pytest.raises(HorizonUnauthorizedError): + await client.get_current_user() + + +async def test_public_routes_do_not_report_a_missing_credential() -> None: + async with HorizonClient( + transport=mock_transport(lambda request: httpx2.Response(401)) + ) as client: + with pytest.raises(HorizonResponseError): + await client.create_device_authorization() + + +async def test_invalid_responses_do_not_include_response_bodies() -> None: + secret_body = "fmcp_response_secret" + async with HorizonClient( + transport=mock_transport(lambda request: httpx2.Response(500, text=secret_body)) + ) as client: + with pytest.raises(HorizonResponseError) as exc_info: + await client.create_device_authorization() + + assert exc_info.value.status_code == 500 + assert secret_body not in str(exc_info.value) + + +@pytest.mark.parametrize( + "value", + [ + "ftp://horizon.prefect.io", + "https://user@example.com", + "https://horizon.prefect.io/path", + "https://horizon.prefect.io?query=value", + "https://horizon.prefect.io:abc", + "https://horizon.prefect.io:99999", + ], +) +def test_api_origin_rejects_values_that_are_not_origins(value: str) -> None: + with pytest.raises(ValueError): + normalize_api_origin(value) + + +def test_api_origin_normalizes_one_trailing_slash() -> None: + assert ( + normalize_api_origin("https://horizon.prefect.io/") + == "https://horizon.prefect.io" + )