gps-denied-onboard/e2e/runner/helpers/cve_probe_evaluator.py

"""OpenCV CVE-2025-53644 probe evaluator for NFT-SEC-04 (AZ-439 / RESTRICT-CVE-1).

The probe feeds the canonical ``cve-jpeg-fixture`` (a single crafted
JPEG known to trigger the upstream OpenCV vulnerability) into the SUT's
nav-camera input and asserts:

* AC-1a: the SUT process remains **alive** after the frame (no crash,
  no abort, no segfault — the FDR archive contains AT LEAST one record
  whose monotonic_ms strictly post-dates the probe);
* AC-1b: the FDR contains a deterministic outcome for the probe frame
  — either a ``decode-success`` record (the patched OpenCV decoded
  the crafted JPEG) OR a ``frame-decode-error`` record (the SUT
  gracefully rejected the malformed payload). A silent drop (no FDR
  record at all) is a failure, because it means the SUT swallowed the
  frame without honoring its error-reporting contract.

Public-boundary discipline: does NOT import any
``src/gps_denied_onboard`` symbol. The evaluator consumes only the
runner-collected ``FdrSurvivalRecord`` summaries the FDR-reader helper
already produces for other scenarios.
"""

from __future__ import annotations

import csv
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Sequence


class ProbeFrameOutcome(str, Enum):
    """The deterministic per-frame outcome the SUT must record."""

    DECODE_SUCCESS = "decode-success"
    FRAME_DECODE_ERROR = "frame-decode-error"
    MISSING = "missing"  # silent drop — fails AC-1b


@dataclass(frozen=True)
class FdrSurvivalRecord:
    """One FDR record from the runner-collected archive."""

    monotonic_ms: int
    kind: str  # e.g. "frame-decode-success" or "frame-decode-error"


@dataclass(frozen=True)
class CveProbeReport:
    """Verdict for one ``cve-jpeg-fixture`` injection."""

    probe_injected_at_ms: int
    last_fdr_record_at_ms: int | None
    probe_outcome: ProbeFrameOutcome

    @property
    def passes_no_crash(self) -> bool:
        return (
            self.last_fdr_record_at_ms is not None
            and self.last_fdr_record_at_ms >= self.probe_injected_at_ms
        )

    @property
    def passes_graceful_outcome(self) -> bool:
        return self.probe_outcome in (
            ProbeFrameOutcome.DECODE_SUCCESS,
            ProbeFrameOutcome.FRAME_DECODE_ERROR,
        )

    @property
    def passes(self) -> bool:
        return self.passes_no_crash and self.passes_graceful_outcome


def classify_probe_outcome(
    fdr_records: Sequence[FdrSurvivalRecord],
    *,
    probe_injected_at_ms: int,
    tolerance_ms: int = 50,
) -> ProbeFrameOutcome:
    """Pick the FDR record nearest the probe injection and classify it.

    A record is considered ``for the probe`` if its monotonic timestamp
    lies within ``[probe_injected_at_ms, probe_injected_at_ms + tolerance_ms]``.
    If no record falls in that window the outcome is ``MISSING`` —
    which fails AC-1b regardless of the no-crash check.
    """
    for record in fdr_records:
        if record.monotonic_ms < probe_injected_at_ms:
            continue
        if record.monotonic_ms > probe_injected_at_ms + tolerance_ms:
            continue
        if "decode-success" in record.kind:
            return ProbeFrameOutcome.DECODE_SUCCESS
        if "decode-error" in record.kind:
            return ProbeFrameOutcome.FRAME_DECODE_ERROR
    return ProbeFrameOutcome.MISSING


def evaluate(
    fdr_records: Sequence[FdrSurvivalRecord],
    *,
    probe_injected_at_ms: int,
    tolerance_ms: int = 50,
) -> CveProbeReport:
    last_record_at = max((r.monotonic_ms for r in fdr_records), default=None)
    outcome = classify_probe_outcome(
        fdr_records,
        probe_injected_at_ms=probe_injected_at_ms,
        tolerance_ms=tolerance_ms,
    )
    return CveProbeReport(
        probe_injected_at_ms=probe_injected_at_ms,
        last_fdr_record_at_ms=last_record_at,
        probe_outcome=outcome,
    )


def write_csv_evidence(out_path: Path, report: CveProbeReport) -> Path:
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", newline="") as fh:
        writer = csv.writer(fh)
        writer.writerow(
            [
                "probe_injected_at_ms",
                "last_fdr_record_at_ms",
                "probe_outcome",
                "passes_no_crash",
                "passes_graceful_outcome",
                "passes",
            ]
        )
        writer.writerow(
            [
                report.probe_injected_at_ms,
                "" if report.last_fdr_record_at_ms is None else report.last_fdr_record_at_ms,
                report.probe_outcome.value,
                "true" if report.passes_no_crash else "false",
                "true" if report.passes_graceful_outcome else "false",
                "true" if report.passes else "false",
            ]
        )
    return out_path