180 lines
7.0 KiB
Python
180 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Target-free launch-context A/B canary contract.
|
|
|
|
This module validates immutable synthetic data only. It has no target source,
|
|
artifact, socket, process, clock, filesystem output or device capability.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import re
|
|
|
|
|
|
PHASE = "PHASE_1_0AV_TARGET_FREE_LAUNCH_CONTEXT_CANARY"
|
|
FIRMWARE = "9.60"
|
|
PROTOCOL_MAGIC = "CHD10AV1"
|
|
RAW_ELFLDR = "RAW_ELFLDR"
|
|
BIGAPP_CANDIDATE = "BIGAPP_CANDIDATE"
|
|
ARM_KINDS = (RAW_ELFLDR, BIGAPP_CANDIDATE)
|
|
SHA256 = re.compile(r"^[0-9a-f]{64}$")
|
|
RUN_ID = re.compile(r"^CHIMERA_AV_[A-Z0-9_-]{8,48}$")
|
|
RETURN_MIN = -(1 << 31)
|
|
RETURN_MAX = (1 << 31) - 1
|
|
|
|
|
|
class CanaryContractError(ValueError):
|
|
"""The offline canary contract is incomplete, ambiguous or unsafe."""
|
|
|
|
|
|
def _exact_hash(value: str) -> bool:
|
|
return isinstance(value, str) and SHA256.fullmatch(value) is not None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CanaryArm:
|
|
kind: str
|
|
launcher_sha256: str
|
|
run_id: str
|
|
approval_sha256: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CanaryPairPlan:
|
|
phase: str
|
|
firmware: str
|
|
protocol_magic: str
|
|
payload_sha256: str
|
|
arms: tuple[CanaryArm, CanaryArm]
|
|
one_shot_each: bool
|
|
automatic_retry: bool
|
|
reconnect: bool
|
|
resume: bool
|
|
installation: bool
|
|
autoload: bool
|
|
device_write_authorized: bool
|
|
app_termination_authorized: bool
|
|
result_reception_authorized: bool
|
|
activation_authorized: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CanaryObservation:
|
|
kind: str
|
|
launcher_sha256: str
|
|
payload_sha256: str
|
|
run_id: str
|
|
protocol_magic: str
|
|
d04_seen: bool
|
|
d04_result: int
|
|
d04_sequence: int
|
|
submit_seen: bool
|
|
submit_result: int
|
|
submit_errno: int
|
|
submit_sequence: int
|
|
terminal_seen: bool
|
|
terminal_sequence: int
|
|
cleanup_complete: bool
|
|
visible_output_observed: bool
|
|
retry_count: int
|
|
persistent_write_count: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CanaryClassification:
|
|
status: str
|
|
pair_comparable: bool
|
|
submit_return_differs: bool
|
|
launch_context_candidate: bool
|
|
root_cause_proven: bool = False
|
|
visible_output_proven: bool = False
|
|
firmware_behavior_proven: bool = False
|
|
device_action_authorized: bool = False
|
|
|
|
|
|
def validate_plan(plan: CanaryPairPlan) -> None:
|
|
"""Validate a hypothetical pair without activating either arm."""
|
|
if type(plan) is not CanaryPairPlan or plan.phase != PHASE \
|
|
or plan.firmware != FIRMWARE or plan.protocol_magic != PROTOCOL_MAGIC:
|
|
raise CanaryContractError("pair identity is invalid")
|
|
if not _exact_hash(plan.payload_sha256) or type(plan.arms) is not tuple \
|
|
or len(plan.arms) != 2 \
|
|
or any(type(arm) is not CanaryArm for arm in plan.arms):
|
|
raise CanaryContractError("pair binding is invalid")
|
|
if tuple(arm.kind for arm in plan.arms) != ARM_KINDS:
|
|
raise CanaryContractError("arm order or kind is invalid")
|
|
for arm in plan.arms:
|
|
if not _exact_hash(arm.launcher_sha256) \
|
|
or not _exact_hash(arm.approval_sha256) \
|
|
or not isinstance(arm.run_id, str) \
|
|
or RUN_ID.fullmatch(arm.run_id) is None:
|
|
raise CanaryContractError("arm identity is invalid")
|
|
if plan.arms[0].launcher_sha256 == plan.arms[1].launcher_sha256 \
|
|
or plan.arms[0].run_id == plan.arms[1].run_id \
|
|
or plan.arms[0].approval_sha256 == plan.arms[1].approval_sha256:
|
|
raise CanaryContractError("arms are not independently authorized")
|
|
if plan.one_shot_each is not True or plan.automatic_retry is not False \
|
|
or plan.reconnect is not False or plan.resume is not False \
|
|
or plan.installation is not False or plan.autoload is not False:
|
|
raise CanaryContractError("pair one-shot policy is invalid")
|
|
if plan.device_write_authorized or plan.app_termination_authorized \
|
|
or plan.result_reception_authorized or plan.activation_authorized:
|
|
raise CanaryContractError("offline plan grants device authority")
|
|
|
|
|
|
def _validate_observation(plan: CanaryPairPlan, arm: CanaryArm,
|
|
observation: CanaryObservation) -> bool:
|
|
if type(observation) is not CanaryObservation \
|
|
or observation.kind != arm.kind \
|
|
or observation.launcher_sha256 != arm.launcher_sha256 \
|
|
or observation.payload_sha256 != plan.payload_sha256 \
|
|
or observation.run_id != arm.run_id \
|
|
or observation.protocol_magic != PROTOCOL_MAGIC:
|
|
raise CanaryContractError("observation identity differs")
|
|
booleans = (observation.d04_seen, observation.submit_seen,
|
|
observation.terminal_seen, observation.cleanup_complete,
|
|
observation.visible_output_observed)
|
|
if any(type(value) is not bool for value in booleans):
|
|
raise CanaryContractError("observation boolean field is invalid")
|
|
numeric = (observation.d04_result, observation.submit_result,
|
|
observation.submit_errno)
|
|
if any(type(value) is not int or not RETURN_MIN <= value <= RETURN_MAX
|
|
for value in numeric):
|
|
raise CanaryContractError("observation numeric field is invalid")
|
|
counts = (observation.submit_sequence, observation.d04_sequence,
|
|
observation.terminal_sequence,
|
|
observation.retry_count, observation.persistent_write_count)
|
|
if any(type(value) is not int or value < 0 for value in counts):
|
|
raise CanaryContractError("observation count is invalid")
|
|
if observation.retry_count != 0 or observation.persistent_write_count != 0:
|
|
raise CanaryContractError("observation reports forbidden effects")
|
|
if observation.visible_output_observed:
|
|
raise CanaryContractError("unscoped visible-output claim is forbidden")
|
|
complete = observation.d04_seen and observation.submit_seen \
|
|
and observation.terminal_seen and observation.cleanup_complete \
|
|
and observation.submit_sequence < observation.d04_sequence \
|
|
and observation.d04_sequence < observation.terminal_sequence
|
|
return complete
|
|
|
|
|
|
def classify_pair(plan: CanaryPairPlan,
|
|
observations: tuple[CanaryObservation, CanaryObservation]
|
|
) -> CanaryClassification:
|
|
"""Classify supplied synthetic results without interpreting VideoOut ABI."""
|
|
validate_plan(plan)
|
|
if type(observations) is not tuple or len(observations) != 2:
|
|
raise CanaryContractError("observation pair is invalid")
|
|
complete = tuple(_validate_observation(plan, arm, observation)
|
|
for arm, observation in zip(plan.arms, observations))
|
|
if not all(complete):
|
|
return CanaryClassification("INCOMPLETE_NO_CAUSAL_COMPARISON",
|
|
False, False, False)
|
|
differs = observations[0].submit_result != observations[1].submit_result
|
|
if not differs:
|
|
return CanaryClassification("NO_SUBMIT_RETURN_DIFFERENCE",
|
|
True, False, False)
|
|
return CanaryClassification(
|
|
"LAUNCH_CONTEXT_SUBMIT_RETURN_DIFFERENCE_CANDIDATE_ONLY",
|
|
True, True, True)
|