103 lines
5.3 KiB
Python
103 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import json
|
|
import secrets
|
|
import unittest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import httpx
|
|
|
|
os.environ.setdefault("GITEA_TOKEN", "test-token")
|
|
os.environ.setdefault("GITEA_OAUTH_CLIENT_ID", "test-client")
|
|
os.environ.setdefault("GITEA_OAUTH_CLIENT_SECRET", "test-secret")
|
|
os.environ.setdefault("JWT_SIGNING_KEY", "test-signing-key-for-unit-tests")
|
|
|
|
from tea_mcp import server
|
|
from tea_mcp.persimmons import create_connection, create_juice
|
|
from fastmcp import Client, FastMCP
|
|
from tests.test_persimmons import FakeGitea
|
|
|
|
|
|
class JuiceHttpTests(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
self.client = httpx.AsyncClient(transport=httpx.ASGITransport(app=server.app), base_url="https://example.invalid")
|
|
|
|
async def asyncTearDown(self):
|
|
await self.client.aclose()
|
|
|
|
async def test_unauthorized_routes_do_not_access_storage_or_duplicate_proxy_cors(self):
|
|
with patch.object(server, "persimmons_actor", AsyncMock(return_value=None)), patch.object(server, "persimmons_requester") as requester:
|
|
for operation, method in (("list", "GET"), ("create", "POST"), ("revoke", "POST")):
|
|
response = await self.client.request(method, "/chat/persimmons/juice/" + operation,
|
|
headers={"Origin": "https://persimmons.app"})
|
|
self.assertEqual(response.status_code, 401)
|
|
self.assertEqual(response.json(), {"error": "unauthorized"})
|
|
self.assertEqual(response.headers["cache-control"], "no-store")
|
|
self.assertNotIn("access-control-allow-origin", response.headers)
|
|
requester.assert_not_called()
|
|
|
|
async def test_authenticated_lifecycle_and_validation(self):
|
|
store = FakeGitea()
|
|
source = await create_connection(store.request, {
|
|
"provider": "Example", "credential_type": "Token", "credential": "hidden-source",
|
|
"vault_kind": "user",
|
|
})
|
|
with patch.object(server, "persimmons_actor", AsyncMock(return_value="example-user")), patch.object(server, "persimmons_requester", return_value=store.request):
|
|
response = await self.client.post("/chat/persimmons/juice/create", json={
|
|
"source_id": source["id"], "subject": "agent:example", "scopes": ["read"],
|
|
})
|
|
self.assertEqual(response.status_code, 201)
|
|
juice = response.json()
|
|
listed = await self.client.get("/chat/persimmons/juice/list", params={"source_id": source["id"]})
|
|
self.assertEqual(listed.json(), {"juice": [juice]})
|
|
revoked = await self.client.post("/chat/persimmons/juice/revoke", json={"id": juice["id"]})
|
|
self.assertEqual(revoked.status_code, 200)
|
|
self.assertEqual(revoked.json()["status"], "revoked")
|
|
self.assertNotIn("hidden-source", response.text + listed.text + revoked.text)
|
|
invalid = await self.client.post("/chat/persimmons/juice/create", json=[])
|
|
self.assertEqual(invalid.status_code, 400)
|
|
|
|
async def test_storage_errors_do_not_leak_details(self):
|
|
with patch.object(server, "persimmons_actor", AsyncMock(return_value="example-user")), patch.object(server, "persimmons_requester", return_value=AsyncMock(side_effect=RuntimeError("private detail"))):
|
|
response = await self.client.get("/chat/persimmons/juice/list")
|
|
self.assertEqual(response.status_code, 502)
|
|
self.assertNotIn("private detail", response.text)
|
|
|
|
|
|
class JuiceMcpTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_registered_agent_route_never_reads_or_returns_source_value(self):
|
|
store = FakeGitea()
|
|
sentinel = secrets.token_hex(24)
|
|
source = await create_connection(store.request, {
|
|
"provider": "Example", "credential_type": "Token", "credential": sentinel,
|
|
"vault_kind": "user",
|
|
})
|
|
juice = await create_juice(store.request, {
|
|
"source_id": source["id"], "subject": "agent:example", "scopes": ["read"],
|
|
})
|
|
key = "PERSIM_JUICE_" + juice["id"]
|
|
poisoned = json.loads(store.user_variables[key])
|
|
poisoned["credential"] = sentinel
|
|
store.user_variables[key] = json.dumps(poisoned)
|
|
calls = []
|
|
|
|
async def requester(method, path, **kwargs):
|
|
calls.append((method, path))
|
|
self.assertEqual(method, "GET")
|
|
self.assertNotIn("/secrets", path)
|
|
return await store.request(method, path, **kwargs)
|
|
|
|
# Exercise the registered tool through MCP without mounting unrelated
|
|
# runtime proxies or starting the production application's background jobs.
|
|
isolated = FastMCP("Juice route test")
|
|
isolated.add_tool(await server.mcp.local_provider.get_tool("persimmons_juice_list"))
|
|
with patch.object(server, "actor_gitea_request", requester):
|
|
async with Client(isolated) as client:
|
|
result = await client.call_tool("persimmons_juice_list", {"source_id": source["id"]})
|
|
self.assertFalse(result.is_error)
|
|
self.assertEqual(result.data, {"juice": [juice]})
|
|
# Use a boolean assertion so a regression cannot echo the sentinel in logs.
|
|
self.assertTrue(sentinel not in str(result))
|
|
self.assertTrue(calls)
|