#!/usr/bin/env python3 """Bounded, attribution-only native Apple keychain lifecycle, without PFX/SDK/CLR.""" import argparse import hashlib import json import os from pathlib import Path import re import shutil import subprocess import sys import time HERE = Path(__file__).resolve().parent STAGES = ['before', 'first', 'repeated', 'idle'] def save(path, value): path.write_text(json.dumps(value, indent=2) + '\n') def complete_rows(path): return [json.loads(line) for line in path.read_text().split('\n')[:-1]] def snapshot(pid, task, output, stage): heap = subprocess.run(['/usr/bin/heap', '--noContent', '-s', str(pid)], text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=20, check=True) (output / ('heap-' + stage + '.txt')).write_text(heap.stdout) database = re.search(r'^\s*(\d+)\s+(\d+)\s+[\d.]+\s+Security::AppleDatabase\s', heap.stdout, re.MULTILINE) manager = re.search(r'^\s*(\d+)\s+(\d+)\s+[\d.]+\s+Security::AppleDatabaseManager\s', heap.stdout, re.MULTILINE) if database is None and manager is None: raise RuntimeError('Native inventory could not classify the database or its manager; do not silently infer zero') descriptors = subprocess.run(['/usr/sbin/lsof', '-a', '-p', str(pid), '-Fnf'], text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=5, check=True) (output / ('lsof-' + stage + '.txt')).write_text(descriptors.stdout) total = temp = keychain = 0 in_descriptor = False for line in descriptors.stdout.splitlines(): if line.startswith('f'): in_descriptor = len(line) > 1 and line[1].isdigit() total += int(in_descriptor) elif in_descriptor and line.startswith('n/'): path = Path(line[1:].removesuffix(' (deleted)')).resolve() owned = path == task / 'tmp' or task / 'tmp' in path.parents temp += int(owned) keychain += int(owned and '.keychain' in path.name) return {'AppleDatabaseCount': int(database[1]) if database else 0, 'AppleDatabaseShallowBytes': int(database[2]) if database else 0, 'AppleDatabaseClassAbsent': database is None, 'databaseManagerClassPresent': manager is not None, 'fdCount': total, 'ownedTemporaryFdCount': temp, 'ownedKeychainFdCount': keychain, 'ownedKeychainFileCount': sum(p.is_file() and '.keychain' in p.name for p in (task / 'tmp').rglob('*'))} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--output', required=True, type=Path, help='Fresh private output directory') options = parser.parse_args() if sys.platform != 'darwin': parser.error('Native control requires macOS') output = options.output.resolve() output.mkdir(parents=True, exist_ok=False) os.chmod(output, 0o700) task = output / 'task' binary = output / 'PureAppleControl' command = ['clang', '-std=c11', '-O0', '-g', '-Wall', '-Wextra', str(HERE / 'PureAppleControl.c'), '-framework', 'Security', '-framework', 'CoreFoundation', '-o', str(binary)] source_hash = hashlib.sha256((HERE / 'PureAppleControl.c').read_bytes()).hexdigest() provenance = {'sourceSha256': source_hash, 'runnerSourceSha256': hashlib.sha256((HERE / 'pure_apple.py').read_bytes()).hexdigest(), 'compileCommand': command, 'compiler': subprocess.check_output(['clang', '--version'], text=True), 'os': subprocess.check_output(['sw_vers'], text=True), 'architecture': os.uname().machine, 'repetitions': 200, 'idleSeconds': 5, 'stages': STAGES, 'fullSoakGate': False, 'readOnlyKeychainQueries': ['SecKeychainCopyDefault', 'SecKeychainCopySearchList', 'SecKeychainGetPath'], 'doesNotCallAnyDefaultOrSearchListSetter': True} save(output / 'provenance.json', provenance) with (output / 'build.log').open('w') as log: subprocess.run(command, stdout=log, stderr=subprocess.STDOUT, timeout=60, check=True) provenance['binarySha256'] = hashlib.sha256(binary.read_bytes()).hexdigest() provenance['linkedLibraries'] = subprocess.check_output(['otool', '-L', str(binary)], text=True) if 'libcoreclr' in provenance['linkedLibraries'] or 'Opc.Ua' in provenance['linkedLibraries']: raise RuntimeError('Pure Apple control must not load SDK or CLR') save(output / 'provenance.json', provenance) task.mkdir(mode=0o700) (task / 'tmp').mkdir(mode=0o700) started = time.monotonic() with (output / 'samples.jsonl').open('w') as log: process = subprocess.Popen([str(binary), str(task)], stdout=log, stderr=subprocess.STDOUT, env=dict(os.environ, TMPDIR=str(task / 'tmp'))) try: deadline = time.monotonic() + 180 samples = [] for stage in STAGES: while True: rows = complete_rows(output / 'samples.jsonl') if any(row['kind'] == 'failed' for row in rows): raise RuntimeError('Native control failed; preserve raw status rows') checkpoints = [row for row in rows if row.get('stage') == stage and row['kind'] == 'checkpoint'] if checkpoints: samples.append({**checkpoints[0], **snapshot(process.pid, task, output, stage)}) (task / ('continue-' + stage)).touch() break if process.poll() is not None or time.monotonic() >= deadline: raise RuntimeError('Native checkpoint exceeded bound: ' + stage) time.sleep(.05) status = process.wait(timeout=15) rows = complete_rows(output / 'samples.jsonl') operations = [r for r in rows if r['kind'] == 'operation'] fingerprints = [r for r in rows if r['kind'] == 'fingerprint'] complete = [r for r in rows if r['kind'] == 'complete'] if status or len(operations) != 200 or len(fingerprints) != 2 or len(complete) != 1: raise RuntimeError('Incomplete native execution count') if any(r['createStatus'] or r['deleteStatus'] or not r['privatePathVerified'] or not r['releasedExactlyOnce'] for r in operations): raise RuntimeError('Native lifecycle status/ownership failed') if any(samples[-1][k] for k in ['ownedKeychainFdCount', 'ownedKeychainFileCount']): raise RuntimeError('Private native keychain files/descriptors retained') result = {'status': 'passed', 'exitCode': status, 'wallSecondsIncludingProbes': time.monotonic() - started, 'samples': samples, 'fingerprints': fingerprints, 'complete': complete[0], 'fullSoakGate': False, 'nativeInventoryIsNotOpenHandlesOrExclusiveRss': True, 'nativeDeleteOsStatusesAllZero': True} save(output / 'result.json', result) finally: if process.poll() is None: process.terminate() try: process.wait(timeout=10) except subprocess.TimeoutExpired: process.kill(); process.wait(timeout=5) save(output / 'fixture-before-harness-cleanup.json', {'remainingOwnedKeychainFiles': sum(p.is_file() and '.keychain' in p.name for p in (task / 'tmp').rglob('*'))}) shutil.rmtree(task) save(output / 'fixture-cleanup.json', {'ownedFixtureRootRemoved': not task.exists()}) print(output) if __name__ == '__main__': main()