Harden protocol support T2 claims

Author: cobycloud

.ssot/evidence/webtransport-h3-draft13-initial-flow-settings/proof-graph-t2.json

@@ -1 +1 @@
-{"claim_ids":["clm:webtransport-h3-draft13-initial-flow-settings.t0","clm:webtransport-h3-draft13-initial-flow-settings.t1","clm:webtransport-h3-draft13-initial-flow-settings.t2"],"feature_id":"feat:webtransport-h3-draft13-initial-flow-settings","schema_version":"ssot.evidence.runtime.v1","status":"passed","target_tier":"T2","test_ids":["tst:pytest.webtransport-h3-draft13-initial-flow-settings.t0.proof-graph","tst:pytest.webtransport-h3-draft13-initial-flow-settings.t1.proof-graph","tst:pytest.webtransport-h3-draft13-initial-flow-settings.t2.proof-graph","tst:webtransport-h3-initial-flow-settings-gap","tst:protocol-support-matrix-drift"]}
+{"claim_ids":["clm:webtransport-h3-draft13-initial-flow-settings.t0","clm:webtransport-h3-draft13-initial-flow-settings.t1","clm:webtransport-h3-draft13-initial-flow-settings.t2"],"feature_id":"feat:webtransport-h3-draft13-initial-flow-settings","schema_version":"ssot.evidence.runtime.v1","status":"passed","target_tier":"T2","test_ids":["tst:pytest.webtransport-h3-draft13-initial-flow-settings.t0.proof-graph","tst:pytest.webtransport-h3-draft13-initial-flow-settings.t1.proof-graph","tst:pytest.webtransport-h3-draft13-initial-flow-settings.t2.proof-graph","tst:webtransport-h3-initial-flow-settings-gap","tst:protocol-support-matrix-drift","tst:webtransport-t2-runtime-robustness"]}

.ssot/registry.json

@@ -276,6 +276,7 @@ class WebTransportSession:
     path: str
     selected_protocol: str | None = None
     flow: WebTransportFlowController = field(default_factory=WebTransportFlowController)
+    flow_control_enabled: bool = True
     state: SessionState = SessionState.OPEN
     streams: dict[int, StreamDirection] = field(default_factory=dict)
     datagrams: list[bytes] = field(default_factory=list)
@@ -341,22 +342,25 @@ def accept(self, request: ConnectRequest) -> ConnectDecision:
             path=_required_header(request.headers, ":path"),
             selected_protocol=request.selected_protocol,
             flow=_flow_from_request(request),
+            flow_control_enabled=_flow_control_enabled(request),
         )
         self.sessions[session.session_id] = session
         return decision
 
     def open_stream(self, session_id: str, stream_id: int, direction: StreamDirection) -> None:
         session = self._session(session_id)
         session.ensure_accepting_traffic()
-        session.flow.open_stream(direction)
+        if session.flow_control_enabled:
+            session.flow.open_stream(direction)
         session.streams[stream_id] = direction
 
     def receive_stream_data(self, session_id: str, stream_id: int, data: bytes, direction: StreamDirection) -> None:
         session = self._session(session_id)
         session.ensure_accepting_traffic()
         if stream_id not in session.streams:
             self.open_stream(session_id, stream_id, direction)
-        session.flow.allow_stream_data(stream_id, direction, len(data))
+        if session.flow_control_enabled:
+            session.flow.allow_stream_data(stream_id, direction, len(data))
 
     def receive_datagram(self, session_id: str, payload: bytes) -> None:
         session = self._session(session_id)
@@ -402,6 +406,8 @@ def apply_capsule(self, session_id: str, capsule: Capsule) -> dict[str, Any]:
             CAPSULE_WT_STREAMS_BLOCKED_BIDI,
             CAPSULE_WT_STREAMS_BLOCKED_UNI,
         }:
+            if not session.flow_control_enabled:
+                return {"event": "webtransport.flow-control-ignored", "capsule_type": capsule.capsule_type}
             session.flow.apply_flow_control_capsule(capsule)
             return {"event": "webtransport.flow-control", "capsule_type": capsule.capsule_type}
         if capsule.capsule_type in {CAPSULE_WT_STREAM, CAPSULE_WT_STREAM_FIN}:
