"""Fixed privileged D09 OS stage. No CLI, shell, repair loop, or unsigned mode. The protected D09 binding is the authority input. APT owns Debian maintainer script execution; those scripts are explicitly reviewed root authority. """ import fcntl import json import os import pathlib import platform import re import subprocess import tarfile from d09_store import atomic, read, secure, sync_dir from os_update_plan import canonical, digest, hashed, keys, protected, require, simulation, validate from os_update_repository import KEYRING, acquire, command from os_update_lock import PackageLease import os_update_custody as custody ROOT = pathlib.Path('/var/lib/archipelago-os-update') VERIFIER = pathlib.Path('/usr/libexec/archipelago-os-release-verifier') INVENTORY = '${binary:Package}\t${Version}\t${Architecture}\t${db:Status-Abbrev}\n' def inventory(): return command(['/usr/bin/dpkg-query', '-W', '-f=' + INVENTORY]).encode() def boot_id(): return pathlib.Path('/proc/sys/kernel/random/boot_id').read_text().strip() def file_hash(path): import hashlib with open(path, 'rb') as source: return hashlib.file_digest(source, 'sha256').hexdigest() class ReconciliationRequired(RuntimeError): pass class Executor: def __init__(self, binding, create=True): keys(binding, 'version plan_sha256 manifest_sha256 authority_digest verifier_sha256') require(type(binding['version']) is int and binding['version'] == 1, 'OS binding version') for field in ('plan_sha256', 'manifest_sha256', 'authority_digest', 'verifier_sha256'): hashed(binding[field]) custody.refuse_legacy() self._pins = [custody.Pin(ROOT)] self.binding = binding self.source = ROOT / 'plans' / binding['plan_sha256'] self._pins.append(custody.Pin(self.source)) raw = read(self.source / 'plan.json') require(digest(raw) == binding['plan_sha256'], 'release plan hash') self.plan = validate(json.loads(raw)) raw = read(self.source / 'manifest.json') require(digest(raw) == binding['manifest_sha256'], 'release manifest hash') manifest = json.loads(raw) require(manifest.get('os_update') == dict(version=1, plan_sha256=binding['plan_sha256']), 'manifest plan binding') require(digest(read(VERIFIER)) == binding['verifier_sha256'], 'appointed verifier changed') command([str(VERIFIER), 'ceremony', 'verify', str(self.source / 'manifest.json')]) require(digest(read(self.source / 'preservation.json')) == self.plan['preservation_sha256'], 'preservation authority') preservation = json.loads(read(self.source / 'preservation.json')) require(preservation.get('authority_digest') == binding['authority_digest'] and preservation.get('service_policy') == self.plan['service_policy'] and preservation.get('preserve_wallets_data_identity_stop_uninstall') is True, 'external maintenance/preservation/writer exclusion authority required') self.root = ROOT / 'transactions' / binding['authority_digest'] self._pins.append(custody.Pin(self.root.parent)) self.status_path = self.root / 'status.json' initial = dict( version=1, plan_sha256=binding['plan_sha256'], authority_digest=binding['authority_digest'], phase=None, sequence=0, events=[]) pin, self.status = custody.transaction(self.root, initial, create) if pin is not None: self._pins.append(pin) require(self.status['plan_sha256'] == binding['plan_sha256'] and self.status['authority_digest'] == binding['authority_digest'], 'authority already bound') def check_custody(self): custody.refuse_legacy() for pin in getattr(self, '_pins', ()): pin.check() if hasattr(self, '_pins') and self.root.exists(): require(digest(read(self.status_path)) == digest(canonical(self.status)), 'OS journal changed or removed; reconciliation required') def close(self): for pin in getattr(self, '_pins', ()): pin.close() def journal(self, phase, **details): self.check_custody() self.status['phase'] = phase self.status['sequence'] += 1 self.status['events'].append(dict(phase=phase, details=details)) atomic(self.status_path, canonical(self.status)) checkpoint = getattr(self, 'checkpoint', None) if checkpoint is not None: checkpoint() # Live D09 owner anchors intent before any dpkg effects. def apt(self, *args): self.check_custody() # APT_CONFIG is read before apt.conf.d. Both host configuration entrypoints # are disabled; no inherited hooks, sources, proxy, shell or apt arguments. env = dict(PATH='/usr/sbin:/usr/bin:/sbin:/bin', LC_ALL='C', APT_CONFIG=str(self.root / 'apt.conf'), DEBIAN_FRONTEND='noninteractive', NEEDRESTART_MODE='l') result = subprocess.run(['/usr/bin/apt-get', *args], env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) atomic(self.root / ('apt-%016d-%s.log' % (self.status['sequence'], os.urandom(8).hex())), (result.stdout + '\n' + result.stderr).encode()) require(result.returncode == 0, 'APT failed; inspect retained apt log') return result.stdout def configure(self): self.check_custody() for name in ('lists', 'archives', 'empty', 'lists/partial', 'archives/partial'): pin, _ = custody.mkdir(self.root / name) pin.close() sources = [] for repo in self.plan['repositories']: components = sorted({x['path'].split('/')[0] for x in repo['indexes']}) sources.append('deb [signed-by=' + KEYRING + '] file:' + str(self.root / 'mirror' / repo['pocket']) + ' ' + repo['pocket'] + ' ' + ' '.join(components)) atomic(self.root / 'sources.list', ('\n'.join(sources) + '\n').encode()) config = { 'Dir::Etc::main': '-', 'Dir::Etc::parts': str(self.root / 'empty'), 'Dir::Etc::sourcelist': str(self.root / 'sources.list'), 'Dir::Etc::sourceparts': str(self.root / 'empty'), 'Dir::Etc::preferences': '-', 'Dir::Etc::preferencesparts': str(self.root / 'empty'), 'Dir::Etc::trusted': '-', 'Dir::Etc::trustedparts': str(self.root / 'empty'), 'Dir::State::lists': str(self.root / 'lists'), 'Dir::Cache::archives': str(self.root / 'archives'), 'Dir::Cache::pkgcache': '', 'Dir::Cache::srcpkgcache': '', 'APT::Architecture': self.plan['architecture'], 'APT::Sandbox::User': 'root', 'APT::Get::AllowUnauthenticated': 'false', 'Acquire::AllowInsecureRepositories': 'false', 'Acquire::AllowDowngradeToInsecureRepositories': 'false', 'Acquire::Check-Valid-Until': 'true', 'Acquire::Check-Date': 'true', 'Acquire::Languages': 'none', 'APT::Get::AutomaticRemove': 'false', } text = '\n'.join(key + ' ' + json.dumps(value) + ';' for key, value in config.items()) text += '\nDPkg::Options { "--force-confold"; };\n' atomic(self.root / 'apt.conf', text.encode()) def preflight(self): self.check_custody() policy = self.plan.get('manager_reexec') if policy and not any(p['name'] == 'systemd' for p in self.plan['packages']): actual = command(['/usr/bin/dpkg-query', '-W', '-f=${Version}\t${Architecture}', 'systemd']) require(actual == policy['package_version'] + '\t' + policy['architecture'], 'manager reexec package differs from installed reviewed version') require(file_hash('/proc/1/exe') == policy['expected_executable_sha256'], 'unchanged manager executable differs from signed plan') require(command(['/usr/bin/dpkg', '--print-architecture']).strip() == self.plan['architecture'], 'host architecture') require(not command(['/usr/bin/dpkg', '--print-foreign-architectures']).strip(), 'multiarch full-suite planning requires a separate contract') osrelease = dict(line.split('=', 1) for line in pathlib.Path('/etc/os-release').read_text().splitlines() if '=' in line) require(osrelease.get('ID', '').strip('"') == 'debian' and osrelease.get('VERSION_CODENAME', '').strip('"') == self.plan['suite'], 'silent distribution upgrade refused') current = inventory() require(digest(current) == self.plan['before_sha256'], 'dpkg inventory changed') require(not command(['/usr/bin/dpkg', '--audit']).strip(), 'partial dpkg state requires reconciliation') installed = set() for line in current.decode().splitlines(): name, version, arch, status = line.split('\t') require(status.startswith(('ii ', 'hi ', 'rc ')), 'unresolved dpkg state') if status.startswith(('ii ', 'hi ')): installed.add(name.split(':')[0]) kernel = self.plan['boot']['package'] if kernel not in self.plan['packages']: identity = command(['/usr/bin/dpkg-query', '-W', '-f=${Version}\t${Architecture}\t${db:Status-Abbrev}', kernel['name']]) require(identity == kernel['version'] + '\t' + kernel['architecture'] + '\tii ', 'retained target kernel not configured') changes = {x['name'].split(':')[0] for x in self.plan['transaction'] if x['operation'] in ('Inst', 'Remv')} custody = {x['name']: x['disposition'] for x in self.plan['custody']} for name in installed | changes: if protected(name): require(custody.get(name) == ('upgrade-reviewed' if name in changes else 'retain'), 'unreviewed custody disposition: ' + name) if re.match(r'linux-image-[0-9]', name) and name in installed: require(name not in changes, 'installed kernel must remain intact') require(platform.release() == self.plan['boot']['running_release'], 'running kernel changed') for path, sha in self.plan['boot']['retained_files'].items(): require(file_hash(path) == sha, 'old boot provenance changed') def solver(self): actual = simulation(self.apt('--simulate', 'dist-upgrade')) require(actual == self.plan['transaction'], 'whole dependency transaction differs from release plan') installs = {(x['name'].split(':')[0], x['version']) for x in actual if x['operation'] == 'Inst'} require(installs == {(x['name'], x['version']) for x in self.plan['packages']}, 'incomplete archive transaction') def prepare(self): self.check_custody() if self.status['phase'] in ('VERIFIED', 'REBOOT_REQUIRED', 'POSTBOOT_CONFIRMED'): return self.status if self.status['phase'] is not None: raise ReconciliationRequired('interrupted preparation/apply requires external reconciliation') self.journal('PREPARE', boot_id=boot_id()) self.preflight() atomic(self.root / 'before-inventory', inventory()) self.journal('DOWNLOAD') acquire(self.plan, self.root / 'mirror') self.configure() self.apt('update') self.solver() self.apt('--download-only', '--assume-yes', 'dist-upgrade') # APT's file method may use the mirror directly instead of populating # archives. Keep independent, durable bytes under its canonical cache name. for package in self.plan['packages']: name = package['name'] + '_' + package['version'].replace(':', '%3a') + '_' + package['architecture'] + '.deb' raw = read(self.root / 'mirror' / package['pocket'] / package['filename']) require(digest(raw) == package['sha256'], 'archive changed during preparation') atomic(self.root / 'archives' / name, raw) self.verify_archives() # Durable retained boot preimages are recovery evidence, not a DB rollback. for path, sha in self.plan['boot']['retained_files'].items(): atomic(self.root / ('retained-' + digest(path.encode())), pathlib.Path(path).read_bytes()) for directory, dirs, files in os.walk(self.root, topdown=False): for name in files: fd = secure(pathlib.Path(directory) / name) try: os.fsync(fd) finally: os.close(fd) sync_dir(directory) self.journal('VERIFIED') return self.status def verify_archives(self): expected = {p['sha256'] for p in self.plan['packages']} actual = {file_hash(p) for p in (self.root / 'archives').glob('*.deb')} require(actual == expected, 'offline archive set mismatch') def apply(self): self.check_custody() if self.status['phase'] in ('REBOOT_REQUIRED', 'POSTBOOT_CONFIRMED'): return self.status if self.status['phase'] != 'VERIFIED': raise ReconciliationRequired('OS authority consumed or preparation incomplete; never retry dpkg') # Re-evaluate signed metadata dates even if VERIFIED sat pending for days. self.apt('update') with PackageLease(self.root / 'apt.conf') as lease: self.preflight() self.verify_archives() self.solver() lease.plan(self.plan['transaction']) self.journal('APPLY_INTENT', boot_id=boot_id(), package_locks=lease.observation) try: self.check_custody() lease.install() self.journal('VERIFY') self.verify_installed() boot = self.boot_provenance() self.journal('REBOOT_REQUIRED', boot_id=boot_id(), boot=boot, inventory_sha256=digest(inventory())) except BaseException as exc: if self.status['phase'] not in ('REBOOT_REQUIRED', 'POSTBOOT_CONFIRMED'): self.journal('RECONCILIATION_REQUIRED', error=type(exc).__name__) raise return self.status def verify_installed(self): require(not command(['/usr/bin/dpkg', '--audit']).strip(), 'dpkg remains partial') rows = {} for line in inventory().decode().splitlines(): name, version, arch, status = line.split('\t') rows[(name.split(':')[0], arch)] = (version, status) for package in self.plan['packages']: require(rows.get((package['name'], package['architecture'])) == (package['version'], 'ii '), 'installed package mismatch') for name in self.plan['removals']: require(not any(n == name.split(':')[0] and status.startswith(('ii', 'hi')) for (n, a), (v, status) in rows.items()), 'removal incomplete') expected = {} for line in read(self.root / 'before-inventory').decode().splitlines(): name, version, arch, status = line.split('\t') if status.startswith(('ii ', 'hi ')): expected[(name.split(':')[0], arch)] = version for name in self.plan['removals']: for identity in list(expected): if identity[0] == name.split(':')[0]: del expected[identity] for package in self.plan['packages']: expected[(package['name'], package['architecture'])] = package['version'] actual = {identity: version for identity, (version, status) in rows.items() if status.startswith(('ii ', 'hi '))} require(actual == expected, 'unexpected whole-system package drift') for path, sha in self.plan['boot']['retained_files'].items(): if path != '/boot/grub/grub.cfg': require(file_hash(path) == sha, 'old boot image lost') def boot_provenance(self): release = self.plan['boot']['expected_release'] package = self.plan['boot']['package'] identity = command(['/usr/bin/dpkg-query', '-W', '-f=${Version}\t${Architecture}\t${db:Status-Abbrev}', package['name']]) require(identity == package['version'] + '\t' + package['architecture'] + '\tii ', 'expected kernel package not configured') archive = self.root / 'mirror' / package['pocket'] / package['filename'] require(file_hash(archive) == package['sha256'], 'kernel archive changed') hashes = {} process = subprocess.Popen(['/usr/bin/dpkg-deb', '--fsys-tarfile', str(archive)], stdout=subprocess.PIPE) try: with tarfile.open(fileobj=process.stdout, mode='r|') as tar: for member in tar: if member.isfile(): path = '/' + member.name.removeprefix('./') if path.startswith(('/boot/', '/lib/modules/', '/usr/lib/modules/')): sha = digest(tar.extractfile(member).read()) require(file_hash(path) == sha, 'kernel payload provenance mismatch') hashes[path] = sha require(process.wait() == 0, 'kernel archive read failed') finally: process.stdout.close() if process.poll() is None: process.kill() process.wait() require('/boot/vmlinuz-' + release in hashes and any('/modules/' in x for x in hashes), 'kernel payload incomplete') for path in ('/boot/initrd.img-' + release, '/boot/grub/grub.cfg'): hashes[path] = file_hash(path) for path in pathlib.Path('/boot').rglob('*'): if path.is_file() and not path.is_symlink() and path.suffix.lower() == '.efi': hashes[str(path)] = file_hash(path) return dict(release=release, package_version=package['version'], archive_sha256=package['sha256'], files=hashes) def confirm_postboot(self): self.check_custody() require(self.status['phase'] == 'REBOOT_REQUIRED', 'no pending OS reboot') observation = self.observe_postboot() self.journal('POSTBOOT_CONFIRMED', boot_id=boot_id(), observation_sha256=digest(canonical(observation))) return self.status def observe_postboot(self): self.check_custody() event = next(e['details'] for e in reversed(self.status['events']) if e['phase'] == 'REBOOT_REQUIRED') require(boot_id() != event['boot_id'] and platform.release() == event['boot']['release'], 'expected new boot not observed') require(re.search(r'\bDebian ' + re.escape(event['boot']['package_version']) + r'(?:\s|$)', platform.version()), 'running Debian build version differs from authenticated kernel archive') for path, sha in event['boot']['files'].items(): require(file_hash(path) == sha, 'boot payload changed') observation = json.loads(read(self.source / 'postboot-preservation.json')) require(observation == dict(version=1, authority_digest=self.binding['authority_digest'], plan_sha256=self.binding['plan_sha256'], boot_id=boot_id(), preservation_verified=True), 'external postboot preservation observation required') self.verify_installed() return observation def reconcile(self): self.check_custody() """Evidence-only reconciliation. Never dpkg --configure -a or apt -f.""" require(self.status['phase'] in ('APPLY_INTENT', 'VERIFY', 'RECONCILIATION_REQUIRED'), 'no interrupted OS effects to reconcile') observed = inventory() atomic(self.root / ('reconciliation-inventory-%016d' % (self.status['sequence'] + 1)), observed) self.journal('RECONCILIATION_REQUIRED', inventory_sha256=digest(observed), dpkg_audit=command(['/usr/bin/dpkg', '--audit']), automatic_retry_permitted=False) return self.status def stage(binding, action, phase='prepare', checkpoint=None): """Called only by privileged D09 worker under its live-owner continuity lock.""" if 'os_update' not in binding: return None require(action == 'APPLY', 'OS update cannot be restored by userspace rollback') require(phase in ('prepare', 'apply', 'postboot', 'reconcile'), 'fixed OS operation') custody.refuse_legacy() os.close(secure(ROOT, True)) root_pin = custody.Pin(ROOT) lock = os.open('executor.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600, dir_fd=root_pin.fd) executor = None try: root_pin.check() checked = secure(ROOT / 'executor.lock') try: require((os.fstat(checked).st_dev, os.fstat(checked).st_ino) == (os.fstat(lock).st_dev, os.fstat(lock).st_ino), 'OS executor lock replaced') finally: os.close(checked) fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) executor = Executor(binding['os_update']) executor.checkpoint = checkpoint return {'prepare': executor.prepare, 'apply': executor.apply, 'postboot': executor.confirm_postboot, 'reconcile': executor.reconcile}[phase]() finally: if executor is not None: executor.close() os.close(lock) root_pin.close()