the-custodian/docs/changes/CUST-WP-0080/test_forge_custody.py

55 lines
2.7 KiB
Python

#!/usr/bin/env python3
"""Exercise the exact startup command in the pinned image, with synthetic volume state."""
import hashlib
import json
from pathlib import Path
import shutil
import subprocess
import tempfile
HERE = Path(__file__).parent
original = json.loads((HERE / 'receive-hook-proposal.json').read_text())
container = json.loads((HERE / 'forge-custody-init.json').read_text())
checks = {}
with tempfile.TemporaryDirectory(prefix='forge-custody-test-') as tmp:
root = Path(tmp)
version = root / 'receive-secret-scan' / Path(original['version_directory']).name
version.mkdir(parents=True)
files = [version / x['name'] for x in original['artifacts'][:2]]
for x, dest in zip(original['artifacts'], files):
shutil.copyfile('/tmp/custodian-forge-lifecycle/' + x['name'], dest)
assert hashlib.sha256(dest.read_bytes()).hexdigest() == x['sha256']
dest.chmod(0o775)
def run():
return subprocess.run(['docker', 'run', '--rm', '--network', 'none', '--read-only',
'--cap-drop', 'ALL', '--security-opt', 'no-new-privileges', '--user', '1000:1000',
'--mount', f'type=bind,src={root},dst=/data/security', '--entrypoint', 'sh',
container['image'], *container['command'][1:]], capture_output=True, timeout=45).returncode
checks['remount_0775_repaired'] = run() == 0 and all(x.stat().st_mode & 0o777 == 0o755 for x in files)
checks['already_correct_idempotent'] = run() == 0
for x in files:
x.chmod(0o775)
saved = files[1].read_bytes()
files[1].write_bytes(b'corrupt fixture')
checks['wrong_hash_refused_before_any_chmod'] = run() != 0 and all(x.stat().st_mode & 0o777 == 0o775 for x in files)
files[1].write_bytes(saved)
files[1].chmod(0o777)
checks['unexpected_world_write_refused'] = run() != 0 and files[0].stat().st_mode & 0o777 == 0o775
files[1].chmod(0o775)
target = version / 'link-target'
files[1].rename(target)
files[1].symlink_to(target.name)
checks['symlink_refused'] = run() != 0 and target.stat().st_mode & 0o777 == 0o775
files[1].unlink()
target.rename(files[1])
checks['recovery_after_refusals'] = run() == 0 and all(x.stat().st_mode & 0o777 == 0o755 for x in files)
receipt = {'result': 'verified' if all(checks.values()) else 'failed', 'checks': checks,
'scope': 'Exact startup command and pinned Forgejo image; simulated fsGroup remount modes, no production mount.',
'container_sha256': hashlib.sha256((HERE / 'forge-custody-init.json').read_bytes()).hexdigest()}
Path('docs/evidence/2026-10-13-netkingdom-security/forge-custody-isolated.json').write_text(json.dumps(receipt, indent=2) + '\n')
print(json.dumps(receipt))
assert all(checks.values())