@@ -734,6 +740,12 @@ def _flow_from_request(request: ConnectRequest) -> WebTransportFlowController:
     )
 
 
+def _flow_control_enabled(request: ConnectRequest) -> bool:
+    if request.carrier is Carrier.H3:
+        return int(request.negotiated_settings.get(SETTING_WT_MAX_SESSIONS, 0)) > 1
+    return True
+
+
 def _header(headers: Mapping[str, str], name: str) -> str | None:
     target = name.lower()
     for key, value in headers.items():

pkgs/tigrcorn-protocols/src/tigrcorn_protocols/webtransport/wire.py

@@ -3,6 +3,7 @@
 import asyncio
 import socket
 import unittest
+from contextlib import closing
 
 from tigrcorn.config.load import build_config
 from tigrcorn.constants import H2_PREFACE
@@ -104,6 +105,17 @@ async def _issue_h3_connect(port: int, authority: str) -> tuple[list[tuple[bytes
         sock.close()
 
 
+def _closed_tcp_port() -> int:
+    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
+        sock.bind(('127.0.0.1', 0))
+        return int(sock.getsockname()[1])
+
+
+def _relay(config):
+    config.http.connect_policy = 'relay'
+    config.read_timeout = 0.1
+
+
 class ConnectRelayIndependentLocalNegativeTests(unittest.IsolatedAsyncioTestCase):
     async def test_http2_connect_policy_deny_and_allowlist_rejection_end_stream(self) -> None:
         upstream = await asyncio.start_server(lambda r, w: None, '127.0.0.1', 0)
@@ -169,6 +181,38 @@ def allowlist(config):
             upstream.close()
             await upstream.wait_closed()
 
+    async def test_http2_connect_bad_authority_and_upstream_failure_end_stream(self) -> None:
+        server, port = await _start_server(http_versions=['2'], config_mutator=_relay)
+        try:
+            headers, body, ended = await _issue_h2_connect(port, 'bad-target')
+            self.assertIn((b':status', b'400'), headers)
+            self.assertEqual(body, b'bad connect target')
+            self.assertTrue(ended)
+
+            closed_port = _closed_tcp_port()
+            headers, body, ended = await _issue_h2_connect(port, f'127.0.0.1:{closed_port}')
+            self.assertIn((b':status', b'502'), headers)
+            self.assertEqual(body, b'bad gateway')
+            self.assertTrue(ended)
+        finally:
+            await server.close()
+
+    async def test_http3_connect_bad_authority_and_upstream_failure_end_stream(self) -> None:
+        server, port = await _start_server(http_versions=['3'], transport='udp', config_mutator=_relay)
+        try:
+            headers, body, ended = await _issue_h3_connect(port, 'bad-target')
+            self.assertIn((b':status', b'400'), headers)
+            self.assertEqual(body, b'bad connect target')
+            self.assertTrue(ended)
+
+            closed_port = _closed_tcp_port()
+            headers, body, ended = await _issue_h3_connect(port, f'127.0.0.1:{closed_port}')
+            self.assertIn((b':status', b'502'), headers)
+            self.assertEqual(body, b'bad gateway')
+            self.assertTrue(ended)
+        finally:
+            await server.close()
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_connect_relay_local_negatives.py

@@ -65,3 +65,27 @@ def test_protocol_support_matrix_named_tests_are_linked_to_evidence() -> None:
         test = tests[test_id]
         assert test["path"] == expected_path
         assert test["evidence_ids"], test_id
+
+
+def test_protocol_support_t2_claims_use_runtime_robustness_tests() -> None:
+    claims = _rows_by_id("claims")
+    tests = _rows_by_id("tests")
+    expected = {
+        "clm:http3-iana-registry-conformance.t2": (
+            "tst:pytest.http3-iana-registry-conformance.t2.proof-graph",
+            "tests/test_webtransport_h3_draft13_iana.py",
+        ),
+        "clm:webtransport-h3-draft13-initial-flow-settings.t2": (
+            "tst:pytest.webtransport-h3-draft13-initial-flow-settings.t2.proof-graph",
+            "tests/test_webtransport_t2_runtime_robustness.py",
+        ),
+        "clm:http2-connect-relay-not-general-proxy.t2": (
+            "tst:pytest.http2-connect-relay-not-general-proxy.t2.proof-graph",
+            "tests/test_connect_relay_local_negatives.py",
+        ),
+    }
+
+    for claim_id, (test_id, expected_path) in expected.items():
+        assert test_id in claims[claim_id]["test_ids"]
+        assert tests[test_id]["path"] == expected_path
+        assert tests[test_id]["status"] == "passing"

tests/test_protocol_support_matrix.py

@@ -15,13 +15,13 @@
 
 
 def _runtime() -> WebTransportWireRuntime:
-    runtime = WebTransportWireRuntime(max_sessions=1)
+    runtime = WebTransportWireRuntime(max_sessions=2)
     runtime.accept(
         ConnectRequest(
             stream_id=4,
             headers={":method": "CONNECT", ":protocol": "webtransport", ":scheme": "https", ":authority": "a", ":path": "/wt"},
             carrier=Carrier.H3,
-            negotiated_settings=h3_draft13_settings(1),
+            negotiated_settings=h3_draft13_settings(2),
         )
     )
     return runtime

tests/test_webtransport_h3_draft13_flow_control.py

@@ -1,12 +1,22 @@
 from __future__ import annotations
 
+import pytest
+
+from tigrcorn_core.errors import ProtocolError
+from tigrcorn_core.utils.bytes import encode_quic_varint
 from tigrcorn.protocols.http3.codec import (
     H3_DATAGRAM_ERROR,
     H3_SETTINGS_ERROR,
+    HTTP3ConnectionError,
+    HTTP3_RESERVED_FRAME_TYPES,
     QPACK_DECODER_STREAM_ERROR,
     SETTING_ENABLE_CONNECT_PROTOCOL,
     SETTING_H3_DATAGRAM,
+    decode_settings,
+    encode_settings,
     http3_iana_registry_snapshot,
+    is_grease_identifier,
+    is_reserved_frame_type,
 )
 from tigrcorn.webtransport.wire import (
     CAPSULE_WT_CLOSE_SESSION,
@@ -50,3 +60,38 @@ def test_http3_iana_registry_snapshot_covers_runtime_constants() -> None:
     assert snapshot["error_codes"]["H3_DATAGRAM_ERROR"] == H3_DATAGRAM_ERROR == 0x33
     assert snapshot["error_codes"]["H3_SETTINGS_ERROR"] == H3_SETTINGS_ERROR == 0x0109
     assert snapshot["error_codes"]["QPACK_DECODER_STREAM_ERROR"] == QPACK_DECODER_STREAM_ERROR == 0x0202
+
+
+def test_http3_iana_registry_snapshot_is_isolated_copy() -> None:
+    snapshot = http3_iana_registry_snapshot()
+    snapshot["settings"]["SETTINGS_H3_DATAGRAM"] = 0
+
+    assert http3_iana_registry_snapshot()["settings"]["SETTINGS_H3_DATAGRAM"] == SETTING_H3_DATAGRAM
+
+
+def test_http3_settings_reserved_and_duplicate_identifiers_fail_closed() -> None:
+    with pytest.raises(ProtocolError, match="reserved HTTP/3 setting"):
+        encode_settings({0x00: 1})
+
+    duplicate_payload = (
+        encode_quic_varint(SETTING_H3_DATAGRAM)
+        + encode_quic_varint(1)
+        + encode_quic_varint(SETTING_H3_DATAGRAM)
+        + encode_quic_varint(1)
+    )
+    with pytest.raises(HTTP3ConnectionError) as exc_info:
+        decode_settings(duplicate_payload)
+    assert exc_info.value.error_code == H3_SETTINGS_ERROR
+
+    reserved_payload = encode_quic_varint(0x02) + encode_quic_varint(1)
+    with pytest.raises(HTTP3ConnectionError) as exc_info:
+        decode_settings(reserved_payload)
+    assert exc_info.value.error_code == H3_SETTINGS_ERROR
+
+
+def test_http3_frame_registry_reserved_and_grease_detection() -> None:
+    assert {0x02, 0x06, 0x08, 0x09} <= set(HTTP3_RESERVED_FRAME_TYPES)
+    assert is_reserved_frame_type(0x02)
+    assert is_grease_identifier(0x21)
+    assert is_grease_identifier(0x40)
+    assert not is_reserved_frame_type(0x41)

tests/test_webtransport_h3_draft13_iana.py

@@ -33,9 +33,9 @@ def test_webtransport_h3_draft13_initial_flow_settings_are_canonical() -> None:
 
 
 def test_webtransport_h3_draft13_accept_applies_initial_flow_settings() -> None:
-    runtime = WebTransportWireRuntime(max_sessions=1)
+    runtime = WebTransportWireRuntime(max_sessions=2)
     settings = h3_draft13_settings(
-        1,
+        2,
         init=WebTransportInit(max_data=128, max_streams_uni=1, max_streams_bidi=2),
     )
 

tests/test_webtransport_h3_draft13_settings.py

@@ -11,11 +11,13 @@
     CAPSULE_WT_DATA_BLOCKED,
     CAPSULE_WT_DRAIN_SESSION,
     CAPSULE_WT_MAX_DATA,
+    CAPSULE_WT_MAX_STREAMS_BIDI,
     CAPSULE_WT_STREAM_DATA_BLOCKED,
     Capsule,
     Carrier,
     ConnectRequest,
     StreamDirection,
+    WebTransportInit,
     WebTransportWireError,
     WebTransportWireRuntime,
     decode_h3_datagram_payload,
@@ -98,6 +100,40 @@ def test_webtransport_t2_h3_datagram_payload_underflow_fails_closed() -> None:
         decode_h3_datagram_payload(b"")
 
 
+def test_webtransport_t2_h3_single_session_ignores_flow_control_capsules() -> None:
+    runtime = WebTransportWireRuntime(max_sessions=1)
+    _accept_h3(runtime)
+
+    assert runtime.apply_capsule("4", Capsule(CAPSULE_WT_MAX_DATA, b"\xff")) == {
+        "event": "webtransport.flow-control-ignored",
+        "capsule_type": CAPSULE_WT_MAX_DATA,
+    }
+    assert runtime.apply_capsule("4", Capsule(CAPSULE_WT_MAX_STREAMS_BIDI, encode_varints(1))) == {
+        "event": "webtransport.flow-control-ignored",
+        "capsule_type": CAPSULE_WT_MAX_STREAMS_BIDI,
+    }
+    runtime.receive_stream_data("4", 8, b"unlimited", StreamDirection.BIDI)
+
+    flow = runtime.sessions["4"].flow
+    assert flow.max_data == 0
+    assert flow.max_streams_bidi == 0
+    assert flow.data_sent == 0
+
+
+def test_webtransport_t2_h3_initial_limits_are_enforced_when_flow_control_enabled() -> None:
+    runtime = WebTransportWireRuntime(max_sessions=2)
+    settings = h3_draft13_settings(2, init=WebTransportInit(max_data=3, max_streams_bidi=1, max_streams_uni=1))
+    assert runtime.accept(_request(4, Carrier.H3, settings)).accepted is True
+
+    runtime.open_stream("4", 8, StreamDirection.BIDI)
+    with pytest.raises(WebTransportWireError, match="WT_MAX_STREAMS bidi exceeded"):
+        runtime.open_stream("4", 12, StreamDirection.BIDI)
+
+    runtime.receive_stream_data("4", 8, b"abc", StreamDirection.BIDI)
+    with pytest.raises(WebTransportWireError, match="WT_MAX_DATA exceeded"):
+        runtime.receive_stream_data("4", 8, b"d", StreamDirection.BIDI)
+
+
 def test_webtransport_t2_buffering_rejects_established_session() -> None:
     runtime = WebTransportWireRuntime(max_sessions=1)
     _accept_h3(runtime)