diff --git a/tests/test_human_vault.py b/tests/test_human_vault.py new file mode 100644 index 0000000..cb59747 --- /dev/null +++ b/tests/test_human_vault.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import base64 +import json +import os +from pathlib import Path +import subprocess +import tempfile +import unittest +from unittest.mock import AsyncMock, patch + +import httpx +from starlette.applications import Starlette +from tea_mcp import human_vault as vault + + +class HumanVaultTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.temp = tempfile.TemporaryDirectory() + self.path = Path(self.temp.name) / 'vault.sqlite' + self.env = patch.dict(os.environ, PERSIMMONS_HUMAN_VAULT_DB=str(self.path)) + self.env.start() + self.actor = AsyncMock(return_value='fixture') + self.app = Starlette(routes=vault.routes(self.actor)) + self.client = httpx.AsyncClient(transport=httpx.ASGITransport(app=self.app), base_url='https://teahub.xyz', + headers={'Origin': vault.EXTENSION_ORIGIN, 'Cookie': 'session=synthetic', 'X-Vault-Owner': 'fixture'}) + # A real Web Crypto ciphertext from the extension, not fake base64 plaintext. + script = """ +import {initialize,unlock,encryptItem} from './persimmons-extension/human-crypto.mjs'; +const r=await initialize('disposable passphrase','fixture'); +const k=await unlock('disposable passphrase','fixture',r); +process.stdout.write(JSON.stringify(await encryptItem(k,'fixture','https://wp.phly.cloud', + {username:'DISPOSABLE_USER_SENTINEL',password:'DISPOSABLE_PASSWORD_SENTINEL'}))); +""" + self.item = json.loads(subprocess.check_output(['node', '--input-type=module', '-e', script])) + + async def asyncTearDown(self): + await self.client.aclose() + self.env.stop() + self.temp.cleanup() + + async def test_crud_persistence_and_owner_isolation(self): + item = self.item + url = vault.PREFIX + '/' + item['id'] + response = await self.client.post(vault.PREFIX, json=item) + self.assertEqual(response.status_code, 201) + listed = await self.client.get(vault.PREFIX) + self.assertEqual(listed.json()['items'], [{k: item[k] for k in ('v', 'id', 'origin', 'username')}]) + self.assertEqual((await self.client.get(url)).json()['item'], item) + # Re-open storage independently; no in-memory substitute. + self.assertEqual(vault.Store(self.path).operate('fixture', 'GET', item['id'])[0]['item'], item) + disk = self.path.read_bytes() + for forbidden in (b'DISPOSABLE_PASSWORD_SENTINEL', b'passphrase', b'wrappedKey', b'PBKDF2'): + self.assertNotIn(forbidden, disk) + self.assertEqual((await self.client.post(vault.PREFIX, json=item)).status_code, 409) + update = {**item, 'iv': base64.b64encode(bytes(range(12))).decode()} + self.assertEqual((await self.client.put(url, json=update)).json()['item'], update) + self.actor.return_value = 'other' + self.assertEqual((await self.client.get(url)).status_code, 409) + for method in ('GET', 'PUT', 'DELETE'): + result = await self.client.request(method, url, headers={'X-Vault-Owner': 'other'}, **({'json': update} if method == 'PUT' else {})) + self.assertEqual(result.status_code, 404) + self.assertEqual((await self.client.get(vault.PREFIX)).json()['items'], []) + self.actor.return_value = 'fixture' + self.assertEqual((await self.client.delete(url)).status_code, 200) + self.assertEqual((await self.client.get(url)).status_code, 404) + + async def test_exact_origin_https_cookie_and_preflight(self): + for origin in ('', 'null', 'https://wp.phly.cloud', 'https://teahub.xyz', vault.EXTENSION_ORIGIN + '/', 'chrome-extension://' + 'a' * 32): + for method in ('GET', 'POST', 'OPTIONS'): + response = await self.client.request(method, vault.PREFIX, headers={'Origin': origin}, json=self.item) + self.assertEqual(response.status_code, 403) + self.assertNotIn('access-control-allow-origin', response.headers) + self.actor.assert_not_awaited() + response = await self.client.get('http://teahub.xyz' + vault.PREFIX) + self.assertEqual(response.status_code, 403) + for headers in ({'Cookie': ''}, {'Authorization': 'Bearer synthetic'}): + self.assertEqual((await self.client.get(vault.PREFIX, headers=headers)).status_code, 401) + self.actor.assert_not_awaited() + self.actor.return_value = None + self.assertEqual((await self.client.get(vault.PREFIX)).status_code, 401) + preflight = await self.client.options(vault.PREFIX, headers={'Cookie': '', 'Access-Control-Request-Method': 'PUT', 'Access-Control-Request-Headers': 'content-type,x-vault-owner'}) + self.assertEqual(preflight.status_code, 204) + self.assertEqual(preflight.headers['access-control-allow-origin'], vault.EXTENSION_ORIGIN) + self.assertEqual(preflight.headers['access-control-allow-credentials'], 'true') + self.assertEqual(preflight.headers['cache-control'], 'no-store') + self.assertFalse(self.path.exists()) + + async def test_validation_and_no_keys_or_plaintext_fields(self): + for extra in ('password', 'key', 'passphrase', 'wrappedKey'): + response = await self.client.post(vault.PREFIX, json={**self.item, extra: 'synthetic'}) + self.assertEqual(response.status_code, 400) + self.assertNotIn('synthetic', response.text) + for item in ({}, [], {**self.item, 'origin': 'https://wp.phly.cloud/path'}, + {**self.item, 'username': ''}, {**self.item, 'username': 'bad\nname'}, + {**self.item, 'iv': 'bad'}, {**self.item, 'v': True}): + self.assertEqual((await self.client.post(vault.PREFIX, json=item)).status_code, 400) + self.assertEqual((await self.client.get(vault.PREFIX + '?password=synthetic')).status_code, 400) + self.assertEqual((await self.client.post(vault.PREFIX, content='x' * 25000, headers={'Content-Type': 'application/json'})).status_code, 400) + self.assertFalse(self.path.exists()) + with patch.dict(os.environ, PERSIMMONS_HUMAN_VAULT_DB=''): + self.assertEqual((await self.client.get(vault.PREFIX)).status_code, 503) + + +class HumanVaultMcpTests(unittest.IsolatedAsyncioTestCase): + async def test_registered_mcp_surface_is_unchanged_and_cannot_fetch_vault(self): + 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 fastmcp import Client, FastMCP + tools = await server.mcp.local_provider.list_tools() + names = {tool.name for tool in tools} + self.assertFalse(any('human_vault' in name or 'wrapped' in name for name in names)) + self.assertEqual({name for name in names if name.startswith('persimmons')}, {'persimmons_list', 'persimmons_juice_list'}) + isolated = FastMCP('human vault boundary test') + for name in ('persimmons_list', 'persimmons_juice_list'): + isolated.add_tool(await server.mcp.local_provider.get_tool(name)) + with patch.object(vault.Store, 'operate', side_effect=AssertionError('MCP must not access human storage')): + async with Client(isolated) as client: + for name in ('human_vault_get', 'persimmons_human_vault', 'wrappedKey'): + with self.assertRaises(Exception): + await client.call_tool(name, {}) + routes = {getattr(route, 'path', None) for route in server.app.routes} + self.assertIn(vault.PREFIX, routes) + source = Path(vault.__file__).read_text() + self.assertNotIn('cryptography', source) + self.assertNotIn('decrypt(', source) + self.assertNotIn('@mcp', source)