Joseph Doherty
169 lines
9.0 KiB
Python
169 lines
9.0 KiB
Python
#!/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()
|