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