Melvin Strobl
6d2e25c2ca
The engine now speaks MCP at /mcp, with a built-in OAuth 2.1 authorization server in front of it: an agent registers itself, sends a human to the browser to approve it, and exchanges the resulting code for a token. PKCE is required, codes are single-use and stored only as hashes, the browser is redirected to the URI that was registered rather than the one asked for, and refresh tokens rotate so that replaying a spent one revokes the whole line. Twenty tools cover reading, building, publishing and running flows, and each one calls the same REST endpoint the dashboard calls, in-process, carrying the caller's own token. That keeps one description of what a flow is and what may be done to it — validation, the draft/publish split, the version check — and means an agent can do nothing a person could not do in the browser. Agent tokens are RS256 with a keypair of their own rather than the secret that signs browser sessions, so deleting the key withdraws every agent without logging anyone out, and deps.decode_token grew the branch that trusting a second issuer will need when the hosted login arrives.
227 lines
7.4 KiB
Python
227 lines
7.4 KiB
Python
"""The authorization flow an agent goes through to reach the MCP endpoint."""
|
|
|
|
import base64
|
|
import hashlib
|
|
import secrets
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.config import settings
|
|
|
|
PREFIX = f"{settings.API_V1_STR}/oauth"
|
|
REDIRECT = "http://127.0.0.1:41234/callback"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mcp_on(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""The OAuth endpoints only answer when the MCP endpoint is wanted."""
|
|
monkeypatch.setattr(settings, "MCP_ENABLED", True)
|
|
|
|
|
|
def pkce() -> tuple[str, str]:
|
|
verifier = secrets.token_urlsafe(48)
|
|
digest = hashlib.sha256(verifier.encode("ascii")).digest()
|
|
challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
|
|
return verifier, challenge
|
|
|
|
|
|
def register(client: TestClient, **overrides) -> dict:
|
|
body = {"client_name": "Test agent", "redirect_uris": [REDIRECT], **overrides}
|
|
return client.post(f"{PREFIX}/register", json=body).json()
|
|
|
|
|
|
def approve(
|
|
client: TestClient, headers: dict[str, str], client_id: str, challenge: str
|
|
) -> str:
|
|
"""Walk the consent step and return the code it hands back."""
|
|
response = client.post(
|
|
f"{PREFIX}/authorize",
|
|
headers=headers,
|
|
json={
|
|
"client_id": client_id,
|
|
"redirect_uri": REDIRECT,
|
|
"code_challenge": challenge,
|
|
"code_challenge_method": "S256",
|
|
"resource": settings.mcp_resource,
|
|
},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
url = response.json()["redirect_url"]
|
|
return parse_qs(urlparse(url).query)["code"][0]
|
|
|
|
|
|
def exchange(client: TestClient, client_id: str, code: str, verifier: str):
|
|
return client.post(
|
|
f"{PREFIX}/token",
|
|
data={
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": REDIRECT,
|
|
"client_id": client_id,
|
|
"code_verifier": verifier,
|
|
},
|
|
)
|
|
|
|
|
|
def test_the_metadata_says_how_to_get_in(client: TestClient) -> None:
|
|
document = client.get("/.well-known/oauth-authorization-server").json()
|
|
|
|
assert document["issuer"] == settings.oauth_issuer
|
|
assert document["code_challenge_methods_supported"] == ["S256"]
|
|
assert document["token_endpoint_auth_methods_supported"] == ["none"]
|
|
assert set(document["grant_types_supported"]) == {
|
|
"authorization_code",
|
|
"refresh_token",
|
|
}
|
|
|
|
|
|
def test_a_full_authorization_reaches_the_api(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
registered = register(client)
|
|
verifier, challenge = pkce()
|
|
code = approve(client, superuser_token_headers, registered["client_id"], challenge)
|
|
|
|
tokens = exchange(client, registered["client_id"], code, verifier).json()
|
|
assert tokens["token_type"] == "Bearer"
|
|
assert tokens["scope"] == "mcp"
|
|
|
|
# The token works against the ordinary API, as the user who approved it.
|
|
whoami = client.post(
|
|
f"{settings.API_V1_STR}/login/test-token",
|
|
headers={"Authorization": f"Bearer {tokens['access_token']}"},
|
|
)
|
|
assert whoami.status_code == 200
|
|
|
|
|
|
def test_a_wrong_verifier_is_refused(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
registered = register(client)
|
|
_, challenge = pkce()
|
|
code = approve(client, superuser_token_headers, registered["client_id"], challenge)
|
|
|
|
response = exchange(client, registered["client_id"], code, secrets.token_urlsafe(48))
|
|
assert response.status_code == 400
|
|
assert response.json()["error"] == "invalid_grant"
|
|
|
|
|
|
def test_replaying_a_code_withdraws_what_it_produced(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
registered = register(client)
|
|
verifier, challenge = pkce()
|
|
code = approve(client, superuser_token_headers, registered["client_id"], challenge)
|
|
first = exchange(client, registered["client_id"], code, verifier).json()
|
|
|
|
replay = exchange(client, registered["client_id"], code, verifier)
|
|
assert replay.status_code == 400
|
|
assert replay.json()["error"] == "invalid_grant"
|
|
|
|
# The refresh token the first exchange handed out is no longer any good:
|
|
# the replay says someone else may be holding a copy.
|
|
refreshed = client.post(
|
|
f"{PREFIX}/token",
|
|
data={
|
|
"grant_type": "refresh_token",
|
|
"refresh_token": first["refresh_token"],
|
|
"client_id": registered["client_id"],
|
|
},
|
|
)
|
|
assert refreshed.status_code == 400
|
|
|
|
|
|
def test_a_refresh_token_rotates_and_cannot_be_reused(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
registered = register(client)
|
|
verifier, challenge = pkce()
|
|
code = approve(client, superuser_token_headers, registered["client_id"], challenge)
|
|
first = exchange(client, registered["client_id"], code, verifier).json()
|
|
|
|
refresh = lambda token: client.post( # noqa: E731
|
|
f"{PREFIX}/token",
|
|
data={
|
|
"grant_type": "refresh_token",
|
|
"refresh_token": token,
|
|
"client_id": registered["client_id"],
|
|
},
|
|
)
|
|
|
|
second = refresh(first["refresh_token"])
|
|
assert second.status_code == 200
|
|
assert second.json()["refresh_token"] != first["refresh_token"]
|
|
|
|
# Presenting the spent one again is what a stolen token looks like.
|
|
assert refresh(first["refresh_token"]).status_code == 400
|
|
assert refresh(second.json()["refresh_token"]).status_code == 400
|
|
|
|
|
|
def test_an_unregistered_redirect_is_refused(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
registered = register(client)
|
|
_, challenge = pkce()
|
|
|
|
response = client.post(
|
|
f"{PREFIX}/authorize",
|
|
headers=superuser_token_headers,
|
|
json={
|
|
"client_id": registered["client_id"],
|
|
"redirect_uri": "https://somewhere.else/callback",
|
|
"code_challenge": challenge,
|
|
"code_challenge_method": "S256",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["error"] == "invalid_request"
|
|
|
|
|
|
def test_registration_refuses_plain_http_off_the_loopback(
|
|
client: TestClient,
|
|
) -> None:
|
|
response = client.post(
|
|
f"{PREFIX}/register",
|
|
json={"client_name": "Nope", "redirect_uris": ["http://example.com/cb"]},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["error"] == "invalid_redirect_uri"
|
|
|
|
|
|
def test_approving_needs_a_signed_in_user(client: TestClient) -> None:
|
|
registered = register(client)
|
|
_, challenge = pkce()
|
|
|
|
response = client.post(
|
|
f"{PREFIX}/authorize",
|
|
json={
|
|
"client_id": registered["client_id"],
|
|
"redirect_uri": REDIRECT,
|
|
"code_challenge": challenge,
|
|
"code_challenge_method": "S256",
|
|
},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_an_unsupported_grant_says_so(client: TestClient) -> None:
|
|
response = client.post(
|
|
f"{PREFIX}/token", data={"grant_type": "client_credentials"}
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["error"] == "unsupported_grant_type"
|
|
|
|
|
|
def test_everything_is_refused_while_mcp_is_off(
|
|
client: TestClient, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(settings, "MCP_ENABLED", False)
|
|
|
|
assert client.post(f"{PREFIX}/register", json={"redirect_uris": [REDIRECT]}).status_code == 403
|
|
assert (
|
|
client.post(f"{PREFIX}/token", data={"grant_type": "authorization_code"}).status_code
|
|
== 403
|
|
)
|