Add Prefect Horizon authentication client and local state (#4785)

* feat: add Horizon authentication client and state * fix: apply Windows state ACLs to existing descriptors * fix: distinguish public route authorization failures * fix: harden Horizon state boundaries
2026-08-28 18:22:07 +02:00 · 2026-08-10 14:06:42 -07:00 · 2026-08-10 14:06:42 -07:00 · bba8c44f7b
commit bba8c44f7b
parent 6475650fc7
10 changed files with 1751 additions and 0 deletions
--- a/tests/cli/deploy/test_authentication.py
+++ 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"
--- a/tests/cli/deploy/test_configuration.py
+++ 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
--- a/tests/cli/deploy/test_credentials.py
+++ 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
--- a/tests/cli/deploy/test_horizon_client.py
+++ 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"
+    )