Native strace preparation preflight

Prospective tool preparation only; NOT RUN. Resolve Fedora44 native strace from Fedora repositories, verify a package signature against the pinned installed Fedora44 public key using an owned private RPM database, then extract under DATA without installing anything. The exact resolved NEVRA, package hash, executable and manual hashes are recorded before native observer use. This establishes tool availability only, not publication or acceptance evidence.

{
  "command": [
    "python3",
    "-B",
    "/run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-trace-tool-01.py",
    "native-strace-preparation-01",
    "1073741824",
    "env",
    "PYTHONPATH=tools",
    "python3",
    "-B",
    "/run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/prepare-native-strace-01.py"
  ],
  "driver_sha256": "60d7b098368bfdcdf4fa17df74b34019b97b1289ba15cdfaad0bb4510ff12b09",
  "preparer_sha256": "f7e08d9c380171998ada159ed4d2b269df58cb11726fee24396df276df6c1793",
  "source": "6bd5250a6223fe3663b254c25627ce4ff2b4c9e0",
  "scope": "Resolve one signed Fedora44 strace RPM from fedora/updates only, validate against pinned host Fedora44 signing key in owned rpmdb, extract on DATA, retain binary/package/manual hashes and actual --version and trace /usr/bin/true exits. No host install or product changes.",
  "limits": "1GiB owned/output reserve sampled every0.5s (sampling overshoot remains possible); wholeDATA95e9 guard; 1/2GiB memory high/max,zeroSwap,CPU200%,tasks256,inner600s/outer1800s; caches/state/repos/logs/tmp/rpmdb explicitly onDATA; serialized."
}

Preparation controller

"""Resolve one signed Fedora tool into DATA; never install a host package."""
import hashlib
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import time

DATA = Path('/run/media/kmosoti/data/FabricO11y')
OUT = DATA / 'results/native-strace-preparation-01'
OWNED = DATA / 'scratch/native-strace-preparation-01'
PINS = {
    '/etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-44-primary': '93642aec521a1e5e96dd715f7ae0ec0850ebc9de09a94ce03cae5263f26cc18a',
    '/etc/yum.repos.d/fedora.repo': '91ec30ca30615b60119587ceb275d5318131fd7b3e56e60677eef5092a09243f',
    '/etc/yum.repos.d/fedora-updates.repo': '762810cb4ab863a7f57ed85602e268d29c8b002128452839d5b415b9f1752019',
}


def need(ok, reason):
    if not ok:
        raise RuntimeError(reason)


def sha(path):
    with open(path, 'rb') as stream:
        return hashlib.file_digest(stream, 'sha256').hexdigest()


from resource_group import require_limits, require_storage
group = require_limits()
need(require_storage() == DATA, 'mounted DATA required')
need(not OUT.exists() and not OWNED.exists(), 'exclusive tool preparation roots')
for path, digest in PINS.items():
    need(sha(path) == digest, 'host distro authority changed')
OUT.mkdir(mode=0o700); OWNED.mkdir(mode=0o700)
(OWNED / '.owned-trace-tool').write_text('native-strace-preparation-01\n')
for name in ('cache', 'system-cache', 'state', 'log', 'tmp', 'repos', 'rpmdb'):
    (OWNED / name).mkdir()
for name in ('packages', 'extract', 'logs', 'authority'):
    (OUT / name).mkdir()
for path in PINS:
    shutil.copy2(path, OUT / 'authority' / Path(path).name)
    need(sha(OUT / 'authority' / Path(path).name) == PINS[path], 'archived distro authority changed')
for name in ('fedora.repo', 'fedora-updates.repo'):
    shutil.copy2(OUT / 'authority' / name, OWNED / 'repos' / name)
environment = dict(os.environ, TMPDIR=str(OWNED / 'tmp'), TMP=str(OWNED / 'tmp'),
                   TEMP=str(OWNED / 'tmp'), XDG_CACHE_HOME=str(OWNED / 'cache'),
                   XDG_STATE_HOME=str(OWNED / 'state'))
receipt = dict(state='preparing', exit=1, source_authorities=PINS, commands=[],
               scope='host tool preparation only; no product acceptance', owned=str(OWNED))
end = time.monotonic() + 600


