Consolidate Persimmons shared backend and tests

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)