190 lines
7.3 KiB
Python
190 lines
7.3 KiB
Python
#!/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)
|