82 lines
3.6 KiB
Python
82 lines
3.6 KiB
Python
"""Refresh tokens are bound to the authorization-server issuer that granted them.
|
|
|
|
The discovered authorization server for an MCP server can change (server migration, protected-resource
|
|
metadata edit, DNS takeover). A stored refresh token must never be sent to a different issuer.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip("mcp")
|
|
|
|
from mcp.shared.auth import OAuthToken # noqa: E402
|
|
|
|
from tools.mcp_oauth import HermesTokenStorage # noqa: E402
|
|
from tools.mcp_oauth_provider import bind_issuer_from_context, enforce_refresh_token_issuer # noqa: E402
|
|
|
|
|
|
def _token_file(tmp_path):
|
|
return tmp_path / "mcp-tokens" / "srv.json"
|
|
|
|
|
|
def _stored(tmp_path, issuer):
|
|
payload = {"access_token": "a", "token_type": "Bearer", "expires_in": 3600, "refresh_token": "r"}
|
|
if issuer is not None:
|
|
payload["hermes_issuer"] = issuer
|
|
_token_file(tmp_path).parent.mkdir(parents=True, exist_ok=True)
|
|
_token_file(tmp_path).write_text(json.dumps(payload))
|
|
storage = HermesTokenStorage("srv", hermes_home=tmp_path)
|
|
tokens = asyncio.run(storage.get_tokens())
|
|
assert tokens is not None
|
|
return storage, tokens
|
|
|
|
|
|
def _context(storage, issuer, tokens=None):
|
|
meta = SimpleNamespace(issuer=issuer) if issuer is not None else None
|
|
return SimpleNamespace(storage=storage, oauth_metadata=meta, current_tokens=tokens)
|
|
|
|
|
|
def test_issuer_mismatch_strips_refresh_token_in_memory_and_on_disk(tmp_path):
|
|
storage, tokens = _stored(tmp_path, "https://old.example.com")
|
|
ctx = _context(storage, "https://evil.example.com", tokens)
|
|
enforce_refresh_token_issuer(ctx)
|
|
assert ctx.current_tokens.refresh_token is None
|
|
assert ctx.current_tokens.access_token == "a" # unexpired access token stays usable
|
|
on_disk = json.loads(_token_file(tmp_path).read_text())
|
|
assert "refresh_token" not in on_disk and on_disk["access_token"] == "a"
|
|
|
|
|
|
def test_matching_issuer_keeps_refresh_token_even_with_trailing_slash(tmp_path):
|
|
storage, tokens = _stored(tmp_path, "https://as.example.com/")
|
|
ctx = _context(storage, "https://as.example.com", tokens)
|
|
enforce_refresh_token_issuer(ctx)
|
|
assert ctx.current_tokens.refresh_token == "r"
|
|
assert json.loads(_token_file(tmp_path).read_text())["refresh_token"] == "r"
|
|
|
|
|
|
def test_legacy_file_without_issuer_adopts_current_issuer_once(tmp_path):
|
|
storage, tokens = _stored(tmp_path, None)
|
|
enforce_refresh_token_issuer(_context(storage, "https://as.example.com", tokens))
|
|
assert tokens.refresh_token == "r"
|
|
assert json.loads(_token_file(tmp_path).read_text())["hermes_issuer"] == "https://as.example.com"
|
|
# Now bound: a later issuer change is rejected.
|
|
storage2 = HermesTokenStorage("srv", hermes_home=tmp_path)
|
|
tokens2 = asyncio.run(storage2.get_tokens())
|
|
assert tokens2 is not None
|
|
enforce_refresh_token_issuer(_context(storage2, "https://evil.example.com", tokens2))
|
|
assert tokens2.refresh_token is None
|
|
|
|
|
|
def test_set_tokens_stamps_bound_issuer_and_get_tokens_keeps_it_out_of_the_sdk_model(tmp_path):
|
|
storage = HermesTokenStorage("srv", hermes_home=tmp_path)
|
|
bind_issuer_from_context(_context(storage, "https://as.example.com"))
|
|
asyncio.run(storage.set_tokens(OAuthToken(access_token="a", token_type="Bearer", expires_in=3600, refresh_token="r")))
|
|
assert json.loads(_token_file(tmp_path).read_text())["hermes_issuer"] == "https://as.example.com"
|
|
fresh = HermesTokenStorage("srv", hermes_home=tmp_path)
|
|
tokens = asyncio.run(fresh.get_tokens())
|
|
assert fresh.loaded_issuer == "https://as.example.com"
|
|
assert not hasattr(tokens, "hermes_issuer") # never leaks into the wire model
|