import binascii from unittest import TestCase from aioquic.buffer import encode_uint_var from aioquic.h3.connection import ( H3_ALPN, ErrorCode, FrameType, FrameUnexpected, H3Connection, StreamType, encode_frame, ) from aioquic.h3.events import DataReceived, HeadersReceived, PushPromiseReceived from aioquic.h3.exceptions import NoAvailablePushIDError from aioquic.quic.configuration import QuicConfiguration from aioquic.quic.events import StreamDataReceived from aioquic.quic.logger import QuicLogger from .test_connection import client_and_server, transfer def h3_client_and_server(): return client_and_server( client_options={"alpn_protocols": H3_ALPN}, server_options={"alpn_protocols": H3_ALPN}, ) def h3_transfer(quic_sender, h3_receiver): quic_receiver = h3_receiver._quic if hasattr(quic_sender, "stream_queue"): quic_receiver._events.extend(quic_sender.stream_queue) quic_sender.stream_queue.clear() else: transfer(quic_sender, quic_receiver) # process QUIC events http_events = [] event = quic_receiver.next_event() while event is not None: http_events.extend(h3_receiver.handle_event(event)) event = quic_receiver.next_event() return http_events class FakeQuicConnection: def __init__(self, configuration): self.closed = None self.configuration = configuration self.stream_queue = [] self._events = [] self._next_stream_bidi = 0 if configuration.is_client else 1 self._next_stream_uni = 2 if configuration.is_client else 3 self._quic_logger = QuicLogger().start_trace( is_client=configuration.is_client, odcid=b"" ) def close(self, error_code, reason_phrase): self.closed = (error_code, reason_phrase) def get_next_available_stream_id(self, is_unidirectional=False): if is_unidirectional: stream_id = self._next_stream_uni self._next_stream_uni += 4 else: stream_id = self._next_stream_bidi self._next_stream_bidi += 4 return stream_id def next_event(self): try: return self._events.pop(0) except IndexError: return None def send_stream_data(self, stream_id, data, end_stream=False): # chop up data into individual bytes for c in data: self.stream_queue.append( StreamDataReceived( data=bytes([c]), end_stream=False, stream_id=stream_id ) ) if end_stream: self.stream_queue.append( StreamDataReceived(data=b"", end_stream=end_stream, stream_id=stream_id) ) class H3ConnectionTest(TestCase): maxDiff = None def _make_request(self, h3_client, h3_server): quic_client = h3_client._quic quic_server = h3_server._quic # send request stream_id = quic_client.get_next_available_stream_id() h3_client.send_headers( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), (b"x-foo", b"client"), ], ) h3_client.send_data(stream_id=stream_id, data=b"", end_stream=True) # receive request events = h3_transfer(quic_client, h3_server) self.assertEqual( events, [ HeadersReceived( headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), (b"x-foo", b"client"), ], stream_id=stream_id, stream_ended=False, ), DataReceived(data=b"", stream_id=stream_id, stream_ended=True), ], ) # send response h3_server.send_headers( stream_id=stream_id, headers=[ (b":status", b"200"), (b"content-type", b"text/html; charset=utf-8"), (b"x-foo", b"server"), ], ) h3_server.send_data( stream_id=stream_id, data=b"hello", end_stream=True, ) # receive response events = h3_transfer(quic_server, h3_client) self.assertEqual( events, [ HeadersReceived( headers=[ (b":status", b"200"), (b"content-type", b"text/html; charset=utf-8"), (b"x-foo", b"server"), ], stream_id=stream_id, stream_ended=False, ), DataReceived( data=b"hello", stream_id=stream_id, stream_ended=True, ), ], ) def test_handle_control_frame_headers(self): """ We should not receive HEADERS on the control stream. """ quic_server = FakeQuicConnection( configuration=QuicConfiguration(is_client=False) ) h3_server = H3Connection(quic_server) h3_server.handle_event( StreamDataReceived( stream_id=2, data=encode_uint_var(StreamType.CONTROL) + encode_frame(FrameType.HEADERS, b""), end_stream=False, ) ) self.assertEqual( quic_server.closed, (ErrorCode.HTTP_FRAME_UNEXPECTED, "Invalid frame type on control stream"), ) def test_handle_control_frame_max_push_id_from_server(self): """ A client should not receive MAX_PUSH_ID on the control stream. """ quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) h3_client = H3Connection(quic_client) h3_client.handle_event( StreamDataReceived( stream_id=3, data=encode_uint_var(StreamType.CONTROL) + encode_frame(FrameType.MAX_PUSH_ID, b""), end_stream=False, ) ) self.assertEqual( quic_client.closed, (ErrorCode.HTTP_FRAME_UNEXPECTED, "Servers must not send MAX_PUSH_ID"), ) def test_handle_control_stream_duplicate(self): """ We must only receive a single control stream. """ quic_server = FakeQuicConnection( configuration=QuicConfiguration(is_client=False) ) h3_server = H3Connection(quic_server) # receive a first control stream h3_server.handle_event( StreamDataReceived( stream_id=2, data=encode_uint_var(StreamType.CONTROL), end_stream=False ) ) # receive a second control stream h3_server.handle_event( StreamDataReceived( stream_id=6, data=encode_uint_var(StreamType.CONTROL), end_stream=False ) ) self.assertEqual( quic_server.closed, ( ErrorCode.HTTP_STREAM_CREATION_ERROR, "Only one control stream is allowed", ), ) def test_handle_push_frame_wrong_frame_type(self): """ We should not received SETTINGS on a push stream. """ quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) h3_client = H3Connection(quic_client) h3_client.handle_event( StreamDataReceived( stream_id=15, data=encode_uint_var(StreamType.PUSH) + encode_uint_var(0) # push ID + encode_frame(FrameType.SETTINGS, b""), end_stream=False, ) ) self.assertEqual( quic_client.closed, (ErrorCode.HTTP_FRAME_UNEXPECTED, "Invalid frame type on push stream"), ) def test_handle_qpack_decoder_duplicate(self): """ We must only receive a single QPACK decoder stream. """ quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) h3_client = H3Connection(quic_client) # receive a first decoder stream h3_client.handle_event( StreamDataReceived( stream_id=11, data=encode_uint_var(StreamType.QPACK_DECODER), end_stream=False, ) ) # receive a second decoder stream h3_client.handle_event( StreamDataReceived( stream_id=15, data=encode_uint_var(StreamType.QPACK_DECODER), end_stream=False, ) ) self.assertEqual( quic_client.closed, ( ErrorCode.HTTP_STREAM_CREATION_ERROR, "Only one QPACK decoder stream is allowed", ), ) def test_handle_qpack_decoder_stream_error(self): """ Receiving garbage on the QPACK decoder stream triggers an exception. """ quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) h3_client = H3Connection(quic_client) h3_client.handle_event( StreamDataReceived( stream_id=11, data=encode_uint_var(StreamType.QPACK_DECODER) + b"\x00", end_stream=False, ) ) self.assertEqual( quic_client.closed, (ErrorCode.HTTP_QPACK_DECODER_STREAM_ERROR, "") ) def test_handle_qpack_encoder_duplicate(self): """ We must only receive a single QPACK encoder stream. """ quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) h3_client = H3Connection(quic_client) # receive a first encoder stream h3_client.handle_event( StreamDataReceived( stream_id=11, data=encode_uint_var(StreamType.QPACK_ENCODER), end_stream=False, ) ) # receive a second encoder stream h3_client.handle_event( StreamDataReceived( stream_id=15, data=encode_uint_var(StreamType.QPACK_ENCODER), end_stream=False, ) ) self.assertEqual( quic_client.closed, ( ErrorCode.HTTP_STREAM_CREATION_ERROR, "Only one QPACK encoder stream is allowed", ), ) def test_handle_qpack_encoder_stream_error(self): """ Receiving garbage on the QPACK encoder stream triggers an exception. """ quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) h3_client = H3Connection(quic_client) h3_client.handle_event( StreamDataReceived( stream_id=7, data=encode_uint_var(StreamType.QPACK_ENCODER) + b"\x00", end_stream=False, ) ) self.assertEqual( quic_client.closed, (ErrorCode.HTTP_QPACK_ENCODER_STREAM_ERROR, "") ) def test_handle_request_frame_bad_headers(self): """ We should not receive HEADERS which cannot be decoded. """ quic_server = FakeQuicConnection( configuration=QuicConfiguration(is_client=False) ) h3_server = H3Connection(quic_server) h3_server.handle_event( StreamDataReceived( stream_id=0, data=encode_frame(FrameType.HEADERS, b""), end_stream=False ) ) self.assertEqual( quic_server.closed, (ErrorCode.HTTP_QPACK_DECOMPRESSION_FAILED, "") ) def test_handle_request_frame_data_before_headers(self): """ We should not receive DATA before receiving headers. """ quic_server = FakeQuicConnection( configuration=QuicConfiguration(is_client=False) ) h3_server = H3Connection(quic_server) h3_server.handle_event( StreamDataReceived( stream_id=0, data=encode_frame(FrameType.DATA, b""), end_stream=False ) ) self.assertEqual( quic_server.closed, ( ErrorCode.HTTP_FRAME_UNEXPECTED, "DATA frame is not allowed in this state", ), ) def test_handle_request_frame_headers_after_trailers(self): """ We should not receive HEADERS after receiving trailers. """ quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) quic_server = FakeQuicConnection( configuration=QuicConfiguration(is_client=False) ) h3_client = H3Connection(quic_client) h3_server = H3Connection(quic_server) stream_id = quic_client.get_next_available_stream_id() h3_client.send_headers( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), ], ) h3_client.send_headers( stream_id=stream_id, headers=[(b"x-some-trailer", b"foo")], end_stream=True ) h3_transfer(quic_client, h3_server) h3_server.handle_event( StreamDataReceived( stream_id=0, data=encode_frame(FrameType.HEADERS, b""), end_stream=False ) ) self.assertEqual( quic_server.closed, ( ErrorCode.HTTP_FRAME_UNEXPECTED, "HEADERS frame is not allowed in this state", ), ) def test_handle_request_frame_push_promise_from_client(self): """ A server should not receive PUSH_PROMISE on a request stream. """ quic_server = FakeQuicConnection( configuration=QuicConfiguration(is_client=False) ) h3_server = H3Connection(quic_server) h3_server.handle_event( StreamDataReceived( stream_id=0, data=encode_frame(FrameType.PUSH_PROMISE, b""), end_stream=False, ) ) self.assertEqual( quic_server.closed, (ErrorCode.HTTP_FRAME_UNEXPECTED, "Clients must not send PUSH_PROMISE"), ) def test_handle_request_frame_wrong_frame_type(self): quic_server = FakeQuicConnection( configuration=QuicConfiguration(is_client=False) ) h3_server = H3Connection(quic_server) h3_server.handle_event( StreamDataReceived( stream_id=0, data=encode_frame(FrameType.SETTINGS, b""), end_stream=False, ) ) self.assertEqual( quic_server.closed, (ErrorCode.HTTP_FRAME_UNEXPECTED, "Invalid frame type on request stream"), ) def test_request(self): with h3_client_and_server() as (quic_client, quic_server): h3_client = H3Connection(quic_client) h3_server = H3Connection(quic_server) # make first request self._make_request(h3_client, h3_server) # make second request self._make_request(h3_client, h3_server) # make third request -> dynamic table self._make_request(h3_client, h3_server) def test_request_headers_only(self): with h3_client_and_server() as (quic_client, quic_server): h3_client = H3Connection(quic_client) h3_server = H3Connection(quic_server) # send request stream_id = quic_client.get_next_available_stream_id() h3_client.send_headers( stream_id=stream_id, headers=[ (b":method", b"HEAD"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), (b"x-foo", b"client"), ], end_stream=True, ) # receive request events = h3_transfer(quic_client, h3_server) self.assertEqual( events, [ HeadersReceived( headers=[ (b":method", b"HEAD"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), (b"x-foo", b"client"), ], stream_id=stream_id, stream_ended=True, ) ], ) # send response h3_server.send_headers( stream_id=stream_id, headers=[ (b":status", b"200"), (b"content-type", b"text/html; charset=utf-8"), (b"x-foo", b"server"), ], end_stream=True, ) # receive response events = h3_transfer(quic_server, h3_client) self.assertEqual( events, [ HeadersReceived( headers=[ (b":status", b"200"), (b"content-type", b"text/html; charset=utf-8"), (b"x-foo", b"server"), ], stream_id=stream_id, stream_ended=True, ) ], ) def test_request_fragmented_frame(self): quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) quic_server = FakeQuicConnection( configuration=QuicConfiguration(is_client=False) ) h3_client = H3Connection(quic_client) h3_server = H3Connection(quic_server) # send request stream_id = quic_client.get_next_available_stream_id() h3_client.send_headers( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), (b"x-foo", b"client"), ], ) h3_client.send_data(stream_id=stream_id, data=b"hello", end_stream=True) # receive request events = h3_transfer(quic_client, h3_server) self.assertEqual( events, [ HeadersReceived( headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), (b"x-foo", b"client"), ], stream_id=stream_id, stream_ended=False, ), DataReceived(data=b"h", stream_id=0, stream_ended=False), DataReceived(data=b"e", stream_id=0, stream_ended=False), DataReceived(data=b"l", stream_id=0, stream_ended=False), DataReceived(data=b"l", stream_id=0, stream_ended=False), DataReceived(data=b"o", stream_id=0, stream_ended=False), DataReceived(data=b"", stream_id=0, stream_ended=True), ], ) # send push promise push_stream_id = h3_server.send_push_promise( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/app.txt"), ], ) self.assertEqual(push_stream_id, 15) # send response h3_server.send_headers( stream_id=stream_id, headers=[ (b":status", b"200"), (b"content-type", b"text/html; charset=utf-8"), ], end_stream=False, ) h3_server.send_data(stream_id=stream_id, data=b"html", end_stream=True) #  fulfill push promise h3_server.send_headers( stream_id=push_stream_id, headers=[(b":status", b"200"), (b"content-type", b"text/plain")], end_stream=False, ) h3_server.send_data(stream_id=push_stream_id, data=b"text", end_stream=True) # receive push promise / reponse events = h3_transfer(quic_server, h3_client) self.assertEqual( events, [ PushPromiseReceived( headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/app.txt"), ], push_id=0, stream_id=stream_id, ), HeadersReceived( headers=[ (b":status", b"200"), (b"content-type", b"text/html; charset=utf-8"), ], stream_id=0, stream_ended=False, ), DataReceived(data=b"h", stream_id=0, stream_ended=False), DataReceived(data=b"t", stream_id=0, stream_ended=False), DataReceived(data=b"m", stream_id=0, stream_ended=False), DataReceived(data=b"l", stream_id=0, stream_ended=False), DataReceived(data=b"", stream_id=0, stream_ended=True), HeadersReceived( headers=[(b":status", b"200"), (b"content-type", b"text/plain")], stream_id=15, stream_ended=False, push_id=0, ), DataReceived(data=b"t", stream_id=15, stream_ended=False, push_id=0), DataReceived(data=b"e", stream_id=15, stream_ended=False, push_id=0), DataReceived(data=b"x", stream_id=15, stream_ended=False, push_id=0), DataReceived(data=b"t", stream_id=15, stream_ended=False, push_id=0), DataReceived(data=b"", stream_id=15, stream_ended=True, push_id=0), ], ) def test_request_with_server_push(self): with h3_client_and_server() as (quic_client, quic_server): h3_client = H3Connection(quic_client) h3_server = H3Connection(quic_server) # send request stream_id = quic_client.get_next_available_stream_id() h3_client.send_headers( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), ], end_stream=True, ) # receive request events = h3_transfer(quic_client, h3_server) self.assertEqual( events, [ HeadersReceived( headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), ], stream_id=stream_id, stream_ended=True, ) ], ) # send push promises push_stream_id_css = h3_server.send_push_promise( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/app.css"), ], ) self.assertEqual(push_stream_id_css, 15) push_stream_id_js = h3_server.send_push_promise( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/app.js"), ], ) self.assertEqual(push_stream_id_js, 19) # send response h3_server.send_headers( stream_id=stream_id, headers=[ (b":status", b"200"), (b"content-type", b"text/html; charset=utf-8"), ], end_stream=False, ) h3_server.send_data( stream_id=stream_id, data=b"hello", end_stream=True, ) #  fulfill push promises h3_server.send_headers( stream_id=push_stream_id_css, headers=[(b":status", b"200"), (b"content-type", b"text/css")], end_stream=False, ) h3_server.send_data( stream_id=push_stream_id_css, data=b"body { color: pink }", end_stream=True, ) h3_server.send_headers( stream_id=push_stream_id_js, headers=[ (b":status", b"200"), (b"content-type", b"application/javascript"), ], end_stream=False, ) h3_server.send_data( stream_id=push_stream_id_js, data=b"alert('howdee');", end_stream=True ) # receive push promises, response and push responses events = h3_transfer(quic_server, h3_client) self.assertEqual( events, [ PushPromiseReceived( headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/app.css"), ], push_id=0, stream_id=stream_id, ), PushPromiseReceived( headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/app.js"), ], push_id=1, stream_id=stream_id, ), HeadersReceived( headers=[ (b":status", b"200"), (b"content-type", b"text/html; charset=utf-8"), ], stream_id=stream_id, stream_ended=False, ), DataReceived( data=b"hello", stream_id=stream_id, stream_ended=True, ), HeadersReceived( headers=[(b":status", b"200"), (b"content-type", b"text/css")], push_id=0, stream_id=push_stream_id_css, stream_ended=False, ), DataReceived( data=b"body { color: pink }", push_id=0, stream_id=push_stream_id_css, stream_ended=True, ), HeadersReceived( headers=[ (b":status", b"200"), (b"content-type", b"application/javascript"), ], push_id=1, stream_id=push_stream_id_js, stream_ended=False, ), DataReceived( data=b"alert('howdee');", push_id=1, stream_id=push_stream_id_js, stream_ended=True, ), ], ) def test_request_with_server_push_max_push_id(self): with h3_client_and_server() as (quic_client, quic_server): h3_client = H3Connection(quic_client) h3_server = H3Connection(quic_server) # send request stream_id = quic_client.get_next_available_stream_id() h3_client.send_headers( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), ], end_stream=True, ) # receive request events = h3_transfer(quic_client, h3_server) self.assertEqual( events, [ HeadersReceived( headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), ], stream_id=stream_id, stream_ended=True, ) ], ) # send push promises for i in range(0, 8): h3_server.send_push_promise( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", "/{}.css".format(i).encode("ascii")), ], ) # send one too many with self.assertRaises(NoAvailablePushIDError): h3_server.send_push_promise( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/8.css"), ], ) def test_send_data_after_trailers(self): """ We should not send DATA after trailers. """ quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) h3_client = H3Connection(quic_client) stream_id = quic_client.get_next_available_stream_id() h3_client.send_headers( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), ], ) h3_client.send_headers( stream_id=stream_id, headers=[(b"x-some-trailer", b"foo")], end_stream=False ) with self.assertRaises(FrameUnexpected): h3_client.send_data(stream_id=stream_id, data=b"hello", end_stream=False) def test_send_data_before_headers(self): """ We should not send DATA before headers. """ quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) h3_client = H3Connection(quic_client) stream_id = quic_client.get_next_available_stream_id() with self.assertRaises(FrameUnexpected): h3_client.send_data(stream_id=stream_id, data=b"hello", end_stream=False) def test_send_headers_after_trailers(self): """ We should not send HEADERS after trailers. """ quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) h3_client = H3Connection(quic_client) stream_id = quic_client.get_next_available_stream_id() h3_client.send_headers( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), ], ) h3_client.send_headers( stream_id=stream_id, headers=[(b"x-some-trailer", b"foo")], end_stream=False ) with self.assertRaises(FrameUnexpected): h3_client.send_headers( stream_id=stream_id, headers=[(b"x-other-trailer", b"foo")], end_stream=False, ) def test_blocked_stream(self): quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) h3_client = H3Connection(quic_client) h3_client.handle_event( StreamDataReceived( stream_id=3, data=binascii.unhexlify( "0004170150000680020000074064091040bcc0000000faceb00c" ), end_stream=False, ) ) h3_client.handle_event( StreamDataReceived(stream_id=7, data=b"\x02", end_stream=False) ) h3_client.handle_event( StreamDataReceived(stream_id=11, data=b"\x03", end_stream=False) ) h3_client.handle_event( StreamDataReceived( stream_id=0, data=binascii.unhexlify("01040280d910"), end_stream=False ) ) h3_client.handle_event( StreamDataReceived( stream_id=0, data=binascii.unhexlify( "00408d796f752072656163686564206d766673742e6e65742c20726561636820" "746865202f6563686f20656e64706f696e7420666f7220616e206563686f2072" "6573706f6e7365207175657279202f3c6e756d6265723e20656e64706f696e74" "7320666f722061207661726961626c652073697a6520726573706f6e73652077" "6974682072616e646f6d206279746573" ), end_stream=True, ) ) self.assertEqual( h3_client.handle_event( StreamDataReceived( stream_id=7, data=binascii.unhexlify( "3fe101c696d07abe941094cb6d0a08017d403971966e32ca98b46f" ), end_stream=False, ) ), [ HeadersReceived( headers=[ (b":status", b"200"), (b"date", b"Mon, 22 Jul 2019 06:33:33 GMT"), ], stream_id=0, stream_ended=False, ), DataReceived( data=( b"you reached mvfst.net, reach the /echo endpoint for an " b"echo response query / endpoints for a variable " b"size response with random bytes" ), stream_id=0, stream_ended=True, ), ], ) def test_blocked_stream_trailer(self): quic_client = FakeQuicConnection( configuration=QuicConfiguration(is_client=True) ) h3_client = H3Connection(quic_client) h3_client.handle_event( StreamDataReceived( stream_id=3, data=binascii.unhexlify( "0004170150000680020000074064091040bcc0000000faceb00c" ), end_stream=False, ) ) h3_client.handle_event( StreamDataReceived(stream_id=7, data=b"\x02", end_stream=False) ) h3_client.handle_event( StreamDataReceived(stream_id=11, data=b"\x03", end_stream=False) ) self.assertEqual( h3_client.handle_event( StreamDataReceived( stream_id=0, data=binascii.unhexlify( "011b0000d95696d07abe941094cb6d0a08017d403971966e32ca98b46f" ), end_stream=False, ) ), [ HeadersReceived( headers=[ (b":status", b"200"), (b"date", b"Mon, 22 Jul 2019 06:33:33 GMT"), ], stream_id=0, stream_ended=False, ) ], ) self.assertEqual( h3_client.handle_event( StreamDataReceived( stream_id=0, data=binascii.unhexlify( "00408d796f752072656163686564206d766673742e6e65742c20726561636820" "746865202f6563686f20656e64706f696e7420666f7220616e206563686f2072" "6573706f6e7365207175657279202f3c6e756d6265723e20656e64706f696e74" "7320666f722061207661726961626c652073697a6520726573706f6e73652077" "6974682072616e646f6d206279746573" ), end_stream=False, ) ), [ DataReceived( data=( b"you reached mvfst.net, reach the /echo endpoint for an " b"echo response query / endpoints for a variable " b"size response with random bytes" ), stream_id=0, stream_ended=False, ) ], ) self.assertEqual( h3_client.handle_event( StreamDataReceived( stream_id=0, data=binascii.unhexlify("0103028010"), end_stream=True ) ), [], ) self.assertEqual( h3_client.handle_event( StreamDataReceived( stream_id=7, data=binascii.unhexlify("6af2b20f49564d833505b38294e7"), end_stream=False, ) ), [ HeadersReceived( headers=[(b"x-some-trailer", b"foo")], stream_id=0, stream_ended=True, push_id=None, ) ], ) def test_uni_stream_grease(self): with h3_client_and_server() as (quic_client, quic_server): h3_server = H3Connection(quic_server) quic_client.send_stream_data( 14, b"\xff\xff\xff\xff\xff\xff\xff\xfeGREASE is the word" ) self.assertEqual(h3_transfer(quic_client, h3_server), []) def test_request_with_trailers(self): with h3_client_and_server() as (quic_client, quic_server): h3_client = H3Connection(quic_client) h3_server = H3Connection(quic_server) # send request with trailers stream_id = quic_client.get_next_available_stream_id() h3_client.send_headers( stream_id=stream_id, headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), ], end_stream=False, ) h3_client.send_headers( stream_id=stream_id, headers=[(b"x-some-trailer", b"foo")], end_stream=True, ) # receive request events = h3_transfer(quic_client, h3_server) self.assertEqual( events, [ HeadersReceived( headers=[ (b":method", b"GET"), (b":scheme", b"https"), (b":authority", b"localhost"), (b":path", b"/"), ], stream_id=stream_id, stream_ended=False, ), HeadersReceived( headers=[(b"x-some-trailer", b"foo")], stream_id=stream_id, stream_ended=True, ), ], ) # send response h3_server.send_headers( stream_id=stream_id, headers=[ (b":status", b"200"), (b"content-type", b"text/html; charset=utf-8"), ], end_stream=False, ) h3_server.send_data( stream_id=stream_id, data=b"hello", end_stream=False, ) h3_server.send_headers( stream_id=stream_id, headers=[(b"x-some-trailer", b"bar")], end_stream=True, ) # receive response events = h3_transfer(quic_server, h3_client) self.assertEqual( events, [ HeadersReceived( headers=[ (b":status", b"200"), (b"content-type", b"text/html; charset=utf-8"), ], stream_id=stream_id, stream_ended=False, ), DataReceived( data=b"hello", stream_id=stream_id, stream_ended=False, ), HeadersReceived( headers=[(b"x-some-trailer", b"bar")], stream_id=stream_id, stream_ended=True, ), ], ) def test_uni_stream_type(self): with h3_client_and_server() as (quic_client, quic_server): h3_server = H3Connection(quic_server) # unknown stream type 9 stream_id = quic_client.get_next_available_stream_id(is_unidirectional=True) self.assertEqual(stream_id, 2) quic_client.send_stream_data(stream_id, b"\x09") self.assertEqual(h3_transfer(quic_client, h3_server), []) self.assertEqual(list(h3_server._stream.keys()), [2]) self.assertEqual(h3_server._stream[2].buffer, b"") self.assertEqual(h3_server._stream[2].stream_type, 9) # unknown stream type 64, one byte at a time stream_id = quic_client.get_next_available_stream_id(is_unidirectional=True) self.assertEqual(stream_id, 6) quic_client.send_stream_data(stream_id, b"\x40") self.assertEqual(h3_transfer(quic_client, h3_server), []) self.assertEqual(list(h3_server._stream.keys()), [2, 6]) self.assertEqual(h3_server._stream[2].buffer, b"") self.assertEqual(h3_server._stream[2].stream_type, 9) self.assertEqual(h3_server._stream[6].buffer, b"\x40") self.assertEqual(h3_server._stream[6].stream_type, None) quic_client.send_stream_data(stream_id, b"\x40") self.assertEqual(h3_transfer(quic_client, h3_server), []) self.assertEqual(list(h3_server._stream.keys()), [2, 6]) self.assertEqual(h3_server._stream[2].buffer, b"") self.assertEqual(h3_server._stream[2].stream_type, 9) self.assertEqual(h3_server._stream[6].buffer, b"") self.assertEqual(h3_server._stream[6].stream_type, 64)