79 lines
4.5 KiB
Python
79 lines
4.5 KiB
Python
|
"""Browser PAT lifecycle and authentication boundaries, with no upstream calls."""
|
||
|
|
|
||
|
|
from datetime import timedelta
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
import pytest
|
||
|
|
from fastapi import HTTPException
|
||
|
|
from sqlalchemy import select
|
||
|
|
|
||
|
|
from app.mcp_api.auth import authenticate
|
||
|
|
from app.models import Account, MCPToken, now
|
||
|
|
|
||
|
|
|
||
|
|
async def test_token_browser_lifecycle(app, logged_in):
|
||
|
|
client = logged_in
|
||
|
|
# Administration remains available when the MCP transport is disabled.
|
||
|
|
initial = (await client.get("/api/v1/mcp-tokens")).json()
|
||
|
|
assert initial["enabled"] is False and initial["can_create"] is False
|
||
|
|
assert (await client.post("/api/v1/mcp-tokens", json={"name": "reader"})).status_code == 422
|
||
|
|
async with app.state.sessions.begin() as db:
|
||
|
|
account = await db.get(Account, 1)
|
||
|
|
account.wq_user_id = "synthetic-user"
|
||
|
|
issued = await client.post("/api/v1/mcp-tokens", json={"name": " reader "})
|
||
|
|
assert issued.status_code == 201
|
||
|
|
assert issued.headers["cache-control"] == "no-store"
|
||
|
|
token = issued.json()
|
||
|
|
assert token["name"] == "reader" and token["scopes"] == ["research:read"]
|
||
|
|
assert token["status"] == "active" and token["token"].startswith("wqmcp_")
|
||
|
|
async with app.state.sessions() as db:
|
||
|
|
principal = await authenticate(db, token["token"])
|
||
|
|
row = await db.get(MCPToken, principal.token_id)
|
||
|
|
assert row.token_hash != token["token"]
|
||
|
|
assert 89 < (row.expires_at.replace(tzinfo=now().tzinfo) - now()).total_seconds() / 86400 <= 90
|
||
|
|
page = await client.get("/api/v1/mcp-tokens?limit=1")
|
||
|
|
assert token["token"] not in page.text and "token_hash" not in page.text
|
||
|
|
assert page.json()["items"][0]["id"] == token["id"]
|
||
|
|
second = await client.post("/api/v1/mcp-tokens", json={"name": "executor", "days": 1, "scopes": ["research:read", "backtests:execute"]})
|
||
|
|
assert second.status_code == 201
|
||
|
|
page = (await client.get("/api/v1/mcp-tokens?limit=1")).json()
|
||
|
|
assert page["total"] == 2 and page["has_more"]
|
||
|
|
assert (await client.get("/api/v1/mcp-tokens?limit=1&offset=1")).json()["items"][0]["id"] == token["id"]
|
||
|
|
revoked = await client.post(f'/api/v1/mcp-tokens/{token["id"]}/revoke')
|
||
|
|
again = await client.post(f'/api/v1/mcp-tokens/{token["id"]}/revoke')
|
||
|
|
assert revoked.json() == again.json() and revoked.json()["status"] == "revoked"
|
||
|
|
assert "token" not in revoked.json()
|
||
|
|
async with app.state.sessions() as db:
|
||
|
|
with pytest.raises(HTTPException) as exc:
|
||
|
|
await authenticate(db, token["token"])
|
||
|
|
assert exc.value.status_code == 401
|
||
|
|
assert (await client.post("/api/v1/mcp-tokens/missing/revoke")).status_code == 404
|
||
|
|
|
||
|
|
|
||
|
|
async def test_token_browser_security_and_validation(app, logged_in):
|
||
|
|
client = logged_in
|
||
|
|
async with app.state.sessions.begin() as db:
|
||
|
|
(await db.get(Account, 1)).wq_user_id = "synthetic-user"
|
||
|
|
for body in [
|
||
|
|
{"name": " "}, {"name": "x", "days": 0}, {"name": "x", "days": 366},
|
||
|
|
{"name": "x", "days": 1.5}, {"name": "x", "scopes": ["backtests:execute"]},
|
||
|
|
{"name": "x", "scopes": ["research:read", "admin"]}, {"name": "x", "account_id": 2},
|
||
|
|
]:
|
||
|
|
assert (await client.post("/api/v1/mcp-tokens", json=body)).status_code == 422
|
||
|
|
assert (await client.post("/api/v1/mcp-tokens", json={"name": "x"}, headers={"X-WQ-Request": ""})).status_code == 403
|
||
|
|
assert (await client.post("/api/v1/mcp-tokens", json={"name": "x"}, headers={"Origin": "https://evil.test"})).status_code == 403
|
||
|
|
token = (await client.post("/api/v1/mcp-tokens", json={"name": "x"})).json()
|
||
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver") as outsider:
|
||
|
|
headers = {"Authorization": f'Bearer {token["token"]}', "X-WQ-Request": "1"}
|
||
|
|
assert (await outsider.get("/api/v1/mcp-tokens")).status_code == 401
|
||
|
|
assert (await outsider.get("/api/v1/mcp-tokens", headers=headers)).status_code == 401
|
||
|
|
assert (await outsider.post("/api/v1/mcp-tokens", json={"name": "escalation"}, headers=headers)).status_code == 401
|
||
|
|
assert (await outsider.post(f'/api/v1/mcp-tokens/{token["id"]}/revoke', headers=headers)).status_code == 401
|
||
|
|
async with app.state.sessions.begin() as db:
|
||
|
|
(await db.get(Account, 1)).wq_user_id = "different-user"
|
||
|
|
assert (await client.get("/api/v1/mcp-tokens")).json()["items"][0]["status"] == "invalid_binding"
|
||
|
|
async with app.state.sessions.begin() as db:
|
||
|
|
row = await db.scalar(select(MCPToken))
|
||
|
|
row.expires_at = now() - timedelta(days=1)
|
||
|
|
assert (await client.get("/api/v1/mcp-tokens")).json()["items"][0]["status"] == "expired"
|