#!/usr/bin/env python3 """Real root/UID1000 filesystem custody tests in private bwrap mounts only. Exercises storage/journal authority, not signature verification or package apply. No fixture signer, release key, unsigned approval, apt, or reboot is involved. """ import json import os import pathlib import subprocess import sys def inside(): assert os.geteuid() == 0 assert str(os.stat('/proc/self/ns/mnt').st_ino) != os.environ['OS_CUSTODY_PARENT_MNT'] mounts = [line.split() for line in pathlib.Path('/proc/self/mountinfo').read_text().splitlines()] assert any(row[4] == '/' and 'ro' in row[5].split(',') for row in mounts) assert any(row[4] == '/var' and row[row.index('-')+1] == 'tmpfs' for row in mounts) sys.path.insert(0, '/mnt/source') import os_update_custody as custody import os_update_executor as executor import d09_store as store from os_update_plan import canonical, digest evidence = pathlib.Path('/mnt/evidence') app = pathlib.Path('/var/lib/archipelago') app.mkdir(parents=True, mode=0o755) os.chown(app, 1000, 1000) canary = app / 'wallet-and-decisions' canary.write_bytes(b'wallet-forward-state; stopped=bitcoin; uninstalled=lnd\n') os.chown(canary, 1000, 1000) before = (app.stat().st_uid, app.stat().st_gid, app.stat().st_mode, canary.read_bytes()) results = {} def refused(name, function): try: function() except (OSError, ValueError, store.ContinuityError): results[name] = True else: raise AssertionError('unsafe operation accepted: ' + name) refused('nonroot_parent_before_mkdir', lambda: custody.mkdir(app / 'os-update')) assert not (app / 'os-update').exists() custody.refuse_legacy() root_pin, created = custody.mkdir(executor.ROOT) assert created and executor.ROOT.stat().st_uid == 0 child = os.fork() if child == 0: try: os.setgid(1000) os.setuid(1000) try: (executor.ROOT / 'forged-status').write_text('authority') except PermissionError: os._exit(0) os._exit(2) except BaseException: os._exit(3) assert os.waitpid(child, 0)[1] == 0 results['actual_uid1000_cannot_write_new_state'] = True parents, _ = custody.mkdir(executor.ROOT / 'transactions') initial = dict(version=1, authority_digest='a'*64, plan_sha256='b'*64, phase=None, sequence=0, events=[]) transaction = executor.ROOT / 'transactions' / ('a'*64) tx_pin, status = custody.transaction(transaction, initial, True) authority = pathlib.Path('/var/lib/custody-d09') authority.mkdir(mode=0o700) binding = dict(transaction_id='a'*32, admission_epoch_id='b'*32, os_update=dict(authority_digest='a'*64, plan_sha256='b'*64)) store.atomic(authority / 'ledger.json', canonical(store.initial(binding))) ledger = store.Store(authority, binding, admitted=True) obj = executor.Executor.__new__(executor.Executor) obj.root, obj.status_path, obj.status = transaction, transaction / 'status.json', status obj._pins = [root_pin, parents, tx_pin] obj.checkpoint = lambda: ledger.checkpoint_os(json.loads(store.read(obj.status_path))) verified = None for phase in ('PREPARE', 'DOWNLOAD', 'VERIFIED', 'APPLY_INTENT'): obj.journal(phase) if phase == 'VERIFIED': verified = store.read(obj.status_path) consumed = store.read(obj.status_path) again, reopened = custody.transaction(transaction, initial, True) assert reopened['phase'] == 'APPLY_INTENT' and reopened['sequence'] == 4 again.close() results['durable_consumed_authority_not_reset'] = True obj.status_path.rename(transaction / 'retained-consumed-status.json') refused('missing_journal_never_reinitialized', lambda: custody.transaction(transaction, initial, True)) assert not obj.status_path.exists() store.atomic(obj.status_path, consumed) # Test-owned restoration of exact retained evidence. store.atomic(obj.status_path, verified) # Adversarial rollback; no package operation follows. refused('live_journal_replacement_refused', lambda: obj.journal('APPLY_INTENT')) obj.status = json.loads(verified) refused('durable_owner_rejects_replayed_apply_intent', lambda: obj.journal('APPLY_INTENT')) assert ledger.value['os_checkpoints'][-1]['sequence'] == 4 store.atomic(obj.status_path, consumed) obj.status = json.loads(consumed) retained = executor.ROOT.with_name('retained-root-for-path-swap-test') executor.ROOT.rename(retained) executor.ROOT.symlink_to(app, target_is_directory=True) refused('symlink_root_swap_refused', obj.check_custody) executor.ROOT.unlink() # Only the disposable adversarial symlink. executor.ROOT.mkdir(mode=0o700) refused('root_owned_directory_replacement_refused', obj.check_custody) executor.ROOT.rmdir() # Only the empty adversarial replacement. retained.rename(executor.ROOT) obj.check_custody() empty = executor.ROOT / 'transactions' / ('c'*64) empty.mkdir(mode=0o700) refused('interrupted_directory_creation_not_reset', lambda: custody.transaction(empty, initial, True)) assert list(empty.iterdir()) == [] # Leave old authority bytes intact as evidence; even an empty legacy entry # would refuse. No application directory changes ownership at any step. legacy = app / 'os-update' legacy.mkdir(mode=0o700) os.chown(legacy, 1000, 1000) refused('empty_legacy_directory_also_refused', custody.refuse_legacy) legacy_record = legacy / 'consumed.json' legacy_record.write_bytes(consumed) refused('legacy_state_blocks_executor', lambda: executor.stage({'os_update': {}}, 'APPLY')) import os_update_owner refused('legacy_state_blocks_provisioning', lambda: os_update_owner.install_inputs({}, pathlib.Path('/missing'))) assert legacy_record.read_bytes() == consumed legacy.rename(app / 'retained-legacy-test-evidence') legacy.symlink_to('/missing/never-follow') refused('dangling_legacy_symlink_refused', custody.refuse_legacy) retained_app = app.with_name('retained-app-for-ancestry-test') app.rename(retained_app) app.symlink_to(retained_app, target_is_directory=True) refused('symlinked_legacy_parent_refused', custody.refuse_legacy) app.unlink() # Disposable adversarial alias only, not retained app data. retained_app.rename(app) assert before == (app.stat().st_uid, app.stat().st_gid, app.stat().st_mode, canary.read_bytes()) results['app_owner_and_forward_data_unchanged'] = True (evidence / 'result.json').write_text(json.dumps(dict(passed=True, checks=results, production_authentication_tested=False, packages_or_boot_tested=False, root=str(executor.ROOT), consumed_status_sha256=digest(consumed)), indent=2) + '\n') (evidence / 'consumed-status.json').write_bytes(consumed) (evidence / 'owner-ledger.json').write_bytes(store.read(authority / 'ledger.json')) obj.close() os.close(ledger.fd) os.close(ledger.head_fd) print(json.dumps(results, sort_keys=True)) def main(): if sys.argv[1:] == ['--inside']: inside() return assert os.geteuid() == 0, 'run this fixture only in its appointed limited root service' report = pathlib.Path(sys.argv[1]).resolve() assert report.parent.is_dir() and not report.exists() report.mkdir(mode=0o755) source = pathlib.Path(__file__).resolve().parents[1] env = dict(os.environ, OS_CUSTODY_PARENT_MNT=str(os.stat('/proc/self/ns/mnt').st_ino)) command = ['bwrap', '--die-with-parent', '--unshare-pid', '--unshare-net', '--unshare-ipc', '--unshare-uts', '--ro-bind', '/', '/', '--tmpfs', '/var', '--tmpfs', '/tmp', '--tmpfs', '/home', '--tmpfs', '/run', '--tmpfs', '/mnt', '--proc', '/proc', '--dev', '/dev', '--cap-add', 'CAP_CHOWN', '--cap-add', 'CAP_SETUID', '--cap-add', 'CAP_SETGID', '--ro-bind', str(source), '/mnt/source', '--bind', str(report), '/mnt/evidence', '/usr/bin/python3', '-B', '/mnt/source/tests/os_update_custody_disposable.py', '--inside'] result = subprocess.run(command, env=env, capture_output=True, text=True) (report / 'test.log').write_text(result.stdout + result.stderr) print(result.stdout + result.stderr) raise SystemExit(result.returncode) if __name__ == '__main__': main()