#!/usr/bin/python3 -I """Appointed root provisioner; accepts the coordinator's exact admitted tuple. Package/prerequisite installation and caller privilege exclusion precede this entry. Inputs are fixed protected incoming snapshots, never updater paths. No overwriting/reprovisioning a ledger and no authority minted by this program. """ import argparse import importlib.util import json import os import pathlib import pwd import shutil import socket import struct import sys sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) from d09_store import atomic, initial, read, secure, sync_dir from d09_wire import MAX, canonical, digest, parse, validate, release_digest spec = importlib.util.spec_from_file_location('d09_owner', pathlib.Path(__file__).with_name('d09-lifecycle-owner.py')) owner = importlib.util.module_from_spec(spec) spec.loader.exec_module(owner) MODULES = ('d09-lifecycle-owner.py','d09_wire.py','d09_store.py','d09_worker.py','d09_handoff.py', 'd09_c01.py','d09-wire-schema.json','c01_d09.py','c01_network.py', 'c01_network_owner.py','c01_network_client.py','c01_runtime_endpoints.py','c01_kiosk_endpoints.py','c01_uplink.py','c01_network_mdns.py','c01_retained_producer.py','c01_network_plan.py','c01_endpoints.py','c01_access.py','c01_access_client.py', 'c01_access_kernel.py','c01_access_policy.py','c01_migration_layout.py','c01_access_generation.py','c01_runtime.py', 'c01_runtime_view.py','c01_runtime_groups.py','runtime_owner.py','c01_children.py','runtime_uplink.py', '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') def copy_snapshot(source, dest): if dest.exists() or dest.is_symlink(): raise ValueError('stage already exists; retained for reconciliation') if source.is_dir(): shutil.copytree(source, dest, symlinks=False) for directory, _, files in os.walk(dest, topdown=False): for name in files: fd = secure(pathlib.Path(directory) / name) os.fsync(fd) os.close(fd) sync_dir(directory) else: atomic(dest, read(source), source.stat().st_mode & 0o777) sync_dir(dest.parent) def install_package(): current = owner.service() if current['MainPID'] != '0' or current['ActiveState'] not in ('inactive', 'failed'): raise ValueError('prerequisite requires externally drained management invocation') package = pathlib.Path(__file__).resolve().parent.parent for directory, mode in ((owner.ROOT, 0o700), (owner.ROOT / 'incoming', 0o700), (owner.ROOT / 'observations', 0o700), (owner.RUN, 0o755), (owner.BINDING.parent, 0o755), (owner.DESTINATIONS['config'].parent, 0o755)): directory.mkdir(mode=mode, exist_ok=True) fd = secure(directory, True) os.close(fd) (owner.DESTINATIONS['config'].parent/'.d09-staging').mkdir(mode=0o700,exist_ok=True) if owner.BINDING.exists() or (owner.ROOT / 'ledger.json').exists(): raise ValueError('existing D09 authority; package replacement needs separate recovery') scripts = {'d09-lifecycle-owner.py': 'archipelago-d09-owner', 'd09-lifecycle-client.py': 'archipelago-d09ctl', 'provision-d09-lifecycle.py': 'archipelago-d09-provision', 'dashboard-public-guard.py': 'archipelago-d09-public-guard'} for source, dest in scripts.items(): atomic(pathlib.Path('/usr/libexec') / dest, read(package / 'scripts' / source), 0o755) # The fixed worker imports the same owner implementation by its source name. for source in MODULES: atomic(pathlib.Path('/usr/libexec') / source, read(package / 'scripts' / source), 0o644) configs = package / 'image-recipe/configs' atomic(pathlib.Path('/usr/libexec/archipelago-d09-prepare-workspace'), read(configs / 'prepare-terminal-workspace.sh'), 0o755) atomic(pathlib.Path('/usr/libexec/archipelago-d09-terminal-workspace.conf'), read(configs / '60-terminal-workspace.conf'), 0o644) atomic(pathlib.Path('/etc/systemd/system/archipelago-d09-lifecycle.service'), read(configs / 'archipelago-d09-lifecycle.service'), 0o644) atomic(owner.DESTINATIONS['config'].with_name('50-d09-lifecycle.conf'), read(configs / '50-d09-lifecycle.conf'), 0o644) owner.command('/usr/bin/systemctl', 'daemon-reload') owner.command('/usr/bin/systemctl', 'enable', 'archipelago-d09-lifecycle.service') def provision(admission, successor=False): binding = json.loads(read(admission)) required = {'schema_version', 'target_binding_digest', 'composition_digest', 'admission_epoch_id', 'transaction_id', 'grants', 'objects', 'artifacts', 'restore_artifacts', 'effective_config', 'exclusion_evidence_digest', 'helper_digest', 'prerequisite', 'release_identity'} if 'c01' in binding: required.add('c01') if 'os_update' in binding: required.add('os_update') if set(binding) != required or binding['schema_version'] != 1: raise ValueError('admission fields') release_digest(binding['release_identity']) backend = [c for c in binding['release_identity'][1] if c[0] == 'archipelago'] if len(backend) != 1 or backend[0][2] != binding['artifacts']['backend']: raise ValueError('release/backend byte-domain mapping') os_admission = None if 'os_update' in binding: import os_update_owner os_update_owner.install_inputs(binding, owner.ROOT / 'incoming/os-update') # Inputs/verifier must already be installed through protected custody. # Qualification verifies signature, all release components, plan, and # actual available Debian locks before any D09 authority is published. os_executor, os_admission = os_update_owner.qualify(binding, lock_probe=True) if os_executor.status['phase'] is not None: raise ValueError('existing OS transaction requires external reconciliation, not new admission') account = pwd.getpwnam('archipelago') if account.pw_dir != '/home/archipelago' or account.pw_uid == 0: raise ValueError('service account') binding.update(requester_uid=account.pw_uid, requester_gid=account.pw_gid) if 'c01' in binding: import d09_c01 c=d09_c01.validate(binding) if digest(read(d09_c01.LAUNCH))!=c['generations']['old']['launch_sha256']: raise ValueError('C01 prerequisite launch pin differs') binding.update(requester_uid=c['application']['uid'],requester_gid=c['application']['gid']) (owner.DESTINATIONS['config'].parent/'.d09-staging').mkdir(mode=0o700,exist_ok=True) # Reuse normative request grammar to check every authority token and route. routes = {'UPDATER': ('APPLY', 'APPLY_REQUEST'), 'AUTOMATIC': ('RECOVER', 'AUTOMATIC_FAILURE'), 'MANUAL': ('RECOVER', 'MANUAL_REQUEST'), 'GUARD': ('RECOVER', 'CRASH_NOTICE'), 'OBSERVER': ('ACK_OBSERVATION', 'OBSERVATION_READY')} if set(binding['grants']) != set(routes): raise ValueError('operation grants') grant_a = binding['prerequisite'] if type(grant_a) is not dict or set(grant_a) != {'grant_digest', 'backend_sha256', 'effective_config_digest'} or any( type(value) is not str or len(value) != 64 or any(c not in '0123456789abcdef' for c in value) for value in grant_a.values()): raise ValueError('exact prerequisite grant-A binding required') if grant_a['grant_digest'] in binding['grants'].values(): raise ValueError('grant A must be distinct from transaction/notice grants') for route, (op, reason) in routes.items(): validate('submit_request', dict(schema_version=1, operation=op, route=route, reason=reason, transaction_id=binding['transaction_id'], request_id=binding['admission_epoch_id'], target_binding_digest=binding['target_binding_digest'], composition_digest=binding['composition_digest'], grant_digest=binding['grants'][route], expected_sequence=0, selected_release_digest=release_digest(binding['release_identity']) if op == 'APPLY' else None, staging_generation=binding['admission_epoch_id'] if op == 'APPLY' else None, requester_invocation_id=None if route == 'OBSERVER' else binding['admission_epoch_id'], observer_receipt_digest=binding['composition_digest'] if route == 'OBSERVER' else None)) if not successor and ((owner.ROOT / 'ledger.json').exists() or owner.BINDING.exists()): raise ValueError('existing authority; use external readmission') if set(binding['objects']) != set(owner.destinations(binding)): raise ValueError('payload/config objects') # The appointed provisioner installs independent exclusion evidence through # its existing coordinator channel; the updater has no access to this input. if digest(read(owner.ROOT / 'incoming/exclusion.json')) != binding['exclusion_evidence_digest']: raise ValueError('prerequisite exclusion evidence') helper = pathlib.Path('/usr/libexec/archipelago-d09-prepare-workspace') if digest(read(helper)) != binding['helper_digest']: raise ValueError('workspace helper binding') current = owner.service() if not successor and (current['MainPID'] != '0' or current['ActiveState'] not in ('inactive', 'failed')): raise ValueError('prerequisite old updater not drained') guard = pathlib.Path('/usr/libexec/archipelago-d09-public-guard') # Validate all supplied objects BEFORE producing any stage files. Both # frontend generations include the containment-preserving nginx template. for name, dest in owner.destinations(binding).items(): obj = binding['objects'][name] if set(obj) != {'old', 'new', 'restore'} or owner.object_digest(dest) != obj['old']: raise ValueError('old object binding') for generation in ('new', 'restore'): source = owner.ROOT / 'incoming' / (generation + '-' + name) if owner.object_digest(source) != obj[generation]: raise ValueError('incoming snapshot binding') if name == 'backend' and digest(read(source)) != binding['artifacts' if generation == 'new' else 'restore_artifacts']['backend']: raise ValueError('backend byte-domain artifact mismatch') if name == 'launch': expected=binding['c01']['generations']['new' if generation=='new' else 'restore'] if digest(read(source))!=expected['launch_sha256'] or json.loads(read(source))['native_sha256']!=expected['native_sha256']: raise ValueError('C01 staged launch/native binding') if name == 'frontend': template = source / 'archipelago-runtime/image-recipe/configs/nginx-archipelago.conf' if not template.exists(): raise ValueError('missing rollback containment template') original = read(template) check = owner.ROOT / ('guard-check-' + digest(original) + '.conf') atomic(check, original, 0o644) owner.command('/usr/bin/python3', '-I', str(guard), '--protect-template', str(check)) if read(check) != original: raise ValueError('frontend must be admitted with protected nginx template') if generation == 'new': for preserved in ('aiui', 'archipelago-companion.apk'): if (dest / preserved).exists() and not (source / preserved).exists(): raise ValueError('frontend omits retained companion asset') if name == 'config': if generation == 'new' and read(source) != read(pathlib.Path('/usr/libexec/archipelago-d09-terminal-workspace.conf')): raise ValueError('owned postimage differs from fixed package') if generation == 'restore' and obj['restore'] != obj['old']: raise ValueError('configuration recovery must restore exact preimage') if 'c01' in binding: d09_c01.stage_records(binding,owner.ROOT/'incoming') for name, dest in owner.destinations(binding).items(): for generation, suffix in [('new', '.new'), ('restore', '.restore')]: source = owner.ROOT / 'incoming' / (generation + '-' + name) if source.exists(): staged = owner.snapshot(binding,name,suffix) if not successor or not staged.exists(): copy_snapshot(source, staged) if owner.object_digest(staged) != binding['objects'][name][generation]: raise ValueError('staged snapshot mismatch') if successor: return binding value = initial(binding) value['os_admission'] = os_admission atomic(owner.BINDING, canonical(binding)) atomic(owner.ROOT / 'ledger.json', canonical(value)) public = {k: binding[k] for k in ('target_binding_digest', 'composition_digest', 'transaction_id', 'grants', 'artifacts', 'restore_artifacts', 'release_identity')} atomic(owner.BINDING.with_name('client.json'), canonical(public), 0o644) atomic(owner.RUN / 'bootstrap.json', canonical({'boot_id': owner.boot_id(), 'binding_digest': digest(read(owner.BINDING)), 'ledger_digest': digest(canonical(value))})) def activate_prerequisite(): """One grant-A start, after --admission and a live owner; no local permit.""" fd = secure(owner.RUN, True) os.close(fd) with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as conn: conn.settimeout(40) conn.connect(str(owner.PRIVATE)) _, uid, _ = struct.unpack('3i', conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) if uid != 0: raise PermissionError('unauthenticated lifecycle owner') conn.sendall(canonical({'method': 'activate_prerequisite'})) conn.shutdown(socket.SHUT_WR) raw = b'' while True: part = conn.recv(8192) if not part: break raw += part if len(raw) > MAX: raise ValueError('oversized prerequisite acknowledgement') response = parse(raw) if set(response) != {'prerequisite_started', 'invocation_id'} or type(response['prerequisite_started']) is not bool: raise PermissionError('owner refused prerequisite activation; preserve grant-A state') return response def successor(admission, record_path, external_history_path): record = parse(read(record_path)) validate('external_successor_record', record) history_raw = read(external_history_path) history = json.loads(history_raw) if digest(history_raw) != record['external_history_digest'] or history['sequence'] != record['external_history_sequence']: raise ValueError('successor external history binding') for key in ('target_binding_digest', 'composition_digest', 'transaction_id', 'predecessor_epoch_id', 'head_sequence', 'preserved_head_digest', 'preserved_journal_digest', 'preserved_consumed_history_digest', 'preserved_target_index_digest', 'terminal_disposition'): if history[key] != record[key]: raise ValueError('successor latest history') proposed = json.loads(read(admission)) account = pwd.getpwnam('archipelago') proposed.update(requester_uid=account.pw_uid, requester_gid=account.pw_gid) if 'c01' in proposed: proposed.update(requester_uid=proposed['c01']['application']['uid'],requester_gid=proposed['c01']['application']['gid']) if digest(canonical(proposed)) != record['successor_binding_digest']: raise ValueError('successor exact admission') retained = owner.ROOT / ('successor-admission-' + record['successor_binding_digest'] + '.json') if retained.exists(): if read(retained) != canonical(proposed): raise ValueError('retained successor admission differs') binding = proposed else: binding = provision(admission, successor=True) atomic(retained, canonical(binding)) atomic(owner.ROOT / 'successor-binding.json', canonical(binding)) atomic(owner.ROOT / 'successor.json', canonical(record)) with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as conn: conn.settimeout(30) conn.connect(str(owner.PRIVATE)) if struct.unpack('3i', conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))[1] != 0: raise PermissionError('owner peer') conn.sendall(canonical({'method': 'successor'})) conn.shutdown(socket.SHUT_WR) if parse(conn.recv(MAX)).get('admitted') is not True: raise ValueError('successor refused') # Publish only the CURRENT target after an old-record retry, never roll the # client pointer back to a predecessor. This file conveys no continuity. value = json.loads(read(owner.ROOT / 'ledger.json')) current = value['active_binding'] public = {k: current[k] for k in ('target_binding_digest', 'composition_digest', 'transaction_id', 'grants', 'artifacts', 'restore_artifacts', 'release_identity')} if value['notice_grant']: public['grants'] = {route: value['notice_grant'] for route in public['grants']} atomic(owner.BINDING.with_name('client.json'), canonical(public), 0o644) def readmit(record_path, external_history_path, activation=False): raw = read(record_path) record = parse(raw) validate('external_activation_record' if activation else 'external_readmission_record', record) history_raw = read(external_history_path) history = json.loads(history_raw) # This file is supplied by the appointed coordinator from OUTSIDE the target # rollback domain. Root installation is the existing authentication channel. if digest(history_raw) != record['external_history_digest'] or history['sequence'] != record['external_history_sequence']: raise ValueError('external history binding') for key in ('target_binding_digest', 'composition_digest', 'transaction_id', 'preserved_head_digest', 'preserved_journal_digest', 'preserved_consumed_history_digest', 'preserved_target_index_digest', 'predecessor_epoch_id', 'head_sequence'): if history[key] != record[key]: raise ValueError('external latest-consumption history') if activation and history.get('disposition_digest') != record['disposition_digest']: raise ValueError('external disposition history') atomic(owner.ROOT / ('activation.json' if activation else 'readmission.json'), canonical(record)) with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as conn: conn.settimeout(40) conn.connect(str(owner.PRIVATE)) if struct.unpack('3i', conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))[1] != 0: raise PermissionError('owner peer') conn.sendall(canonical({'method': 'reactivate' if activation else 'readmit'})) conn.shutdown(socket.SHUT_WR) response = conn.recv(65536) if json.loads(response).get('readmitted') is not True: raise ValueError('owner refused readmission') public_path = owner.BINDING.with_name('client.json') public = json.loads(read(public_path)) public['grants'] = {route: record['new_grant_digest'] for route in public['grants']} atomic(public_path, canonical(public), 0o644) def main(): if os.geteuid() != 0: raise SystemExit('root required') p = argparse.ArgumentParser() group = p.add_mutually_exclusive_group(required=True) group.add_argument('--install-package', action='store_true') group.add_argument('--activate-prerequisite', action='store_true') group.add_argument('--admission', type=pathlib.Path) group.add_argument('--readmission', type=pathlib.Path) group.add_argument('--reactivation', type=pathlib.Path) group.add_argument('--successor', type=pathlib.Path) group.add_argument('--observation', type=pathlib.Path) group.add_argument('--revocation', type=pathlib.Path) group.add_argument('--os-postboot', action='store_true') p.add_argument('--external-history', type=pathlib.Path) p.add_argument('--successor-admission', type=pathlib.Path) args = p.parse_args() if args.install_package: install_package() elif args.os_postboot: with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as conn: conn.settimeout(300) conn.connect(str(owner.PRIVATE)) if struct.unpack('3i', conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))[1] != 0: raise PermissionError('owner peer') conn.sendall(canonical({'method': 'os_postboot'})) conn.shutdown(socket.SHUT_WR) result = parse(conn.recv(MAX)) if result.get('os_postboot_verified') is not True: raise ValueError('OS boot observation refused') print(json.dumps(result, sort_keys=True)) elif args.activate_prerequisite: print(json.dumps(activate_prerequisite(), sort_keys=True)) elif args.admission: provision(args.admission) elif args.successor: if not args.external_history or not args.successor_admission: p.error('--successor-admission and --external-history required') successor(args.successor_admission, args.successor, args.external_history) elif args.readmission or args.reactivation: if not args.external_history: p.error('--external-history is required for readmission') readmit(args.reactivation or args.readmission, args.external_history, activation=bool(args.reactivation)) elif args.revocation: record = json.loads(read(args.revocation)) if set(record) != {'target_binding_digest', 'transaction_id', 'grant_digest'}: raise ValueError('revocation fields') atomic(owner.ROOT / 'revocation.json', canonical(record)) else: # The separate observer supplies namespace/default-terminal and Work/ # history/custom-fragment evidence. This program does not self-approve it. raw = read(args.observation) atomic(owner.ROOT / 'observations' / (digest(raw) + '.json'), raw) print(digest(raw)) if __name__ == '__main__': main()