From 55bd70a7c00d352cb0e504c35b07086156b3fcc1 Mon Sep 17 00:00:00 2001 From: avalyset Date: Mon, 21 Sep 2026 07:51:36 +0200 Subject: [PATCH] Validate header names on the outbound path Split _reject_illegal_characters into a name half and a value half, and run the name half from validate_outbound_headers as well. The four name classes from RFC 9113 section 8.2.1 were checked on the inbound path only. Classes 2, 3 and 4 went out on the wire untouched; class 1 was normalised away by _lowercase_header_names with defaults, but not with normalize_outbound_headers=False. The value rules (NUL/LF/CR and surrounding whitespace) stay inbound-only: _strip_surrounding_whitespace already normalises the whitespace case on the outbound path. "Received uppercase header name" is now direction-neutral, since the check runs both ways. Co-Authored-By: Claude Opus 5 --- src/h2/utilities.py | 26 +++++++-- tests/test_invalid_headers.py | 100 ++++++++++++++++++++++++++++++++++ 2 files changed, 121 insertions(+), 5 deletions(-) diff --git a/src/h2/utilities.py b/src/h2/utilities.py index c4e62f2f3..b40ae395c 100644 --- a/src/h2/utilities.py +++ b/src/h2/utilities.py @@ -195,7 +195,10 @@ def validate_headers(headers: Iterable[Header], hdr_validation_flags: HeaderVali # For example, we avoid tuple unpacking in loops because it represents a # fixed cost that we don't want to spend, instead indexing into the header # tuples. - headers = _reject_illegal_characters( + headers = _reject_illegal_name_characters( + headers, hdr_validation_flags, + ) + headers = _reject_illegal_value_characters( headers, hdr_validation_flags, ) headers = _reject_empty_header_names( @@ -216,10 +219,10 @@ def validate_headers(headers: Iterable[Header], hdr_validation_flags: HeaderVali return _check_path_header(headers, hdr_validation_flags) -def _reject_illegal_characters(headers: Iterable[Header], - hdr_validation_flags: HeaderValidationFlags) -> Generator[Header, None, None]: +def _reject_illegal_name_characters(headers: Iterable[Header], + hdr_validation_flags: HeaderValidationFlags) -> Generator[Header, None, None]: """ - Raises a ProtocolError if any header names or values contain illegal characters. + Raises a ProtocolError if any header names contain illegal characters. See . """ for header in headers: @@ -227,7 +230,7 @@ def _reject_illegal_characters(headers: Iterable[Header], # > or 0x7f-0xff (all ranges inclusive). for c in header[0]: if 0x41 <= c <= 0x5a: - msg = f"Received uppercase header name {header[0]!r}." + msg = f"Uppercase header name present: {header[0]!r}." raise ProtocolError(msg) if c <= 0x20 or c >= 0x7f: msg = f"Illegal character '{chr(c)}' in header name: {header[0]!r}" @@ -240,6 +243,16 @@ def _reject_illegal_characters(headers: Iterable[Header], msg = f"Illegal character ':' in header name: {header[0]!r}" raise ProtocolError(msg) + yield header + + +def _reject_illegal_value_characters(headers: Iterable[Header], + hdr_validation_flags: HeaderValidationFlags) -> Generator[Header, None, None]: + """ + Raises a ProtocolError if any header values contain illegal characters. + See . + """ + for header in headers: # For compatibility with RFC 7230 header fields, we need to allow the field # value to be an empty string. This is ludicrous, but technically allowed. if field_value := header[1]: @@ -690,6 +703,9 @@ def validate_outbound_headers(headers: Iterable[Header], :param headers: The HTTP header set. :param hdr_validation_flags: An instance of HeaderValidationFlags. """ + headers = _reject_illegal_name_characters( + headers, hdr_validation_flags, + ) headers = _reject_te( headers, hdr_validation_flags, ) diff --git a/tests/test_invalid_headers.py b/tests/test_invalid_headers.py index 2a68c6fc6..ff354ea26 100644 --- a/tests/test_invalid_headers.py +++ b/tests/test_invalid_headers.py @@ -405,6 +405,106 @@ def test_push_promise_skip_normalization(self, frame_factory, headers) -> None: ) assert c.data_to_send() == pp_frame.serialize() + illegal_name_header_blocks = [ + [*base_request_headers, ("foo bar", "baz")], + [*base_request_headers, ("foo\x7f", "bar")], + [*base_request_headers, ("foo:bar", "baz")], + ] + + @pytest.mark.parametrize("headers", illegal_name_header_blocks) + def test_headers_event_illegal_name_characters(self, frame_factory, headers) -> None: + """ + Sending header names containing illegal characters raises a + ProtocolError, even though normalization leaves them untouched. + """ + c = h2.connection.H2Connection() + c.initiate_connection() + + # Clear the data, then try to send headers. + c.clear_outbound_data_buffer() + with pytest.raises(h2.exceptions.ProtocolError): + c.send_headers(1, headers) + + @pytest.mark.parametrize("headers", illegal_name_header_blocks) + def test_send_push_promise_illegal_name_characters(self, frame_factory, headers) -> None: + """ + Sending header names containing illegal characters in a push promise + raises a ProtocolError. + """ + c = h2.connection.H2Connection(config=self.server_config) + c.initiate_connection() + c.receive_data(frame_factory.preamble()) + + header_frame = frame_factory.build_headers_frame( + self.base_request_headers, + ) + c.receive_data(header_frame.serialize()) + + # Clear the data, then try to send a push promise. + c.clear_outbound_data_buffer() + with pytest.raises(h2.exceptions.ProtocolError): + c.push_stream( + stream_id=1, promised_stream_id=2, request_headers=headers, + ) + + @pytest.mark.parametrize("headers", illegal_name_header_blocks) + def test_headers_event_illegal_name_characters_skipping_validation(self, frame_factory, headers) -> None: + """ + If we have ``validate_outbound_headers`` disabled, header names + containing illegal characters are allowed to pass. + """ + config = h2.config.H2Configuration( + validate_outbound_headers=False, + ) + + c = h2.connection.H2Connection(config=config) + c.initiate_connection() + + # Clear the data, then send headers. + c.clear_outbound_data_buffer() + c.send_headers(1, headers) + + headers = h2.utilities.utf8_encode_headers(headers) + norm_headers = h2.utilities.normalize_outbound_headers( + headers, None, False, + ) + f = frame_factory.build_headers_frame(norm_headers) + assert c.data_to_send() == f.serialize() + + def test_headers_event_uppercase_name_skipping_normalization(self, frame_factory) -> None: + """ + With ``normalize_outbound_headers`` disabled, an uppercase header name + is no longer lowercased before validation and is rejected. + """ + config = h2.config.H2Configuration( + normalize_outbound_headers=False, + ) + + c = h2.connection.H2Connection(config=config) + c.initiate_connection() + + # Clear the data, then try to send headers. + c.clear_outbound_data_buffer() + with pytest.raises(h2.exceptions.ProtocolError): + c.send_headers(1, [*self.base_request_headers, ("X-Foo", "bar")]) + + def test_headers_event_uppercase_name_is_lowercased(self, frame_factory) -> None: + """ + With normalization enabled an uppercase header name is still + lowercased and sent, rather than rejected. + """ + c = h2.connection.H2Connection() + c.initiate_connection() + + # Clear the data, then send headers. + c.clear_outbound_data_buffer() + c.send_headers(1, [*self.base_request_headers, ("X-Foo", "bar")]) + + f = frame_factory.build_headers_frame( + [*self.base_request_headers, ("x-foo", "bar")], + ) + assert c.data_to_send() == f.serialize() + @pytest.mark.parametrize("headers", strippable_header_blocks) def test_strippable_headers(self, frame_factory, headers) -> None: """