"""Protected admission and authenticated D09 OS projection (no public mutation).""" import json import os import pathlib import platform import subprocess from d09_store import atomic, read, secure, sync_dir from d09_wire import release_digest from os_update_plan import digest, require from os_update_executor import Executor, ROOT, VERIFIER, command, boot_id, stage from os_update_lock import probe import os_update_custody as custody MODULES = ('os_update_plan.py', 'os_update_repository.py', 'os_update_executor.py', 'os_update_owner.py', 'os_update_lock.py', 'os_update_custody.py', 'os_update_bus.py', 'os_update_manager.py') OS_RUNTIME_SECONDS = 21600 def install_inputs(binding, incoming): """Appointed root provisioning only; exact fixed names, no overwrite/reset.""" custody.refuse_legacy() # Check/create only the independent root before any verifier/input writes. root_pin, _ = custody.mkdir(ROOT) root_pin.check() spec = binding['os_update'] from os_update_plan import hashed, keys, validate keys(spec, 'version plan_sha256 manifest_sha256 authority_digest verifier_sha256') require(type(spec['version']) is int and spec['version'] == 1, 'OS binding version') for key in ('plan_sha256', 'manifest_sha256', 'authority_digest', 'verifier_sha256'): hashed(spec[key]) raw = read(incoming / 'plan.json') require(digest(raw) == spec['plan_sha256'], 'incoming plan hash') plan = validate(json.loads(raw)) files = {'plan.json': raw, 'manifest.json': read(incoming / 'manifest.json'), 'preservation.json': read(incoming / 'preservation.json')} require(digest(files['manifest.json']) == spec['manifest_sha256'] and digest(files['preservation.json']) == plan['preservation_sha256'], 'incoming OS evidence hash') verifier = read(incoming / 'verifier') require(digest(verifier) == spec['verifier_sha256'] and verifier.startswith(b'\x7fELF'), 'incoming native verifier hash/type') if VERIFIER.exists() or VERIFIER.is_symlink(): require(read(VERIFIER) == verifier, 'existing verifier differs; separate custody transition required') else: atomic(VERIFIER, verifier, 0o755) for directory in (ROOT, ROOT / 'plans', ROOT / 'transactions', ROOT / 'plans' / spec['plan_sha256']): root_pin.check() pin, _ = custody.mkdir(directory) pin.close() custody_path = ROOT / 'verifier-custody.json' expected = dict(version=1, verifier_sha256=spec['verifier_sha256'], prerequisite_backend_sha256=spec['verifier_sha256']) if custody_path.exists() or custody_path.is_symlink(): require(json.loads(read(custody_path)) == expected, 'verifier custody changed') else: require(spec['verifier_sha256'] == binding['prerequisite']['backend_sha256'], 'initial verifier must be the already appointed prerequisite native artifact') from os_update_plan import canonical atomic(custody_path, canonical(expected)) for name, data in files.items(): root_pin.check() target = ROOT / 'plans' / spec['plan_sha256'] / name if target.exists() or target.is_symlink(): require(read(target) == data, 'existing OS input differs; preserve evidence') else: atomic(target, data) root_pin.close() def qualify(binding, lock_probe=False): custody.refuse_legacy() root_pin = custody.Pin(ROOT) spec = binding['os_update'] require(json.loads(read(ROOT / 'verifier-custody.json')) == dict(version=1, verifier_sha256=spec['verifier_sha256'], prerequisite_backend_sha256=spec['verifier_sha256']), 'release verifier has no appointed prerequisite custody') executor = Executor(spec, create=False) root_pin.check() root_pin.close() manifest = json.loads(read(executor.source / 'manifest.json')) identity = [manifest['version'], sorted([[c[k] for k in ('name', 'new_version', 'sha256', 'size_bytes')] for c in manifest['components']])] require(release_digest(identity) == release_digest(binding['release_identity']), 'OS signed manifest/release identity differs') component = [c for c in identity[1] if c[0] == 'os-update-plan.json'] require(component == [['os-update-plan.json', identity[0], spec['plan_sha256'], len(read(executor.source / 'plan.json'))]], 'OS plan release component differs') # The appointed verifier is an existing native pinned-root verifier, not an # application-selected script. Protected incoming provisioning pins its hash. require(read(VERIFIER).startswith(b'\x7fELF') and os.access(VERIFIER, os.X_OK), 'native release verifier custody') for name in MODULES: os.close(secure(pathlib.Path(__file__).with_name(name))) import apt_pkg # Installed runtime prerequisite, never silently apt-installed. require(hasattr(apt_pkg, 'SystemLock'), 'python3-apt lock runtime missing') if lock_probe: with probe() as locks: executor.preflight() return executor, dict(version=1, plan_sha256=spec['plan_sha256'], manifest_sha256=spec['manifest_sha256'], verifier_sha256=spec['verifier_sha256'], authority_digest=spec['authority_digest'], boot_id=boot_id(), package_locks=locks) return executor, None def projection(binding, checkpoints=None): value = dict(version=1, supported=False, plan_sha256=None, phase='UNKNOWN', installed_os=None, installed_kernel=None, running_kernel=None, reboot_required=None, post_boot_verified=False) if 'os_update' not in binding: return value try: executor, _ = qualify(binding) if checkpoints is not None and executor.status['phase'] is not None: history = [c for c in checkpoints if c['authority_digest'] == binding['os_update']['authority_digest']] require(history and history[-1]['status_sha256'] == digest(read(executor.status_path)), 'unanchored OS journal') phase = executor.status['phase'] or 'NO_OS_TRANSACTION' require(phase in ('NO_OS_TRANSACTION', 'PREPARE', 'DOWNLOAD', 'VERIFIED', 'APPLY_INTENT', 'VERIFY', 'REBOOT_REQUIRED', 'POSTBOOT_CONFIRMED', 'RECONCILIATION_REQUIRED'), 'unknown OS phase') value.update(supported=True, plan_sha256=binding['os_update']['plan_sha256'], phase=phase) # Installed versions are live dpkg observations, never desired plan values. version = command(['/usr/bin/dpkg-query', '-W', '-f=${Version}\t${db:Status-Abbrev}', 'base-files']) installed, status = version.split('\t') if status == 'ii ': value['installed_os'] = 'Debian ' + executor.plan['suite'] + ' / base-files ' + installed package = executor.plan['boot']['package'] try: actual = command(['/usr/bin/dpkg-query', '-W', '-f=${Version}\t${db:Status-Abbrev}', package['name']]) except subprocess.CalledProcessError: actual = None # A new target kernel is normally not installed yet. if actual == package['version'] + '\tii ': value['installed_kernel'] = executor.plan['boot']['expected_release'] value['running_kernel'] = platform.release() if phase == 'REBOOT_REQUIRED': value['reboot_required'] = True if phase == 'POSTBOOT_CONFIRMED': try: executor.observe_postboot() except Exception: value.update(phase='REBOOT_REQUIRED', reboot_required=True) else: value.update(reboot_required=False, post_boot_verified=True) except Exception: # A malformed/stale/unqualified projection must not leave a stale green. value.update(phase='UNKNOWN', reboot_required=None, post_boot_verified=False) return value def postboot(binding): executor, _ = qualify(binding) if executor.status['phase'] == 'POSTBOOT_CONFIRMED': executor.observe_postboot() # Lost-reply retry is fresh observation only. else: stage(binding, 'APPLY', 'postboot') return dict(plan_sha256=binding['os_update']['plan_sha256'], boot_id=boot_id(), status_sha256=digest(read(executor.status_path)))