Rooba
/
mitmproxy
mirror of https://github.com/mitmproxy/mitmproxy.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

http2: add protocol tests

This commit is contained in:
Thomas Kriechbaumer 2015-06-05 13:28:09 +02:00
parent f003f87197
commit fdc908cb98
5 changed files with 231 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
netlib/http2/__init__.py
View File

@ -30,7 +30,7 @@ class HTTP2Protocol(object):
# "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
CLIENT_CONNECTION_PREFACE = '505249202a20485454502f322e300d0a0d0a534d0d0a0d0a'
ALPN_PROTO_H2 = b'h2'
ALPN_PROTO_H2 = 'h2'
HTTP2_DEFAULT_SETTINGS = {
SettingsFrame.SETTINGS.SETTINGS_HEADER_TABLE_SIZE: 4096,
@ -53,18 +53,25 @@ class HTTP2Protocol(object):
alp = self.tcp_client.get_alpn_proto_negotiated()
if alp != self.ALPN_PROTO_H2:
raise NotImplementedError(
"H2Client can not handle unknown ALP: %s" % alp)
"HTTP2Protocol can not handle unknown ALP: %s" % alp)
log.debug("ALP 'h2' successfully negotiated.")
return True
def send_connection_preface(self):
def perform_connection_preface(self):
self.tcp_client.wfile.write(
bytes(self.CLIENT_CONNECTION_PREFACE.decode('hex')))
self.send_frame(SettingsFrame(state=self))
# read server settings frame
frame = Frame.from_file(self.tcp_client.rfile, self)
assert isinstance(frame, SettingsFrame)
self._apply_settings(frame.settings)
self.read_frame() # read setting ACK frame
# read setting ACK frame
settings_ack_frame = self.read_frame()
assert isinstance(settings_ack_frame, SettingsFrame)
assert settings_ack_frame.flags & Frame.FLAG_ACK
assert len(settings_ack_frame.settings) == 0
log.debug("Connection Preface completed.")
@ -94,9 +101,9 @@ class HTTP2Protocol(object):
old_value = '-'
self.http2_settings[setting] = value
log.debug("Setting changed: %s to %d (was %s)" % (
log.debug("Setting changed: %s to %s (was %s)" % (
SettingsFrame.SETTINGS.get_name(setting),
value,
str(value),
str(old_value)))
self.send_frame(SettingsFrame(state=self, flags=Frame.FLAG_ACK))
@ -157,9 +164,6 @@ class HTTP2Protocol(object):
header_block_fragment += frame.header_block_fragment
if frame.flags | Frame.FLAG_END_HEADERS:
break
else:
log.debug("Unexpected frame received:")
log.debug(frame.human_readable())
while True:
frame = self.read_frame()
@ -167,9 +171,6 @@ class HTTP2Protocol(object):
body += frame.payload
if frame.flags | Frame.FLAG_END_STREAM:
break
else:
log.debug("Unexpected frame received:")
log.debug(frame.human_readable())
headers = {}
for header, value in self.decoder.decode(header_block_fragment):

2
netlib/test.py
View File

@ -4,7 +4,7 @@ import Queue
import cStringIO
import OpenSSL
from . import tcp, certutils
import tutils
from test import tutils
class ServerThread(threading.Thread):

0
test/h2/__init__.py → test/__init__.py
View File

2
test/h2/test_frames.py → test/http2/test_frames.py
View File

@ -1,4 +1,4 @@
import tutils
from test import tutils
from nose.tools import assert_equal
from netlib.http2.frame import *

216
test/http2/test_http2_protocol.py Normal file
View File

@ -0,0 +1,216 @@
import OpenSSL
from netlib import http2
from netlib import tcp
from netlib import test
from netlib.http2.frame import *
from test import tutils
class EchoHandler(tcp.BaseHandler):
sni = None
def handle(self):
v = self.rfile.readline()
self.wfile.write(v)
self.wfile.flush()
class TestCheckALPNMatch(test.ServerTestBase):
handler = EchoHandler
ssl = dict(
alpn_select=http2.HTTP2Protocol.ALPN_PROTO_H2,
)
if OpenSSL._util.lib.Cryptography_HAS_ALPN:
def test_check_alpn(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
c.connect()
c.convert_to_ssl(alpn_protos=[http2.HTTP2Protocol.ALPN_PROTO_H2])
protocol = http2.HTTP2Protocol(c)
assert protocol.check_alpn()
class TestCheckALPNMismatch(test.ServerTestBase):
handler = EchoHandler
ssl = dict(
alpn_select=None,
)
if OpenSSL._util.lib.Cryptography_HAS_ALPN:
def test_check_alpn(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
c.connect()
c.convert_to_ssl(alpn_protos=[http2.HTTP2Protocol.ALPN_PROTO_H2])
protocol = http2.HTTP2Protocol(c)
tutils.raises(NotImplementedError, protocol.check_alpn)
class TestPerformConnectionPreface(test.ServerTestBase):
class handler(tcp.BaseHandler):
def handle(self):
# check magic
assert self.rfile.read(24) ==\
'505249202a20485454502f322e300d0a0d0a534d0d0a0d0a'.decode('hex')
# check empty settings frame
assert self.rfile.read(9) ==\
'000000040000000000'.decode('hex')
# send empty settings frame
self.wfile.write('000000040000000000'.decode('hex'))
self.wfile.flush()
# check settings acknowledgement
assert self.rfile.read(9) == \
'000000040100000000'.decode('hex')
# send settings acknowledgement
self.wfile.write('000000040100000000'.decode('hex'))
self.wfile.flush()
ssl = True
def test_perform_connection_preface(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
c.connect()
c.convert_to_ssl()
protocol = http2.HTTP2Protocol(c)
protocol.perform_connection_preface()
class TestStreamIds():
c = tcp.TCPClient(("127.0.0.1", 0))
protocol = http2.HTTP2Protocol(c)
def test_stream_ids(self):
assert self.protocol.current_stream_id is None
assert self.protocol.next_stream_id() == 1
assert self.protocol.current_stream_id == 1
assert self.protocol.next_stream_id() == 3
assert self.protocol.current_stream_id == 3
assert self.protocol.next_stream_id() == 5
assert self.protocol.current_stream_id == 5
class TestApplySettings(test.ServerTestBase):
class handler(tcp.BaseHandler):
def handle(self):
# check settings acknowledgement
assert self.rfile.read(9) == '000000040100000000'.decode('hex')
self.wfile.write("OK")
self.wfile.flush()
ssl = True
def test_apply_settings(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
c.connect()
c.convert_to_ssl()
protocol = http2.HTTP2Protocol(c)
protocol._apply_settings({
SettingsFrame.SETTINGS.SETTINGS_ENABLE_PUSH: 'foo',
SettingsFrame.SETTINGS.SETTINGS_MAX_CONCURRENT_STREAMS: 'bar',
SettingsFrame.SETTINGS.SETTINGS_INITIAL_WINDOW_SIZE: 'deadbeef',
})
assert c.rfile.safe_read(2) == "OK"
assert protocol.http2_settings[
SettingsFrame.SETTINGS.SETTINGS_ENABLE_PUSH] == 'foo'
assert protocol.http2_settings[
SettingsFrame.SETTINGS.SETTINGS_MAX_CONCURRENT_STREAMS] == 'bar'
assert protocol.http2_settings[
SettingsFrame.SETTINGS.SETTINGS_INITIAL_WINDOW_SIZE] == 'deadbeef'
class TestCreateHeaders():
c = tcp.TCPClient(("127.0.0.1", 0))
def test_create_headers(self):
headers = [
(b':method', b'GET'),
(b':path', b'index.html'),
(b':scheme', b'https'),
(b'foo', b'bar')]
bytes = http2.HTTP2Protocol(self.c)._create_headers(
headers, 1, end_stream=True)
assert b''.join(bytes) ==\
'000014010500000001824488355217caf3a69a3f87408294e7838c767f'\
.decode('hex')
bytes = http2.HTTP2Protocol(self.c)._create_headers(
headers, 1, end_stream=False)
assert b''.join(bytes) ==\
'000014010400000001824488355217caf3a69a3f87408294e7838c767f'\
.decode('hex')
# TODO: add test for too large header_block_fragments
class TestCreateBody():
c = tcp.TCPClient(("127.0.0.1", 0))
protocol = http2.HTTP2Protocol(c)
def test_create_body_empty(self):
bytes = self.protocol._create_body(b'', 1)
assert b''.join(bytes) == ''.decode('hex')
def test_create_body_single_frame(self):
bytes = self.protocol._create_body('foobar', 1)
assert b''.join(bytes) == '000006000100000001666f6f626172'.decode('hex')
def test_create_body_multiple_frames(self):
pass
# bytes = self.protocol._create_body('foobar' * 3000, 1)
# TODO: add test for too large frames
class TestCreateRequest():
c = tcp.TCPClient(("127.0.0.1", 0))
def test_create_request_simple(self):
bytes = http2.HTTP2Protocol(self.c).create_request('GET', '/')
assert len(bytes) == 1
assert bytes[0] == '000003010500000001828487'.decode('hex')
def test_create_request_with_body(self):
bytes = http2.HTTP2Protocol(self.c).create_request(
'GET', '/', [(b'foo', b'bar')], 'foobar')
assert len(bytes) == 2
assert bytes[0] ==\
'00000b010400000001828487408294e7838c767f'.decode('hex')
assert bytes[1] ==\
'000006000100000001666f6f626172'.decode('hex')
class TestReadResponse(test.ServerTestBase):
class handler(tcp.BaseHandler):
def handle(self):
self.wfile.write(
b'00000801040000000188628594e78c767f'.decode('hex'))
self.wfile.write(
b'000006000100000001666f6f626172'.decode('hex'))
self.wfile.flush()
ssl = True
def test_read_response(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
c.connect()
c.convert_to_ssl()
protocol = http2.HTTP2Protocol(c)
status, headers, body = protocol.read_response()
assert headers == {':status': '200', 'etag': 'foobar'}
assert status == '200'
assert body == b'foobar'