import importlib.util import json import os from pathlib import Path import stat import tempfile import unittest from unittest import mock from dataclasses import replace from control_sidecar.auth import Auth from control_sidecar.config import Config from control_sidecar.protocol import BoundaryError import auth_control as auth class AuthToolTests(unittest.TestCase): def test_cancel_preserves_files_and_never_posts(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) operation = root / 'operation' operation.mkdir() pending = root / 'pending.json' pending.write_text(json.dumps(dict(operation=str(operation)))) secret = root / 'unchanged' secret.write_bytes(b'fixture-original') session = root / 'session' session.write_text(json.dumps(dict(token='fixture-saved-token', expires=None))) (operation / 'operation.json').write_text(json.dumps(dict(phase='UNCONFIRMED', sidecar_was_running=True, files=[dict(path=str(secret), before_sha256=auth.digest(secret.read_bytes()))]))) cfg = mock.Mock(session_file=str(session)) with mock.patch.object(auth, 'ROOT', root), mock.patch.object(auth, 'PENDING', pending), \ mock.patch.object(auth, 'context', return_value=({'State': {'Running': False}}, {}, cfg)), \ mock.patch.object(auth, 'command') as command: auth.cancel(confirmed=True) command.assert_called_once_with(['docker', 'start', auth.SIDE]) self.assertEqual(secret.read_bytes(), b'fixture-original') self.assertFalse(pending.exists()) def test_credential_file_requires_private_mode(self): with tempfile.TemporaryDirectory() as directory: p = Path(directory) / 'credentials.json' p.write_text(json.dumps(dict(loginId='fixture-ai', password='fixture-value'))) p.chmod(0o600) self.assertEqual(auth.read_credentials(p)['loginId'], 'fixture-ai') p.chmod(0o644) with self.assertRaises(ValueError): auth.read_credentials(p) def test_publish_and_partial_resume_without_new_login(self): for partial in (False, True): with self.subTest(partial=partial), tempfile.TemporaryDirectory() as directory: root = Path(directory) operation = root / 'operation' operation.mkdir() active = root / 'active' active.write_bytes(b'fixture-active') pending = root / 'pending' pending.write_text('{}') cfg = Config(server_id='fixture-ai', manager='https://manager.invalid', runtime='http://runtime.invalid', manager_ca='/fixture/ca', runtime_token='fixture-runtime', relay_token='fixture-relay', login_body={'loginId': 'before'}, password='fixture-before', session_file=str(root / 'session')) (operation / 'new-password').write_text('fixture-after\n') (operation / 'new-body.json').write_text(json.dumps({'loginId': 'after'})) candidate = replace(cfg, password='fixture-after', login_body={'loginId': 'after'}, session_file=str(operation / 'new-session.json')) state = Auth(candidate, None) state.token = 'fixture-session-token-after' state.save_session() files = [] for name in ('password', 'body.json', 'session.json'): new = 'new-' + name p = root / name p.write_bytes(b'fixture-old') files.append(dict(path=str(p), new=new, before_sha256=auth.digest(b'fixture-old'), owner=[os.getuid(), os.getgid()])) if partial: (root / 'password').write_bytes((operation / 'new-password').read_bytes()) data = dict(phase='APPLYING' if partial else 'LOGIN_OK', container_id='fixture-container', active_sha256=auth.digest(active.read_bytes()), files=files, sidecar_was_running=False) (operation / 'operation.json').write_text(json.dumps(data)) with mock.patch.object(auth, 'ACTIVE', active), mock.patch.object(auth, 'PENDING', pending), \ mock.patch.object(auth, 'context', return_value=({'Id': 'fixture-container'}, {}, cfg)), \ mock.patch.object(auth, 'stop_sidecar'), mock.patch.object(Auth, 'login') as login: auth.publish(operation) login.assert_not_called() self.assertFalse(pending.exists()) for row in files: self.assertEqual(Path(row['path']).read_bytes(), (operation / row['new']).read_bytes()) self.assertEqual(stat.S_IMODE(Path(row['path']).stat().st_mode), 0o600) def test_private_write_keeps_mode_owner_and_hides_temp(self): with tempfile.TemporaryDirectory() as directory: p = Path(directory) / 'private' auth.private_write(p, b'fixture', (os.getuid(), os.getgid())) self.assertEqual(p.read_bytes(), b'fixture') self.assertEqual(stat.S_IMODE(p.stat().st_mode), 0o600) self.assertFalse(p.with_name('private.auth-next').exists()) auth.private_write(p, b'updated-fixture') self.assertEqual(p.read_bytes(), b'updated-fixture') def test_private_write_refuses_symlink_and_foreign_interruption(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / 'real').write_text('preserve') (root / 'link').symlink_to(root / 'real') with self.assertRaises(ValueError): auth.private_write(root / 'link', b'new') (root / 'private.auth-next').write_text('foreign') with self.assertRaises(ValueError): auth.private_write(root / 'private', b'new') self.assertEqual((root / 'real').read_text(), 'preserve') def test_private_matching_interruption_is_resumable(self): with tempfile.TemporaryDirectory() as directory: p = Path(directory) / 'private' p.with_name('private.auth-next').write_bytes(b'fixture') auth.private_write(p, b'fixture') self.assertEqual(p.read_bytes(), b'fixture') def test_map_is_limited_to_known_bind(self): mounts = [dict(Type='bind', Source='/fixture/private', Destination='/run/bridge')] self.assertEqual(auth.map_file('/run/bridge/password', mounts), '/fixture/private/password') with self.assertRaises(ValueError): auth.map_file('/run/bridge-other/password', mounts) def test_id_and_status_do_not_require_login(self): self.assertEqual(auth.login_id('fixture-ai'), 'fixture-ai') for value in ('', ' ', 'a\nb', 'x' * 257): with self.assertRaises(ValueError): auth.login_id(value) with mock.patch.object(auth, 'command', side_effect=subprocess_error()): self.assertFalse(auth.wss()['connected']) def test_publish_refuses_unknown_outcome_before_writes(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / 'operation.json').write_text(json.dumps(dict(phase='LOGIN_INTENT'))) with mock.patch.object(auth, 'context', return_value=({}, {}, None)), mock.patch.object(auth, 'private_write') as write: with self.assertRaises(ValueError): auth.publish(root) write.assert_not_called() def subprocess_error(): import subprocess return subprocess.CalledProcessError(1, 'fixture') class LoginContractTests(unittest.IsolatedAsyncioTestCase): async def test_actor_and_server_identity_and_rejection(self): for actor, server, code, success in (('AI', 'fixture-ai', 200, True), ('MANAGER', 'fixture-ai', 200, False), ('AI', 'different-ai', 200, False), ('AI', 'fixture-ai', 401, False)): cfg = Config(server_id='fixture-ai', manager='https://manager.invalid', runtime='http://runtime.invalid', manager_ca='/fixture/ca', runtime_token='fixture-runtime', relay_token='fixture-relay', login_body={'loginId': 'fixture'}, password='fixture-password') class Transport: calls = 0 async def request(self, *args, **kwargs): self.calls += 1 doc = dict(result=True, error=None, serverTimeMs=1, data=dict(actorType=actor, serverId=server, accountId='fixture', token='fixture-valid-token-value', mustChangePassword=False, expiresAtWallMs=None, permission='admin', replacedSession=False, revokedPreviousTokens=0)) return code, json.dumps(doc).encode() transport = Transport() state = Auth(cfg, transport) if success: await state.login() self.assertIsNotNone(state.token) else: with self.assertRaises(BoundaryError): await state.login() self.assertIsNone(state.token) self.assertEqual(transport.calls, 1) if __name__ == '__main__': unittest.main()