diff --git a/tests/test_persimmons_host.py b/tests/test_persimmons_host.py new file mode 100644 index 0000000..ce22949 --- /dev/null +++ b/tests/test_persimmons_host.py @@ -0,0 +1,88 @@ +import base64 +import json +import os +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch +import httpx +from persimmons import server +from tests.persimmons_gitea_fixture import GiteaFixture + +class HostTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.data = Path(self.tmp.name) + self.patch = patch.object(server, 'DATA', self.data) + self.patch.start() + self.gitea = GiteaFixture() + self.backend = patch.object(server.gitea, 'call', side_effect=self.gitea) + self.backend.start() + self.env = patch.dict(os.environ, PERSIMMONS_HUMAN_VAULT_DB=str(self.data / 'vault.sqlite')) + self.env.start() + self.client = httpx.AsyncClient(transport=httpx.ASGITransport(app=server.app), base_url=server.ORIGIN) + + async def asyncTearDown(self): + await self.client.aclose() + self.env.stop() + self.patch.stop() + self.backend.stop() + self.tmp.cleanup() + + async def test_home_health_assets_and_no_runtime_tea_calls(self): + response = await self.client.get('/') + self.assertEqual(response.status_code, 200) + self.assertIn('put plainly', response.text) + for path in ['/healthz', '/assets/app.js', '/assets/version.mjs', '/assets/version.json', '/assets/extension.zip', '/assets/update.html', '/assets/install.html', '/assets/update-extension.py', '/signin']: + self.assertEqual((await self.client.get(path)).status_code, 200) + self.assertNotIn('https://teahub.xyz/api', (await self.client.get('/assets/app.js')).text) + self.assertEqual((await self.client.get('/mcp')).status_code, 404) + self.assertEqual((await self.client.get('/', headers={'Host':'evil.invalid'})).status_code, 400) + + async def test_session_vault_boundary_persistence_logout(self): + body = {'name':'taste-fixture', 'password':'disposable synthetic password'} + auth = '/api/auth/register' + self.assertEqual((await self.client.post(auth, json=body)).status_code, 403) + self.assertEqual((await self.client.post(auth, json=body, headers={'Origin':'https://evil.invalid'})).status_code, 403) + response = await self.client.post(auth, json=body, headers={'Origin':server.ORIGIN}) + self.assertEqual(response.status_code, 200) + for flag in ['HttpOnly', 'Secure', 'SameSite=none', 'Path=/']: + self.assertIn(flag, response.headers['set-cookie']) + identity = (await self.client.get('/api/session')).json() + owner = identity['owner'] + self.assertEqual(identity['name'], body['name']) + url = '/api/human-vault/items' + headers = {'Origin':server.human_vault.EXTENSION_ORIGIN, 'X-Vault-Owner':owner} + self.assertEqual((await self.client.get(url)).status_code, 403) + self.assertEqual((await self.client.post(url+'/read')).status_code, 403) + self.assertEqual((await self.client.post(url+'/read', headers=headers)).json()['items'], []) + self.assertEqual((await self.client.post(url+'/read', content='unexpected', headers=headers)).status_code, 400) + self.assertEqual((await self.client.options(url, headers={**headers, 'Access-Control-Request-Method':'POST', 'Access-Control-Request-Headers':'content-type,x-vault-owner'})).headers['access-control-allow-origin'], headers['Origin']) + # Validation accepts opaque ciphertext; accepted vault suite uses Web Crypto. + item = {'v':1,'id':'a'*32,'origin':'https://example.com','iv':base64.b64encode(b'i'*12).decode(),'ciphertext':base64.b64encode(b'c'*24).decode()} + self.assertEqual((await self.client.post(url, json=item, headers=headers)).status_code, 201) + self.assertEqual((await self.client.post(url+'/'+item['id']+'/read', headers=headers)).json()['item'], item) + self.assertEqual((await self.client.post(url+'/'+item['id']+'/read', headers={**headers,'X-Vault-Owner':'other'})).status_code, 409) + self.assertEqual(server.human_vault.Store(self.data / 'vault.sqlite').operate(owner,'GET',item['id'])[0]['item'], item) + self.assertEqual((await self.client.get(url+'/'+item['id'], headers={**headers,'X-Vault-Owner':'other'})).status_code, 409) + cookie = self.client.cookies.get(server.COOKIE) + self.assertEqual((await self.client.post('/api/logout', headers={'Origin':server.ORIGIN})).status_code, 200) + self.assertEqual((await self.client.get(url, headers={**headers,'Cookie':server.COOKIE+'='+cookie})).status_code, 401) + self.assertEqual((await self.client.post('/api/auth/login',json=body,headers={'Origin':server.ORIGIN})).status_code, 200) + self.assertEqual((await self.client.get(url,headers=headers)).json()['items'][0]['id'],item['id']) + self.assertEqual((await self.client.delete('/api/vault-index/'+item['id'])).status_code, 403) + self.assertEqual((await self.client.delete('/api/vault-index/'+item['id'], headers={ + 'Origin': server.ORIGIN, 'X-Requested-With': 'Persimmons'})).status_code, 200) + self.assertEqual((await self.client.get(url,headers=headers)).json()['items'], []) + self.assertNotIn(body['password'].encode(), (self.data/'sessions.sqlite').read_bytes()) + self.assertNotIn(cookie.encode(), (self.data/'sessions.sqlite').read_bytes()) + + async def test_different_accounts_and_untrusted_preflight(self): + for name in ['fixture-one','fixture-two']: + self.assertEqual((await self.client.post('/api/auth/register',json={'name':name,'password':'disposable synthetic password'},headers={'Origin':server.ORIGIN})).status_code,200) + owner=(await self.client.get('/api/session')).json()['owner'] + if name=='fixture-one': first=owner + else: self.assertNotEqual(first,owner) + response=await self.client.options('/api/human-vault/items',headers={'Origin':'chrome-extension://'+'a'*32,'Access-Control-Request-Method':'GET'}) + self.assertEqual(response.status_code,403) + self.assertNotIn('access-control-allow-origin',response.headers)