#!/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())