add client-session t2 runtime proof

Author: cobycloud

pkgs/tigrcorn-protocols/src/tigrcorn_protocols/client_session_coverage.py

@@ -190,6 +190,8 @@ class ClientSession:
     session_id: str
     closed: bool = False
     payloads: list[str] = field(default_factory=list)
+    streams_seen: set[str] = field(default_factory=set)
+    datagrams_seen: set[str] = field(default_factory=set)
 
 
 class ClientSessionTopologyHarness:
@@ -285,6 +287,186 @@ def session_for(self, client_id: str, connection_id: str, session_id: str) -> Cl
         return session
 
 
+class ClientSessionRobustnessHarness:
+    """Carrier-aware runtime harness for T2 pressure, fault, and cleanup proof."""
+
+    def __init__(
+        self,
+        carrier: ProtocolCarrier,
+        scope: SessionScope | None = None,
+        *,
+        max_sessions: int = 32,
+        max_streams: int = 32,
+        max_queue: int = 32,
+        max_message_size: int = 65536,
+        max_datagram_size: int = 1200,
+    ) -> None:
+        self.carrier = carrier
+        self.scope = scope
+        self.max_sessions = max_sessions
+        self.max_streams = max_streams
+        self.max_queue = max_queue
+        self.max_message_size = max_message_size
+        self.max_datagram_size = max_datagram_size
+        self.sessions: dict[str, ClientSession] = {}
+        self.errors: list[dict[str, Any]] = []
+
+    def open(self, client_id: str, connection_id: str, session_id: str) -> None:
+        if len([session for session in self.sessions.values() if not session.closed]) >= self.max_sessions:
+            self.fail_closed(
+                client_id,
+                connection_id,
+                session_id,
+                ClientTopology.CONCURRENT_CLIENTS,
+                "max_sessions_exceeded",
+            )
+            raise BufferError("session pressure budget exceeded")
+        if session_id in self.sessions and not self.sessions[session_id].closed:
+            self.fail_closed(
+                client_id,
+                connection_id,
+                session_id,
+                ClientTopology.CONCURRENT_CLIENTS,
+                "duplicate_session",
+            )
+            raise ValueError("duplicate session rejected")
+        self.sessions[session_id] = ClientSession(
+            client_id=client_id,
+            connection_id=connection_id,
+            session_id=session_id,
+        )
+
+    def send(
+        self,
+        client_id: str,
+        connection_id: str,
+        session_id: str,
+        topology: ClientTopology,
+        payload: object,
+        *,
+        stream_id: str | int | None = None,
+        datagram_id: str | int | None = None,
+    ) -> None:
+        session = self.session_for(client_id, connection_id, session_id, topology)
+        identifiers = {"stream_id": stream_id, "datagram_id": datagram_id}
+        if session.closed:
+            self.fail_closed(client_id, connection_id, session_id, topology, "post_close_send", **identifiers)
+            raise RuntimeError("post-close send rejected")
+        if not isinstance(payload, (str, bytes)) or not payload:
+            self.fail_closed(client_id, connection_id, session_id, topology, "malformed_payload", **identifiers)
+            raise ValueError("malformed payload rejected")
+        payload_size = len(payload)
+        if payload_size > self.max_message_size:
+            self.fail_closed(
+                client_id,
+                connection_id,
+                session_id,
+                topology,
+                "message_pressure_budget_exceeded",
+                **identifiers,
+            )
+            raise BufferError("message pressure budget exceeded")
+        if datagram_id is not None and payload_size > self.max_datagram_size:
+            self.fail_closed(
+                client_id,
+                connection_id,
+                session_id,
+                topology,
+                "datagram_pressure_budget_exceeded",
+                **identifiers,
+            )
+            raise BufferError("datagram pressure budget exceeded")
+        if len(session.payloads) >= self.max_queue:
+            self.fail_closed(
+                client_id,
+                connection_id,
+                session_id,
+                topology,
+                "queue_pressure_budget_exceeded",
+                **identifiers,
+            )
+            raise BufferError("queue pressure budget exceeded")
+        if stream_id is not None:
+            session.streams_seen.add(str(stream_id))
+            if len(session.streams_seen) > self.max_streams:
+                self.fail_closed(
+                    client_id,
+                    connection_id,
+                    session_id,
+                    topology,
+                    "max_streams_exceeded",
+                    **identifiers,
+                )
+                session.streams_seen.remove(str(stream_id))
+                raise BufferError("stream pressure budget exceeded")
+        if datagram_id is not None:
+            session.datagrams_seen.add(str(datagram_id))
+        session.payloads.append(payload if isinstance(payload, str) else payload.decode("latin1"))
+
+    def close(self, client_id: str, connection_id: str, session_id: str, topology: ClientTopology) -> None:
+        self.session_for(client_id, connection_id, session_id, topology).closed = True
+
+    def cancel(self, client_id: str, connection_id: str, session_id: str, topology: ClientTopology) -> None:
+        self.session_for(client_id, connection_id, session_id, topology).closed = True
+        self.fail_closed(client_id, connection_id, session_id, topology, "cancelled")
+
+    def timeout(self, client_id: str, connection_id: str, session_id: str, topology: ClientTopology) -> None:
+        self.session_for(client_id, connection_id, session_id, topology).closed = True
+        self.fail_closed(client_id, connection_id, session_id, topology, "timeout")
+
+    def session_for(
+        self,
+        client_id: str,
+        connection_id: str,
+        session_id: str,
+        topology: ClientTopology,
+    ) -> ClientSession:
+        try:
+            session = self.sessions[session_id]
+        except KeyError as exc:
+            self.fail_closed(client_id, connection_id, session_id, topology, "unknown_session")
+            raise KeyError("unknown session rejected") from exc
+        if session.client_id != client_id or session.connection_id != connection_id:
+            self.fail_closed(
+                client_id,
+                connection_id,
+                session_id,
+                topology,
+                "cross_client_session_access",
+            )
+            raise PermissionError("cross-client or cross-connection session access rejected")
+        return session
+
+    def fail_closed(
+        self,
+        client_id: str,
+        connection_id: str,
+        session_id: str,
+        topology: ClientTopology,
+        error_kind: str,
+        **identifiers: Any,
+    ) -> dict[str, Any]:
+        identifiers = {key: value for key, value in identifiers.items() if value is not None}
+        row = build_matrix_row(
+            protocol_carrier=self.carrier,
+            client_topology=topology,
+            session_scope=self.scope,
+            disposition=CoverageDisposition.FAIL_CLOSED,
+            lifecycle_behavior=CoverageDisposition.COVERED,
+            identity_isolation=CoverageDisposition.COVERED,
+            ordering_behavior=CoverageDisposition.COVERED,
+            pressure_mode=CoverageDisposition.COVERED,
+            fault_mode=CoverageDisposition.COVERED,
+            client_id=client_id,
+            connection_id=connection_id,
+            session_id=session_id,
+            error_kind=error_kind,
+            **identifiers,
+        )
+        self.errors.append(row)
+        return row
+
+
 def sequential_pair(carrier: ProtocolCarrier, scope: SessionScope | None = None) -> ClientSessionTopologyHarness:
     topology = ClientTopology.SEQUENTIAL_CLIENTS
     harness = ClientSessionTopologyHarness(carrier, scope)