def run(label, command, data=None):
    need(time.monotonic() < end, 'tool preparation deadline')
    input_path = OWNED / 'tmp/command-input'
    input_path.write_bytes(data or b'')
    limit = min(end, time.monotonic() + 300)
    stdout, stderr = OUT / 'logs' / (label + '.stdout'), OUT / 'logs' / (label + '.stderr')
    with input_path.open('rb') as incoming, stdout.open('wb') as output, stderr.open('wb') as errors:
        process = subprocess.Popen(command, stdin=incoming, stdout=output, stderr=errors, env=environment)
        try:
            while True:
                size = sum((Path(root) / name).lstat().st_size for tree in (OWNED, OUT)
                           for root, _, files in os.walk(tree) for name in files)
                need(size < 1024**3, 'one GiB owned tool preparation allowance')
                if process.poll() is not None:
                    break
                need(time.monotonic() < limit, 'tool command deadline')
                time.sleep(.5)
        finally:
            if process.poll() is None:
                process.terminate()
                try:
                    process.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    process.kill(); process.wait(timeout=5)
            receipt['commands'].append(dict(label=label, command=command, exit=process.returncode))
    need(process.returncode == 0, label + ' failed')
    need(stdout.stat().st_size < 32 * 1024**2, 'bounded command output')
    return stdout.read_bytes()


try:
    run('caps', ['systemctl', '--user', 'set-property', '--runtime', group.name,
                 'MemoryHigh=1G', 'MemoryMax=2G', 'MemorySwapMax=0', 'CPUQuota=200%', 'TasksMax=256'])
    caps = {name: (group / name).read_text().strip() for name in
            ('memory.high', 'memory.max', 'memory.swap.max', 'cpu.max', 'pids.max')}
    need(caps == {'memory.high': str(1024**3), 'memory.max': str(2 * 1024**3),
                  'memory.swap.max': '0', 'cpu.max': '200000 100000', 'pids.max': '256'}, 'tool caps')
    receipt['caps'] = caps
    dnf = ['dnf5', '--releasever=44', '--repo=fedora,updates']
    for key, value in {'cachedir': 'cache', 'system_cachedir': 'system-cache',
                       'persistdir': 'state', 'logdir': 'log', 'reposdir': 'repos'}.items():
        dnf.append('--setopt=' + key + '=' + str(OWNED / value))
    run('download', [*dnf, 'download', '--arch=x86_64', '--destdir=' + str(OUT / 'packages'), 'strace'])
    packages = list((OUT / 'packages').glob('*.rpm'))
    need(len(packages) == 1, 'exactly one strace package')
    package = packages[0]
    run('rpmdb-init', ['rpm', '--dbpath', str(OWNED / 'rpmdb'), '--initdb'])
    run('key-import', ['rpmkeys', '--dbpath', str(OWNED / 'rpmdb'), '--import',
                       str(OUT / 'authority/RPM-GPG-KEY-fedora-44-primary')])
    signature = run('signature', ['rpmkeys', '--dbpath', str(OWNED / 'rpmdb'), '--checksig', '--verbose', str(package)])
    need(any(b'signature' in line.lower() and line.strip().endswith(b': OK')
             for line in signature.splitlines()), 'verified Fedora signature required, not digest alone')
    identity = run('package-identity', ['rpm', '-qp', '--qf', '%{NAME}|%{VERSION}|%{RELEASE}|%{ARCH}\n', str(package)]).decode().strip()
    need(identity.split('|')[0] == 'strace' and identity.split('|')[-1] == 'x86_64', 'native strace identity')
    archive = run('cpio', ['rpm2cpio', str(package)])
    need(len(archive) < 32 * 1024**2, 'bounded strace archive')
    names = run('archive-list', ['cpio', '-t'], archive).decode().splitlines()
    need(names and all(name.startswith('./usr/') and '..' not in Path(name).parts for name in names), 'tool archive namespace')
    run('extract', ['cpio', '-idm', '--no-absolute-filenames', '--no-preserve-owner', '-D', str(OUT / 'extract')], archive)
    binary = OUT / 'extract/usr/bin/strace'
    need(binary.is_file() and not binary.is_symlink(), 'regular extracted strace')
    run('version', [str(binary), '--version'])
    run('native-trace', [str(binary), '-f', '-s', '0', '-o', str(OUT / 'true.trace'),
                        '-e', 'trace=execve,exit_group', '/usr/bin/true'])
    receipt.update(state='prepared', exit=0, package=str(package), package_sha256=sha(package),
                   package_identity=identity, binary=str(binary), binary_sha256=sha(binary))
    receipt['extracted_files'] = {str(path.relative_to(OUT / 'extract')): sha(path)
                                 for path in (OUT / 'extract').rglob('*') if path.is_file() and not path.is_symlink()}
    shutil.rmtree(OWNED)
    receipt['cleanup'] = 'owned cache/state/rpmdb removed; signed RPM, extracted tool/manuals and raw logs retained'
