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)