@@ -327,6 +509,7 @@ def bounded_interleaved_pair(
     "SESSION_SCOPE_VALUES",
     "BehaviorAxis",
     "ClientSession",
+    "ClientSessionRobustnessHarness",
     "ClientSessionTopologyHarness",
     "ClientTopology",
     "CoverageDisposition",

tests/support/client_session_matrix.py

@@ -2,13 +2,15 @@
 
 from tigrcorn_protocols.client_session_coverage import (
     ClientSession,
+    ClientSessionRobustnessHarness,
     ClientSessionTopologyHarness,
     bounded_interleaved_pair,
     sequential_pair,
 )
 
 __all__ = [
     "ClientSession",
+    "ClientSessionRobustnessHarness",
     "ClientSessionTopologyHarness",
     "bounded_interleaved_pair",
     "sequential_pair",

tests/test_client_session_fault_cleanup_modes_t2.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from tests.support.client_session_matrix import ClientSessionTopologyHarness
+from tests.support.client_session_matrix import ClientSessionRobustnessHarness
 from tigrcorn_protocols.client_session_coverage import (
     ClientTopology,
     CoverageDisposition,
@@ -25,40 +25,59 @@ def test_governed_records_reject_internal_lane_field() -> None:
 
 
 def test_cross_client_session_access_fails_closed() -> None:
-    harness = ClientSessionTopologyHarness(ProtocolCarrier.WEBTRANSPORT_H3_QUIC)
+    harness = ClientSessionRobustnessHarness(ProtocolCarrier.WEBTRANSPORT_H3_QUIC)
     topology = ClientTopology.CONCURRENT_CLIENTS
-    harness.open("client-a", "wt-conn-a", "wt-session-a", topology)
+    harness.open("client-a", "wt-conn-a", "wt-session-a")
 
     with pytest.raises(PermissionError, match="cross-client"):
         harness.send("client-b", "wt-conn-a", "wt-session-a", topology, "stolen")
 
-    row = build_matrix_row(
-        protocol_carrier=ProtocolCarrier.WEBTRANSPORT_H3_QUIC,
-        client_topology=topology,
-        disposition=CoverageDisposition.FAIL_CLOSED,
-        lifecycle_behavior=CoverageDisposition.COVERED,
-        identity_isolation=CoverageDisposition.COVERED,
-        ordering_behavior=CoverageDisposition.REQUIRED,
-        pressure_mode=CoverageDisposition.REQUIRED,
-        fault_mode=CoverageDisposition.COVERED,
-        client_id="client-b",
-        connection_id="wt-conn-a",
-        session_id="wt-session-a",
-        error_kind="cross_client_session_access",
-    )
-    assert row["fault_mode"] == "covered"
+    assert harness.errors[-1]["fault_mode"] == "covered"
+    assert harness.errors[-1]["error_kind"] == "cross_client_session_access"
 
 
 def test_post_close_send_is_rejected_and_session_cleanup_is_visible() -> None:
-    harness = ClientSessionTopologyHarness(ProtocolCarrier.WEBSOCKET_H1)
+    harness = ClientSessionRobustnessHarness(ProtocolCarrier.WEBSOCKET_H1)
     topology = ClientTopology.CHURN_CLIENTS
-    harness.open("client-a", "ws-conn-a", "ws-session-a", topology)
+    harness.open("client-a", "ws-conn-a", "ws-session-a")
     harness.close("client-a", "ws-conn-a", "ws-session-a", topology)
 
     with pytest.raises(RuntimeError, match="post-close"):
         harness.send("client-a", "ws-conn-a", "ws-session-a", topology, "late")
 
     assert harness.sessions["ws-session-a"].closed is True
+    assert harness.errors[-1]["error_kind"] == "post_close_send"
+
+
+def test_timeout_cancel_and_unknown_session_cleanup_fail_closed() -> None:
+    harness = ClientSessionRobustnessHarness(ProtocolCarrier.HTTP3_QUIC)
+    topology = ClientTopology.CHURN_CLIENTS
+    harness.open("client-a", "h3-conn-a", "h3-session-a")
+    harness.open("client-b", "h3-conn-b", "h3-session-b")
+
+    harness.timeout("client-a", "h3-conn-a", "h3-session-a", topology)
+    harness.cancel("client-b", "h3-conn-b", "h3-session-b", topology)
+    with pytest.raises(KeyError, match="unknown session"):
+        harness.send("client-c", "h3-conn-c", "missing-session", topology, "late")
+
+    assert [error["error_kind"] for error in harness.errors] == [
+        "timeout",
+        "cancelled",
+        "unknown_session",
+    ]
+    assert harness.sessions["h3-session-a"].closed is True
+    assert harness.sessions["h3-session-b"].closed is True
+
+
+def test_malformed_payload_is_rejected_fail_closed() -> None:
+    harness = ClientSessionRobustnessHarness(ProtocolCarrier.HTTP1)
+    topology = ClientTopology.SINGLE_CLIENT
+    harness.open("client-a", "http1-conn", "request-session")
+
+    with pytest.raises(ValueError, match="malformed payload"):
+        harness.send("client-a", "http1-conn", "request-session", topology, {})
+
+    assert harness.errors[-1]["error_kind"] == "malformed_payload"
 
 
 def test_native_webtransport_message_event_is_unsupported_fault() -> None: