#!/usr/bin/env python3 """Stage or explicitly import the unsigned candidate image; never activate apps. Import uses a live appointed C01/D09 runtime lease and its mount/storage view. Missing runtime admission defers (exit 75); there is no default-store fallback. """ import argparse import importlib.util import json import os from pathlib import Path import re import shutil import subprocess import sys import tempfile import time import tomllib sys.dont_write_bytecode = True import uat_oci def stage(bundle, target): receipt = uat_oci.bundle(bundle) destination = target / 'opt/archipelago/unsigned-uat-oci' if not target.is_absolute() or destination.parent.resolve() != destination.parent: raise ValueError('staging destination must be an absolute non-symlink path') destination.parent.mkdir(parents=True, exist_ok=True) if destination.exists(): if uat_oci.sha(destination / 'BUNDLE.json') != uat_oci.sha(bundle / 'BUNDLE.json'): raise ValueError('different candidate already staged; preserve it for review') uat_oci.bundle(destination) else: shutil.copytree(bundle, destination) uat_oci.bundle(destination) # No chroot Podman, no service enable/start, no desired-state mutations. return dict(status='staged-import-deferred', bundle_sha256=uat_oci.sha(destination / 'BUNDLE.json'), oci=receipt['oci'], reason='explicit live appointed runtime import required') def import_image(call, archive, proof): """Fixed image-only operation set. call supplies the verified storage context.""" call(['load', '--input', str(archive)]) image_id = proof['config_id'].removeprefix('sha256:') tag = uat_oci.REPOSITORY + ':uat-' + proof['manifest_digest'].removeprefix('sha256:') call(['tag', image_id, tag]) actual = json.loads(call(['image', 'inspect', proof['reference']])) if len(actual) != 1: raise ValueError('ambiguous imported image') image = actual[0] if (image['Id'].removeprefix('sha256:') != image_id or image['Digest'] != proof['manifest_digest'] or (image['Os'], image['Architecture'], image['Config']['User']) != ('linux', 'amd64', '100:100')): raise ValueError('imported image digest/config/platform/user mismatch') return dict(reference=proof['reference'], manifest_digest=image['Digest'], config_id='sha256:' + image_id, platform='linux/amd64', user=image['Config']['User']) def storage_declaration(config, storage): selected = storage['storage'] runroot = Path(selected['runroot']) if (selected['graphroot'] != config['graphroot'] or not Path(selected['graphroot']).is_absolute() or selected['driver'] not in ('vfs', 'overlay') or '..' in runroot.parts or not runroot.is_absolute() or not runroot.is_relative_to(Path(config['runtime_dir']))): raise ValueError('appointed storage declaration mismatch; refuse before invoking Podman') return selected def storage_check(info, config, storage, config_path): store = info['store'] selected = storage_declaration(config, storage) if (not info['host']['security']['rootless'] or store['graphRoot'] != config['graphroot'] or store['runRoot'] != selected['runroot'] or store['configFile'] != str(config_path) or store['graphDriverName'] != selected['driver'] ): raise ValueError('appointed rootless storage mismatch; no default-store fallback') def appointed_import(bundle, bundle_hash, runtime_hash, storage_hash): if os.getresuid() != (0, 0, 0): raise ValueError('root coordinator required; Podman itself must run as appointed non-root account') # These installed modules are the existing admission validators, not copies # provided by an untrusted image or an alternate implementation of admission. sys.path.insert(0, '/usr/libexec') if not Path('/usr/libexec/c01_runtime_view.py').is_file(): raise FileNotFoundError('installed D09 runtime admission helpers unavailable; defer') import c01_runtime from c01_runtime_view import RuntimeLease config, digest = c01_runtime.load_config() if digest != runtime_hash or config['version'] != 2: raise ValueError('appointed runtime configuration digest/version mismatch') receipt = uat_oci.bundle(bundle, bundle_hash) if not isinstance(receipt.get('source_commit'), str) or not re.fullmatch('[0-9a-f]{40}', receipt['source_commit']): raise ValueError('fixture/unqualified bundle cannot be imported into an appointed runtime') lease = RuntimeLease(config['manager_uid'], config['manager_gid'], config['runtime_uid'], config['runtime_gid'], digest) namespace = None manager = None try: pid = lease.identity['pid'] namespace = os.open(f'/proc/{pid}/ns/mnt', os.O_RDONLY | os.O_CLOEXEC) if os.fstat(namespace).st_ino != lease.identity['mount_namespace']: raise ValueError('runtime mount namespace changed') lease.check() home = Path(config['home']) conf = home / '.config/containers/storage.conf' viewed_conf = Path(f'/proc/{pid}/root') / conf.relative_to('/') if uat_oci.sha(viewed_conf) != storage_hash: raise ValueError('appointed storage configuration digest mismatch') storage = tomllib.loads(viewed_conf.read_text()) selected = storage_declaration(config, storage) helper=Path(__file__).resolve().with_name('c01_oci_job.py') c01_runtime.protected(helper) helper_hash=uat_oci.sha(helper) if receipt['files'].get('c01_oci_job.py')!=helper_hash: raise ValueError('reviewed bundle/import job implementation mismatch') spec=importlib.util.spec_from_file_location('appointed_oci_job',helper) jobs=importlib.util.module_from_spec(spec);spec.loader.exec_module(jobs) manager=jobs.UserManager(lease,config) # Retain the verified read-only artifact descriptor. No second archive # copy, chroot store, UID impersonation or out-of-hierarchy process. fd=os.open(bundle/'lnd-tor.oci.tar',os.O_RDONLY|os.O_NOFOLLOW|os.O_CLOEXEC) with os.fdopen(fd,'rb') as archive: import hashlib if hashlib.file_digest(archive, 'sha256').hexdigest() != receipt['oci']['archive_sha256']: raise ValueError('OCI archive changed before import') archive.seek(0) context={'uid':config['runtime_uid'],'gid':config['runtime_gid'],'home':str(home), 'username':config['runtime_user'],'runtime_dir':config['runtime_dir'], 'mount_namespace':lease.identity['mount_namespace'], 'network_namespace':os.stat(f'/proc/{pid}/ns/net').st_ino, 'helper_sha256':helper_hash,'podman_sha256':uat_oci.sha(Path(f'/proc/{pid}/root/usr/bin/podman')), 'config_path':str(conf),'storage_sha256':storage_hash,'driver':selected['driver'], 'graphroot':selected['graphroot'],'runroot':selected['runroot'], 'image_id':receipt['oci']['config_id'].removeprefix('sha256:'), 'tag':uat_oci.REPOSITORY+':uat-'+receipt['oci']['manifest_digest'].removeprefix('sha256:'), 'reference':receipt['oci']['reference']} context['archive_sha256']=receipt['oci']['archive_sha256'] def call(args): lease.check() if uat_oci.sha(viewed_conf) != storage_hash or c01_runtime.load_config()[1] != digest: raise ValueError('runtime/storage config changed during import') if args[0] == 'load': archive.seek(0) with tempfile.TemporaryFile() as log, tempfile.TemporaryFile() as errors, open('/dev/null','rb') as empty: try: manager.execute(helper,context,args,archive if args[0]=='load' else empty,log,errors) lease.check() if uat_oci.sha(viewed_conf)!=storage_hash or c01_runtime.load_config()[1]!=digest: raise ValueError('runtime/storage changed during image command') log.seek(0) output=log.read(4*1024*1024+1) if len(output)>4*1024*1024:raise ValueError('image command output exceeds bound') return output.decode() except Exception as error: errors.seek(0) raise RuntimeError('appointed image job failed: '+str(error)+': '+errors.read(16384).decode(errors='replace')) from error info = json.loads(call(['info', '--format', 'json'])) storage_check(info, config, storage, conf) imported = import_image(call, Path('/dev/stdin'), receipt['oci']) storage_check(json.loads(call(['info', '--format', 'json'])), config, storage, conf) uat_oci.bundle(bundle, bundle_hash) return dict(status='imported-verified-no-activation', image=imported, runtime_sha256=digest, storage_sha256=storage_hash, store=info['store'], runtime_uid=config['runtime_uid'], bundle_sha256=bundle_hash, jobs=manager.records, lifecycle='unchanged; stopped/uninstalled/held intent preserved') finally: try: if manager is not None:manager.close() finally: if namespace is not None:os.close(namespace) lease.close() def main(): parser = argparse.ArgumentParser(description=__doc__) commands = parser.add_subparsers(dest='action', required=True) staging = commands.add_parser('stage') staging.add_argument('--bundle', type=Path, required=True) staging.add_argument('--target-root', type=Path, required=True) importing = commands.add_parser('import') importing.add_argument('--bundle', type=Path, required=True) for name in ('bundle-sha256', 'runtime-sha256', 'storage-sha256'): importing.add_argument('--' + name, required=True) args = parser.parse_args() try: if args.action == 'stage': result = stage(args.bundle, args.target_root) else: result = appointed_import(args.bundle, args.bundle_sha256, args.runtime_sha256, args.storage_sha256) print(json.dumps(result, sort_keys=True)) except (OSError, ValueError, KeyError, ImportError, subprocess.SubprocessError, RuntimeError) as error: parser.exit(75 if args.action == 'import' else 1, f'DEFER/FAIL: {error}\n') if __name__ == '__main__': main()