58 lines
2.8 KiB
Python
58 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Offline endpoint error analysis. Does not authenticate inputs or qualify a profile.
|
|
|
|
Inputs are preserved independent-observer receipts, not recovery grants. Review
|
|
their NTS reports, process restrictions and exact platform provenance separately.
|
|
"""
|
|
import argparse
|
|
from decimal import Decimal
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
|
|
|
|
def analyze(before, after):
|
|
a, b = before['observation'], after['observation']
|
|
if not re.fullmatch('[0-9a-f]{32}', a['boot_id']) or a['boot_id'] == '0' * 32 or a['boot_id'] != b['boot_id']:
|
|
raise ValueError('different or missing boot identity')
|
|
for receipt, observation in ((before, a), (after, b)):
|
|
for field in ('boottime_ns', 'utc_ns', 'sampling_bracket_ns', 'observation_error_ns', 'collected_at_ns'):
|
|
if type(observation[field]) is not int or observation[field] < 0:
|
|
raise ValueError('invalid endpoint time/error')
|
|
if observation['collected_at_ns'] != observation['boottime_ns'] + observation['sampling_bracket_ns']:
|
|
raise ValueError('incoherent endpoint bracket')
|
|
if observation['observation_error_ns'] < observation['sampling_bracket_ns']:
|
|
raise ValueError('endpoint uncertainty excludes its bracket')
|
|
reference = Decimal(observation['assessment']['reference_time'])
|
|
started = receipt['observer_started_wall_ns']
|
|
if not reference.is_finite() or type(started) is not int or reference * 10**9 <= started:
|
|
raise ValueError('no new reference after observer start')
|
|
if a['collected_at_ns'] >= b['boottime_ns']:
|
|
raise ValueError('overlapping or reordered endpoints')
|
|
elapsed = b['boottime_ns'] - a['boottime_ns']
|
|
residual = elapsed - (b['utc_ns'] - a['utc_ns'])
|
|
endpoint_error = a['observation_error_ns'] + b['observation_error_ns']
|
|
envelope = abs(residual) + endpoint_error
|
|
return dict(boot_id=a['boot_id'], boottime_elapsed_ns=elapsed,
|
|
elapsed_error_interval_ns=[residual-endpoint_error, residual+endpoint_error],
|
|
endpoint_error_envelope_ns=envelope,
|
|
observed_interval_envelope_ppb=(envelope*10**9+elapsed-1)//elapsed,
|
|
qualified=False, max_elapsed_error_ppb=None,
|
|
limitation='Endpoint interval only; no future or intermediate-time guarantee; input authenticity is external.')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('before', type=Path)
|
|
parser.add_argument('after', type=Path)
|
|
args = parser.parse_args()
|
|
raw = [p.read_bytes() for p in (args.before, args.after)]
|
|
result = analyze(*(json.loads(data) for data in raw))
|
|
result['receipt_sha256'] = [hashlib.sha256(data).hexdigest() for data in raw]
|
|
print(json.dumps(result, indent=2))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|