except BaseException as error:
    receipt.update(state='failed', error=type(error).__name__ + ': ' + str(error),
                   cleanup='owned failure evidence retained')
finally:
    receipt['memory'] = {name: (group / name).read_text().strip() for name in
                         ('memory.peak', 'memory.events', 'memory.max', 'memory.swap.max')}
    (OUT / 'receipt.json').write_text(json.dumps(receipt, indent=2) + '\n')
    print(json.dumps(receipt))
sys.exit(receipt['exit'])

Guarded launcher

import datetime, hashlib, json, os, re, signal, subprocess, sys, time
from pathlib import Path
ROOT = Path('/run/media/kmosoti/data/FabricO11y')
REPO = ROOT / 'scratch/trace-tool-preparation'
OUT = ROOT / 'results/service-opening-candidate'
STORE = ROOT / 'scratch/service-opening-census'
FROZEN = '6bd5250a6223fe3663b254c25627ce4ff2b4c9e0'
def require(value, reason):
    if not value:
        raise ValueError(reason)
label, reserve_text, *command = sys.argv[1:]
reserve = int(reserve_text)
delegated = command[:1] == ['--delegate']
if delegated: command = command[1:]
require(re.fullmatch(r'[a-z0-9-]{1,80}', label) and reserve > 0 and command, 'ordinary workload admission guard')
record_path = OUT / (label + '.json')
require(not record_path.exists(), 'unique run label required')
def census():
    # Rootless container files can belong to subordinate UIDs. Inspect all of
    # them in the same user namespace; never skip unreadable directories.
    require(not STORE.is_symlink(), 'ordinary workload admission guard')
    require((STORE / '.owned-census-store').read_text() == 'service-opening-census-v1\n', 'ordinary workload admission guard')
    args = ['podman', '--root', str(STORE/'graph'), '--runroot', str(STORE/'run'),
            '--tmpdir', str(STORE/'tmp'), '--storage-driver', 'vfs',
            'unshare', 'python3', '-B', str(OUT/'census.py')]
    env = dict(os.environ, TMPDIR=str(STORE/'tmp'), TMP=str(STORE/'tmp'), TEMP=str(STORE/'tmp'))
    result = subprocess.run(args, env=env, text=True, capture_output=True, timeout=25)
    if result.returncode:
        raise RuntimeError(f'strict namespace census failed ({result.returncode}): {result.stderr}')
    observed = json.loads(result.stdout)
    require(observed['uid'] == 0 and observed['gid'] == 0, 'ordinary workload admission guard')
    require(type(observed['data_bytes']) is int and observed['data_bytes'] >= 0, 'ordinary workload admission guard')
    return observed
def git(*args):
    return subprocess.check_output(['git', *args], cwd=REPO, text=True).strip()
require(git('rev-parse', 'HEAD') == FROZEN and not git('status', '--porcelain'), 'clean frozen source required')
require(not subprocess.check_output(['systemctl', '--user', 'list-units', 'fabric-work-*', '--state=running', '--plain', '--no-legend'], text=True).strip(), 'another project workload is active')
initial = census()
used = initial['data_bytes']
record = {'label': label, 'command': command, 'delegated': delegated, 'frozen': None, 'scope': 'harness controls only; no candidate acceptance evidence', 'harness_revision': FROZEN, 'source_status_before': git('status', '--porcelain'), 'storage_root': str(ROOT), 'admission_bytes': used, 'admission_observation': initial, 'reserved_bytes': reserve, 'observations': [], 'started_utc': datetime.datetime.now(datetime.timezone.utc).isoformat()}
if used + reserve > 95_000_000_000:
    record.update(state='not_run', reason='95 GB storage admission threshold')
    record_path.write_text(json.dumps(record, indent=2)+'\n')
    raise SystemExit('storage admission denied; project command not launched')
