Publish Chimera GFX source

tools/phase10v_shsrv_collector_model.py

@@ -0,0 +1,207 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-3.0-or-later
+"""Offline-only, one-shot shsrv transcript collection model.
+
+This module has no network transport. It accepts bounded bytes from stdin or
+tests, removes Telnet control traffic, and delegates sanitized classification
+to the Phase-1.0T parser. It does not prove memory erasure or device behavior.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import posixpath
+import re
+import sys
+from typing import Any
+
+from phase10t_shsrv_transcript import parse_transcript
+
+
+MAX_RAW_BYTES = 65_536
+MAX_SANITIZED_BYTES = 65_536
+MAX_CHUNKS = 256
+EXPECTED_FIRMWARE = "9.60"
+IAC = 0xFF
+SE = 0xF0
+SB = 0xFA
+WILL = 0xFB
+WONT = 0xFC
+DO = 0xFD
+DONT = 0xFE
+
+
+class CollectorError(RuntimeError):
+    """Fail-closed model error that never embeds input bytes."""
+
+
+class TelnetFilter:
+    """Incrementally remove Telnet negotiations and subnegotiations."""
+
+    def __init__(self) -> None:
+        self.state = "DATA"
+
+    def feed(self, chunk: bytes) -> bytes:
+        output = bytearray()
+        for value in chunk:
+            if self.state == "DATA":
+                if value == IAC:
+                    self.state = "IAC"
+                else:
+                    output.append(value)
+            elif self.state == "IAC":
+                if value == IAC:
+                    output.append(IAC)
+                    self.state = "DATA"
+                elif value in {WILL, WONT, DO, DONT}:
+                    self.state = "NEGOTIATION_OPTION"
+                elif value == SB:
+                    self.state = "SUBNEGOTIATION"
+                else:
+                    self.state = "DATA"
+            elif self.state == "NEGOTIATION_OPTION":
+                self.state = "DATA"
+            elif self.state == "SUBNEGOTIATION":
+                if value == IAC:
+                    self.state = "SUBNEGOTIATION_IAC"
+            elif self.state == "SUBNEGOTIATION_IAC":
+                if value == SE:
+                    self.state = "DATA"
+                elif value == IAC:
+                    self.state = "SUBNEGOTIATION"
+                else:
+                    self.state = "SUBNEGOTIATION"
+        return bytes(output)
+
+    def is_complete(self) -> bool:
+        return self.state == "DATA"
+
+
+class OfflineCollector:
+    """Bounded one-shot state model with no connection capability."""
+
+    def __init__(self) -> None:
+        self.state = "READY"
+        self.raw_bytes_received = 0
+        self.chunk_count = 0
+        self._filter = TelnetFilter()
+        self._sanitized = bytearray()
+
+    def _discard_buffer(self) -> None:
+        for index in range(len(self._sanitized)):
+            self._sanitized[index] = 0
+        self._sanitized.clear()
+
+    def _invalidate(self, message: str) -> None:
+        self._discard_buffer()
+        self.state = "INVALID"
+        raise CollectorError(message)
+
+    def feed(self, chunk: bytes) -> None:
+        if self.state not in {"READY", "RECEIVING"}:
+            raise CollectorError("collector is not accepting input")
+        if not isinstance(chunk, bytes):
+            self._invalidate("input must be bytes")
+        if not chunk:
+            return
+        self.chunk_count += 1
+        self.raw_bytes_received += len(chunk)
+        if self.chunk_count > MAX_CHUNKS:
+            self._invalidate("chunk limit exceeded")
+        if self.raw_bytes_received > MAX_RAW_BYTES:
+            self._invalidate("byte limit exceeded")
+        filtered = self._filter.feed(chunk)
+        if len(self._sanitized) + len(filtered) > MAX_SANITIZED_BYTES:
+            self._invalidate("sanitized byte limit exceeded")
+        self._sanitized.extend(filtered)
+        self.state = "RECEIVING"
+
+    def abort(self) -> None:
+        if self.state in {"SEALED", "INVALID"}:
+            raise CollectorError("collector can no longer abort")
+        self._discard_buffer()
+        self.state = "ABORTED"
+
+    @staticmethod
+    def _validate_expected_paths(expected_paths: set[str]) -> None:
+        path_pattern = re.compile(r"^/[A-Za-z0-9._/-]{1,511}$")
+        for path in expected_paths:
+            if not isinstance(path, str) or not path_pattern.fullmatch(path):
+                raise CollectorError("expected path is not a safe absolute path")
+            if posixpath.normpath(path) != path or "//" in path:
+                raise CollectorError("expected path is not normalized")
+
+    @staticmethod
+    def _validate_metadata(result: dict[str, Any]) -> None:
+        metadata = result["compile_metadata"]
+        date = metadata["date"]
+        time = metadata["time"]
+        firmware = metadata["firmware"]
+        if date is not None and not re.fullmatch(
+                r"[A-Z][a-z]{2} [ 0-3][0-9] [0-9]{4}", date):
+            raise CollectorError("compile date format is invalid")
+        if time is not None and not re.fullmatch(
+                r"(?:[01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]", time):
+            raise CollectorError("compile time format is invalid")
+        if firmware is not None and firmware != EXPECTED_FIRMWARE:
+            raise CollectorError("firmware metadata does not match the gate")
+
+    def finalize(self, expected_paths: set[str] | None = None) -> dict[str, Any]:
+        if self.state not in {"READY", "RECEIVING"}:
+            raise CollectorError("collector cannot be finalized")
+        if not self._filter.is_complete():
+            self._invalidate("incomplete Telnet control sequence")
+        try:
+            transcript = bytes(self._sanitized).decode("utf-8", errors="strict")
+        except UnicodeDecodeError:
+            self._invalidate("transcript is not valid UTF-8")
+        allowed_paths = expected_paths or set()
+        try:
+            self._validate_expected_paths(allowed_paths)
+            result = parse_transcript(transcript, allowed_paths)
+            self._validate_metadata(result)
+        except (ValueError, OverflowError, CollectorError):
+            self._invalidate("transcript validation failed")
+        self._discard_buffer()
+        self.state = "SEALED"
+        result["collector_model"] = {
+            "offline_only": True,
+            "one_shot": True,
+            "network_transport_present": False,
+            "raw_bytes_received": self.raw_bytes_received,
+            "chunk_count": self.chunk_count,
+            "raw_transcript_persisted": False,
+            "logical_buffer_discard_performed": True,
+            "physical_memory_erasure_proven": False,
+            "device_behavior_proven": False,
+        }
+        return result
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--offline-transcript", action="store_true", required=True,
+        help="Confirm that stdin is an already-supplied offline transcript")
+    parser.add_argument(
+        "--expected-path", action="append", default=[],
+        help="Literal pre-approved path whose metadata may be retained")
+    args = parser.parse_args()
+    collector = OfflineCollector()
+    try:
+        while True:
+            chunk = sys.stdin.buffer.read(4096)
+            if not chunk:
+                break
+            collector.feed(chunk)
+        result = collector.finalize(set(args.expected_path))
+    except CollectorError as error:
+        print(f"ERROR: {error}", file=sys.stderr)
+        return 1
+    print(json.dumps(result, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())