[h3] Adapt datagram API for RFC 9297 (fixes: #420)

Our HTTP/3 code dealing with datagrams dates back to before RFC 9297 was
published. This is reflected in the the use of the `flow_id` terminology
which no longer exists. Update the following interfaces:

- `DatagramReceived` now has a `stream_id` attribute
  instead of `flow_id`.
- `H3Connection.send_datagram` now has a `stream_id` argument
  instead of `flow_id`

Encoding and decoding to / from quarter stream ID is handled by the
connection, so the user does not need to multiply or divide by four.

README.rst

@@ -51,14 +51,16 @@ different concurrency models.
 Features
 --------
 
+- minimal TLS 1.3 implementation conforming with `RFC 8446`_
 - QUIC stack conforming with `RFC 9000`_
+  * IPv4 and IPv6 support
+  * connection migration and NAT rebinding
+  * logging TLS traffic secrets
+  * logging QUIC events in QLOG format
 - HTTP/3 stack conforming with `RFC 9114`_
-- minimal TLS 1.3 implementation conforming with `RFC 8446`_
-- IPv4 and IPv6 support
-- connection migration and NAT rebinding
-- logging TLS traffic secrets
-- logging QUIC events in QLOG format
-- HTTP/3 server push support
+  * server push support
+  * WebSocket bootstrapping conforming with `RFC 9220`_
+  * datagram support conforming with `RFC 9297`_
 
 Requirements
 ------------
@@ -132,3 +134,5 @@ License
 .. _RFC 8446: https://datatracker.ietf.org/doc/html/rfc8446
 .. _RFC 9000: https://datatracker.ietf.org/doc/html/rfc9000
 .. _RFC 9114: https://datatracker.ietf.org/doc/html/rfc9114
+.. _RFC 9220: https://datatracker.ietf.org/doc/html/rfc9220
+.. _RFC 9297: https://datatracker.ietf.org/doc/html/rfc9297

examples/http3_server.py

@@ -302,7 +302,9 @@ async def send(self, message: Dict) -> None:
                 )
             end_stream = True
         elif message["type"] == "webtransport.datagram.send":
-            self.connection.send_datagram(flow_id=self.stream_id, data=message["data"])
+            self.connection.send_datagram(
+                stream_id=self.stream_id, data=message["data"]
+            )
         elif message["type"] == "webtransport.stream.send":
             self.connection._quic.send_stream_data(
                 stream_id=message["stream"], data=message["data"]
@@ -438,7 +440,7 @@ def http_event_received(self, event: H3Event) -> None:
             handler = self._handlers[event.stream_id]
             handler.http_event_received(event)
         elif isinstance(event, DatagramReceived):
-            handler = self._handlers[event.flow_id]
+            handler = self._handlers[event.stream_id]
             handler.http_event_received(event)
         elif isinstance(event, WebTransportStreamDataReceived):
             handler = self._handlers[event.session_id]

src/aioquic/h3/connection.py

@@ -28,6 +28,7 @@
 
 
 class ErrorCode(IntEnum):
+    H3_DATAGRAM_ERROR = 0x33
     H3_NO_ERROR = 0x100
     H3_GENERAL_PROTOCOL_ERROR = 0x101
     H3_INTERNAL_ERROR = 0x102
@@ -76,7 +77,7 @@ class Setting(IntEnum):
 
     # https://datatracker.ietf.org/doc/html/rfc9220#section-5
     ENABLE_CONNECT_PROTOCOL = 0x8
-    # https://www.rfc-editor.org/rfc/rfc9297.html#section-5.1
+    # https://datatracker.ietf.org/doc/html/rfc9297#section-5.1
     H3_DATAGRAM = 0x33
     # https://datatracker.ietf.org/doc/html/draft-ietf-webtrans-http2-02#section-10.1
     ENABLE_WEBTRANSPORT = 0x2B603742
@@ -124,6 +125,10 @@ class ClosedCriticalStream(ProtocolError):
     error_code = ErrorCode.H3_CLOSED_CRITICAL_STREAM
 
 
+class DatagramError(ProtocolError):
+    error_code = ErrorCode.H3_DATAGRAM_ERROR
+
+
 class FrameUnexpected(ProtocolError):
     error_code = ErrorCode.H3_FRAME_UNEXPECTED
 
@@ -384,14 +389,17 @@ def handle_event(self, event: QuicEvent) -> List[H3Event]:
 
         return []
 
-    def send_datagram(self, flow_id: int, data: bytes) -> None:
+    def send_datagram(self, stream_id: int, data: bytes) -> None:
         """
-        Send a datagram for the specified flow.
+        Send a datagram for the specified stream.
 
-        :param flow_id: The flow ID.
+        :param stream_id: The stream ID.
         :param data: The HTTP/3 datagram payload.
         """
-        self._quic.send_datagram_frame(encode_uint_var(flow_id) + data)
+        assert (
+            stream_id % 4 == 0
+        ), "Datagrams can only be sent for client-initiated bidirectional streams"
+        self._quic.send_datagram_frame(encode_uint_var(stream_id // 4) + data)
 
     def send_push_promise(self, stream_id: int, headers: Headers) -> int:
         """
@@ -767,10 +775,12 @@ def _receive_datagram(self, data: bytes) -> List[H3Event]:
         """
         buf = Buffer(data=data)
         try:
-            flow_id = buf.pull_uint_var()
+            quarter_stream_id = buf.pull_uint_var()
         except BufferReadError:
-            raise ProtocolError("Could not parse flow ID")
-        return [DatagramReceived(data=data[buf.tell() :], flow_id=flow_id)]
+            raise DatagramError("Could not parse quarter stream ID")
+        return [
+            DatagramReceived(data=data[buf.tell() :], stream_id=quarter_stream_id * 4)
+        ]
 
     def _receive_request_or_push_data(
         self, stream: H3Stream, data: bytes, stream_ended: bool

src/aioquic/h3/events.py

@@ -40,8 +40,8 @@ class DatagramReceived(H3Event):
     data: bytes
     "The data which was received."
 
-    flow_id: int
-    "The ID of the flow the data was received for."
+    stream_id: int
+    "The ID of the stream the data was received for."
 
 
 @dataclass

tests/test_webtransport.py

@@ -282,13 +282,13 @@ def test_datagram(self):
             session_id = self._make_session(h3_client, h3_server)
 
             # send datagram
-            h3_client.send_datagram(data=b"foo", flow_id=session_id)
+            h3_client.send_datagram(data=b"foo", stream_id=session_id)
 
             # receive datagram
             events = h3_transfer(quic_client, h3_server)
             self.assertEqual(
                 events,
-                [DatagramReceived(data=b"foo", flow_id=session_id)],
+                [DatagramReceived(data=b"foo", stream_id=session_id)],
             )
 
     def test_handle_datagram_truncated(self):
@@ -301,8 +301,5 @@ def test_handle_datagram_truncated(self):
         h3_server.handle_event(DatagramFrameReceived(data=b"\xff"))
         self.assertEqual(
             quic_server.closed,
-            (
-                ErrorCode.H3_GENERAL_PROTOCOL_ERROR,
-                "Could not parse flow ID",
-            ),
+            (ErrorCode.H3_DATAGRAM_ERROR, "Could not parse quarter stream ID"),
         )