178 lines
8.3 KiB
Python
178 lines
8.3 KiB
Python
#!/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()
|