test: isolate native certificate reload retention on macOS
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,168 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user