#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Host-only CHD10AV1 frame and cleanup-terminal reference model.""" from __future__ import annotations from dataclasses import dataclass import struct import zlib MAGIC = b"CHD10AV1" VERSION = 1 FRAME_SIZE = 64 D04 = 4 D07 = 7 D12 = 12 D14 = 30 KIND_RAW = 1 KIND_PAIR = 2 RAW0_VALID = 1 << 0 RAW1_VALID = 1 << 1 TERMINAL = 1 << 2 ALLOWED_FLAGS = RAW0_VALID | RAW1_VALID | TERMINAL S15_COMPLETE = 15 UINT32_MAX = (1 << 32) - 1 INT32_MIN = -(1 << 31) INT32_MAX = (1 << 31) - 1 class CanaryProtocolError(ValueError): """A synthetic AV frame or cleanup state is invalid.""" @dataclass(frozen=True) class Frame: sequence: int stage: int kind: int flags: int raw0: int raw1: int result: int aux0: int aux1: int @dataclass(frozen=True) class CleanupSnapshot: rarch_main_returned: bool d04_emitted: bool phase: int initialized_mask: int cleaned_mask: int cleanup_order_errors: int cleanup_failure_count: int rarch_main_result: int @dataclass(frozen=True) class TraceResult: complete: bool submit_result: int sdl_result: int rarch_main_result: int cleaned_mask: int visible_output_proven: bool = False firmware_behavior_proven: bool = False device_action_authorized: bool = False def _u32(value: int, label: str) -> int: if type(value) is not int or not 0 <= value <= UINT32_MAX: raise CanaryProtocolError(f"{label} is not uint32") return value def _i32(value: int, label: str) -> int: if type(value) is not int or not INT32_MIN <= value <= INT32_MAX: raise CanaryProtocolError(f"{label} is not int32") return value def encode_frame(frame: Frame) -> bytes: """Encode one exact frame; AV has exactly one possible terminal stage.""" if type(frame) is not Frame: raise CanaryProtocolError("frame type is invalid") _u32(frame.sequence, "sequence") if frame.sequence == 0 or type(frame.stage) is not int \ or not 0 <= frame.stage <= D14 or frame.kind not in {KIND_RAW, KIND_PAIR}: raise CanaryProtocolError("frame identity is invalid") if type(frame.flags) is not int or frame.flags & ~ALLOWED_FLAGS: raise CanaryProtocolError("frame flags are invalid") if frame.stage == D14: if frame.kind != KIND_PAIR or frame.flags != (RAW0_VALID | RAW1_VALID | TERMINAL): raise CanaryProtocolError("D14 is not the exact terminal frame") elif frame.flags & TERMINAL: raise CanaryProtocolError("only D14 may be terminal") for value, label in ((frame.raw0, "raw0"), (frame.raw1, "raw1"), (frame.result, "result")): _i32(value, label) _u32(frame.aux0, "aux0") _u32(frame.aux1, "aux1") output = bytearray(FRAME_SIZE) output[:8] = MAGIC struct.pack_into(">HHIBBHiiiII", output, 8, VERSION, FRAME_SIZE, frame.sequence, frame.stage, frame.kind, frame.flags, frame.raw0, frame.raw1, frame.result, frame.aux0, frame.aux1) struct.pack_into(">I", output, 60, zlib.crc32(output[:60]) & UINT32_MAX) return bytes(output) def parse_frame(raw: bytes) -> Frame: if type(raw) is not bytes or len(raw) != FRAME_SIZE or raw[:8] != MAGIC: raise CanaryProtocolError("frame envelope is invalid") if any(raw[40:60]): raise CanaryProtocolError("reserved frame bytes are nonzero") expected = zlib.crc32(raw[:60]) & UINT32_MAX if struct.unpack_from(">I", raw, 60)[0] != expected: raise CanaryProtocolError("frame CRC differs") version, size, sequence, stage, kind, flags, raw0, raw1, result, aux0, aux1 = \ struct.unpack_from(">HHIBBHiiiII", raw, 8) if version != VERSION or size != FRAME_SIZE: raise CanaryProtocolError("frame version or size differs") frame = Frame(sequence, stage, kind, flags, raw0, raw1, result, aux0, aux1) if encode_frame(frame) != raw: raise CanaryProtocolError("frame is not canonical") return frame def build_cleanup_terminal(sequence: int, snapshot: CleanupSnapshot) -> bytes | None: """Return D14 only when the modeled lifecycle is completely closed.""" if type(snapshot) is not CleanupSnapshot: raise CanaryProtocolError("cleanup snapshot type is invalid") booleans = (snapshot.rarch_main_returned, snapshot.d04_emitted) if any(type(value) is not bool for value in booleans): raise CanaryProtocolError("cleanup booleans are invalid") for value, label in ((snapshot.phase, "phase"), (snapshot.initialized_mask, "initialized mask"), (snapshot.cleaned_mask, "cleaned mask"), (snapshot.cleanup_order_errors, "cleanup order errors"), (snapshot.cleanup_failure_count, "cleanup failure count")): _u32(value, label) _i32(snapshot.rarch_main_result, "rarch_main result") complete = snapshot.rarch_main_returned and snapshot.d04_emitted \ and snapshot.phase == S15_COMPLETE and snapshot.initialized_mask == 0 \ and snapshot.cleanup_order_errors == 0 \ and snapshot.cleanup_failure_count == 0 if not complete: return None return encode_frame(Frame(sequence, D14, KIND_PAIR, RAW0_VALID | RAW1_VALID | TERMINAL, snapshot.initialized_mask, snapshot.cleaned_mask, snapshot.rarch_main_result, snapshot.cleanup_order_errors, snapshot.cleanup_failure_count)) def validate_trace(raw_frames: tuple[bytes, ...]) -> TraceResult: """Require one ordered D07/D04/D14 path and no bytes after terminal.""" if type(raw_frames) is not tuple or not raw_frames: raise CanaryProtocolError("trace is empty or not immutable") frames = tuple(parse_frame(raw) for raw in raw_frames) if any(right.sequence <= left.sequence for left, right in zip(frames, frames[1:])): raise CanaryProtocolError("frame sequence is not strictly increasing") terminal_indexes = [index for index, frame in enumerate(frames) if frame.flags & TERMINAL] if terminal_indexes != [len(frames) - 1]: raise CanaryProtocolError("terminal is absent, duplicated or not final") for stage in (D07, D04, D14): if sum(frame.stage == stage for frame in frames) != 1: raise CanaryProtocolError("required stage is absent or duplicated") if sum(frame.stage == D12 for frame in frames) > 1: raise CanaryProtocolError("D12 is duplicated") positions = {frame.stage: index for index, frame in enumerate(frames) if frame.stage in {D07, D04, D14}} if not positions[D07] < positions[D04] < positions[D14]: raise CanaryProtocolError("submit/D04/D14 order differs") terminal = frames[-1] if terminal.raw0 != 0 or terminal.aux0 != 0 or terminal.aux1 != 0: raise CanaryProtocolError("D14 cleanup predicate differs") submit = frames[positions[D07]] sdl = frames[positions[D04]] if any(frame.kind != KIND_RAW or frame.flags != RAW0_VALID for frame in (submit, sdl)): raise CanaryProtocolError("D07 or D04 semantics differ") return TraceResult(True, submit.raw0, sdl.raw0, terminal.result, terminal.raw1)