#!/usr/bin/env python3 """GKDOE2MM: classify, then optionally reap only two finished target subtrees.""" import datetime import hashlib import json import os from pathlib import Path import shutil import stat import subprocess import sys ROOT = Path('/home/reavus/projects/spt-core/spt-core') PROOF = Path(__file__).resolve().parent LANES = [ ('s4', ROOT / '.worktrees/304-release-note-s4', ROOT / '.spt/preserved/304-handoff/s4-release-note'), ('registry308', ROOT / '.worktrees/308-registry-process-lock', ROOT / '.spt/preserved/308-registry-process-lock/linux-validation'), ] TARGETS = [tree / 'target' for _, tree, _ in LANES] WARM = ROOT / '.worktrees/consumer-linux-49a08a07' def utc(): return datetime.datetime.now(datetime.timezone.utc).isoformat() def digest(path): with path.open('rb') as stream: return hashlib.file_digest(stream, 'sha256').hexdigest() def save(name, value): (PROOF / name).write_text(json.dumps(value, indent=2) + '\n') def source_snapshot(tree): files = subprocess.check_output(['git', 'ls-files', '-z', '--cached', '--others', '--exclude-standard'], cwd=tree).split(b'\0') values = {} for name in files: if not name: continue relative = name.decode() path = tree / relative info = path.lstat() values[relative] = {'mode': info.st_mode, 'content': os.readlink(path) if path.is_symlink() else digest(path) if path.is_file() else None} return {'inode': tree.stat().st_ino, 'git_file_sha256': digest(tree / '.git'), 'files': values} def touches(value): return any(value == str(target) or value.startswith(str(target) + '/') for target in TARGETS) def census(): active, references, limits = [], [], [] for proc in Path('/proc').iterdir(): if not proc.name.isdigit(): continue try: name = (proc / 'comm').read_text().strip() command = (proc / 'cmdline').read_bytes().replace(b'\0', b' ').decode(errors='replace') exe = None try: exe = os.readlink(proc / 'exe') except PermissionError: limits.append({'pid': int(proc.name), 'field': 'exe'}) except FileNotFoundError: pass if any(word in name for word in ('cargo', 'rustc', 'nextest', 'Runner.Worker', 'xtask')) or exe and '/deps/' in exe: active.append({'pid': int(proc.name), 'name': name, 'exe': exe, 'command': command}) if exe and touches(exe) or any(str(target) in command for target in TARGETS): references.append({'pid': int(proc.name), 'kind': 'exe-or-command', 'exe': exe, 'command': command}) for field in ('cwd', 'root'): try: value = os.readlink(proc / field) if touches(value): references.append({'pid': int(proc.name), 'kind': field, 'path': value}) except PermissionError: limits.append({'pid': int(proc.name), 'field': field}) except FileNotFoundError: pass try: for fd in (proc / 'fd').iterdir(): try: value = os.readlink(fd) if touches(value): references.append({'pid': int(proc.name), 'kind': 'fd', 'path': value}) except FileNotFoundError: pass except PermissionError: limits.append({'pid': int(proc.name), 'field': 'fd'}) try: maps = (proc / 'maps').read_text() if any(str(target) in maps for target in TARGETS): references.append({'pid': int(proc.name), 'kind': 'maps'}) except PermissionError: limits.append({'pid': int(proc.name), 'field': 'maps'}) except (FileNotFoundError, ProcessLookupError): continue except PermissionError: limits.append({'pid': int(proc.name), 'field': 'process'}) protected = [] for pid in sorted({row['pid'] for row in limits}): try: proc = Path('/proc') / str(pid) status = (proc / 'status').read_text() uid = int(next(line for line in status.splitlines() if line.startswith('Uid:')).split()[1]) if uid == os.getuid(): protected.append({'pid': pid, 'name': (proc / 'comm').read_text().strip(), 'command': (proc / 'cmdline').read_bytes().replace(b'\0', b' ').decode(errors='replace')}) except FileNotFoundError: pass assert all(row['name'] in ('(sd-pam)', 'fusermount3', 'ssh-agent') or row['name'] == 'sshd' and row['command'].startswith('sshd: reavus@') for row in protected), protected return {'utc': utc(), 'active': active, 'target_references': references, 'permission_limits': limits, 'protected_same_uid_services': protected} def classify(): lanes = [] for name, tree, proofs in LANES: target = tree / 'target' info = target.lstat() assert stat.S_ISDIR(info.st_mode) and not target.is_symlink() and target.resolve() == target owner = json.loads((target / 'POOL-OWNER.json').read_text()) assert owner == {'owner_tree': str(tree), 'written_by': 'spt-poolguard'}, owner transfer = json.loads((proofs / 'proof-transfer-verification.json').read_text()) for relative, expected in transfer['files'].items(): assert digest(proofs / relative) == expected, (name, relative, 'preserved proof changed') hashes = {str(path.relative_to(proofs)): digest(path) for path in proofs.rglob('*') if path.is_file()} release = json.loads((proofs / 'pool-release-receipt.json').read_text()) claim = json.loads((proofs / 'pool-claim-receipt.json').read_text()) result = json.loads((proofs / 'receipt.json').read_text()) assert result['success'] and release['exit'] == 0 and claim['exit'] == 0 assert digest(proofs / 'pool-release.log') == release['log_sha256'] assert digest(proofs / 'pool-claim.log') == claim['log_sha256'] claim_argv = claim.get('command', claim.get('argv')) assert str(target) in claim_argv assert ('todlando-304-s4-docs' if name == 's4' else 'registry308-linux') in claim_argv creation = json.loads(((proofs / 'setup.json') if name == 's4' else (PROOF / 'registry308-creation.json')).read_text()) assert creation['tree'] == str(tree) commit = json.loads((proofs / 'commit-receipt.json').read_text()) assert commit['sha'] == ('35d6f7a38f2189455f1a9e1a85a86b1e4bc0481a' if name == 's4' else 'fefc4d64298baf2300a790e9b5ab201d23cb0405') allocated = int(subprocess.check_output(['du', '-s', '-B1', str(target)], text=True).split()[0]) lanes.append({'name': name, 'tree': str(tree), 'target': str(target), 'outbound': 'real-directory', 'inode': info.st_ino, 'device': info.st_dev, 'owner': owner, 'allocated_bytes': allocated, 'proof_root': str(proofs), 'proof_hashes': hashes, 'prior_manifest_verified': len(transfer['files']), 'release': release, 'creation': creation, 'commit': commit['sha'], 'source': source_snapshot(tree)}) inbound, errors = [], [] for directory, dirs, files in os.walk('/home/reavus', followlinks=False, onerror=lambda e: errors.append(str(e))): for name in dirs + files: path = Path(directory) / name if not path.is_symlink(): continue resolved = os.path.realpath(path) for target in TARGETS: if (resolved == str(target) or resolved.startswith(str(target) + '/')) and not path.is_relative_to(target): inbound.append({'path': str(path), 'resolved': resolved}) report = {'authority': 'GKDOE2MM', 'utc': utc(), 'lanes': lanes, 'inbound_scope': '/home/reavus', 'inbound': inbound, 'scan_errors': errors, 'census': census(), 'free_bytes': shutil.disk_usage(ROOT).free} save('classification-final.json' if '--apply' in sys.argv else 'classification.json', report) print(json.dumps({'utc': report['utc'], 'free_bytes': report['free_bytes'], 'lanes': [{'name': row['name'], 'bytes': row['allocated_bytes'], 'proof_files': len(row['proof_hashes'])} for row in lanes], 'inbound': inbound, 'scan_errors': errors, 'active': report['census']['active'], 'target_references': report['census']['target_references']}), flush=True) assert not inbound and not errors and not report['census']['active'] and not report['census']['target_references'] return report report = classify() if '--apply' in sys.argv: assert not (PROOF / 'reclaim.json').exists() prior = json.loads((PROOF / 'classification.json').read_text()) for before, now in zip(prior['lanes'], report['lanes']): assert all(before[key] == now[key] for key in ('target', 'inode', 'device', 'owner', 'proof_hashes', 'source')) warm_before = {'tree_inode': WARM.stat().st_ino, 'target_inode': (WARM / 'target').stat().st_ino, 'target_mtime_ns': (WARM / 'target').stat().st_mtime_ns, 'owner_sha256': digest(WARM / 'target/POOL-OWNER.json')} receipt = {'authority': 'GKDOE2MM', 'start_utc': utc(), 'free_bytes_before': shutil.disk_usage(ROOT).free, 'classification_sha256': digest(PROOF / 'classification-final.json'), 'targets': [], 'warm_before': warm_before} save('reclaim-before.json', receipt) for row in report['lanes']: target = Path(row['target']) assert target in TARGETS and target.name == 'target' and target.resolve() == target and not target.is_symlink() assert target.lstat().st_ino == row['inode'] assert json.loads((target / 'POOL-OWNER.json').read_text()) == row['owner'] fresh = census() assert not fresh['active'] and not fresh['target_references'], fresh before = shutil.disk_usage(ROOT).free shutil.rmtree(target) os.sync() after = shutil.disk_usage(ROOT).free assert not target.exists() assert source_snapshot(Path(row['tree'])) == row['source'] receipt['targets'].append({'target': str(target), 'allocated_bytes_before': row['allocated_bytes'], 'free_bytes_before': before, 'free_bytes_after': after, 'free_delta': after - before, 'target_exists': False, 'source_unchanged': True}) save('reclaim-progress.json', receipt) warm_after = {'tree_inode': WARM.stat().st_ino, 'target_inode': (WARM / 'target').stat().st_ino, 'target_mtime_ns': (WARM / 'target').stat().st_mtime_ns, 'owner_sha256': digest(WARM / 'target/POOL-OWNER.json')} assert warm_before == warm_after for row in report['lanes']: assert all(digest(Path(row['proof_root']) / name) == value for name, value in row['proof_hashes'].items()) receipt.update(end_utc=utc(), free_bytes_after=shutil.disk_usage(ROOT).free, warm_after=warm_after, proofs_unchanged=True, warm_s3_unchanged=True, final_census=census()) receipt['free_delta'] = receipt['free_bytes_after'] - receipt['free_bytes_before'] save('reclaim.json', receipt) print('RECLAIM_DONE ' + json.dumps({k: v for k, v in receipt.items() if k != 'final_census'}), flush=True)