#!/usr/bin/env python3 """Disposable installed Debian disks. Run in a durable CPU2/8G/no-swap job. No host chroot, host package installation, containers, or writable host shares. Direct-kernel cold boots are reported as such, never as GRUB/ISO acceptance. All heavyweight work, including hashing/copying, is inside release-heavy.lock. """ import argparse import fcntl import hashlib import io import json import os import re from pathlib import Path import shutil import subprocess import tarfile import time ENTRY_SCRIPT = '''#!/bin/bash set -u ip link set ens3 up 2>/dev/null || ip link set eth0 up iface=$(ls /sys/class/net | grep -v '^lo$' | head -1) ip addr add 10.0.2.15/24 dev "$iface" || true ip route add default via 10.0.2.2 || true status=0 if [ -f /opt/os-ota-payload.tar ]; then mkdir -p /opt/ota-source tar -xf /opt/os-ota-payload.tar -C /opt/ota-source || status=$? fi if [ "$status" -eq 0 ]; then python3 -B /opt/os-ota-guest.py > /var/log/os-ota-guest.log 2>&1 status=$? fi echo "$status" > /var/log/os-ota-guest.exit sync echo "OS_OTA_GUEST_EXIT=$status" cat /var/log/os-ota-guest.log sync systemctl poweroff ''' def sha(path): with Path(path).open('rb') as stream: return hashlib.file_digest(stream, 'sha256').hexdigest() HARNESS_SHA256 = sha(__file__) def run(cmd, log, timeout=1800): with log.open('xb') as stream: result = subprocess.run([str(x) for x in cmd], stdout=stream, stderr=subprocess.STDOUT, timeout=timeout) if result.returncode: raise RuntimeError(f'{cmd[0]} exit {result.returncode}: {log}') def inject(disk, source, target, log, private=False): # debugfs commands are not shell commands; restrict their grammar nevertheless. for value in (str(source), target): if any(c.isspace() or c in '\"\\' for c in value): raise ValueError('unsupported debugfs path') run(['/usr/sbin/debugfs', '-w', '-R', f'rm {target}', disk], log.with_suffix('.rm.log')) run(['/usr/sbin/debugfs', '-w', '-R', f'write {source} {target}', disk], log) verified=log.with_suffix('.readback') if private: verified.touch(mode=0o600,exist_ok=False) run(['/usr/sbin/debugfs','-R',f'dump {target} {verified}',disk],log.with_suffix('.readback.log')) if sha(verified) != sha(source): raise ValueError('offline disk injection readback mismatch') def owned_disk(path): if path.is_symlink(): raise ValueError('symlink disk refused') disk = path.resolve(strict=True) if not disk.is_file() or disk.name != 'installed.raw' or not (disk.parent/'result.json').is_file(): raise ValueError('only harness-owned regular installed.raw disks accepted') preparation = json.loads((disk.parent/'result.json').read_text()) if preparation.get('action') != 'prepare' or not preparation.get('passed') or preparation.get('disk') != str(disk): raise ValueError('missing successful disk ownership receipt') return disk def source_archive(root, destination): manifest = {} with tarfile.open(destination, 'x') as archive: for directory in ('scripts', 'image-recipe/configs', 'tests/regression'): for path in sorted((root/directory).rglob('*')): if path.is_file() and not path.is_symlink() and '__pycache__' not in str(path) and path.suffix in ('.py','.json','.sh','.service','.conf','.in'): relative=str(path.relative_to(root)) raw=path.read_bytes() manifest[relative]=hashlib.sha256(raw).hexdigest() info=archive.gettarinfo(path,arcname=relative) info.size=len(raw) info.uid=info.gid=0 info.uname=info.gname='root' info.mode=0o755 if info.mode & 0o111 else 0o644 archive.addfile(info,io.BytesIO(raw)) public=root/'release-manifest.json' if public.is_file() and not public.is_symlink(): raw=public.read_bytes() manifest['release-manifest.json']=hashlib.sha256(raw).hexdigest() info=tarfile.TarInfo('release-manifest.json') info.size=len(raw) info.mode=0o644 archive.addfile(info,io.BytesIO(raw)) return manifest def main(): p = argparse.ArgumentParser(description=__doc__) p.add_argument('--output', type=Path, required=True) sub = p.add_subparsers(dest='action', required=True) prep = sub.add_parser('prepare') prep.add_argument('--rootfs', type=Path, required=True) prep.add_argument('--receipt', type=Path, required=True) prep.add_argument('--kernel-release', required=True) boot = sub.add_parser('boot') boot.add_argument('--disk', type=Path, required=True) boot.add_argument('--kernel', type=Path, required=True) boot.add_argument('--initrd', type=Path, required=True) boot.add_argument('--guest', type=Path, required=True) boot.add_argument('--payload', type=Path) boot.add_argument('--source-root', type=Path) boot.add_argument('--native-artifact',type=Path,help='retained real non-libtest candidate artifact directory') boot.add_argument('--crypto-fixture',type=Path,help='separate compile-time disposable test-root artifact; NOT candidate') boot.add_argument('--preboot-evidence',type=Path,help='external retained completed ledger before guest reboot') boot.add_argument('--timeout', type=int, default=1800) boot.add_argument('--offline', action='store_true') boot.add_argument('--firmware', action='store_true', help='boot private GRUB disk instead of direct kernel') boot.add_argument('--scenario', default='none', choices=['none', 'locked-apt', 'offline', 'space', 'space-prepare', 'bad-hash', 'missing-dependency', 'partial-dpkg', 'interrupted-before']) export = sub.add_parser('export') export.add_argument('--disk', type=Path, required=True) export.add_argument('--kernel-release', required=True) clone = sub.add_parser('clone') clone.add_argument('--disk', type=Path, required=True) args = p.parse_args() if os.geteuid() != 0: p.error('requires root only for owned disk preparation; run in durable limited service') # Refuse accidentally unrestricted execution even when root. cg = Path('/proc/self/cgroup').read_text().strip().split('::')[-1] cgroup = Path('/sys/fs/cgroup') / cg.lstrip('/') cpu = (cgroup / 'cpu.max').read_text().split() memory = (cgroup / 'memory.max').read_text().strip() swap = (cgroup / 'memory.swap.max').read_text().strip() if cpu[0] == 'max' or int(cpu[0]) > 2 * int(cpu[1]) or memory == 'max' or int(memory) > 8*1024**3 or swap != '0': p.error('job must enforce CPUQuota<=200%, MemoryMax<=8G, MemorySwapMax=0') with open('/tmp/opencode/release-heavy.lock', 'a') as lock: fcntl.flock(lock, fcntl.LOCK_EX) if shutil.disk_usage(args.output.parent).free < 24*1024**3: p.error('24 GiB free required; preserve previous receipts') args.output.mkdir(mode=0o755, exist_ok=False) out = args.output.resolve() result = {'scope': 'disposable-installed-Debian; unsigned root-appointed fixture only', 'production_approved': False, 'harness_sha256': HARNESS_SHA256, 'action': args.action, 'started': time.time(), 'passed': False} try: if args.action == 'prepare': receipt = json.loads(args.receipt.read_text()) expected = receipt['files'][args.rootfs.name]['sha256'] if sha(args.rootfs) != expected: raise ValueError('rootfs receipt hash mismatch') result['rootfs_sha256'] = expected root = out / 'rootfs' root.mkdir() run(['tar', '--numeric-owner', '-xf', args.rootfs, '-C', root], out/'extract.log') release = args.kernel_release if '/' in release or not release.startswith('6.'): raise ValueError('invalid kernel release') shutil.copytree('/usr/lib/modules/'+release, root/'usr/lib/modules'/release, symlinks=True) for name in ('vmlinuz-', 'initrd.img-'): shutil.copyfile('/boot/'+name+release, out/(name+release)) (root/'etc/fstab').write_text('/dev/vda / ext4 defaults 0 1\n') (root/'etc/resolv.conf').unlink(missing_ok=True) (root/'etc/resolv.conf').write_text('nameserver 10.0.2.3\n') (root/'etc/systemd/system/os-ota-fixture.service').write_text('''[Unit] Description=Disposable OS OTA qualification After=local-fs.target [Service] Type=oneshot ExecStart=/bin/bash /opt/os-ota-entry.sh StandardOutput=journal+console StandardError=journal+console [Install] WantedBy=multi-user.target ''') wants = root/'etc/systemd/system/multi-user.target.wants' wants.mkdir(exist_ok=True) (wants/'os-ota-fixture.service').symlink_to('../os-ota-fixture.service') (root/'opt/os-ota-entry.sh').write_text(ENTRY_SCRIPT) disk = out/'installed.raw' with disk.open('xb') as stream: stream.truncate(8*1024**3) run(['/usr/sbin/mkfs.ext4', '-F', '-L', 'os-ota-private', '-d', root, disk], out/'mkfs.log') result.update(disk=str(disk), boot_mode='direct kernel; ext4 installed root') elif args.action == 'clone': disk = owned_disk(args.disk) destination = out/'installed.raw' run(['cp', '--reflink=auto', '--sparse=always', disk, destination], out/'clone.log') if (disk.parent/'boot.raw').is_file(): run(['cp','--reflink=auto','--sparse=always',disk.parent/'boot.raw',out/'boot.raw'],out/'clone-boot.log') result.update(action='prepare', disk=str(destination), clone_origin=str(disk), scope='independent disposable fault disk; never restore onto source VM') elif args.action == 'export': disk = owned_disk(args.disk) release = args.kernel_release if '/' in release or not release.startswith('6.') or any(c.isspace() for c in release): raise ValueError('invalid kernel release') result['files'] = {} for prefix in ('vmlinuz-', 'initrd.img-'): name = prefix + release run(['/usr/sbin/debugfs', '-R', f'dump /boot/{name} {out}/{name}', disk], out/(name+'.log')) result['files'][name] = sha(out/name) run(['/usr/sbin/debugfs', '-R', f'dump /var/log/os-ota-baseline/receipt.json {out}/baseline.json', disk], out/'baseline-export.log') run(['/usr/sbin/debugfs', '-R', f'dump /boot/grub/grub.cfg {out}/grub.cfg', disk], out/'grub-export.log') result['disk'] = str(disk) else: disk = owned_disk(args.disk) (out/'entry.sh').write_text(ENTRY_SCRIPT) inject(disk,out/'entry.sh','/opt/os-ota-entry.sh',out/'entry.log') # Never accept a previous boot's exit after an interrupted run. (out/'exit.pending').write_text('PENDING\n') inject(disk,out/'exit.pending','/var/log/os-ota-guest.exit',out/'exit-reset.log') inject(disk, args.guest.resolve(), '/opt/os-ota-guest.py', out/'inject.log') (out/'scenario').write_text(args.scenario) inject(disk, out/'scenario', '/opt/os-ota-scenario', out/'scenario.log') if args.preboot_evidence: inject(disk,args.preboot_evidence,'/opt/os-ota-external-preboot-ledger.json',out/'preboot-ledger.log') result['external_preboot_ledger_sha256']=sha(args.preboot_evidence) if args.crypto_fixture: proof=json.loads((args.crypto_fixture/'fixture-proof.json').read_text()) if proof['scope']!='ISOLATED_COMPILETIME_TEST_ROOT_NOT_CANDIDATE' or proof['binary_sha256']!=sha(args.crypto_fixture/'fixture-native'): raise ValueError('isolated crypto fixture identity mismatch') if (args.crypto_fixture/'fixture.seed').stat().st_size!=32: raise ValueError('disposable seed must be exactly 32 bytes') result['crypto_fixture']=proof inject(disk,args.crypto_fixture/'fixture-native','/opt/os-ota-crypto-native',out/'crypto-native.log') inject(disk,args.crypto_fixture/'fixture-proof.json','/opt/os-ota-crypto-fixture.json',out/'crypto-proof.log') inject(disk,args.crypto_fixture/'fixture.seed','/root/os-ota-crypto.seed',out/'crypto-seed.log',private=True) if args.native_artifact: proof=json.loads((args.native_artifact/'native-derivation.json').read_text()) if proof['packaged_sha256']!=sha(args.native_artifact/'normal-elf') or proof['source_manifest_sha256']!=sha(args.native_artifact/'source-sha256.json'): raise ValueError('native artifact/source proof mismatch') if json.loads((args.native_artifact/'compile-result.json').read_text())['exit']!=0: raise ValueError('native build failed') result['native_artifact']=proof inject(disk,args.native_artifact/'normal-elf','/opt/os-ota-native',out/'native.log') (out/'native-proof.json').write_text(json.dumps(proof,indent=2)+'\n') inject(disk,out/'native-proof.json','/opt/os-ota-native-proof.json',out/'native-proof.log') if args.source_root: if args.payload: raise ValueError('choose source-root or payload') args.payload = out/'source.tar' manifest = source_archive(args.source_root,args.payload) (out/'source-sha256.json').write_text(json.dumps(manifest, indent=2)+'\n') if args.payload: inject(disk, args.payload.resolve(), '/opt/os-ota-payload.tar', out/'payload.log') result['payload_sha256'] = sha(args.payload) result.update(guest_sha256=sha(args.guest), kernel_sha256=sha(args.kernel), initrd_sha256=sha(args.initrd), disk=str(disk)) bootdisk = disk.parent/'boot.raw' if bootdisk.is_symlink(): raise ValueError('boot disk symlink refused') if not bootdisk.exists(): with bootdisk.open('xb') as stream: stream.truncate(64*1024**2) if not bootdisk.is_file() or bootdisk.stat().st_size != 64*1024**2: raise ValueError('only owned 64 MiB regular boot disk accepted') cmd = ['qemu-system-x86_64', '-machine', 'pc', '-accel', 'kvm', '-m', '2048', '-smp', '2', '-display', 'none', '-serial', 'stdio', '-monitor', 'none', '-no-reboot', '-drive', f'file={disk},format=raw,if=virtio', '-drive', f'file={bootdisk},format=raw,if=ide,index=0', '-nic', 'none' if args.offline else 'user,model=virtio-net-pci'] if args.firmware: cmd += ['-boot','c'] else: cmd += ['-kernel',str(args.kernel),'-initrd',str(args.initrd), '-append','root=/dev/vda rw console=ttyS0 net.ifnames=0 noresume systemd.unit=multi-user.target'] result['boot_mode'] = 'BIOS/GRUB private boot disk' if args.firmware else 'direct kernel' result['command'] = cmd run(cmd, out/'serial.log', args.timeout) text = (out/'serial.log').read_text(errors='replace') run(['/usr/sbin/debugfs', '-R', f'dump /var/log/os-ota-guest.log {out}/guest.log', disk], out/'extract-guest.log') run(['/usr/sbin/debugfs', '-R', f'rdump /var/log/os-ota-stage {out}', disk], out/'extract-evidence.log') exits = re.findall(r'OS_OTA_GUEST_EXIT=(\d+)', text) result['serial_guest_exit'] = int(exits[-1]) if exits else None run(['/usr/sbin/debugfs','-R',f'dump /var/log/os-ota-guest.exit {out}/durable-guest-exit.txt',disk],out/'extract-exit.log') raw_exit=(out/'durable-guest-exit.txt').read_text().strip() result['guest_exit']=int(raw_exit) if raw_exit.isdigit() else None result['guest_exit_source']='private disk durable exit file; reset before boot' if result['guest_exit'] != 0: raise RuntimeError('guest did not report success; inspect serial.log') result['passed'] = True except Exception as error: result['error'] = str(error) raise finally: result['ended'] = time.time() (out/'result.json').write_text(json.dumps(result, indent=2)+'\n') if __name__ == '__main__': main()