diff --git a/tests/test_persimmons_gitea.py b/tests/test_persimmons_gitea.py new file mode 100644 index 0000000..e2b50d6 --- /dev/null +++ b/tests/test_persimmons_gitea.py @@ -0,0 +1,151 @@ +import json +import os +import sqlite3 +import unittest +from contextlib import closing +from unittest.mock import patch +from persimmons import gitea, server +from tests.test_persimmons_host import HostTests + + +class IdentityVaultTests(HostTests): + async def signup(self, name='vault-fixture'): + result = await self.client.post('/api/auth/register', json={'name': name, 'password': 'synthetic-long-password'}, headers={'Origin': server.ORIGIN}) + self.assertEqual(result.status_code, 200) + return (await self.client.get('/api/session')).json() + + async def create(self, body=None): + return await self.client.post('/api/vaults', json=body or {'name': 'work-test'}, + headers={'Origin': server.ORIGIN, 'X-Requested-With': 'Persimmons'}) + + async def test_gitea_identity_duplicate_login_and_session(self): + identity = await self.signup() + self.assertEqual(identity, {'owner': 'pg-2', 'name': 'vault-fixture', 'gitea_id': 2}) + self.assertFalse(self.gitea.users[identity['name']][0]['is_admin']) + for name, password, path, expected in [ + ('vault-fixture', 'synthetic-long-password', 'register', 409), + ('vault-fixture', 'wrong-long-password', 'login', 401), + ('unknown-fixture', 'synthetic-long-password', 'login', 401), + ('vault-fixture', 'synthetic-long-password', 'login', 200)]: + response = await self.client.post('/api/auth/' + path, json={'name': name, 'password': password}, headers={'Origin': server.ORIGIN}) + self.assertEqual(response.status_code, expected) + with closing(server.database()) as db: + self.assertIsNone(db.execute("SELECT name FROM sqlite_master WHERE name='users'").fetchone()) + self.assertEqual(db.execute('SELECT gitea_id,login FROM sessions LIMIT 1').fetchone(), (2, 'vault-fixture')) + self.assertNotIn(b'synthetic-long-password', (self.data / 'sessions.sqlite').read_bytes()) + + async def test_create_private_empty_owned_repo_and_no_autoimport(self): + await self.signup() + self.assertEqual((await self.client.get('/api/vaults')).json(), {'vaults': []}) + response = await self.create({'name': 'Work test'}) + self.assertEqual(response.status_code, 201) + repo = response.json()['vault'] + self.assertEqual((repo['owner'], repo['name'], repo['private'], repo['empty']), ('vault-fixture', 'Work-test', True, True)) + self.assertEqual((await self.client.get('/api/vaults')).json(), {'vaults': [repo]}) + self.assertEqual((await self.create({'name': 'Work-test'})).status_code, 409) + self.gitea('/admin/users/vault-fixture/repos', 'POST', {'name': 'external', 'private': True, 'auto_init': False}) + self.assertEqual((await self.create({'name': 'external'})).status_code, 409) + self.assertEqual(len((await self.client.get('/api/vaults')).json()['vaults']), 1) + + async def test_cross_user_and_request_boundaries(self): + self.assertEqual((await self.create()).status_code, 401) + await self.signup('first-fixture') + self.assertEqual((await self.create()).status_code, 201) + await self.signup('second-fixture') + self.assertEqual((await self.client.get('/api/vaults')).json(), {'vaults': []}) + for extra in ({'owner': 'first-fixture'}, {'gitea_id': 2}, {'repo_id': 1}, {'private': False}): + self.assertEqual((await self.create({'name': 'work-test', **extra})).status_code, 400) + self.assertEqual((await self.client.post('/api/vaults', json={'name': 'work-test'})).status_code, 403) + self.assertEqual((await self.client.get('/api/vaults?owner=first-fixture')).status_code, 403) + self.assertEqual((await self.create()).status_code, 201) + self.assertEqual((await self.client.get('/api/vaults')).json()['vaults'][0]['owner'], 'second-fixture') + self.assertEqual((await self.client.get('/api/human-vault/items', headers={ + 'Origin': server.human_vault.EXTENSION_ORIGIN, 'X-Vault-Owner': 'pg-2'})).json()['items'], []) + + async def test_wrong_owner_public_nonempty_or_replaced_repo_fails_closed(self): + identity = await self.signup() + user = {'id': identity['gitea_id'], 'login': identity['name'], 'owner': identity['owner']} + for field, value in [('private', False), ('empty', False), ('owner', {'id': 999, 'login': 'foreign'})]: + with self.subTest(field=field): + original = self.gitea + def corrupt(path, method='GET', body=None, login=None): + result = original(path, method, body, login) + if '/repos' in path: + result[field] = value + return result + with patch.object(gitea, 'call', side_effect=corrupt): + result = await self.create({'name': 'invalid-' + field}) + self.assertEqual(result.status_code, 503) + self.assertEqual((await self.client.get('/api/vaults')).json(), {'vaults': []}) + self.assertEqual((await self.create()).status_code, 201) + self.gitea.repos[user['login'], 'work-test']['id'] = 999 + self.assertEqual((await self.client.get('/api/vaults')).status_code, 503) + self.gitea.users[user['login']][0]['id'] = 999 + self.assertEqual((await self.create({'name': 'replacement'})).status_code, 503) + self.assertNotIn((user['login'], 'replacement'), self.gitea.repos) + + async def test_admin_and_backend_failures_never_issue_session_or_leak(self): + await self.signup() + self.gitea.users['vault-fixture'][0]['is_admin'] = True + result = await self.client.post('/api/auth/login', json={'name': 'vault-fixture', 'password': 'synthetic-long-password'}, headers={'Origin': server.ORIGIN}) + self.assertEqual(result.status_code, 403) + self.assertNotIn('set-cookie', result.headers) + with patch.object(gitea, 'call', side_effect=gitea.Failure(500)), patch.dict(os.environ, PERSIMMONS_GITEA_TOKEN='secret-provisioner-fixture'): + result = await self.create() + self.assertEqual(result.status_code, 503) + self.assertNotIn('secret-provisioner-fixture', result.text) + self.assertNotIn('Authorization', result.text) + self.assertNotIn(b'secret-provisioner-fixture', (self.data / 'sessions.sqlite').read_bytes()) + + async def test_authorized_reset_is_once_and_bounded(self): + with closing(sqlite3.connect(self.data / 'sessions.sqlite')) as db, db: + db.execute('CREATE TABLE users (name TEXT, salt TEXT, verifier TEXT)') + db.execute("INSERT INTO users VALUES ('legacy','old','old-verifier')") + db.execute('CREATE TABLE sessions (digest TEXT, owner TEXT, expires INTEGER)') + with closing(sqlite3.connect(self.data / 'human-vault.sqlite')) as db, db: + db.execute('CREATE TABLE human_items (owner TEXT)') + db.execute("INSERT INTO human_items VALUES ('legacy')") + db.execute('CREATE TABLE unrelated (value TEXT)') + with closing(server.database()) as db: + self.assertEqual(db.execute('PRAGMA user_version').fetchone()[0], 2) + with closing(sqlite3.connect(self.data / 'human-vault.sqlite')) as db, db: + self.assertIsNone(db.execute("SELECT name FROM sqlite_master WHERE name='human_items'").fetchone()) + self.assertIsNotNone(db.execute("SELECT name FROM sqlite_master WHERE name='unrelated'").fetchone()) + db.execute('CREATE TABLE human_items (owner TEXT)') + db.execute("INSERT INTO human_items VALUES ('fresh')") + server.database().close() + with closing(sqlite3.connect(self.data / 'human-vault.sqlite')) as db, db: + self.assertEqual(db.execute('SELECT owner FROM human_items').fetchall(), [('fresh',)]) + + +class ClientTests(unittest.TestCase): + def test_fixed_https_credentials_and_redirect_refusal(self): + from io import BytesIO + with patch.dict(os.environ, PERSIMMONS_GITEA_TOKEN='admin-fixture'), patch('urllib.request.OpenerDirector.open', return_value=BytesIO(b'{}')) as opened: + gitea.call('/admin/users', 'POST', {'username': 'fixture'}) + req = opened.call_args.args[0] + self.assertEqual(req.full_url, 'https://teahub.xyz/api/v1/admin/users') + self.assertEqual(req.get_header('Authorization'), 'token admin-fixture') + with patch('urllib.request.OpenerDirector.open', return_value=BytesIO(b'{}')) as opened: + gitea.call('/user', login=('fixture', 'password-fixture')) + self.assertTrue(opened.call_args.args[0].get_header('Authorization').startswith('Basic ')) + self.assertNotIn('admin-fixture', str(opened.call_args)) + self.assertIsNone(gitea.NoRedirect().redirect_request(None, None, 302, '', {}, 'https://evil.invalid')) + + def test_shared_gitea_mode_uses_teahub_origin_and_disables_registration(self): + from io import BytesIO + with patch.object(gitea, 'ORIGIN', 'https://teahub.xyz'), patch('urllib.request.OpenerDirector.open', return_value=BytesIO(b'{}')) as opened: + gitea.call('/user', login=('chai', 'password-fixture')) + self.assertEqual(opened.call_args.args[0].full_url, 'https://teahub.xyz/api/v1/user') + with patch.object(gitea, 'ALLOW_REGISTRATION', False): + with self.assertRaises(gitea.Failure) as failed: + gitea.authenticate('new-user', 'synthetic-long-password', True) + self.assertEqual(failed.exception.code, 403) + + def test_normalization_rejects_paths_and_preserves_valid_names(self): + for name in ('Work-test', 'scope_One', 'a.b'): + self.assertEqual(gitea.repo_name(name), name) + self.assertEqual(gitea.repo_name(' Work test '), 'Work-test') + for name in ('../other', 'a/b', 'https://evil', '.git', 'name.git', '', 'a' * 101): + with self.subTest(name=name), self.assertRaises(ValueError): + gitea.repo_name(name)