#!/usr/bin/env python3 """Independent root application-pair owner. No app/package/OS operations. Production is bundled with public_update_prepare.py by the installed binary. Only a freshly authenticated immutable snapshot may enter submit(). Recovery uses root-owned preimages and a write-ahead journal, never legacy app backups. """ import contextlib import shutil import time import urllib.request TRANSACTIONS = Path('/var/lib/archipelago-public-transactions') BINARY = Path('/usr/local/bin/archipelago') FRONTEND = Path('/opt/archipelago/web-ui') UNIT_DIR = Path('/etc/systemd/system') DROPIN = UNIT_DIR / 'archipelago.service.d/70-public-application-update.conf' OWNER_DROPINS = (DROPIN, *(UNIT_DIR / (unit + '.d/70-public-application-update.conf') for unit in MAINTENANCE_UNITS)) BOOT_WANTS = UNIT_DIR / 'basic.target.wants' TERMINAL = {'committed', 'restored', 'aborted'} ENV = {'PATH': '/usr/sbin:/usr/bin:/sbin:/bin', 'HOME': '/nonexistent', 'LANG': 'C'} def command(*args, timeout=90): return subprocess.check_output(args, env=ENV, cwd='/', timeout=timeout, text=True) def atomic(path, data, mode=0o600): temporary = path.with_name('.' + path.name + '-' + uuid.uuid4().hex) with temporary.open('xb') as stream: stream.write(data) stream.flush() os.fchmod(stream.fileno(), mode) os.fsync(stream.fileno()) os.replace(temporary, path) sync_directory(path.parent) def journal(tx, value, phase): value = dict(value, phase=phase) atomic(tx / 'state.json', json.dumps(value, sort_keys=True).encode()) return value def secure_read(path, limit=MAX_DOCUMENT): fd = trusted_open(path) with os.fdopen(fd, 'rb') as stream: raw = stream.read(limit + 1) if len(raw) > limit: raise ValueError('oversized protected record') return raw @contextlib.contextmanager def owner_lock(): os.close(trusted_open(TRANSACTIONS.parent, directory=True)) TRANSACTIONS.mkdir(mode=0o755, exist_ok=True) os.close(trusted_open(TRANSACTIONS, directory=True)) fd = os.open(TRANSACTIONS / 'lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) try: info = os.fstat(fd) if not stat.S_ISREG(info.st_mode) or info.st_uid or info.st_mode & 0o077: raise ValueError('untrusted worker lock') fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) yield finally: os.close(fd) def load_transaction(name): if not re.fullmatch('[0-9a-f]{32}', name): raise ValueError('invalid transaction pointer') tx = TRANSACTIONS / name state = json.loads(secure_read(tx / 'state.json'), object_pairs_hook=pairs) if state.get('schema') != 1 or state.get('id') != name: raise ValueError('invalid transaction state') return tx, state def current(): try: name = secure_read(TRANSACTIONS / 'current').decode() except FileNotFoundError: return None, None return load_transaction(name) def check_owner_dropins(previous, incoming=None): for path in OWNER_DROPINS: allowed = [owner_dropin(value, path) for value in (previous, incoming) if value] if os.path.lexists(path) and secure_read(path) not in allowed: raise ValueError('foreign public-update service drop-in') def owner_dropin(state, path): value = state['dropin'] if state.get('runtime_startup_guard'): if path == DROPIN: # Historical binaries unconditionally repromote the frontend's # runtime on every startup, deleting locally retained app sources. # Hide only that bootstrap input inside the manager namespace. The # host owner and nginx still see the exact authenticated frontend. value += '[Service]\nInaccessiblePaths=-/opt/archipelago/web-ui/archipelago-runtime\n' else: value += 'ConditionPathExists=!/var/lib/archipelago-public-transactions/maintenance-paused\n' return value.encode() def install_owner_dropins(state, *, reload=True): previous = load_transaction(state['previous_id'])[1] if state.get('previous_id') else None check_owner_dropins(previous, state) for path in OWNER_DROPINS: path.parent.mkdir(mode=0o755, exist_ok=True) os.close(trusted_open(path.parent, directory=True)) atomic(path, owner_dropin(state, path), 0o644) if reload: command('/usr/bin/systemctl', 'daemon-reload') def retain_boot_recovery(unit): BOOT_WANTS.mkdir(mode=0o755, exist_ok=True) os.close(trusted_open(BOOT_WANTS, directory=True)) link = BOOT_WANTS / unit target = str(UNIT_DIR / unit) if os.path.lexists(link): info = link.lstat() if not stat.S_ISLNK(info.st_mode) or info.st_uid or os.readlink(link) != target: raise ValueError('foreign public recovery boot link') else: link.symlink_to(target) sync_directory(BOOT_WANTS) def digest_file(path): fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) with os.fdopen(fd, 'rb') as stream: if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode): raise ValueError('nonregular code file') return hashlib.file_digest(stream, 'sha256').hexdigest() def copy_file_fd(fd, destination, executable=False): before = os.fstat(fd) if not stat.S_ISREG(before.st_mode) or before.st_size > MAX_COMPONENT: raise ValueError('invalid preimage file') digest = hashlib.sha256() with os.fdopen(os.dup(fd), 'rb') as source, destination.open('xb') as target: copied = 0 while block := source.read(1024 * 1024): copied += len(block) if copied > before.st_size: raise ValueError('preimage grew during snapshot') target.write(block) digest.update(block) target.flush() os.fchmod(target.fileno(), 0o555 if executable else 0o444) os.fsync(target.fileno()) after = os.fstat(fd) if copied != before.st_size or (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns) != (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns): raise ValueError('preimage mutated during snapshot') return digest.hexdigest() def durable_mode(path, mode): fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) try: os.fchmod(fd, mode) os.fsync(fd) finally: os.close(fd) def tree(path, destination=None, with_size=False): """No-follow fd walk; snapshot or identify code, including executable bits. Never follows app-controlled links or trusts path-based copytree. A changing directory/file refuses, and partial copies are retained as failure evidence. """ entries = [] size = 0 def walk(fd, prefix, out): nonlocal size before = os.fstat(fd) for name in sorted(os.listdir(fd)): child = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=fd) try: info = os.fstat(child) relative = prefix + name if stat.S_ISDIR(info.st_mode): entries.append([relative, 'directory']) target = out / name if out else None if target: target.mkdir(mode=0o700) walk(child, relative + '/', target) if target: target.chmod(0o555) sync_directory(target) elif stat.S_ISREG(info.st_mode) and not info.st_mode & 0o6000: size += info.st_size executable = bool(info.st_mode & 0o111) if out: digest = copy_file_fd(child, out / name, executable) else: with os.fdopen(os.dup(child), 'rb') as source: digest = hashlib.file_digest(source, 'sha256').hexdigest() entries.append([relative, 'file', executable, info.st_size, digest]) else: raise ValueError('unsupported frontend preimage inode') if len(entries) > 20000 or size > 4 * 1024**3: raise ValueError('frontend preimage exceeds bounds') finally: os.close(child) after = os.fstat(fd) if (before.st_mtime_ns, before.st_ctime_ns) != (after.st_mtime_ns, after.st_ctime_ns): raise ValueError('frontend directory mutated during snapshot') fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: if destination: destination.mkdir(mode=0o700) walk(fd, '', destination) if destination: destination.chmod(0o555) sync_directory(destination) finally: os.close(fd) digest = hashlib.sha256(json.dumps(entries, separators=(',', ':')).encode()).hexdigest() return (digest, size + 4096 * len(entries)) if with_size else digest def extract(archive_path, destination, require_declaration=True): validate_archive(archive_path, require_declaration) destination.mkdir(mode=0o700) with tarfile.open(archive_path, 'r:gz') as archive: for member in archive: name = member.name.removeprefix('./').rstrip('/') if name in ('', '.'): continue target = destination / name target.parent.mkdir(parents=True, exist_ok=True) if member.isdir(): target.mkdir(exist_ok=True) else: with archive.extractfile(member) as source, target.open('xb') as output: shutil.copyfileobj(source, output, 1024 * 1024) output.flush() os.fchmod(output.fileno(), 0o555 if member.mode & 0o111 else 0o444) os.fsync(output.fileno()) for directory, _, _ in os.walk(destination, topdown=False): Path(directory).chmod(0o555) sync_directory(Path(directory)) return tree(destination) def pair_identity(binary, frontend): return {'binary': digest_file(binary), 'frontend': tree(frontend)} def require_space(tx): # Budget both forward and restore copies without deleting retired evidence. # Sum requirements for shared filesystems rather than counting their free # space independently. No reservation is inferred from app-writable state. old_frontend = tree(FRONTEND, with_size=True)[1] new_frontend = tree(tx / 'new-frontend', with_size=True)[1] budgets = {} for path, amount in [(tx, BINARY.stat().st_size + 2 * old_frontend), (FRONTEND.parent, old_frontend + new_frontend), (BINARY.parent, BINARY.stat().st_size + (tx / 'new-binary').stat().st_size)]: device = path.stat().st_dev previous = budgets.get(device, (path, 256 * 1024**2)) budgets[device] = (path, previous[1] + amount) if json.loads(secure_read(tx / 'signed-manifest.json')).get('application_update') == RUNTIME_POLICY: for destination, relative, kind, required in RUNTIME_OBJECTS.values(): old = Path(destination) new = tx / 'new-frontend/archipelago-runtime' / relative if relative else None def size(path): if path is None or not os.path.lexists(path): return 0 identity = runtime_identity(path) return tree(path, with_size=True)[1] if identity['kind'] == 'directory' else path.stat().st_size + 4096 old_size, new_size = size(old), size(new) parent = old.parent while not parent.exists(): parent = parent.parent for path, amount in ((tx, old_size + new_size), (parent, old_size + new_size)): device = path.stat().st_dev previous = budgets.get(device, (path, 256 * 1024**2)) budgets[device] = (path, previous[1] + amount) for path, required in budgets.values(): free = os.statvfs(path) if free.f_bavail * free.f_frsize < required: raise ValueError('insufficient space for protected preimages, installation and recovery') def check_preimages(tx, state): for label in ('old', 'new'): if pair_identity(tx / (label + '-binary'), tx / (label + '-frontend')) != state[label]: raise ValueError('protected pair preimage identity changed') def custody_frontend_parent(): os.close(trusted_open(FRONTEND.parent.parent, directory=True)) fd = os.open(FRONTEND.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: info = os.fstat(fd) import pwd if info.st_uid not in (0, pwd.getpwnam('archipelago').pw_uid) or info.st_mode & 0o002: raise ValueError('nonstandard application prefix owner') # Only the code prefix directory, never its children or any app data. # Root custody prevents the service user renaming our installation paths. os.fchown(fd, 0, 0) os.fchmod(fd, 0o755) os.fsync(fd) finally: os.close(fd) def replace_pair(tx, state, label, *, unprepared_adoption=False): if unprepared_adoption: if (not state.get('adoption') or label != 'old' or state['phase'] not in ('queued', 'preparing') or pair_identity(tx / 'old-binary', tx / 'old-frontend') != state['old']): raise ValueError('first-generation abort lacks authenticated old preimages') # Derived candidate staging may be incomplete. It is never a source for # this restore; both restored objects are the authenticated old pair. else: check_preimages(tx, state) os.close(trusted_open(BINARY.parent, directory=True)) os.close(trusted_open(FRONTEND.parent, directory=True)) # Each target may be absent at an interrupted directory rename, or match # either transaction image. Foreign modifications are never overwritten. for key, target, identify in [('binary', BINARY, digest_file), ('frontend', FRONTEND, tree)]: allowed = [state['old'][key], state['new'][key]] if state.get('adoption'): allowed.append(state['initial'][key]) if os.path.lexists(target) and identify(target) not in allowed: raise ValueError('foreign installed code edit; recovery requires operator review') for key, target in [('frontend', FRONTEND), ('binary', BINARY)]: temporary = target.with_name('.public-new-' + uuid.uuid4().hex) if key == 'frontend': tree(tx / (label + '-frontend'), temporary) # Normalize readable deployed modes; preserve executable bits. for directory, _, files in os.walk(temporary): for name in files: p = Path(directory) / name durable_mode(p, 0o755 if p.stat().st_mode & 0o111 else 0o644) durable_mode(Path(directory), 0o755) if os.path.lexists(target): target.rename(target.with_name('.public-retired-' + uuid.uuid4().hex)) sync_directory(target.parent) else: fd = trusted_open(tx / (label + '-binary')) try: copy_file_fd(fd, temporary, True) finally: os.close(fd) durable_mode(temporary, 0o755) os.replace(temporary, target) sync_directory(target.parent) if pair_identity(BINARY, FRONTEND) != state[label]: raise ValueError('installed pair identity mismatch') def manager_active(): return command('/usr/bin/systemctl', 'show', 'archipelago.service', '-p', 'ActiveState', '--value').strip() == 'active' def manager_adoptable(): # Type=notify remains activating until the initial migration hands off. # Bind that starting process to the actual installed executable inode. fields = command('/usr/bin/systemctl', 'show', 'archipelago.service', '-p', 'ActiveState', '-p', 'MainPID') properties = dict(line.split('=', 1) for line in fields.splitlines() if '=' in line) if properties.get('ActiveState') not in ('active', 'activating'): return False pid = int(properties.get('MainPID', '0')) return pid > 0 and os.path.samefile(f'/proc/{pid}/exe', BINARY) def stop_manager(): command('/usr/bin/systemctl', 'stop', 'archipelago.service', timeout=180) if manager_active(): raise ValueError('manager did not stop') def verify_running(tx, state, label): deadline = time.monotonic() + 180 last = None while time.monotonic() < deadline: if command('/usr/bin/systemctl', 'show', 'archipelago.service', '-p', 'ActiveState', '--value').strip() == 'failed': raise RuntimeError('manager failed after pair replacement') try: if not manager_active(): raise ValueError('manager not active') pid = int(command('/usr/bin/systemctl', 'show', 'archipelago.service', '-p', 'MainPID', '--value')) # Hash the actually executing inode, not a version string or stale # socket response. /proc/PID/exe is deliberately followed here. with open(f'/proc/{pid}/exe', 'rb') as stream: actual = hashlib.file_digest(stream, 'sha256').hexdigest() if actual != state[label]['binary']: raise ValueError('manager executing a different binary') opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) with opener.open('http://127.0.0.1:5678/health', timeout=3) as response: health = json.loads(response.read(65537)) if health.get('version') != state[label + '_version']: raise RuntimeError('running backend version differs from the admitted pair') if health.get('status') != 'ok': raise ValueError('backend health not ready') with opener.open('http://127.0.0.1/', timeout=3) as response: index = response.read(8 * 1024**2 + 1) expected_frontend = tx / (label + '-frontend') if tx else FRONTEND if hashlib.sha256(index).hexdigest() != digest_file(expected_frontend / 'index.html'): raise ValueError('served frontend differs from installed pair') if pair_identity(BINARY, FRONTEND) != state[label]: raise ValueError('installed pair changed during verification') if state.get('runtime'): runtime_verify_installed(state, label) return except (OSError, ValueError, subprocess.SubprocessError) as error: if str(last) != str(error): state['last_probe_error'] = str(error) if tx: journal(tx, state, state['phase']) last = error time.sleep(1) raise ValueError(f'post-restart pair verification failed: {last}') def recovery(tx, state, boot=False): require_legacy_service() if state['phase'] in ('queued', 'preparing'): if state.get('adoption'): # The historical installer replaced its pair before this owner # existed. Abort this first-generation handoff to its authenticated # old pair even if runtime preparation never became durable. if not boot: stop_manager() command('/usr/bin/systemctl', 'daemon-reload') replace_pair(tx, state, 'old', unprepared_adoption=True) retain_legacy_marker(tx, state) state = journal(tx, state, 'restored-pending-start') else: # Normal updates cannot mutate code before prepared is durable. state = journal(tx, state, 'aborted') elif state['phase'] not in ('restored', 'aborted'): state = journal(tx, state, 'recovering') if not boot: stop_manager() check_preimages(tx, state) if state.get('runtime'): runtime_hold_maintenance(state) runtime_replace(tx, state, 'old') replace_pair(tx, state, 'old') retain_legacy_marker(tx, state) state = journal(tx, state, 'restored-pending-start') if not boot and state['was_active']: command('/usr/bin/systemctl', 'start', 'archipelago.service', timeout=180) if state['phase'] == 'restored-pending-start': verify_running(tx, state, 'old') runtime_resume_units(state) state = journal(tx, state, 'restored') elif state['phase'] == 'aborted': runtime_resume_units(state) elif boot and state['phase'] == 'restored-pending-start': command('/usr/bin/systemctl', 'start', '--no-block', state['worker_unit']) return state def run_worker(expected_id=None): with owner_lock(): tx, state = current() if state and expected_id is not None and state['id'] != expected_id: return if not state or state['phase'] in TERMINAL: return if state['phase'] != 'queued' or state.get('recovery_requested'): try: recovery(tx, state) except Exception as error: tx, state = current() journal(tx, dict(state, failure=str(error)), state['phase']) raise return try: require_legacy_service() install_owner_dropins(state, reload=not state.get('adoption')) if not (manager_adoptable() if state.get('adoption') else manager_active()): if state.get('adoption'): recovery(tx, state) else: journal(tx, dict(state, was_active=False), 'aborted') return state = journal(tx, state, 'preparing') custody_frontend_parent() baseline = pair_identity(BINARY, FRONTEND) if state.get('adoption'): if baseline != state['initial']: raise ValueError('first-generation installed code changed before admission') check_preimages(tx, state) else: state['old'] = baseline verify_running(None, state, 'old') require_space(tx) stop_manager() if state.get('adoption'): # Reload only after the first manager has stopped. Reloading # file-backed units discards its EXTEND_TIMEOUT_USEC deadline. command('/usr/bin/systemctl', 'daemon-reload') if state.get('runtime_requested'): state = runtime_pause_units(tx, state) if not state.get('adoption'): fd = trusted_open(BINARY) try: copy_file_fd(fd, tx / 'old-binary', True) finally: os.close(fd) tree(FRONTEND, tx / 'old-frontend') state['old'] = pair_identity(tx / 'old-binary', tx / 'old-frontend') if pair_identity(BINARY, FRONTEND) != baseline: raise ValueError('installed code changed after baseline verification') # Preserve separately installed UI extras only if not superseded by # the signed archive. Copies originate from the frozen old tree. for name in ('aiui', 'archipelago-companion.apk'): old, new = tx / 'old-frontend' / name, tx / 'new-frontend' / name if old.exists() and not new.exists(): if old.is_dir(): tree(old, new) else: fd = trusted_open(old) try: copy_file_fd(fd, new, bool(old.stat().st_mode & 0o111)) finally: os.close(fd) state['new'] = pair_identity(tx / 'new-binary', tx / 'new-frontend') if state.get('runtime_requested'): state = runtime_prepare(tx, state) sync_directory(tx / 'new-frontend') sync_directory(tx) state = journal(tx, state, 'prepared') require_legacy_service() state = journal(tx, state, 'installing') check_preimages(tx, state) if state.get('runtime'): runtime_replace(tx, state, 'new') replace_pair(tx, state, 'new') state = journal(tx, state, 'verifying') command('/usr/bin/systemctl', 'start', 'archipelago.service', timeout=180) verify_running(tx, state, 'new') retain_legacy_marker(tx, state) runtime_resume_units(state) journal(tx, state, 'committed') except Exception as error: # Preserve the exact cause even when recovery itself fails. A killed # worker is retried by PID1; a reboot invokes recovery before manager. tx, state = current() state['failure'] = str(error) state = journal(tx, state, state['phase']) recovery(tx, state) raise def submit(identity, bundle): if not re.fullmatch('[0-9a-f]{64}', identity): raise ValueError('invalid prepared identity') require_legacy_service() ready = SNAPSHOTS / identity raw = secure_read(ready / 'signed-manifest.json') document = json.loads(raw, object_pairs_hook=pairs) old_version = authenticate(raw) components = validate_document(document, old_version) expected = hashlib.sha256(json.dumps([document['version'], sorted([ [c['name'], c['new_version'], c['sha256'], c['size_bytes']] for c in components])], separators=(',', ':'), ensure_ascii=True).encode()).hexdigest() if expected != identity: raise ValueError('prepared identity differs from authenticated release') with owner_lock(): previous_tx, previous = current() if previous and previous['phase'] not in TERMINAL: raise ValueError('public application transaction already pending') require_legacy_service() if os.path.lexists(STAGING.parent / 'update-pending-verify.json'): raise ValueError('unresolved historical OTA verification marker') if not manager_active(): raise ValueError('only an active standard manager may request activation') check_owner_dropins(previous) tx = TRANSACTIONS / uuid.uuid4().hex tx.mkdir(mode=0o700) for component in components: os.close(trusted_open(ready / component['name'])) source = os.open(ready, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: for component in components: copy_component(source, tx / component['name'], component) finally: os.close(source) (tx / 'archipelago').rename(tx / 'new-binary') durable_mode(tx / 'new-binary', 0o555) extract(tx / f"archipelago-frontend-{document['version']}.tar.gz", tx / 'new-frontend') write_manifest(tx / 'signed-manifest.json', raw) state = {'schema': 1, 'id': tx.name, 'release': identity, 'version': document['version'], 'old_version': old_version, 'new_version': document['version'], 'was_active': True, 'runtime_requested': document.get('application_update') == RUNTIME_POLICY, 'new': pair_identity(tx / 'new-binary', tx / 'new-frontend')} return schedule_transaction(tx, state, bundle) def schedule_transaction(tx, state, bundle): # Frozen authenticated inputs and the installed owner's source; the caller # holds the root lock. No candidate-supplied worker is executed. runner = TRANSACTIONS / ('worker-' + hashlib.sha256(bundle.encode()).hexdigest() + '.py') if runner.exists(): if secure_read(runner, 1024**2) != bundle.encode(): raise ValueError('worker code identity changed') else: write_manifest(runner, bundle.encode()) _, previous = current() check_owner_dropins(previous) recovery_unit = 'archipelago-public-recovery-' + tx.name + '.service' worker_unit = 'archipelago-public-worker-' + tx.name + '.service' dropin = f'[Unit]\nRequires={recovery_unit}\nAfter={recovery_unit}\n' state.update(dropin=dropin, worker_unit=worker_unit, runtime_startup_guard=bool(state.get('runtime_requested')), previous_id=previous['id'] if previous else None) state = journal(tx, state, 'queued') sync_directory(TRANSACTIONS) os.close(trusted_open(UNIT_DIR, directory=True)) mounts = 'RequiresMountsFor=/var/lib/archipelago /var/lib/archipelago-public-transactions /opt/archipelago\n' before = ' '.join(('basic.target', 'archipelago.service', *MAINTENANCE_UNITS)) recovery_prefix = ('[Unit]\nDescription=Recover interrupted public application transaction\n' 'DefaultDependencies=no\nAfter=local-fs.target\n' + mounts + f'Before={before}\n' '[Service]\nType=oneshot\nRemainAfterExit=yes\nTimeoutStartSec=900\n') # Validate/arm this boot's dependency without replaying an upgrade whose # pointer is not published yet. Its persistent boot command is installed # and enabled BEFORE the pointer can make the transaction authoritative. # The live arm must not wait behind the manager's in-progress start job. # Its persistent boot ordering is installed durably before current below. arm_prefix = recovery_prefix.replace(f'Before={before}\n', '') if state.get('adoption') else recovery_prefix atomic(UNIT_DIR / recovery_unit, (arm_prefix + f'ExecStart=/usr/bin/python3 -I {runner} arm {tx.name}\n').encode(), 0o644) atomic(UNIT_DIR / worker_unit, ( '[Unit]\nDescription=Independent public application transaction worker\n' + mounts + 'StartLimitIntervalSec=600\nStartLimitBurst=3\n[Service]\nType=exec\n' 'ExecStartPre=/usr/bin/sleep 2\n' f'ExecStart=/usr/bin/python3 -I {runner} run {tx.name}\nRestart=on-failure\nRestartSec=2\n' 'TimeoutStopSec=10\nRuntimeMaxSec=900\nUMask=0077\n').encode(), 0o644) # Fresh transaction unit names are loaded on demand by start. During # adoption, a daemon-reload would reset the starting manager's timeout. if not state.get('adoption'): command('/usr/bin/systemctl', 'daemon-reload') command('/usr/bin/systemctl', 'start', recovery_unit) atomic(UNIT_DIR / recovery_unit, (recovery_prefix + f'ExecStart=/usr/bin/python3 -I {runner} recover-boot {tx.name}\n').encode(), 0o644) retain_boot_recovery(recovery_unit) if not state.get('adoption'): command('/usr/bin/systemctl', 'daemon-reload') atomic(TRANSACTIONS / 'current', tx.name.encode()) install_owner_dropins(state, reload=not state.get('adoption')) command('/usr/bin/systemctl', 'start', '--no-block', worker_unit) return state def worker_main(): if os.geteuid() != 0: raise PermissionError('root public pair worker required') os.umask(0o077) if sys.argv[1:] == ['apply']: # Source-only bootstrap for a legacy binary with ceremony verification # but without the new backend RPC dispatch. The reviewed bundle must be # installed under root custody first; never run it from a downloaded UI. import io bundle = globals().get('BUNDLE') if bundle is None: bundle = secure_read(Path(__file__), 1024**2).decode() sys.argv = [sys.argv[0]] output = io.StringIO() with contextlib.redirect_stdout(output): main() # fixed-root preparation, original signed JSON on stdin prepared = json.loads(output.getvalue()) print(json.dumps(submit(Path(prepared['path']).name, bundle))) elif len(sys.argv) == 3 and sys.argv[1] == 'submit': bundle = globals().get('BUNDLE') if bundle is None: bundle = secure_read(Path(__file__), 1024**2).decode() print(json.dumps(submit(sys.argv[2], bundle))) elif sys.argv[1:] == ['adopt']: bundle = globals().get('BUNDLE') if bundle is None: bundle = secure_read(Path(__file__), 1024**2).decode() print(json.dumps(adopt_legacy_pair(bundle))) elif sys.argv[1:] == ['run'] or (len(sys.argv) == 3 and sys.argv[1] == 'run' and re.fullmatch('[0-9a-f]{32}', sys.argv[2])): run_worker(sys.argv[2] if len(sys.argv) == 3 else None) elif len(sys.argv) == 3 and sys.argv[1] == 'arm': tx, state = load_transaction(sys.argv[2]) require_legacy_service() if state['phase'] != 'queued' or pair_identity(tx / 'new-binary', tx / 'new-frontend') != state['new']: raise ValueError('unprepared public recovery dependency') if state.get('adoption'): check_preimages(tx, state) elif sys.argv[1:] == ['request-recovery']: with owner_lock(): require_legacy_service() tx, state = current() if not state or state['phase'] in ('restored', 'aborted'): raise ValueError('no public pair available for recovery') state['recovery_requested'] = True journal(tx, state, 'recovery-requested' if state['phase'] == 'committed' else state['phase']) command('/usr/bin/systemctl', 'start', '--no-block', state['worker_unit']) elif len(sys.argv) == 3 and sys.argv[1] == 'recover-boot': own_tx, own = load_transaction(sys.argv[2]) _, selected = current() if selected and selected['id'] != own['id']: # An old enabled recovery unit must never operate on a later job. # Only a durable, not-yet-published admission whose predecessor is # still current may acquire its pointer after a scheduling crash. if (own['phase'] != 'queued' or own.get('previous_id') != selected['id'] or selected['phase'] not in TERMINAL): return elif selected is None and own.get('previous_id') is not None: raise ValueError('public recovery predecessor disappeared') with owner_lock(): _, selected = current() if not selected or selected['id'] != own['id']: if (own['phase'] != 'queued' or (selected and (selected['id'] != own.get('previous_id') or selected['phase'] not in TERMINAL))): raise ValueError('public recovery pointer changed') atomic(TRANSACTIONS / 'current', own['id'].encode()) tx, state = current() if state['phase'] not in TERMINAL: recovery(tx, state, True) elif sys.argv[1:] == ['status']: _, state = current() print(json.dumps(state)) else: raise ValueError('fixed public worker command required') if __name__ == '__main__': worker_main()