receipts = REPO / 'target/resource-containment/runs'
before = set(receipts.glob('*.json'))
env = dict(os.environ, FABRIC_STORAGE_ROOT=str(ROOT), CARGO_HOME=str(ROOT/'toolchain-cache/cargo'), RUSTUP_HOME=str(ROOT/'toolchain-cache/rustup'), CARGO_BUILD_JOBS='2')
env['PATH'] = str(ROOT/'tools/candidate-dpkg/bin') + ':' + env['PATH']
log_path = OUT / (label + '.log')
record.update(state='running', log=str(log_path))
record_path.write_text(json.dumps(record, indent=2)+'\n')
def find_unit():
    with log_path.open() as stream:
        prefix = stream.read(65536)
    match = re.search(r'Running as unit: (fabric-work-[0-9a-f]+)\.service', prefix)
    return match.group(1) if match else None
def stop_owned(process):
    unit = find_unit()
    if unit:
        stopped = subprocess.run(['systemctl', '--user', 'stop', unit],
                                 text=True, capture_output=True, timeout=25)
        record['stop'] = {'unit': unit, 'exit': stopped.returncode, 'stderr': stopped.stderr}
        if stopped.returncode:
            # Interrupt the launcher too; its finally block independently
            # confirms that this exact service has no surviving descendants.
            process.send_signal(signal.SIGINT)
    else:
        process.send_signal(signal.SIGINT)
        record['stop'] = {'unit': None, 'launcher_interrupted': True}
with log_path.open('wb') as log:
    launcher = ['python3', '-B', 'tools/resource_group.py', *(['--delegate'] if delegated else []), '--', *command]
    process = subprocess.Popen(launcher, cwd=REPO, env=env, stdout=log, stderr=subprocess.STDOUT)
    next_census = time.monotonic() + 30
    try:
        while process.poll() is None:
            time.sleep(1)
            if time.monotonic() >= next_census:
                observed = census()
                record['observations'].append({'unix':time.time(), **observed})
                record_path.write_text(json.dumps(record, indent=2)+'\n')
                if observed['data_bytes'] >= 95_000_000_000:
                    record['interruption'] = '95 GB threshold reached'
                    break
                next_census = time.monotonic() + 30
    except BaseException as error:
        record['interruption'] = f'{type(error).__name__}: {error}'
    finally:
        if record.get('interruption') and process.poll() is None:
            try:
                stop_owned(process)
            except BaseException as error:
                record['stop_error'] = f'{type(error).__name__}: {error}'
                process.send_signal(signal.SIGINT)
        status = process.wait()
record['launcher_exit'] = status
if record.get('interruption'): status = 2
with log_path.open('rb') as stream:
    log_digest = hashlib.file_digest(stream, 'sha256').hexdigest()
record.update(state='interrupted' if record.get('interruption') else 'completed', exit=status, source_status_after=git('status', '--porcelain'), finished_utc=datetime.datetime.now(datetime.timezone.utc).isoformat(), log_sha256=log_digest, resource_receipts=[])
archive = OUT/'resource-containment'; archive.mkdir(exist_ok=True)
for p in sorted(set(receipts.glob('*.json')) - before):
    value=json.loads(p.read_text())
    if value['command'] != command: continue
    q=archive/p.name; q.write_bytes(p.read_bytes())
    record['resource_receipts'].append({'path':str(q),'sha256':hashlib.sha256(q.read_bytes()).hexdigest(),'exit':value['exit'],'stop_confirmed':value['stop_confirmed'],'retained_failure_evidence':value['retained_failure_evidence']})
if not record['resource_receipts'] or any(not r['stop_confirmed'] or r['exit'] != status for r in record['resource_receipts']):
    record['receipt_error'] = 'missing or inconsistent outer resource receipt'
    status = 2
    record['exit'] = status
record_path.write_text(json.dumps(record, indent=2)+'\n')
print(json.dumps(record, indent=2))
if status: print(log_path.read_text()[-8000:])
raise SystemExit(status)