"""Unprivileged filesystem/control tests, NOT installed or OS-transaction UAT.""" import importlib.util import json import os from pathlib import Path import tempfile import unittest from unittest.mock import patch SPEC = importlib.util.spec_from_file_location('bootstrap', Path(__file__).parents[1] / 'os_updater_bootstrap.py') B = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(B) class BootstrapTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) for name in ('etc', 'usr/local/bin', 'var/lib', 'payload', 'evidence', 'wallet'): (self.root / name).mkdir(parents=True) (self.root / 'etc/machine-id').write_bytes(b'fixture-machine\n') (self.root / 'usr/local/bin/archipelago').write_bytes(b'old native') (self.root / 'payload/owner.py').write_bytes(b'# inert fixture owner\n') (self.root / 'evidence/inventory.json').write_bytes(b'public inventory') (self.root / 'wallet/sentinel').write_bytes(b'never restored') self.plan = {'schema': 1, 'machine_id_sha256': B.digest(b'fixture-machine\n'), 'current': {'version': '2.0.0-alpha+uat', 'source_sha256': 'a'*64, 'backend_sha256': B.digest(b'old native')}, 'candidate': {'version': '2.0.1-alpha', 'source_sha256': 'b'*64, 'backend_sha256': 'c'*64}, 'admin': {'uid': 1001, 'service_uid': 1000, 'access_receipt_sha256': B.digest(b'public inventory')}, 'observations': {'evidence/inventory.json': B.digest(b'public inventory')}, 'files': {'owner.py': B.digest(b'# inert fixture owner\n')}} self.plan['candidate']['source_sha256'] = B.digest(json.dumps( self.plan['files'], sort_keys=True, separators=(',', ':')).encode()) def prepare(self): approval = self.root / 'approval.json' raw = json.dumps(self.plan).encode() approval.write_bytes(raw) # Emulate root custody only in this unprivileged filesystem fixture. # Production secure() is tested separately and is never bypassed by CLI. def fixture_open(path, uid=0, directory=False): return os.open(path, os.O_RDONLY | os.O_NOFOLLOW) with patch.object(B.os, 'geteuid', return_value=0), patch.object(B, 'secure', fixture_open): return B.prepare(self.root, approval, B.digest(raw), self.root / 'payload') def test_atomic_inert_generation_preserves_old_code_and_data(self): result = self.prepare() self.assertEqual(result['state'], 'PREPARED_NOT_ACTIVATED') prepared = self.root / B.BASE / 'prepared' self.assertEqual(json.loads((prepared / 'control.json').read_bytes()), result) self.assertEqual((self.root / 'wallet/sentinel').read_bytes(), b'never restored') self.assertEqual((self.root / 'usr/local/bin/archipelago').read_bytes(), b'old native') with self.assertRaisesRegex(ValueError, 'existing bootstrap'): self.prepare() def test_unresolved_ledgers_refuse_without_rewriting(self): for name in B.BLOCKERS: with self.subTest(name=name): p = self.root / name p.parent.mkdir(parents=True, exist_ok=True) p.write_bytes(b'unresolved') with self.assertRaisesRegex(ValueError, 'existing authority'): self.prepare() self.assertEqual(p.read_bytes(), b'unresolved') p.unlink() def test_interruption_retained_and_retry_refused(self): original = B.write def fail_control(path, raw): if path.name == 'control.json': raise OSError('injected disk failure') original(path, raw) with patch.object(B, 'write', fail_control), self.assertRaises(OSError): self.prepare() self.assertFalse((self.root / B.BASE / 'prepared').exists()) self.assertEqual(len(list((self.root / B.BASE).glob('pending-*'))), 1) with self.assertRaisesRegex(ValueError, 'unresolved bootstrap'): self.prepare() def test_downgrade_and_same_version_refused(self): for version in ('1.9.0-alpha', '2.0.0-alpha', '2.0.0-alpha+uat'): self.plan['candidate']['version'] = version with self.assertRaisesRegex(ValueError, 'downgrade'): self.prepare() def test_source_inventory_and_backend_drift(self): for name in ('payload/owner.py', 'evidence/inventory.json', 'usr/local/bin/archipelago'): with self.subTest(name=name): p = self.root / name before = p.read_bytes() p.write_bytes(b'changed') with self.assertRaisesRegex(ValueError, 'drift'): self.prepare() p.write_bytes(before) self.assertFalse((self.root / B.BASE).exists()) def test_shared_admin_and_traversal_refuse(self): self.plan['admin']['uid'] = 1000 with self.assertRaisesRegex(ValueError, 'shared'): self.prepare() self.plan['admin']['uid'] = 1001 self.plan['files'] = {'../wallet/sentinel': 'a'*64} with self.assertRaisesRegex(ValueError, 'payload names'): self.prepare() def test_real_custody_rejects_untrusted_input(self): with self.assertRaisesRegex(ValueError, 'custody'): B.read(self.root / 'payload/owner.py') def test_duplicate_keys_refused(self): with self.assertRaisesRegex(ValueError, 'duplicate'): B.decode(b'{"schema":1,"schema":1}') def test_unauthenticated_approval_and_nonroot_refuse(self): with patch.object(B.os, 'geteuid', return_value=1000): with self.assertRaisesRegex(ValueError, 'root administrator'): B.prepare(self.root, self.root / 'absent', '0'*64, self.root / 'payload') with patch.object(B.os, 'geteuid', return_value=0), patch.object( B, 'secure', side_effect=lambda p, **kw: os.open(p, os.O_RDONLY)), patch.object( B, 'read', return_value=b'not the approved bytes'): with self.assertRaisesRegex(ValueError, 'authenticated approval digest'): B.prepare(self.root, self.root / 'absent', '0'*64, self.root / 'payload') def test_candidate_source_inventory_pin_refuses(self): self.plan['candidate']['source_sha256'] = '0'*64 with self.assertRaisesRegex(ValueError, 'source inventory digest'): self.prepare() if __name__ == '__main__': unittest.main()