138 lines
7.5 KiB
Python
138 lines
7.5 KiB
Python
"""Launch-time source snapshots shared by the runner, archiver and release audit."""
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import re
|
|
import time
|
|
import uuid
|
|
from source_manifest import addon_manifest
|
|
|
|
MANIFEST = 'launch-source-sha256.json'
|
|
METADATA = 'launch-source.json'
|
|
PROVENANCE = 'source-provenance.json'
|
|
|
|
|
|
def require(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def strict_json(raw):
|
|
def pairs(items):
|
|
result = {}
|
|
for key, value in items:
|
|
require(key not in result, f'duplicate JSON key: {key}')
|
|
result[key] = value
|
|
return result
|
|
def constant(value):
|
|
raise ValueError(f'invalid JSON number: {value}')
|
|
return json.loads(raw, object_pairs_hook=pairs, parse_constant=constant)
|
|
|
|
|
|
def encoded(value):
|
|
return (json.dumps(value, indent=2, allow_nan=False) + '\n').encode()
|
|
|
|
|
|
def snapshot(root, command):
|
|
"""Call synchronously before Popen; never reconstruct this after a run."""
|
|
manifest = encoded(addon_manifest(root))
|
|
metadata = {'schema_version': 1, 'launch_id': str(uuid.uuid4()),
|
|
'captured_unix': time.time(), 'source_root': str(Path(root).resolve()),
|
|
'command': list(command), 'manifest_sha256': hashlib.sha256(manifest).hexdigest()}
|
|
return manifest, encoded(metadata)
|
|
|
|
|
|
def persist(session, captured):
|
|
"""Persist the pre-launch bytes once a new engine session directory exists."""
|
|
for name, raw in zip((MANIFEST, METADATA), captured):
|
|
with (Path(session) / name).open('xb') as stream:
|
|
stream.write(raw)
|
|
|
|
|
|
def bindings(captured):
|
|
metadata = strict_json(captured[1])
|
|
return {'source_provenance': 'launch_snapshot', 'launch_id': metadata['launch_id'],
|
|
'launch_source_manifest': MANIFEST, 'launch_source_metadata': METADATA,
|
|
'launch_source_sha256': metadata['manifest_sha256']}
|
|
|
|
|
|
def seal_run(root, session, captured):
|
|
"""After process cleanup, bind final sources and exact raw log to this launch."""
|
|
metadata = strict_json(captured[1])
|
|
final_sources = encoded(addon_manifest(root))
|
|
final_digest = hashlib.sha256(final_sources).hexdigest()
|
|
return {'stopped_unix': time.time(), 'console_sha256': hashlib.sha256(read_file(Path(session) / 'console.log')).hexdigest(),
|
|
'end_source_sha256': final_digest,
|
|
'source_unchanged_after_run': final_digest == metadata['manifest_sha256']}
|
|
|
|
|
|
def read_file(path):
|
|
require(path.is_file() and not path.is_symlink(), f'missing or symlinked launch evidence: {path.name}')
|
|
return path.read_bytes()
|
|
|
|
|
|
def read_launch(session, runner_name='runner.json', log_name='console.log'):
|
|
session = Path(session)
|
|
raw_manifest = read_file(session / MANIFEST)
|
|
raw_metadata = read_file(session / METADATA)
|
|
manifest, metadata = strict_json(raw_manifest), strict_json(raw_metadata)
|
|
runner = strict_json(read_file(session / runner_name))
|
|
require(isinstance(manifest, dict) and bool(manifest), 'launch manifest must be nonempty')
|
|
require(all(isinstance(path, str) and isinstance(digest, str) and re.fullmatch('[0-9a-f]{64}', digest)
|
|
for path, digest in manifest.items()), 'invalid launch source hash')
|
|
require(isinstance(metadata, dict) and set(metadata) == {'schema_version', 'launch_id', 'captured_unix', 'source_root', 'command', 'manifest_sha256'}, 'invalid launch metadata fields')
|
|
require(type(metadata['schema_version']) is int and metadata['schema_version'] == 1, 'invalid launch metadata version')
|
|
require(isinstance(metadata['launch_id'], str), 'invalid launch id')
|
|
try:
|
|
uuid.UUID(metadata['launch_id'])
|
|
except (ValueError, AttributeError) as error:
|
|
raise ValueError('invalid launch id') from error
|
|
require(isinstance(metadata['source_root'], str) and bool(metadata['source_root']), 'missing launch source root')
|
|
require(isinstance(metadata['command'], list) and bool(metadata['command']) and all(isinstance(arg, str) for arg in metadata['command']), 'invalid launch command')
|
|
captured = metadata['captured_unix']
|
|
require(type(captured) in (int, float) and math.isfinite(captured) and captured > 0, 'invalid launch capture time')
|
|
require(hashlib.sha256(raw_manifest).hexdigest() == metadata['manifest_sha256'], 'launch manifest byte hash disagrees with launch metadata')
|
|
require(isinstance(runner, dict), 'runner metadata must be an object')
|
|
for key, value in bindings((raw_manifest, raw_metadata)).items():
|
|
require(runner.get(key) == value, f'runner launch binding disagrees: {key}')
|
|
started = runner.get('started_unix')
|
|
require(type(started) in (int, float) and math.isfinite(started) and started >= captured, 'source snapshot must precede process launch')
|
|
stopped = runner.get('stopped_unix')
|
|
require(type(stopped) in (int, float) and math.isfinite(stopped) and stopped >= started, 'invalid stopped runner timestamp')
|
|
require(runner.get('source_unchanged_after_run') is True and runner.get('end_source_sha256') == metadata['manifest_sha256'], 'addon sources changed between launch and stopped-run snapshot')
|
|
require(runner.get('console_sha256') == hashlib.sha256(read_file(session / log_name)).hexdigest(), 'raw native log differs from stopped runner log hash')
|
|
return manifest, metadata, runner
|
|
|
|
|
|
def archive_source(session, root, allow_legacy=False):
|
|
"""Refuse changed launch sources; legacy provenance requires explicit opt-in."""
|
|
session = Path(session)
|
|
runner_path = session / 'runner.json'
|
|
runner = strict_json(read_file(runner_path)) if runner_path.exists() else {}
|
|
marked = isinstance(runner, dict) and any(key in runner for key in ('source_provenance', 'launch_id', 'launch_source_sha256'))
|
|
present = (session / MANIFEST).exists() or (session / METADATA).exists() or marked
|
|
current = addon_manifest(root)
|
|
if present:
|
|
manifest, metadata, runner = read_launch(session)
|
|
require(manifest == current, 'current addon sources differ from pre-launch snapshot; refusing to relabel an old run')
|
|
return read_file(session / MANIFEST), {'schema_version': 1, 'kind': 'launch_snapshot'}, True
|
|
require(allow_legacy, 'session has no pre-launch source snapshot; use --allow-legacy-archive-time-source only for historical evidence')
|
|
return encoded(current), {'schema_version': 1, 'kind': 'archive_time_only',
|
|
'qualification': 'historical_only',
|
|
'reason': 'No pre-launch snapshot; hashes describe archive-time files, not proven launch files.'}, False
|
|
|
|
|
|
def qualify_archive(archive, current):
|
|
archive = Path(archive)
|
|
provenance = strict_json(read_file(archive / PROVENANCE))
|
|
require(provenance == {'schema_version': 1, 'kind': 'launch_snapshot'}
|
|
and type(provenance.get('schema_version')) is int, 'native qualification requires launch_snapshot provenance; archive-time-only evidence is historical')
|
|
manifest, metadata, runner = read_launch(archive, 'runner.json.txt', 'console.log.txt')
|
|
require(manifest == current, 'launch source snapshot does not match current addon sources/resources')
|
|
require(read_file(archive / 'source-sha256.json') == read_file(archive / MANIFEST), 'archived source manifest differs from exact launch snapshot bytes')
|
|
require(runner.get('result') == 'smoke-completed', 'native qualification requires a completed runner session')
|
|
rounds = runner.get('rounds_required')
|
|
require(type(rounds) is int and 1 <= rounds <= 10, 'invalid completed runner round requirement')
|
|
return rounds
|