62 lines
3.0 KiB
Python
62 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Summarize raw collector evidence without inventing qualification bounds."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
def summarize(data):
|
|
records = [json.loads(line[7:]) for line in data.splitlines() if line.startswith('TIMING ')]
|
|
if not records or records[-1] != {'kind': 'complete', 'qualified': False}:
|
|
raise ValueError('missing completed collection')
|
|
samples = [r for r in records if r['kind'] == 'sample']
|
|
if len(samples) < 2 or [r['index'] for r in samples] != list(range(len(samples))):
|
|
raise ValueError('missing or reordered samples')
|
|
metadata = {phase: {r['path']: r for r in records
|
|
if r['kind'] == 'metadata' and r['phase'] == phase}
|
|
for phase in ('start', 'end')}
|
|
for path in ('/proc/sys/kernel/osrelease', '/proc/sys/kernel/random/boot_id',
|
|
'/sys/devices/system/clocksource/clocksource0/current_clocksource'):
|
|
a, b = (metadata[p].get(path) for p in ('start', 'end'))
|
|
if not a or not b or a['errno'] or b['errno'] or not a['value'].strip() or a['value'] != b['value']:
|
|
raise ValueError('missing/changed platform identity: ' + path)
|
|
previous = 0
|
|
for s in samples:
|
|
if not previous <= s['before_ns'] <= s['after_ns'] <= s['end_ns']:
|
|
raise ValueError('BOOTTIME regression')
|
|
previous = s['end_ns']
|
|
first, last = samples[0], samples[-1]
|
|
# mono is read between after and end. Thus BOOTTIME-minus-MONOTONIC
|
|
# offset lies in [after-mono, end-mono]; subtract intervals, not midpoints.
|
|
suspend_low = (last['after_ns'] - last['mono_ns']) - (first['end_ns'] - first['mono_ns'])
|
|
suspend_high = (last['end_ns'] - last['mono_ns']) - (first['after_ns'] - first['mono_ns'])
|
|
return dict(qualified=False, samples=len(samples), metadata=metadata,
|
|
measured_max_bracket_ns=max(s['after_ns'] - s['before_ns'] for s in samples),
|
|
measured_boottime_elapsed_ns=last['before_ns'] - first['before_ns'],
|
|
suspend_offset_change_interval_ns=[suspend_low, suspend_high],
|
|
independent_reference=False, max_elapsed_error_ppb=None,
|
|
max_observation_error_ns=None,
|
|
note='Observed extrema only; correlated clocks cannot establish UTC accuracy or future bounds.')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('evidence', type=Path)
|
|
parser.add_argument('--output', type=Path, help='create a new JSON evidence summary')
|
|
args = parser.parse_args()
|
|
raw = args.evidence.read_bytes()
|
|
result = summarize(raw.decode())
|
|
result['evidence_sha256'] = hashlib.sha256(raw).hexdigest()
|
|
encoded = json.dumps(result, indent=2) + '\n'
|
|
if args.output:
|
|
with args.output.open('x') as stream:
|
|
stream.write(encoded)
|
|
print(json.dumps({k: v for k, v in result.items() if k != 'metadata'}))
|
|
else:
|
|
print(encoded, end='')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|