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