Source code
Revision control
Copy as Markdown
Other Tools
# -*- coding: utf-8 -*-
"""
test_header_indexing.py
~~~~~~~~~~~~~~~~~~~~~~~
This module contains tests that use HPACK header tuples that provide additional
metadata to the hpack module about how to encode the headers.
"""
import pytest
from hpack import HeaderTuple, NeverIndexedHeaderTuple
import h2.config
import h2.connection
def assert_header_blocks_actually_equal(block_a, block_b):
"""
Asserts that two header bocks are really, truly equal, down to the types
of their tuples. Doesn't return anything.
"""
assert len(block_a) == len(block_b)
for a, b in zip(block_a, block_b):
assert a == b
assert a.__class__ is b.__class__
class TestHeaderIndexing(object):
"""
Test that Hyper-h2 can correctly handle never indexed header fields using
the appropriate hpack data structures.
"""
example_request_headers = [
HeaderTuple(u':authority', u'example.com'),
HeaderTuple(u':path', u'/'),
HeaderTuple(u':scheme', u'https'),
HeaderTuple(u':method', u'GET'),
]
bytes_example_request_headers = [
HeaderTuple(b':authority', b'example.com'),
HeaderTuple(b':path', b'/'),
HeaderTuple(b':scheme', b'https'),
HeaderTuple(b':method', b'GET'),
]
extended_request_headers = [
HeaderTuple(u':authority', u'example.com'),
HeaderTuple(u':path', u'/'),
HeaderTuple(u':scheme', u'https'),
HeaderTuple(u':method', u'GET'),
NeverIndexedHeaderTuple(u'authorization', u'realpassword'),
]
bytes_extended_request_headers = [
HeaderTuple(b':authority', b'example.com'),
HeaderTuple(b':path', b'/'),
HeaderTuple(b':scheme', b'https'),
HeaderTuple(b':method', b'GET'),
NeverIndexedHeaderTuple(b'authorization', b'realpassword'),
]
example_response_headers = [
HeaderTuple(u':status', u'200'),
HeaderTuple(u'server', u'fake-serv/0.1.0')
]
bytes_example_response_headers = [
HeaderTuple(b':status', b'200'),
HeaderTuple(b'server', b'fake-serv/0.1.0')
]
extended_response_headers = [
HeaderTuple(u':status', u'200'),
HeaderTuple(u'server', u'fake-serv/0.1.0'),
NeverIndexedHeaderTuple(u'secure', u'you-bet'),
]
bytes_extended_response_headers = [
HeaderTuple(b':status', b'200'),
HeaderTuple(b'server', b'fake-serv/0.1.0'),
NeverIndexedHeaderTuple(b'secure', b'you-bet'),
]
server_config = h2.config.H2Configuration(client_side=False)
@pytest.mark.parametrize(
'headers', (
example_request_headers,
bytes_example_request_headers,
extended_request_headers,
bytes_extended_request_headers,
)
)
def test_sending_header_tuples(self, headers, frame_factory):
"""
Providing HeaderTuple and HeaderTuple subclasses preserves the metadata
about indexing.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
# Clear the data, then send headers.
c.clear_outbound_data_buffer()
c.send_headers(1, headers)
f = frame_factory.build_headers_frame(headers=headers)
assert c.data_to_send() == f.serialize()
@pytest.mark.parametrize(
'headers', (
example_request_headers,
bytes_example_request_headers,
extended_request_headers,
bytes_extended_request_headers,
)
)
def test_header_tuples_in_pushes(self, headers, frame_factory):
"""
Providing HeaderTuple and HeaderTuple subclasses to push promises
preserves metadata about indexing.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.receive_data(frame_factory.preamble())
# We can use normal headers for the request.
f = frame_factory.build_headers_frame(
self.example_request_headers
)
c.receive_data(f.serialize())
frame_factory.refresh_encoder()
expected_frame = frame_factory.build_push_promise_frame(
stream_id=1,
promised_stream_id=2,
headers=headers,
flags=['END_HEADERS'],
)
c.clear_outbound_data_buffer()
c.push_stream(
stream_id=1,
promised_stream_id=2,
request_headers=headers
)
assert c.data_to_send() == expected_frame.serialize()
@pytest.mark.parametrize(
'headers,encoding', (
(example_request_headers, 'utf-8'),
(bytes_example_request_headers, None),
(extended_request_headers, 'utf-8'),
(bytes_extended_request_headers, None),
)
)
def test_header_tuples_are_decoded_request(self,
headers,
encoding,
frame_factory):
"""
The indexing status of the header is preserved when emitting
RequestReceived events.
"""
config = h2.config.H2Configuration(
client_side=False, header_encoding=encoding
)
c = h2.connection.H2Connection(config=config)
c.receive_data(frame_factory.preamble())
f = frame_factory.build_headers_frame(headers)
data = f.serialize()
events = c.receive_data(data)
assert len(events) == 1
event = events[0]
assert isinstance(event, h2.events.RequestReceived)
assert_header_blocks_actually_equal(headers, event.headers)
@pytest.mark.parametrize(
'headers,encoding', (
(example_response_headers, 'utf-8'),
(bytes_example_response_headers, None),
(extended_response_headers, 'utf-8'),
(bytes_extended_response_headers, None),
)
)
def test_header_tuples_are_decoded_response(self,
headers,
encoding,
frame_factory):
"""
The indexing status of the header is preserved when emitting
ResponseReceived events.
"""
config = h2.config.H2Configuration(
header_encoding=encoding
)
c = h2.connection.H2Connection(config=config)
c.initiate_connection()
c.send_headers(stream_id=1, headers=self.example_request_headers)
f = frame_factory.build_headers_frame(headers)
data = f.serialize()
events = c.receive_data(data)
assert len(events) == 1
event = events[0]
assert isinstance(event, h2.events.ResponseReceived)
assert_header_blocks_actually_equal(headers, event.headers)
@pytest.mark.parametrize(
'headers,encoding', (
(example_response_headers, 'utf-8'),
(bytes_example_response_headers, None),
(extended_response_headers, 'utf-8'),
(bytes_extended_response_headers, None),
)
)
def test_header_tuples_are_decoded_info_response(self,
headers,
encoding,
frame_factory):
"""
The indexing status of the header is preserved when emitting
InformationalResponseReceived events.
"""
# Manipulate the headers to send 100 Continue. We need to copy the list
# to avoid breaking the example headers.
headers = headers[:]
if encoding:
headers[0] = HeaderTuple(u':status', u'100')
else:
headers[0] = HeaderTuple(b':status', b'100')
config = h2.config.H2Configuration(
header_encoding=encoding
)
c = h2.connection.H2Connection(config=config)
c.initiate_connection()
c.send_headers(stream_id=1, headers=self.example_request_headers)
f = frame_factory.build_headers_frame(headers)
data = f.serialize()
events = c.receive_data(data)
assert len(events) == 1
event = events[0]
assert isinstance(event, h2.events.InformationalResponseReceived)
assert_header_blocks_actually_equal(headers, event.headers)
@pytest.mark.parametrize(
'headers,encoding', (
(example_response_headers, 'utf-8'),
(bytes_example_response_headers, None),
(extended_response_headers, 'utf-8'),
(bytes_extended_response_headers, None),
)
)
def test_header_tuples_are_decoded_trailers(self,
headers,
encoding,
frame_factory):
"""
The indexing status of the header is preserved when emitting
TrailersReceived events.
"""
# Manipulate the headers to remove the status, which shouldn't be in
# the trailers. We need to copy the list to avoid breaking the example
# headers.
headers = headers[1:]
config = h2.config.H2Configuration(
header_encoding=encoding
)
c = h2.connection.H2Connection(config=config)
c.initiate_connection()
c.send_headers(stream_id=1, headers=self.example_request_headers)
f = frame_factory.build_headers_frame(self.example_response_headers)
data = f.serialize()
c.receive_data(data)
f = frame_factory.build_headers_frame(headers, flags=['END_STREAM'])
data = f.serialize()
events = c.receive_data(data)
assert len(events) == 2
event = events[0]
assert isinstance(event, h2.events.TrailersReceived)
assert_header_blocks_actually_equal(headers, event.headers)
@pytest.mark.parametrize(
'headers,encoding', (
(example_request_headers, 'utf-8'),
(bytes_example_request_headers, None),
(extended_request_headers, 'utf-8'),
(bytes_extended_request_headers, None),
)
)
def test_header_tuples_are_decoded_push_promise(self,
headers,
encoding,
frame_factory):
"""
The indexing status of the header is preserved when emitting
PushedStreamReceived events.
"""
config = h2.config.H2Configuration(
header_encoding=encoding
)
c = h2.connection.H2Connection(config=config)
c.initiate_connection()
c.send_headers(stream_id=1, headers=self.example_request_headers)
f = frame_factory.build_push_promise_frame(
stream_id=1,
promised_stream_id=2,
headers=headers,
flags=['END_HEADERS'],
)
data = f.serialize()
events = c.receive_data(data)
assert len(events) == 1
event = events[0]
assert isinstance(event, h2.events.PushedStreamReceived)
assert_header_blocks_actually_equal(headers, event.headers)
class TestSecureHeaders(object):
"""
Certain headers should always be transformed to their never-indexed form.
"""
example_request_headers = [
(u':authority', u'example.com'),
(u':path', u'/'),
(u':scheme', u'https'),
(u':method', u'GET'),
]
bytes_example_request_headers = [
(b':authority', b'example.com'),
(b':path', b'/'),
(b':scheme', b'https'),
(b':method', b'GET'),
]
possible_auth_headers = [
(u'authorization', u'test'),
(u'Authorization', u'test'),
(u'authorization', u'really long test'),
HeaderTuple(u'authorization', u'test'),
HeaderTuple(u'Authorization', u'test'),
HeaderTuple(u'authorization', u'really long test'),
NeverIndexedHeaderTuple(u'authorization', u'test'),
NeverIndexedHeaderTuple(u'Authorization', u'test'),
NeverIndexedHeaderTuple(u'authorization', u'really long test'),
(b'authorization', b'test'),
(b'Authorization', b'test'),
(b'authorization', b'really long test'),
HeaderTuple(b'authorization', b'test'),
HeaderTuple(b'Authorization', b'test'),
HeaderTuple(b'authorization', b'really long test'),
NeverIndexedHeaderTuple(b'authorization', b'test'),
NeverIndexedHeaderTuple(b'Authorization', b'test'),
NeverIndexedHeaderTuple(b'authorization', b'really long test'),
(u'proxy-authorization', u'test'),
(u'Proxy-Authorization', u'test'),
(u'proxy-authorization', u'really long test'),
HeaderTuple(u'proxy-authorization', u'test'),
HeaderTuple(u'Proxy-Authorization', u'test'),
HeaderTuple(u'proxy-authorization', u'really long test'),
NeverIndexedHeaderTuple(u'proxy-authorization', u'test'),
NeverIndexedHeaderTuple(u'Proxy-Authorization', u'test'),
NeverIndexedHeaderTuple(u'proxy-authorization', u'really long test'),
(b'proxy-authorization', b'test'),
(b'Proxy-Authorization', b'test'),
(b'proxy-authorization', b'really long test'),
HeaderTuple(b'proxy-authorization', b'test'),
HeaderTuple(b'Proxy-Authorization', b'test'),
HeaderTuple(b'proxy-authorization', b'really long test'),
NeverIndexedHeaderTuple(b'proxy-authorization', b'test'),
NeverIndexedHeaderTuple(b'Proxy-Authorization', b'test'),
NeverIndexedHeaderTuple(b'proxy-authorization', b'really long test'),
]
secured_cookie_headers = [
(u'cookie', u'short'),
(u'Cookie', u'short'),
(u'cookie', u'nineteen byte cooki'),
HeaderTuple(u'cookie', u'short'),
HeaderTuple(u'Cookie', u'short'),
HeaderTuple(u'cookie', u'nineteen byte cooki'),
NeverIndexedHeaderTuple(u'cookie', u'short'),
NeverIndexedHeaderTuple(u'Cookie', u'short'),
NeverIndexedHeaderTuple(u'cookie', u'nineteen byte cooki'),
NeverIndexedHeaderTuple(u'cookie', u'longer manually secured cookie'),
(b'cookie', b'short'),
(b'Cookie', b'short'),
(b'cookie', b'nineteen byte cooki'),
HeaderTuple(b'cookie', b'short'),
HeaderTuple(b'Cookie', b'short'),
HeaderTuple(b'cookie', b'nineteen byte cooki'),
NeverIndexedHeaderTuple(b'cookie', b'short'),
NeverIndexedHeaderTuple(b'Cookie', b'short'),
NeverIndexedHeaderTuple(b'cookie', b'nineteen byte cooki'),
NeverIndexedHeaderTuple(b'cookie', b'longer manually secured cookie'),
]
unsecured_cookie_headers = [
(u'cookie', u'twenty byte cookie!!'),
(u'Cookie', u'twenty byte cookie!!'),
(u'cookie', u'substantially longer than 20 byte cookie'),
HeaderTuple(u'cookie', u'twenty byte cookie!!'),
HeaderTuple(u'cookie', u'twenty byte cookie!!'),
HeaderTuple(u'Cookie', u'twenty byte cookie!!'),
(b'cookie', b'twenty byte cookie!!'),
(b'Cookie', b'twenty byte cookie!!'),
(b'cookie', b'substantially longer than 20 byte cookie'),
HeaderTuple(b'cookie', b'twenty byte cookie!!'),
HeaderTuple(b'cookie', b'twenty byte cookie!!'),
HeaderTuple(b'Cookie', b'twenty byte cookie!!'),
]
server_config = h2.config.H2Configuration(client_side=False)
@pytest.mark.parametrize(
'headers', (example_request_headers, bytes_example_request_headers)
)
@pytest.mark.parametrize('auth_header', possible_auth_headers)
def test_authorization_headers_never_indexed(self,
headers,
auth_header,
frame_factory):
"""
Authorization and Proxy-Authorization headers are always forced to be
never-indexed, regardless of their form.
"""
# Regardless of what we send, we expect it to be never indexed.
send_headers = headers + [auth_header]
expected_headers = headers + [
NeverIndexedHeaderTuple(auth_header[0].lower(), auth_header[1])
]
c = h2.connection.H2Connection()
c.initiate_connection()
# Clear the data, then send headers.
c.clear_outbound_data_buffer()
c.send_headers(1, send_headers)
f = frame_factory.build_headers_frame(headers=expected_headers)
assert c.data_to_send() == f.serialize()
@pytest.mark.parametrize(
'headers', (example_request_headers, bytes_example_request_headers)
)
@pytest.mark.parametrize('auth_header', possible_auth_headers)
def test_authorization_headers_never_indexed_push(self,
headers,
auth_header,
frame_factory):
"""
Authorization and Proxy-Authorization headers are always forced to be
never-indexed, regardless of their form, when pushed by a server.
"""
# Regardless of what we send, we expect it to be never indexed.
send_headers = headers + [auth_header]
expected_headers = headers + [
NeverIndexedHeaderTuple(auth_header[0].lower(), auth_header[1])
]
c = h2.connection.H2Connection(config=self.server_config)
c.receive_data(frame_factory.preamble())
# We can use normal headers for the request.
f = frame_factory.build_headers_frame(
self.example_request_headers
)
c.receive_data(f.serialize())
frame_factory.refresh_encoder()
expected_frame = frame_factory.build_push_promise_frame(
stream_id=1,
promised_stream_id=2,
headers=expected_headers,
flags=['END_HEADERS'],
)
c.clear_outbound_data_buffer()
c.push_stream(
stream_id=1,
promised_stream_id=2,
request_headers=send_headers
)
assert c.data_to_send() == expected_frame.serialize()
@pytest.mark.parametrize(
'headers', (example_request_headers, bytes_example_request_headers)
)
@pytest.mark.parametrize('cookie_header', secured_cookie_headers)
def test_short_cookie_headers_never_indexed(self,
headers,
cookie_header,
frame_factory):
"""
Short cookie headers, and cookies provided as NeverIndexedHeaderTuple,
are never indexed.
"""
# Regardless of what we send, we expect it to be never indexed.
send_headers = headers + [cookie_header]
expected_headers = headers + [
NeverIndexedHeaderTuple(cookie_header[0].lower(), cookie_header[1])
]
c = h2.connection.H2Connection()
c.initiate_connection()
# Clear the data, then send headers.
c.clear_outbound_data_buffer()
c.send_headers(1, send_headers)
f = frame_factory.build_headers_frame(headers=expected_headers)
assert c.data_to_send() == f.serialize()
@pytest.mark.parametrize(
'headers', (example_request_headers, bytes_example_request_headers)
)
@pytest.mark.parametrize('cookie_header', secured_cookie_headers)
def test_short_cookie_headers_never_indexed_push(self,
headers,
cookie_header,
frame_factory):
"""
Short cookie headers, and cookies provided as NeverIndexedHeaderTuple,
are never indexed when pushed by servers.
"""
# Regardless of what we send, we expect it to be never indexed.
send_headers = headers + [cookie_header]
expected_headers = headers + [
NeverIndexedHeaderTuple(cookie_header[0].lower(), cookie_header[1])
]
c = h2.connection.H2Connection(config=self.server_config)
c.receive_data(frame_factory.preamble())
# We can use normal headers for the request.
f = frame_factory.build_headers_frame(
self.example_request_headers
)
c.receive_data(f.serialize())
frame_factory.refresh_encoder()
expected_frame = frame_factory.build_push_promise_frame(
stream_id=1,
promised_stream_id=2,
headers=expected_headers,
flags=['END_HEADERS'],
)
c.clear_outbound_data_buffer()
c.push_stream(
stream_id=1,
promised_stream_id=2,
request_headers=send_headers
)
assert c.data_to_send() == expected_frame.serialize()
@pytest.mark.parametrize(
'headers', (example_request_headers, bytes_example_request_headers)
)
@pytest.mark.parametrize('cookie_header', unsecured_cookie_headers)
def test_long_cookie_headers_can_be_indexed(self,
headers,
cookie_header,
frame_factory):
"""
Longer cookie headers can be indexed.
"""
# Regardless of what we send, we expect it to be indexed.
send_headers = headers + [cookie_header]
expected_headers = headers + [
HeaderTuple(cookie_header[0].lower(), cookie_header[1])
]
c = h2.connection.H2Connection()
c.initiate_connection()
# Clear the data, then send headers.
c.clear_outbound_data_buffer()
c.send_headers(1, send_headers)
f = frame_factory.build_headers_frame(headers=expected_headers)
assert c.data_to_send() == f.serialize()
@pytest.mark.parametrize(
'headers', (example_request_headers, bytes_example_request_headers)
)
@pytest.mark.parametrize('cookie_header', unsecured_cookie_headers)
def test_long_cookie_headers_can_be_indexed_push(self,
headers,
cookie_header,
frame_factory):
"""
Longer cookie headers can be indexed.
"""
# Regardless of what we send, we expect it to be never indexed.
send_headers = headers + [cookie_header]
expected_headers = headers + [
HeaderTuple(cookie_header[0].lower(), cookie_header[1])
]
c = h2.connection.H2Connection(config=self.server_config)
c.receive_data(frame_factory.preamble())
# We can use normal headers for the request.
f = frame_factory.build_headers_frame(
self.example_request_headers
)
c.receive_data(f.serialize())
frame_factory.refresh_encoder()
expected_frame = frame_factory.build_push_promise_frame(
stream_id=1,
promised_stream_id=2,
headers=expected_headers,
flags=['END_HEADERS'],
)
c.clear_outbound_data_buffer()
c.push_stream(
stream_id=1,
promised_stream_id=2,
request_headers=send_headers
)
assert c.data_to_send() == expected_frame.serialize()