241 lines
10 KiB
Python
241 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Offline-only passive LF-batch contract for synthetic shsrv inputs.
|
|
|
|
The module has no transport, CLI, address, clock, file output or prompt
|
|
detector. It formats one bounded batch from an already validated Phase-1.0W
|
|
plan and seals supplied bytes only when the caller supplies a synthetic hard
|
|
deadline event.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from phase10v_shsrv_collector_model import CollectorError, OfflineCollector
|
|
from phase10w_shsrv_client_policy import SessionPlan, WINDOW_COMMANDS
|
|
|
|
|
|
IAC = 0xFF
|
|
MAX_LITERAL_PATH_BYTES = 512
|
|
MAX_BATCH_BYTES = 1035
|
|
KNOWN_HELP_FAMILIES = {
|
|
"OFFICIAL_V07_SOURCE_FAMILY_CANDIDATE",
|
|
"OFFICIAL_V019_SOURCE_FAMILY_CANDIDATE",
|
|
}
|
|
|
|
|
|
class PassiveContractError(RuntimeError):
|
|
"""Fail-closed contract error that never embeds supplied bytes or paths."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PassiveBatch:
|
|
"""Target-free immutable outbound bytes for one synthetic session."""
|
|
|
|
window: str
|
|
payload: bytes
|
|
expected_paths: tuple[str, ...]
|
|
command_count: int
|
|
deadline_seconds: int
|
|
outbound_batches: int = 1
|
|
lf_only: bool = True
|
|
telnet_negotiation: bool = False
|
|
retry_allowed: bool = False
|
|
reconnect_allowed: bool = False
|
|
resume_allowed: bool = False
|
|
|
|
|
|
def _validate_payload(payload: bytes) -> None:
|
|
if not payload or len(payload) > MAX_BATCH_BYTES:
|
|
raise PassiveContractError("batch length is invalid")
|
|
if any(value in payload for value in (0x00, 0x0D, IAC)):
|
|
raise PassiveContractError("batch contains a forbidden byte")
|
|
if any(value > 0x7F for value in payload):
|
|
raise PassiveContractError("batch is not ASCII")
|
|
if not payload.endswith(b"\n"):
|
|
raise PassiveContractError("batch is not LF terminated")
|
|
|
|
|
|
def _validate_batch(batch: PassiveBatch) -> None:
|
|
_validate_payload(batch.payload)
|
|
if batch.outbound_batches != 1 or batch.lf_only is not True or \
|
|
batch.telnet_negotiation is not False or \
|
|
batch.retry_allowed is not False or \
|
|
batch.reconnect_allowed is not False or batch.resume_allowed is not False:
|
|
raise PassiveContractError("batch safety fields are invalid")
|
|
if not isinstance(batch.deadline_seconds, int) or isinstance(
|
|
batch.deadline_seconds, bool) or not 1 <= batch.deadline_seconds <= 10:
|
|
raise PassiveContractError("batch deadline is invalid")
|
|
if batch.window == "T2_GREETING_AND_HELP":
|
|
if batch.payload != b"help\n" or batch.expected_paths != () or \
|
|
batch.command_count != 1:
|
|
raise PassiveContractError("help batch contract is invalid")
|
|
return
|
|
if batch.window == "T3_ONE_EXACT_PATH" and \
|
|
len(batch.expected_paths) == 1 and batch.command_count == 2:
|
|
path = batch.expected_paths[0]
|
|
try:
|
|
OfflineCollector._validate_expected_paths({path})
|
|
encoded = path.encode("ascii", errors="strict")
|
|
except (CollectorError, UnicodeEncodeError) as error:
|
|
raise PassiveContractError("path batch contract is invalid") from error
|
|
expected = b"stat " + encoded + b"\nsum " + encoded + b"\n"
|
|
if len(encoded) <= MAX_LITERAL_PATH_BYTES and batch.payload == expected:
|
|
return
|
|
raise PassiveContractError("path batch contract is invalid")
|
|
|
|
|
|
def build_passive_batch(plan: SessionPlan) -> PassiveBatch:
|
|
"""Build one target-free LF batch from an exact Phase-1.0W plan."""
|
|
if not isinstance(plan, SessionPlan):
|
|
raise PassiveContractError("session plan type is invalid")
|
|
expected_commands = WINDOW_COMMANDS.get(plan.window)
|
|
if expected_commands is None or plan.commands != expected_commands:
|
|
raise PassiveContractError("window command sequence is invalid")
|
|
if not isinstance(plan.deadline_seconds, int) or isinstance(
|
|
plan.deadline_seconds, bool) or not 1 <= plan.deadline_seconds <= 10:
|
|
raise PassiveContractError("deadline is invalid")
|
|
|
|
expected_paths: tuple[str, ...]
|
|
if plan.window == "T2_GREETING_AND_HELP":
|
|
if plan.exact_literal_path is not None:
|
|
raise PassiveContractError("help window contains a path")
|
|
payload = b"help\n"
|
|
expected_paths = ()
|
|
elif plan.window == "T3_ONE_EXACT_PATH":
|
|
path = plan.exact_literal_path
|
|
if not isinstance(path, str):
|
|
raise PassiveContractError("exact path is missing")
|
|
try:
|
|
OfflineCollector._validate_expected_paths({path})
|
|
encoded = path.encode("ascii", errors="strict")
|
|
except (CollectorError, UnicodeEncodeError) as error:
|
|
raise PassiveContractError("exact path is invalid") from error
|
|
if len(encoded) > MAX_LITERAL_PATH_BYTES:
|
|
raise PassiveContractError("exact path is too long")
|
|
payload = b"stat " + encoded + b"\nsum " + encoded + b"\n"
|
|
expected_paths = (path,)
|
|
else:
|
|
raise PassiveContractError("window is not allowlisted")
|
|
|
|
batch = PassiveBatch(
|
|
window=plan.window,
|
|
payload=payload,
|
|
expected_paths=expected_paths,
|
|
command_count=len(plan.commands),
|
|
deadline_seconds=plan.deadline_seconds,
|
|
)
|
|
_validate_batch(batch)
|
|
return batch
|
|
|
|
|
|
class PassiveResultAccumulator:
|
|
"""Consume supplied chunks once; never observes prompts or remote EOF."""
|
|
|
|
def __init__(self, batch: PassiveBatch) -> None:
|
|
if not isinstance(batch, PassiveBatch):
|
|
raise PassiveContractError("batch type is invalid")
|
|
_validate_batch(batch)
|
|
self.batch = batch
|
|
self._collector = OfflineCollector()
|
|
self.state = "READY"
|
|
|
|
def _invalidate(self, message: str) -> None:
|
|
if self._collector.state not in {"SEALED", "INVALID", "ABORTED"}:
|
|
try:
|
|
self._collector.abort()
|
|
except CollectorError:
|
|
pass
|
|
self.state = "INVALID"
|
|
raise PassiveContractError(message)
|
|
|
|
def feed_supplied_chunk(self, chunk: bytes) -> None:
|
|
if self.state not in {"READY", "RECEIVING"}:
|
|
raise PassiveContractError("accumulator is not accepting input")
|
|
if not isinstance(chunk, bytes):
|
|
self._invalidate("input must be bytes")
|
|
if IAC in chunk:
|
|
self._invalidate("unexpected Telnet control byte")
|
|
try:
|
|
self._collector.feed(chunk)
|
|
except CollectorError as error:
|
|
self._invalidate("collector rejected supplied input")
|
|
raise AssertionError("unreachable") from error
|
|
if chunk:
|
|
self.state = "RECEIVING"
|
|
|
|
def abort(self) -> None:
|
|
if self.state not in {"READY", "RECEIVING"}:
|
|
raise PassiveContractError("accumulator can no longer abort")
|
|
try:
|
|
self._collector.abort()
|
|
except CollectorError as error:
|
|
raise PassiveContractError("collector abort failed") from error
|
|
self.state = "ABORTED"
|
|
|
|
def _validate_complete_result(self, result: dict[str, Any]) -> None:
|
|
if self.batch.window == "T2_GREETING_AND_HELP":
|
|
fingerprint = result.get("command_fingerprint", {})
|
|
if fingerprint.get("source_family_match") not in KNOWN_HELP_FAMILIES:
|
|
raise PassiveContractError("help response is incomplete or unknown")
|
|
if result.get("classification") != "SOURCE_FAMILY_FINGERPRINT_ONLY":
|
|
raise PassiveContractError("help response classification is invalid")
|
|
elif self.batch.window == "T3_ONE_EXACT_PATH":
|
|
observations = result.get("file_observations")
|
|
if not isinstance(observations, list) or len(observations) != 1:
|
|
raise PassiveContractError("file response is incomplete")
|
|
observation = observations[0]
|
|
required = {
|
|
"path", "metadata_seen", "size", "weak_checksum",
|
|
"weak_checksum_algorithm", "cryptographic_checksum",
|
|
"proves_exact_binary",
|
|
}
|
|
if not required.issubset(observation) or observation.get("path") != \
|
|
self.batch.expected_paths[0]:
|
|
raise PassiveContractError("file response fields are incomplete")
|
|
if observation.get("metadata_seen") is not True or not isinstance(
|
|
observation.get("size"), int) or isinstance(
|
|
observation.get("size"), bool) or observation["size"] < 0:
|
|
raise PassiveContractError("stat response is incomplete")
|
|
if observation.get("weak_checksum_algorithm") != "BSD_ROTATE_16" or \
|
|
observation.get("cryptographic_checksum") is not False:
|
|
raise PassiveContractError("sum response is incomplete")
|
|
if result.get("classification") != "WEAK_FILE_CORRELATION_ONLY":
|
|
raise PassiveContractError("file response classification is invalid")
|
|
else:
|
|
raise PassiveContractError("batch window is invalid")
|
|
|
|
def seal_at_hard_deadline(
|
|
self, hard_deadline_reached: bool,
|
|
) -> dict[str, Any]:
|
|
"""Seal once only after an externally supplied synthetic deadline."""
|
|
if hard_deadline_reached is not True:
|
|
self._invalidate("hard deadline event is absent")
|
|
if self.state not in {"READY", "RECEIVING"}:
|
|
raise PassiveContractError("accumulator cannot be sealed")
|
|
try:
|
|
result = self._collector.finalize(set(self.batch.expected_paths))
|
|
self._validate_complete_result(result)
|
|
except (CollectorError, PassiveContractError) as error:
|
|
self.state = "INVALID"
|
|
raise PassiveContractError("deadline result is invalid") from error
|
|
self.state = "SEALED"
|
|
result["passive_batch_contract"] = {
|
|
"offline_only": True,
|
|
"one_outbound_batch": True,
|
|
"plain_lf_only": True,
|
|
"telnet_negotiation_emitted": False,
|
|
"telnet_control_received": False,
|
|
"prompt_completion_used": False,
|
|
"remote_eof_completion_used": False,
|
|
"sealed_by_synthetic_hard_deadline": True,
|
|
"source_family_selected": False,
|
|
"retry_allowed": False,
|
|
"reconnect_allowed": False,
|
|
"device_behavior_proven": False,
|
|
"exact_identity_proven": False,
|
|
}
|
|
return result
|