import secrets
import time
import unittest
from unittest.mock import patch
import jwt
from fastmcp import Client
from fastmcp.exceptions import ToolError
from fastmcp.server.auth import AccessToken
from fastmcp.server.auth.providers.jwt import JWTVerifier
import readonly_server as app
class ReadOnlyTests(unittest.IsolatedAsyncioTestCase):
def setUp(self):
self.secret = secrets.token_urlsafe(48)
self.verifier = JWTVerifier(
public_key=self.secret, algorithm="HS256",
issuer="test-issuer", audience="test-mcp",
)
self.server = app.build_server(self.verifier)
def identity(self, subject="alice", scopes=None):
return AccessToken(
token="synthetic-test-only", client_id="test-client",
scopes=["docs:read"] if scopes is None else scopes,
claims={"sub": subject},
)
async def call(self, arguments, identity):
with patch.object(app, "get_access_token", return_value=identity):
async with Client(self.server) as client:
return await client.call_tool("get_document", arguments)
async def denied(self, arguments, identity):
with self.assertRaises(ToolError) as caught:
await self.call(arguments, identity)
return str(caught.exception)
async def test_schema_and_readonly_hint(self):
async with Client(self.server) as client:
tools = await client.list_tools()
self.assertEqual([t.name for t in tools], ["get_document"])
self.assertEqual(set(tools[0].input_schema["properties"]), {"doc_id"})
self.assertTrue(tools[0].annotations.read_only_hint)
self.assertFalse(tools[0].annotations.destructive_hint)
async def test_own_document(self):
result = await self.call({"doc_id": "a-001"}, self.identity())
self.assertEqual(result.data,
{"doc_id": "a-001", "title": "A team onboarding"})
async def test_cross_tenant_and_missing_share_error(self):
foreign = await self.denied({"doc_id": "b-001"}, self.identity())
missing = await self.denied({"doc_id": "a-999"}, self.identity())
self.assertEqual(foreign, missing)
self.assertNotIn("B private handbook", foreign)
async def test_missing_identity(self):
await self.denied({"doc_id": "a-001"}, None)
async def test_missing_scope(self):
await self.denied({"doc_id": "a-001"}, self.identity(scopes=[]))
async def test_unknown_member(self):
await self.denied({"doc_id": "a-001"}, self.identity(subject="eve"))
async def test_invalid_inputs_and_tenant_injection(self):
for args in [{}, {"doc_id": 1}, {"doc_id": "a-001-extra"},
{"doc_id": "../../x"}, {"doc_id": "c-001"},
{"doc_id": "a-001", "tenant_id": "tenant-b"}]:
with self.subTest(args=args):
await self.denied(args, self.identity())
async def test_repeat_calls_leave_fixture_unchanged(self):
before = tuple(app.DOCUMENTS)
first = await self.call({"doc_id": "a-001"}, self.identity())
second = await self.call({"doc_id": "a-001"}, self.identity())
self.assertEqual(first.data, second.data)
self.assertEqual(tuple(app.DOCUMENTS), before)
async def test_native_jwt_verifier(self):
now = int(time.time())
claims = {"sub": "alice", "iss": "test-issuer", "aud": "test-mcp",
"exp": now + 300, "scope": "docs:read"}
signed = jwt.encode(claims, self.secret, algorithm="HS256")
verified = await self.verifier.verify_token(signed)
self.assertIsNotNone(verified)
self.assertEqual(verified.claims["sub"], "alice")
self.assertIn("docs:read", verified.scopes)
invalid = [jwt.encode(claims, secrets.token_urlsafe(48), algorithm="HS256")]
for change in [{"iss": "other"}, {"aud": "other"}, {"exp": now - 300}]:
invalid.append(jwt.encode(claims | change, self.secret, algorithm="HS256"))
for value in invalid:
with self.subTest(kind="invalid token"):
self.assertIsNone(await self.verifier.verify_token(value))
if __name__ == "__main__":
unittest.main(verbosity=2)
