harden tigrcorn client session coverage t2

Author: cobycloud

.ssot/registry.json

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import pytest
+
+from tests.support.client_session_matrix import ClientSessionTopologyHarness
+from tigrcorn_protocols.client_session_coverage import (
+    ClientTopology,
+    CoverageDisposition,
+    ProtocolCarrier,
+    build_matrix_row,
+    validate_governed_identifiers,
+)
+
+
+def test_governed_records_reject_internal_lane_field() -> None:
+    with pytest.raises(ValueError, match="lane"):
+        validate_governed_identifiers(
+            {
+                "client_id": "client-a",
+                "connection_id": "wt-conn",
+                "session_id": "wt-session",
+                "lane": "bidi_stream",
+            }
+        )
+
+
+def test_cross_client_session_access_fails_closed() -> None:
+    harness = ClientSessionTopologyHarness(ProtocolCarrier.WEBTRANSPORT_H3_QUIC)
+    topology = ClientTopology.CONCURRENT_CLIENTS
+    harness.open("client-a", "wt-conn-a", "wt-session-a", topology)
+
+    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"
+
+
+def test_post_close_send_is_rejected_and_session_cleanup_is_visible() -> None:
+    harness = ClientSessionTopologyHarness(ProtocolCarrier.WEBSOCKET_H1)
+    topology = ClientTopology.CHURN_CLIENTS
+    harness.open("client-a", "ws-conn-a", "ws-session-a", topology)
+    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
+
+
+def test_native_webtransport_message_event_is_unsupported_fault() -> None:
+    event = {"type": "webtransport.message.receive", "session_id": "wt-session"}
+
+    row = build_matrix_row(
+        protocol_carrier=ProtocolCarrier.WEBTRANSPORT_H3_QUIC,
+        client_topology=ClientTopology.SINGLE_CLIENT,
+        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-a",
+        connection_id="wt-conn",
+        session_id=event["session_id"],
+        error_kind="unsupported_native_webtransport_message",
+    )
+
+    assert event["type"].startswith("webtransport.message.")
+    assert row["disposition"] == "fail_closed"

tests/test_client_session_fault_cleanup_modes_t2.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from tigrcorn.config.model import WebSocketConfig, WebTransportConfig
+from tigrcorn.sessions.limits import SessionLimits
+from tigrcorn_protocols.client_session_coverage import (
+    ClientTopology,
+    CoverageDisposition,
+    ProtocolCarrier,
+    build_matrix_row,
+)
+
+
+def test_session_limits_bound_concurrent_stream_pressure() -> None:
+    limits = SessionLimits(max_streams=2)
+
+    assert limits.allow_stream(0) is True
+    assert limits.allow_stream(1) is True
+    assert limits.allow_stream(2) is False
+
+    row = build_matrix_row(
+        protocol_carrier=ProtocolCarrier.HTTP2,
+        client_topology=ClientTopology.CONCURRENT_CLIENTS,
+        disposition=CoverageDisposition.FAIL_CLOSED,
+        lifecycle_behavior=CoverageDisposition.COVERED,
+        identity_isolation=CoverageDisposition.COVERED,
+        ordering_behavior=CoverageDisposition.COVERED,
+        pressure_mode=CoverageDisposition.COVERED,
+        fault_mode=CoverageDisposition.REQUIRED,
+        client_id="client-a",
+        connection_id="h2-conn",
+        stream_id=3,
+        error_kind="max_streams_exceeded",
+    )
+    assert row["pressure_mode"] == "covered"
+    assert row["error_kind"] == "max_streams_exceeded"
+
+
+def test_websocket_queue_and_message_pressure_have_configured_bounds() -> None:
+    config = WebSocketConfig(max_message_size=4, max_queue=2)
+    accepted = [b"one", b"two"]
+    rejected_payload = b"three"
+
+    assert len(accepted) == config.max_queue
+    assert len(rejected_payload) > config.max_message_size
+
+    row = build_matrix_row(
+        protocol_carrier=ProtocolCarrier.WEBSOCKET_H1,
+        client_topology=ClientTopology.CONCURRENT_CLIENTS,
+        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-a",
+        connection_id="ws-conn",
+        error_kind="websocket_pressure_budget_exceeded",
+    )
+    assert row["session_scope"] == "websocket_connection_scoped"
+
+
+def test_webtransport_session_stream_and_datagram_pressure_are_bounded() -> None:
+    config = WebTransportConfig(max_sessions=1, max_streams=1, max_datagram_size=4)
+
+    assert config.max_sessions == 1
+    assert config.max_streams == 1
+    assert len(b"toolong") > config.max_datagram_size
+
+    row = build_matrix_row(
+        protocol_carrier=ProtocolCarrier.WEBTRANSPORT_H3_QUIC,
+        client_topology=ClientTopology.CONCURRENT_CLIENTS,
+        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-a",
+        connection_id="wt-conn",
+        session_id="wt-session",
+        datagram_id="d-too-large",
+        error_kind="webtransport_pressure_budget_exceeded",
+    )
+    assert row["datagram_id"] == "d-too-large"