Commit c6145329a016d1bd3c23459edf648abfd20ecd68 - python-hpack

+6

-6

hpack/hpack.py less more

30	30
31	31	max_number = (2 ** prefix_bits) - 1
32	32
33		if (integer < max_number):
	33	if integer < max_number:
34	34	return bytearray([integer]) # Seriously?
35	35	else:
36	36	elements = [max_number]
37		integer = integer - max_number
	37	integer -= max_number
38	38
39	39	while integer >= 128:
40	40	elements.append((integer % 128) + 128)
41		integer = integer // 128 # We need integer division
	41	integer //= 128 # We need integer division
42	42
43	43	elements.append(integer)
44	44

60	60	try:
61	61	number = to_byte(data[index]) & mask
62	62
63		if (number == max_number):
	63	if number == max_number:
64	64
65	65	while True:
66	66	index += 1

78	78
79	79	log.debug("Decoded %d, consumed %d bytes", number, index + 1)
80	80
81		return (number, index + 1)
	81	return number, index + 1
82	82
83	83
84	84	def _to_bytes(string):

219	219	Encodes a header using the indexed representation.
220	220	"""
221	221	field = encode_integer(index, 7)
222		field[0] = field[0] \| 0x80 # we set the top bit
	222	field[0] \|= 0x80 # we set the top bit
223	223	return bytes(field)
224	224
225	225	def _encode_literal(self, name, value, indexbit, huffman=False):

+1

-1

hpack/huffman.py less more

115	115	# Pad out to an octet with ones.
116	116	bits_to_be_padded = (8 - (final_int_len % 8)) % 8
117	117	final_num <<= bits_to_be_padded
118		final_num \|= (1 << (bits_to_be_padded)) - 1
	118	final_num \|= (1 << bits_to_be_padded) - 1
119	119
120	120	# Convert the number to hex and strip off the leading '0x' and the
121	121	# trailing 'L', if present.

+4

-3

hpack/table.py less more

	0	# -- coding: utf-8 --
0	1	from collections import deque
1	2	import logging
2	3

11	12
12	13	This size is mostly irrelevant to us and defined
13	14	specifically to accommodate memory management for
14		lower level implementions. The 32 extra bytes are
	15	lower level implementations. The 32 extra bytes are
15	16	considered the "maximum" overhead that would be
16	17	required to represent each entry in the table.
17	18

163	164	for (i, (n, v)) in enumerate(HeaderTable.STATIC_TABLE):
164	165	if n == name:
165	166	if v == value:
166		return (i + 1, n, v)
	167	return i + 1, n, v
167	168	elif partial is None:
168	169	partial = (i + 1, n, None)
169	170	for (i, (n, v)) in enumerate(self.dynamic_entries):
170	171	if n == name:
171	172	if v == value:
172		return (i + offset + 1, n, v)
	173	return i + offset + 1, n, v
173	174	elif partial is None:
174	175	partial = (i + offset + 1, n, None)
175	176	return partial

+7

-3

tasks.py less more

	0	# -- coding: utf-8 --
0	1	import json
1	2	import os
2	3
3	4	from binascii import hexlify
4	5	from invoke import task
5	6	from hyper.http20.hpack import Encoder
	7
6	8
7	9	@task
8	10	def hpack():

31	33	# For each file, build our output.
32	34	for source, outname in raw_story_files:
33	35	with open(source, 'rb') as f:
34		indata = json.loads(f.read())
	36	indata = json.load(f)
35	37
36	38	# Prepare the output and the encoder.
37	39	output = {

41	43	e = Encoder()
42	44
43	45	for case in indata['cases']:
44		outcase = {'header_table_size': e.header_table_size}
45		outcase['headers'] = case['headers']
	46	outcase = {
	47	'header_table_size': e.header_table_size,
	48	'headers': case['headers'],
	49	}
46	50	headers = []
47	51
48	52	for header in case['headers']:

+7

-12

test/conftest.py less more

21	21	for name in os.listdir('test/test_fixtures/raw-data')
22	22	)
23	23
24		@pytest.fixture(scope="class",
25		params=story_files)
	24
	25	@pytest.fixture(scope='class', params=story_files)
26	26	def story(request):
27	27	"""
28	28	Provides a detailed HPACK story to test with.
29	29	"""
30		path = request.param
31		with open(path, 'r', encoding='utf-8') as f:
32		details = json.loads(f.read())
	30	with open(request.param, 'r', encoding='utf-8') as f:
	31	return json.load(f)
33	32
34		return details
35	33
36		@pytest.fixture(scope="class", params=raw_story_files)
	34	@pytest.fixture(scope='class', params=raw_story_files)
37	35	def raw_story(request):
38	36	"""
39	37	Provides a detailed HPACK story to test the encoder with.
40	38	"""
41		path = request.param
42		with open(path, 'r', encoding='utf-8') as f:
43		details = json.loads(f.read())
44
45		return details
	39	with open(request.param, 'r', encoding='utf-8') as f:
	40	return json.load(f)

+1

-2

test/test_hpack.py less more

43	43	header_set = [
44	44	(':method', 'GET', True),
45	45	(':path', '/jimiscool/', True),
46		('customkey','sensitiveinfo',True)
	46	('customkey', 'sensitiveinfo', True)
47	47	]
48	48	assert e.encode(header_set, huffman=True) == result
49	49

403	403	(':path', '/',),
404	404	(':authority', 'www.example.com'),
405	405	]
406		first_header_table = first_header_set[::-1]
407	406	first_data = (
408	407	b'\x82\x86\x84\x01\x8c\xf1\xe3\xc2\xe5\xf2:k\xa0\xab\x90\xf4\xff'
409	408	)

+1

-0

test/test_hpack_integration.py less more

6	6	from hpack.hpack import Decoder, Encoder
7	7	from binascii import unhexlify
8	8	from pytest import skip
	9
9	10
10	11	class TestHPACKDecoderIntegration(object):
11	12	def test_can_decode_a_story(self, story):

+8

-6

test/test_huffman.py less more

	0	# -- coding: utf-8 --
0	1	from hpack.huffman import HuffmanDecoder, HuffmanEncoder
1	2	from hpack.huffman_constants import REQUEST_CODES, REQUEST_CODES_LENGTH
2	3
3	4
4	5	class TestHuffman(object):
	6
5	7	def test_request_huffman_decoder(self):
6		decoder = HuffmanDecoder(REQUEST_CODES,REQUEST_CODES_LENGTH)
	8	decoder = HuffmanDecoder(REQUEST_CODES, REQUEST_CODES_LENGTH)
7	9	assert decoder.decode(b'\xf1\xe3\xc2\xe5\xf2:k\xa0\xab\x90\xf4\xff') == b"www.example.com"
8	10	assert decoder.decode(b'\xa8\xeb\x10d\x9c\xbf') == b"no-cache"
9	11	assert decoder.decode(b'%\xa8I\xe9[\xa9}\x7f') == b"custom-key"

11	13
12	14	def test_request_huffman_encode(self):
13	15	encoder = HuffmanEncoder(REQUEST_CODES, REQUEST_CODES_LENGTH)
14		assert encoder.encode(b"www.example.com") == (b'\xf1\xe3\xc2\xe5\xf2:k\xa0\xab\x90\xf4\xff')
15		assert encoder.encode(b"no-cache") == (b'\xa8\xeb\x10d\x9c\xbf')
16		assert encoder.encode(b"custom-key") == (b'%\xa8I\xe9[\xa9}\x7f')
17		assert encoder.encode(b"custom-value") == (b'%\xa8I\xe9[\xb8\xe8\xb4\xbf')
	16	assert encoder.encode(b"www.example.com") == b'\xf1\xe3\xc2\xe5\xf2:k\xa0\xab\x90\xf4\xff'
	17	assert encoder.encode(b"no-cache") == b'\xa8\xeb\x10d\x9c\xbf'
	18	assert encoder.encode(b"custom-key") == b'%\xa8I\xe9[\xa9}\x7f'
	19	assert encoder.encode(b"custom-value") == b'%\xa8I\xe9[\xb8\xe8\xb4\xbf'
18	20
19	21	def test_eos_terminates_decode_request(self):
20		decoder = HuffmanDecoder(REQUEST_CODES,REQUEST_CODES_LENGTH)
	22	decoder = HuffmanDecoder(REQUEST_CODES, REQUEST_CODES_LENGTH)
21	23	assert decoder.decode(b'\xff\xff\xff\xfc') == b''

+11

-9

test/test_table.py less more

	0	# -- coding: utf-8 --
0	1	from hpack.table import HeaderTable, table_entry_size
1	2	from hpack.exceptions import InvalidTableIndex
2	3	import pytest

5	6	is_py2 = _ver[0] == 2
6	7	is_py3 = _ver[0] == 3
7	8
	9
8	10	class TestPackageFunctions(object):
9	11	def test_table_entry_size(self):
10	12	res = table_entry_size(b'TestValue', b'TestName')
11	13	assert res == 49
	14
12	15
13	16	class TestHeaderTable(object):
14	17	def test_get_by_index_dynamic_table(self):

32	35	def test_get_by_index_zero_index(self):
33	36	tbl = HeaderTable()
34	37	with pytest.raises(InvalidTableIndex):
35		res = tbl.get_by_index(0)
	38	tbl.get_by_index(0)
36	39
37	40	def test_get_by_index_out_of_range(self):
38	41	tbl = HeaderTable()
39	42	off = len(HeaderTable.STATIC_TABLE)
40	43	tbl.add(b'TestName', b'TestValue')
41	44	with pytest.raises(InvalidTableIndex):
42		res = tbl.get_by_index(off+2)
	45	tbl.get_by_index(off + 2)
43	46
44	47	def test_repr(self):
45	48	tbl = HeaderTable()

48	51	tbl.add(b'TestName2', b'TestValue2')
49	52	# Meh, I hate that I have to do this to test
50	53	# repr
51		if(is_py3):
	54	if is_py3:
52	55	exp = ("HeaderTable(4096, False, deque([" +
53	56	"(b'TestName2', b'TestValue2'), " +
54	57	"(b'TestName2', b'TestValue2'), " +

88	91	tbl = HeaderTable()
89	92	idx = len(HeaderTable.STATIC_TABLE) + 1
90	93	tbl.add(b'TestName', b'TestValue')
91		exp = (idx , b'TestName', b'TestValue')
	94	exp = (idx, b'TestName', b'TestValue')
92	95	res = tbl.search(b'TestName', b'TestValue')
93	96	assert res == exp
94	97

96	99	tbl = HeaderTable()
97	100	idx = len(HeaderTable.STATIC_TABLE) + 1
98	101	tbl.add(b'TestName', b'TestValue')
99		exp = (idx , b'TestName', None)
	102	exp = (idx, b'TestName', None)
100	103	res = tbl.search(b'TestName', b'NotInTable')
101	104	assert res == exp
102	105
103	106	def test_search_no_match(self):
104	107	tbl = HeaderTable()
105		idx = len(HeaderTable.STATIC_TABLE)
106	108	tbl.add(b'TestName', b'TestValue')
107	109	res = tbl.search(b'NotInTable', b'NotInTable')
108	110	assert res is None

115	117	tbl = HeaderTable()
116	118	exp = int(HeaderTable.DEFAULT_SIZE / 2)
117	119	tbl.maxsize = exp
118		assert tbl.resized == True
	120	assert tbl.resized is True
119	121	assert tbl.maxsize == exp
120	122	tbl.resized = False
121	123	tbl.maxsize = exp
122		assert tbl.resized == False
	124	assert tbl.resized is False
123	125	assert tbl.maxsize == exp
124	126
125	127	def test_size(self):

131	133
132	134	def test_shrink_maxsize_is_zero(self):
133	135	tbl = HeaderTable()
134		tbl.add(b'TestName',b'TestValue')
	136	tbl.add(b'TestName', b'TestValue')
135	137	assert len(tbl.dynamic_entries) == 1
136	138	tbl.maxsize = 0
137	139	assert len(tbl.dynamic_entries) == 0