#!/usr/bin/env python3 """Bounded attribution-only macOS SDK private-key reload/registry controls.""" import argparse import hashlib import json import os from pathlib import Path import re import shutil import subprocess import sys import time import xml.etree.ElementTree as ET HERE = Path(__file__).resolve().parent STAGES = ['ready', 'first', 'repeated', 'disposed', 'idle'] def save(path, value): path.write_text(json.dumps(value, indent=2) + '\n') def snapshots(pid, root, stage): # These raw, no-content inventory files remain private/local and are not release artifacts. heap = subprocess.run(['/usr/bin/heap', '--noContent', '-s', str(pid)], text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=20, check=True) (root / ('heap-' + stage + '.txt')).write_text(heap.stdout) match = re.search(r'^\s*(\d+)\s+(\d+)\s+[\d.]+\s+Security::AppleDatabase\s', heap.stdout, re.MULTILINE) if not match: raise RuntimeError('AppleDatabase inventory class missing; diagnostic cannot silently report zero') descriptors = subprocess.run(['/usr/sbin/lsof', '-a', '-p', str(pid), '-Fnf'], text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=5, check=True) (root / ('lsof-' + stage + '.txt')).write_text(descriptors.stdout) total = pki = temporary = keychain = 0 in_descriptor = False pki_root = (root / 'pki').resolve() tmp_root = (root / 'tmp').resolve() 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/'): target = Path(line[1:].removesuffix(' (deleted)')).resolve() pki += int(target == pki_root or pki_root in target.parents) in_temp = target == tmp_root or tmp_root in target.parents temporary += int(in_temp) keychain += int(in_temp and '.keychain' in target.name) if pki: raise RuntimeError('Quiescent task-owned PKI descriptors are open') if stage in ['disposed', 'idle'] and keychain: raise RuntimeError('Owned temporary keychain descriptors remain after application disposal') return {'stage': stage, 'AppleDatabaseCount': int(match[1]), 'AppleDatabaseShallowBytes': int(match[2]), 'fdCount': total, 'pkiFdCount': pki, 'temporaryDirectoryFdCount': temporary, 'temporaryKeychainFdCount': keychain, 'temporaryKeychainFileCount': sum(p.is_file() and '.keychain' in p.name for p in (root / 'tmp').rglob('*'))} def arm(app, root, mode): (root / 'tmp').mkdir() (root / 'data').mkdir() env = dict(os.environ, OPCUA_CERTIFICATE_LEAK_TRACKING='1', TMPDIR=str(root / 'tmp'), XDG_DATA_HOME=str(root / 'data')) with (root / 'samples.jsonl').open('w') as log: process = subprocess.Popen(['dotnet', str(app), mode, str(root)], stdout=log, stderr=subprocess.STDOUT, env=env) try: deadline = time.monotonic() + 180 values = [] for stage in STAGES: while True: text = (root / 'samples.jsonl').read_text() rows = [json.loads(line) for line in text.split('\n')[:-1]] if any(row.get('kind') == 'failed' for row in rows): raise RuntimeError('Control failed; see samples.jsonl') matching = [row for row in rows if row.get('stage') == stage] if matching: values.append({**matching[0], **snapshots(process.pid, root, stage)}) (root / ('continue-' + stage)).touch() break if process.poll() is not None or time.monotonic() >= deadline: raise RuntimeError('Control checkpoint did not complete within bound: ' + stage) time.sleep(.05) status = process.wait(timeout=15) rows = [json.loads(line) for line in (root / 'samples.jsonl').read_text().splitlines()] complete = [row for row in rows if row.get('kind') == 'complete'] if status or len(complete) != 1 or complete[0]['operations'] != 200: raise RuntimeError('Control failed or acquisition count changed') result = {'exitCode': status, 'complete': complete[0], 'samples': values, 'nativeCountIsNotOpenHandleOrExclusiveRssAttribution': True, 'registryArmIsNotApprovedProductionMitigation': True} save(root / 'result.json', result) return result finally: if process.poll() is None: process.terminate() try: process.wait(timeout=10) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--references', required=True, type=Path, help='Actual immutable published SDK2 DLL directory') parser.add_argument('--output', required=True, type=Path, help='Fresh private output; contains PKI and native snapshots') options = parser.parse_args() if sys.platform != 'darwin': parser.error('This native attribution control requires macOS') out = options.output.resolve() out.mkdir(parents=True, exist_ok=False) os.chmod(out, 0o700) sdk = sorted(options.references.resolve().glob('Opc.Ua*.dll')) if not sdk: raise RuntimeError('Missing actual SDK assemblies') project = ET.Element('Project', Sdk='Microsoft.NET.Sdk') props = ET.SubElement(project, 'PropertyGroup') for k, v in {'OutputType': 'Exe', 'TargetFramework': 'net10.0', 'ImplicitUsings': 'enable', 'Nullable': 'enable', 'EnableDefaultCompileItems': 'false', 'TreatWarningsAsErrors': 'true'}.items(): ET.SubElement(props, k).text = v items = ET.SubElement(project, 'ItemGroup') ET.SubElement(items, 'FrameworkReference', Include='Microsoft.AspNetCore.App') ET.SubElement(items, 'Compile', Include=str(HERE / 'ReloadControl.cs')) for dll in sdk: ref = ET.SubElement(items, 'Reference', Include=dll.stem) ET.SubElement(ref, 'HintPath').text = str(dll) ET.indent(project) ET.ElementTree(project).write(out / 'Control.csproj', encoding='unicode') sha = lambda p: hashlib.sha256(p.read_bytes()).hexdigest() provenance = {'controlSourceSha256': sha(HERE / 'ReloadControl.cs'), 'runnerSourceSha256': sha(HERE / 'run.py'), 'repetitionsPerArm': 200, 'stages': STAGES, 'idleSeconds': 5, 'fullSoakGate': False, 'leakTracking': True, 'trackingOverhead': 'Both A/B processes capture allocation stacks; additional allocation and GC overhead is included in observed memory.', 'nativeReferences': {p.name: sha(p) for p in sdk}, 'hostDotnet': subprocess.check_output(['dotnet', '--info'], text=True), 'os': subprocess.check_output(['sw_vers'], text=True), 'architecture': os.uname().machine} save(out / 'provenance.json', provenance) with (out / 'build.log').open('w') as log: subprocess.run(['dotnet', 'build', str(out / 'Control.csproj'), '-c', 'Release', '-m:1', '--nologo'], stdout=log, stderr=subprocess.STDOUT, timeout=120, check=True) app = out / 'bin/Release/net10.0/Control.dll' seed = out / 'seed' seed.mkdir() (seed / 'tmp').mkdir() (seed / 'data').mkdir() results = {} try: with (seed / 'seed.jsonl').open('w') as log: subprocess.run(['dotnet', str(app), 'seed', str(seed)], stdout=log, stderr=subprocess.STDOUT, timeout=60, check=True, env=dict(os.environ, TMPDIR=str(seed / 'tmp'), XDG_DATA_HOME=str(seed / 'data'))) seed_rows = [json.loads(x) for x in (seed / 'seed.jsonl').read_text().splitlines()] assert len(seed_rows) == 1 and seed_rows[0]['kind'] == 'seed' provenance['cryptographyRuntime'] = seed_rows[0]['cryptographyRuntime'] save(out / 'provenance.json', provenance) for mode in ['store', 'registry']: root = out / mode root.mkdir() shutil.copytree(seed / 'pki', root / 'pki') results[mode] = arm(app, root, mode) assert results['store']['complete']['sameLeafSha256'] == results['registry']['complete']['sameLeafSha256'] == seed_rows[0]['leafSha256'] save(out / 'result.json', {'status': 'passed', 'seed': seed_rows[0], 'arms': results, 'fullSoakGate': False}) finally: for root in [seed, out / 'store', out / 'registry']: for name in ['pki', 'tmp', 'data']: if (root / name).exists(): shutil.rmtree(root / name) save(out / 'fixture-cleanup.json', {'allOwnedFixtureDirectoriesRemoved': all(not (p / name).exists() for p in [seed, out / 'store', out / 'registry'] for name in ['pki', 'tmp', 'data'])}) print(out) if __name__ == '__main__': main()