#!/usr/bin/env python3 """Real APT/dpkg transaction and failing postinst, ONLY in a fresh bwrap namespace. Fixture repository is deliberately unsigned and trusted only in fixture config. Production authentication is not bypassed: Executor.__init__/prepare are not used. This qualifies apply/journal/reconciliation, not release admission or boot. """ import hashlib import json import os import pathlib import subprocess import sys def run(*args): result = subprocess.run(args, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode: raise RuntimeError(str(args) + '\n' + result.stdout + '\n' + result.stderr) return result.stdout def durable(path, raw, mode=0o600): temporary = path.with_suffix('.new') with open(temporary, 'wb') as out: out.write(raw) out.flush() os.fsync(out.fileno()) os.replace(temporary, path) fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) os.fsync(fd) os.close(fd) def inside(): assert os.getuid() == 0 os.environ['PATH'] = '/usr/sbin:/usr/bin:/sbin:/bin' assert os.environ['OS_OTA_DISPOSABLE_PARENT_USERNS'] != str(os.stat('/proc/self/ns/user').st_ino) assert any(line.split()[4] == '/' and 'ro' in line.split()[5].split(',') for line in pathlib.Path('/proc/self/mountinfo').read_text().splitlines()) sys.path.insert(0, '/mnt/src') import os_update_executor as engine from os_update_plan import simulation, digest engine.atomic = durable # Fixture filesystem has mapped root, not host UID0 ancestry. engine.read = lambda path: pathlib.Path(path).read_bytes() root = pathlib.Path('/mnt/work') for name in ('/var/lib/dpkg/updates', '/var/log', '/opt/preserved', '/mnt/work/repo'): pathlib.Path(name).mkdir(parents=True, exist_ok=True) pathlib.Path('/var/lib/dpkg/status').write_text('') wallet = pathlib.Path('/opt/preserved/wallet') wallet.write_text('wallet sentinel; never snapshot/restore') decisions = pathlib.Path('/opt/preserved/decisions.json') decisions.write_text('{"stopped":["bitcoin"],"uninstalled":["lnd"]}') preservation = {str(p): digest(p.read_bytes()) for p in (wallet, decisions)} def package(name, version, extra='', fail=False): directory = root / ('build-' + name + '-' + version) (directory / 'DEBIAN').mkdir(parents=True) (directory / 'DEBIAN/control').write_text( f'Package: {name}\nVersion: {version}\nArchitecture: all\nMaintainer: Disposable \nDescription: OS updater disposable fixture\n' + extra) destination = directory / 'opt/os-ota-fixture' destination.mkdir(parents=True) (destination / name).write_text(version) if fail: script = directory / 'DEBIAN/postinst' script.write_text('#!/bin/sh\necho invoked >> /opt/os-ota-fixture/postinst-count\nexit 42\n') script.chmod(0o755) target = root / 'repo' / f'{name}_{version}_all.deb' run('/usr/bin/dpkg-deb', '--build', '--root-owner-group', str(directory), str(target)) return target old = package('os-ota-base', '1') obsolete = package('os-ota-obsolete', '1') run('/usr/bin/dpkg', '-i', str(old), str(obsolete)) base = package('os-ota-base', '2', 'Depends: os-ota-helper (= 1)\nConflicts: os-ota-obsolete\n') helper = package('os-ota-helper', '1') def repository(archives): paragraphs = [] for archive in archives: control = run('/usr/bin/dpkg-deb', '-f', str(archive)).strip() paragraphs.append(control + f'\nFilename: {archive.name}\nSize: {archive.stat().st_size}\nSHA256: {digest(archive.read_bytes())}\n') (root / 'repo/Packages').write_text('\n'.join(paragraphs)) def executor(name, archives): obj = engine.Executor.__new__(engine.Executor) obj.root = root / name obj.root.mkdir() for directory in ('lists/partial', 'archives/partial', 'empty'): (obj.root / directory).mkdir(parents=True) obj.status_path = obj.root / 'status.json' (obj.root / 'before-inventory').write_bytes(engine.inventory()) obj.status = dict(version=1, plan_sha256='a'*64, authority_digest=digest(name.encode()), phase='VERIFIED', sequence=0, events=[]) sources = obj.root / 'sources.list' sources.write_text('deb [trusted=yes] file:/mnt/work/repo ./\n') config = dict(**{'Dir::Etc::main': '-', 'Dir::Etc::parts': str(obj.root / 'empty'), 'Dir::Etc::sourcelist': str(sources), 'Dir::Etc::sourceparts': str(obj.root / 'empty'), 'Dir::Etc::preferences': '-', 'Dir::Etc::preferencesparts': str(obj.root / 'empty'), 'Dir::State::lists': str(obj.root / 'lists'), 'Dir::Cache::archives': str(obj.root / 'archives'), 'Dir::Cache::pkgcache': '', 'Dir::Cache::srcpkgcache': '', 'APT::Sandbox::User': 'root'}) (obj.root / 'apt.conf').write_text('\n'.join(k + ' ' + json.dumps(v) + ';' for k, v in config.items())) obj.apt('update') transaction = simulation(obj.apt('--simulate', 'dist-upgrade')) obj.plan = dict(transaction=transaction, packages=[dict(name=run('/usr/bin/dpkg-deb', '-f', str(a), 'Package').strip(), version=run('/usr/bin/dpkg-deb', '-f', str(a), 'Version').strip(), architecture='all', sha256=digest(a.read_bytes())) for a in archives], removals=[x['name'] for x in transaction if x['operation'] == 'Remv'], boot={'retained_files': {}}) obj.apt('--download-only', '--assume-yes', 'dist-upgrade') # file repositories may use the original file instead of copying to cache. for a in archives: (obj.root / 'archives' / a.name).write_bytes(a.read_bytes()) obj.preflight = lambda: None # Host suite/kernel admission is out of this fixture's scope. obj.boot_provenance = lambda: {'fixture_only': True} return obj repository([base, helper]) success = executor('success', [base, helper]) assert success.plan['removals'] == ['os-ota-obsolete'] success.apply() assert success.status['phase'] == 'REBOOT_REQUIRED' assert pathlib.Path('/opt/os-ota-fixture/os-ota-base').read_text() == '2' assert not pathlib.Path('/opt/os-ota-fixture/os-ota-obsolete').exists() before_repeat = engine.inventory() success.apply() assert engine.inventory() == before_repeat broken = package('os-ota-base', '3', 'Depends: os-ota-helper (= 1)\n', fail=True) repository([broken, helper]) failure = executor('failure', [broken]) try: failure.apply() raise AssertionError('failing maintainer script accepted') except ValueError: pass assert failure.status['phase'] == 'RECONCILIATION_REQUIRED' assert run('/usr/bin/dpkg', '--audit').strip() failure.status = json.loads(failure.status_path.read_bytes()) try: failure.apply() raise AssertionError('replayed partial dpkg transaction') except engine.ReconciliationRequired: pass assert pathlib.Path('/opt/os-ota-fixture/postinst-count').read_text() == 'invoked\n' assert preservation == {p: digest(pathlib.Path(p).read_bytes()) for p in preservation} result = dict(real_apt_dpkg=True, dependency_upgrade=True, explicit_conflict_removal=True, partial_dpkg_reconciliation=True, maintainer_script_invocations=1, preserved_fixture_wallet_and_decisions=True, production_authentication_tested=False, boot_tested=False) (root / 'result.json').write_text(json.dumps(result, indent=2) + '\n') print(json.dumps(result, sort_keys=True)) def main(): if sys.argv[1:] == ['--inside']: inside() return target = pathlib.Path(sys.argv[1]).resolve() assert not target.exists() and target.parent.is_dir() target.mkdir(mode=0o700) source = pathlib.Path(__file__).resolve().parents[1] env = dict(os.environ, OS_OTA_DISPOSABLE_PARENT_USERNS=str(os.stat('/proc/self/ns/user').st_ino), CARGO_BUILD_JOBS='1', CARGO_INCREMENTAL='0') args = ['bwrap', '--unshare-all', '--die-with-parent', '--ro-bind', '/', '/', '--tmpfs', '/tmp', '--tmpfs', '/run', '--tmpfs', '/home', '--tmpfs', '/var', '--tmpfs', '/opt', '--tmpfs', '/mnt', '--proc', '/proc', '--dev', '/dev', '--uid', '0', '--gid', '0', '--ro-bind', str(source), '/mnt/src', '--bind', str(target), '/mnt/work', '/usr/bin/python3', '/mnt/src/tests/os_update_disposable.py', '--inside'] result = subprocess.run(args, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) (target / 'integration.log').write_text(result.stdout) print(result.stdout) raise SystemExit(result.returncode) if __name__ == '__main__': main()