Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions src/late/mcp/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,8 +66,16 @@ def is_allowed_origin(request: Request) -> bool:

logger = logging.getLogger(__name__)

_VERIFY_URL = "https://zernio.com/api/v1/accounts"
_VERIFY_TIMEOUT = 5.0
# /v1/auth/verify authenticates and returns, nothing else. The old target,
# /v1/accounts, answered the same question by running a full account listing
# (team resolution, ads status, counts): during the 2026-08-03 API degradation
# 29% of those calls crossed the timeout below, and every one of them reached a
# user as "your token is invalid, clear it and re-register". Overridable so a
# bad endpoint is an env-var flip on Railway, not a redeploy.
_VERIFY_URL = os.getenv("MCP_VERIFY_URL", "https://zernio.com/api/v1/auth/verify")
# Generous because a slow answer is still an answer: only an exception here
# costs the caller its verdict.
_VERIFY_TIMEOUT = 10.0

# Positive-only verification cache: sha256(token) -> monotonic timestamp of the
# last upstream confirmation. Positives only, so an attacker cannot grow it by
Expand Down
19 changes: 19 additions & 0 deletions tests/test_mcp_auth_verification.py
Original file line number Diff line number Diff line change
Expand Up @@ -166,3 +166,22 @@ def test_verification_cache_is_bounded():
# Eviction is LRU, so the oldest inserts are the ones that went.
assert "token-0" not in auth._VERIFIED_AT
assert f"token-{auth._VERIFIED_CACHE_MAX + 49}" in auth._VERIFIED_AT


async def test_verification_targets_the_auth_only_endpoint():
"""A data endpoint would make every MCP request pay for a listing it
discards, which is what pushed verification past the client timeout during
the 2026-08-03 API degradation."""
seen: list[str] = []

def handler(request: httpx.Request) -> httpx.Response:
seen.append(str(request.url))
return httpx.Response(200)

client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
verifier = ZernioTokenVerifier(client=client)

await verifier.verify_token("some-api-key")

assert seen == ["https://zernio.com/api/v1/auth/verify"]
await client.aclose()
Loading