Merge pull request #1523 from Kriechi/http2

improve error handling in http2

10 files changed, 112 insertions, 70 deletions

diff --git a/.appveyor.yml b/.appveyor.yml
index 13782ee8..1acc5abb 100644
--- a/.appveyor.yml
+++ b/.appveyor.yml
@@ -25,7 +25,7 @@ install:
   - "pip install -U tox"
 
 test_script:
-  - ps: "tox -- --cov netlib --cov mitmproxy --cov pathod | Select-String -NotMatch Cryptography_locking_cb"
+  - ps: "tox -- --cov netlib --cov mitmproxy --cov pathod"
 
 deploy_script:
   ps: |
diff --git a/mitmproxy/exceptions.py b/mitmproxy/exceptions.py
index 94876514..4d1d90c3 100644
--- a/mitmproxy/exceptions.py
+++ b/mitmproxy/exceptions.py
@@ -62,6 +62,10 @@ class Http2ProtocolException(ProtocolException):
     pass
 
 
+class Http2ZombieException(ProtocolException):
+    pass
+
+
 class ServerException(ProxyException):
     pass
 
diff --git a/mitmproxy/protocol/http2.py b/mitmproxy/protocol/http2.py
index 0e42d619..1595fb61 100644
--- a/mitmproxy/protocol/http2.py
+++ b/mitmproxy/protocol/http2.py
@@ -96,15 +96,17 @@ class Http2Layer(base.Layer):
         self.server_to_client_stream_ids = dict([(0, 0)])
         self.client_conn.h2 = SafeH2Connection(self.client_conn, client_side=False, header_encoding=False)
 
-        # make sure that we only pass actual SSL.Connection objects in here,
-        # because otherwise ssl_read_select fails!
-        self.active_conns = [self.client_conn.connection]
-
     def _initiate_server_conn(self):
-        self.server_conn.h2 = SafeH2Connection(self.server_conn, client_side=True, header_encoding=False)
-        self.server_conn.h2.initiate_connection()
-        self.server_conn.send(self.server_conn.h2.data_to_send())
-        self.active_conns.append(self.server_conn.connection)
+        if self.server_conn:
+            self.server_conn.h2 = SafeH2Connection(self.server_conn, client_side=True, header_encoding=False)
+            self.server_conn.h2.initiate_connection()
+            self.server_conn.send(self.server_conn.h2.data_to_send())
+
+    def _complete_handshake(self):
+        preamble = self.client_conn.rfile.read(24)
+        self.client_conn.h2.initiate_connection()
+        self.client_conn.h2.receive_data(preamble)
+        self.client_conn.send(self.client_conn.h2.data_to_send())
 
     def next_layer(self):  # pragma: no cover
         # WebSockets over HTTP/2?
@@ -126,7 +128,7 @@ class Http2Layer(base.Layer):
                 eid = event.stream_id
 
         if isinstance(event, events.RequestReceived):
-            return self._handle_request_received(eid, event)
+            return self._handle_request_received(eid, event, source_conn.h2)
         elif isinstance(event, events.ResponseReceived):
             return self._handle_response_received(eid, event)
         elif isinstance(event, events.DataReceived):
@@ -138,9 +140,9 @@ class Http2Layer(base.Layer):
         elif isinstance(event, events.RemoteSettingsChanged):
             return self._handle_remote_settings_changed(event, other_conn)
         elif isinstance(event, events.ConnectionTerminated):
-            return self._handle_connection_terminated(event)
+            return self._handle_connection_terminated(event, is_server)
         elif isinstance(event, events.PushedStreamReceived):
-            return self._handle_pushed_stream_received(event)
+            return self._handle_pushed_stream_received(event, source_conn.h2)
         elif isinstance(event, events.PriorityUpdated):
             return self._handle_priority_updated(eid, event)
         elif isinstance(event, events.TrailersReceived):
@@ -149,9 +151,9 @@ class Http2Layer(base.Layer):
         # fail-safe for unhandled events
         return True
 
-    def _handle_request_received(self, eid, event):
+    def _handle_request_received(self, eid, event, h2_connection):
         headers = netlib.http.Headers([[k, v] for k, v in event.headers])
-        self.streams[eid] = Http2SingleStreamLayer(self, eid, headers)
+        self.streams[eid] = Http2SingleStreamLayer(self, h2_connection, eid, headers)
         self.streams[eid].timestamp_start = time.time()
         self.streams[eid].no_body = (event.stream_ended is not None)
         if event.priority_updated is not None:
@@ -173,7 +175,7 @@ class Http2Layer(base.Layer):
     def _handle_data_received(self, eid, event, source_conn):
         bsl = self.config.options.body_size_limit
         if bsl and self.streams[eid].queued_data_length > bsl:
-            self.streams[eid].zombie = time.time()
+            self.streams[eid].kill()
             source_conn.h2.safe_reset_stream(
                 event.stream_id,
                 h2.errors.REFUSED_STREAM
@@ -194,7 +196,7 @@ class Http2Layer(base.Layer):
         return True
 
     def _handle_stream_reset(self, eid, event, is_server, other_conn):
-        self.streams[eid].zombie = time.time()
+        self.streams[eid].kill()
         if eid in self.streams and event.error_code == h2.errors.CANCEL:
             if is_server:
                 other_stream_id = self.streams[eid].client_stream_id
@@ -209,7 +211,13 @@ class Http2Layer(base.Layer):
         other_conn.h2.safe_update_settings(new_settings)
         return True
 
-    def _handle_connection_terminated(self, event):
+    def _handle_connection_terminated(self, event, is_server):
+        self.log("HTTP/2 connection terminated by {}: error code: {}, last stream id: {}, additional data: {}".format(
+            "server" if is_server else "client",
+            event.error_code,
+            event.last_stream_id,
+            event.additional_data), "info")
+
         if event.error_code != h2.errors.NO_ERROR:
             # Something terrible has happened - kill everything!
             self.client_conn.h2.close_connection(
@@ -226,7 +234,7 @@ class Http2Layer(base.Layer):
             """
         return False
 
-    def _handle_pushed_stream_received(self, event):
+    def _handle_pushed_stream_received(self, event, h2_connection):
         # pushed stream ids should be unique and not dependent on race conditions
         # only the parent stream id must be looked up first
         parent_eid = self.server_to_client_stream_ids[event.parent_stream_id]
@@ -235,7 +243,7 @@ class Http2Layer(base.Layer):
             self.client_conn.send(self.client_conn.h2.data_to_send())
 
         headers = netlib.http.Headers([[k, v] for k, v in event.headers])
-        self.streams[event.pushed_stream_id] = Http2SingleStreamLayer(self, event.pushed_stream_id, headers)
+        self.streams[event.pushed_stream_id] = Http2SingleStreamLayer(self, h2_connection, event.pushed_stream_id, headers)
         self.streams[event.pushed_stream_id].timestamp_start = time.time()
         self.streams[event.pushed_stream_id].pushed = True
         self.streams[event.pushed_stream_id].parent_stream_id = parent_eid
@@ -253,7 +261,7 @@ class Http2Layer(base.Layer):
         with self.server_conn.h2.lock:
             mapped_stream_id = event.stream_id
             if mapped_stream_id in self.streams and self.streams[mapped_stream_id].server_stream_id:
-                # if the stream is already up and running and was sent to the server
+                # if the stream is already up and running and was sent to the server,
                 # use the mapped server stream id to update priority information
                 mapped_stream_id = self.streams[mapped_stream_id].server_stream_id
 
@@ -294,37 +302,36 @@ class Http2Layer(base.Layer):
 
     def _kill_all_streams(self):
         for stream in self.streams.values():
-            if not stream.zombie:
-                stream.zombie = time.time()
-                stream.request_data_finished.set()
-                stream.response_arrived.set()
-                stream.data_finished.set()
+            stream.kill()
 
     def __call__(self):
-        if self.server_conn:
-            self._initiate_server_conn()
+        self._initiate_server_conn()
+        self._complete_handshake()
 
-        preamble = self.client_conn.rfile.read(24)
-        self.client_conn.h2.initiate_connection()
-        self.client_conn.h2.receive_data(preamble)
-        self.client_conn.send(self.client_conn.h2.data_to_send())
+        client = self.client_conn.connection
+        server = self.server_conn.connection
+        conns = [client, server]
 
         try:
             while True:
-                r = tcp.ssl_read_select(self.active_conns, 1)
+                r = tcp.ssl_read_select(conns, 1)
                 for conn in r:
-                    source_conn = self.client_conn if conn == self.client_conn.connection else self.server_conn
-                    other_conn = self.server_conn if conn == self.client_conn.connection else self.client_conn
+                    source_conn = self.client_conn if conn == client else self.server_conn
+                    other_conn = self.server_conn if conn == client else self.client_conn
                     is_server = (conn == self.server_conn.connection)
 
                     with source_conn.h2.lock:
                         try:
-                            raw_frame = b''.join(http2.framereader.http2_read_raw_frame(source_conn.rfile))
+                            raw_frame = b''.join(http2.read_raw_frame(source_conn.rfile))
                         except:
                             # read frame failed: connection closed
                             self._kill_all_streams()
                             return
 
+                        if source_conn.h2.state_machine.state == h2.connection.ConnectionState.CLOSED:
+                            self.log("HTTP/2 connection entered closed state already", "debug")
+                            return
+
                         incoming_events = source_conn.h2.receive_data(raw_frame)
                         source_conn.send(source_conn.h2.data_to_send())
 
@@ -354,10 +361,11 @@ def detect_zombie_stream(func):
 
 class Http2SingleStreamLayer(http._HttpTransmissionLayer, basethread.BaseThread):
 
-    def __init__(self, ctx, stream_id, request_headers):
+    def __init__(self, ctx, h2_connection, stream_id, request_headers):
         super(Http2SingleStreamLayer, self).__init__(
             ctx, name="Http2SingleStreamLayer-{}".format(stream_id)
         )
+        self.h2_connection = h2_connection
         self.zombie = None
         self.client_stream_id = stream_id
         self.server_stream_id = None
@@ -365,6 +373,9 @@ class Http2SingleStreamLayer(http._HttpTransmissionLayer, basethread.BaseThread)
         self.response_headers = None
         self.pushed = False
 
+        self.timestamp_start = None
+        self.timestamp_end = None
+
         self.request_data_queue = queue.Queue()
         self.request_queued_data_length = 0
         self.request_data_finished = threading.Event()
@@ -381,6 +392,13 @@ class Http2SingleStreamLayer(http._HttpTransmissionLayer, basethread.BaseThread)
         self.priority_weight = None
         self.handled_priority_event = None
 
+    def kill(self):
+        if not self.zombie:
+            self.zombie = time.time()
+            self.request_data_finished.set()
+            self.response_arrived.set()
+            self.response_data_finished.set()
+
     def connect(self):  # pragma: no cover
         raise exceptions.Http2ProtocolException("HTTP2 layer should already have a connection.")
 
@@ -421,10 +439,11 @@ class Http2SingleStreamLayer(http._HttpTransmissionLayer, basethread.BaseThread)
         self.request_queued_data_length = v
 
     def raise_zombie(self, pre_command=None):
-        if self.zombie is not None:
+        connection_closed = self.h2_connection.state_machine.state == h2.connection.ConnectionState.CLOSED
+        if self.zombie is not None or not hasattr(self.server_conn, 'h2') or connection_closed:
             if pre_command is not None:
                 pre_command()
-            raise exceptions.Http2ProtocolException("Zombie Stream")
+            raise exceptions.Http2ZombieException("Connection already dead")
 
     @detect_zombie_stream
     def read_request(self):
@@ -508,6 +527,7 @@ class Http2SingleStreamLayer(http._HttpTransmissionLayer, basethread.BaseThread)
         except Exception as e:  # pragma: no cover
             raise e
         finally:
+            self.raise_zombie()
             self.server_conn.h2.lock.release()
 
         if not self.no_body:
@@ -581,6 +601,8 @@ class Http2SingleStreamLayer(http._HttpTransmissionLayer, basethread.BaseThread)
 
         try:
             layer()
+        except exceptions.Http2ZombieException as e:  # pragma: no cover
+            pass
         except exceptions.ProtocolException as e:  # pragma: no cover
             self.log(repr(e), "info")
             self.log(traceback.format_exc(), "debug")
@@ -589,5 +611,4 @@ class Http2SingleStreamLayer(http._HttpTransmissionLayer, basethread.BaseThread)
         except exceptions.Kill:
             self.log("Connection killed", "info")
 
-        if not self.zombie:
-            self.zombie = time.time()
+        self.kill()
diff --git a/netlib/debug.py b/netlib/debug.py
index 29c7f655..f9c700de 100644
--- a/netlib/debug.py
+++ b/netlib/debug.py
@@ -37,7 +37,7 @@ def sysinfo():
     return "\n".join(data)
 
 
-def dump_info(sig, frm, file=sys.stdout):  # pragma: no cover
+def dump_info(signal=None, frame=None, file=sys.stdout):  # pragma: no cover
     print("****************************************************", file=file)
     print("Summary", file=file)
     print("=======", file=file)
@@ -81,7 +81,7 @@ def dump_info(sig, frm, file=sys.stdout):  # pragma: no cover
     print("****************************************************", file=file)
 
 
-def dump_stacks(signal, frame, file=sys.stdout):
+def dump_stacks(signal=None, frame=None, file=sys.stdout):
     id2name = dict([(th.ident, th.name) for th in threading.enumerate()])
     code = []
     for threadId, stack in sys._current_frames().items():
diff --git a/netlib/http/http2/__init__.py b/netlib/http/http2/__init__.py
index 60064190..7f84a1ab 100644
--- a/netlib/http/http2/__init__.py
+++ b/netlib/http/http2/__init__.py
@@ -1,8 +1,10 @@
 from __future__ import absolute_import, print_function, division
-from netlib.http.http2 import framereader
+
+from netlib.http.http2.framereader import read_raw_frame, parse_frame
 from netlib.http.http2.utils import parse_headers
 
 __all__ = [
-    "framereader",
+    "read_raw_frame",
+    "parse_frame",
     "parse_headers",
 ]
diff --git a/netlib/http/http2/framereader.py b/netlib/http/http2/framereader.py
index eb9b069a..8b7cfffb 100644
--- a/netlib/http/http2/framereader.py
+++ b/netlib/http/http2/framereader.py
@@ -4,7 +4,7 @@ import hyperframe
 from ...exceptions import HttpException
 
 
-def http2_read_raw_frame(rfile):
+def read_raw_frame(rfile):
     header = rfile.safe_read(9)
     length = int(codecs.encode(header[:3], 'hex_codec'), 16)
 
@@ -15,8 +15,11 @@ def http2_read_raw_frame(rfile):
     return [header, body]
 
 
-def http2_read_frame(rfile):
-    header, body = http2_read_raw_frame(rfile)
+def parse_frame(header, body=None):
+    if body is None:
+        body = header[9:]
+        header = header[:9]
+
     frame, length = hyperframe.frame.Frame.parse_frame_header(header)
     frame.parse_body(memoryview(body))
     return frame
diff --git a/pathod/protocols/http2.py b/pathod/protocols/http2.py
index 5ad120de..a2aa91b4 100644
--- a/pathod/protocols/http2.py
+++ b/pathod/protocols/http2.py
@@ -254,7 +254,7 @@ class HTTP2StateProtocol(object):
 
     def read_frame(self, hide=False):
         while True:
-            frm = http2.framereader.http2_read_frame(self.tcp_handler.rfile)
+            frm = http2.parse_frame(*http2.read_raw_frame(self.tcp_handler.rfile))
             if not hide and self.dump_frames:  # pragma no cover
                 print(frm.human_readable("<<"))
 
diff --git a/setup.py b/setup.py
index b86a48c0..f54d9037 100644
--- a/setup.py
+++ b/setup.py
@@ -107,6 +107,7 @@ setup(
             "pytest-cov>=2.2.1, <3",
             "pytest-timeout>=1.0.0, <2",
             "pytest-xdist>=1.14, <2",
+            "pytest-faulthandler>=1.3.0, <2",
             "sphinx>=1.3.5, <1.5",
             "sphinx-autobuild>=0.5.2, <0.7",
             "sphinxcontrib-documentedlist>=0.4.0, <0.5",
diff --git a/test/mitmproxy/protocol/test_http2.py b/test/mitmproxy/protocol/test_http2.py
index 1eabebf1..c4bd2049 100644
--- a/test/mitmproxy/protocol/test_http2.py
+++ b/test/mitmproxy/protocol/test_http2.py
@@ -15,7 +15,7 @@ from mitmproxy.proxy.config import ProxyConfig
 import netlib
 from ...netlib import tservers as netlib_tservers
 from netlib.exceptions import HttpException
-from netlib.http.http2 import framereader
+from netlib.http import http1, http2
 
 from .. import tservers
 
@@ -33,6 +33,11 @@ requires_alpn = pytest.mark.skipif(
     reason='requires OpenSSL with ALPN support')
 
 
+# inspect the log:
+#   for msg in self.proxy.tmaster.tlog:
+#       print(msg)
+
+
 class _Http2ServerBase(netlib_tservers.ServerTestBase):
     ssl = dict(alpn_select=b'h2')
 
@@ -55,7 +60,7 @@ class _Http2ServerBase(netlib_tservers.ServerTestBase):
             done = False
             while not done:
                 try:
-                    raw = b''.join(framereader.http2_read_raw_frame(self.rfile))
+                    raw = b''.join(http2.read_raw_frame(self.rfile))
                     events = h2_conn.receive_data(raw)
                 except HttpException:
                     print(traceback.format_exc())
@@ -124,11 +129,17 @@ class _Http2TestBase(object):
         client.connect()
 
         # send CONNECT request
-        client.wfile.write(
-            b"CONNECT localhost:%d HTTP/1.1\r\n"
-            b"Host: localhost:%d\r\n"
-            b"\r\n" % (self.server.server.address.port, self.server.server.address.port)
-        )
+        client.wfile.write(http1.assemble_request(netlib.http.Request(
+            'authority',
+            b'CONNECT',
+            b'',
+            b'localhost',
+            self.server.server.address.port,
+            b'/',
+            b'HTTP/1.1',
+            [(b'host', b'localhost:%d' % self.server.server.address.port)],
+            b'',
+        )))
         client.wfile.flush()
 
         # read CONNECT response
@@ -242,7 +253,7 @@ class TestSimple(_Http2Test):
         done = False
         while not done:
             try:
-                raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+                raw = b''.join(http2.read_raw_frame(client.rfile))
                 events = h2_conn.receive_data(raw)
             except HttpException:
                 print(traceback.format_exc())
@@ -319,7 +330,7 @@ class TestRequestWithPriority(_Http2Test):
         done = False
         while not done:
             try:
-                raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+                raw = b''.join(http2.read_raw_frame(client.rfile))
                 events = h2_conn.receive_data(raw)
             except HttpException:
                 print(traceback.format_exc())
@@ -358,7 +369,7 @@ class TestRequestWithPriority(_Http2Test):
         done = False
         while not done:
             try:
-                raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+                raw = b''.join(http2.read_raw_frame(client.rfile))
                 events = h2_conn.receive_data(raw)
             except HttpException:
                 print(traceback.format_exc())
@@ -430,7 +441,7 @@ class TestPriority(_Http2Test):
         done = False
         while not done:
             try:
-                raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+                raw = b''.join(http2.read_raw_frame(client.rfile))
                 events = h2_conn.receive_data(raw)
             except HttpException:
                 print(traceback.format_exc())
@@ -507,7 +518,7 @@ class TestPriorityWithExistingStream(_Http2Test):
         done = False
         while not done:
             try:
-                raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+                raw = b''.join(http2.read_raw_frame(client.rfile))
                 events = h2_conn.receive_data(raw)
             except HttpException:
                 print(traceback.format_exc())
@@ -558,7 +569,7 @@ class TestStreamResetFromServer(_Http2Test):
         done = False
         while not done:
             try:
-                raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+                raw = b''.join(http2.read_raw_frame(client.rfile))
                 events = h2_conn.receive_data(raw)
             except HttpException:
                 print(traceback.format_exc())
@@ -608,7 +619,7 @@ class TestBodySizeLimit(_Http2Test):
         done = False
         while not done:
             try:
-                raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+                raw = b''.join(http2.read_raw_frame(client.rfile))
                 events = h2_conn.receive_data(raw)
             except HttpException:
                 print(traceback.format_exc())
@@ -693,7 +704,7 @@ class TestPushPromise(_Http2Test):
         responses = 0
         while not done:
             try:
-                raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+                raw = b''.join(http2.read_raw_frame(client.rfile))
                 events = h2_conn.receive_data(raw)
             except HttpException:
                 print(traceback.format_exc())
@@ -746,7 +757,7 @@ class TestPushPromise(_Http2Test):
         responses = 0
         while not done:
             try:
-                raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+                raw = b''.join(http2.read_raw_frame(client.rfile))
                 events = h2_conn.receive_data(raw)
             except HttpException:
                 print(traceback.format_exc())
@@ -806,7 +817,7 @@ class TestConnectionLost(_Http2Test):
         done = False
         while not done:
             try:
-                raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+                raw = b''.join(http2.read_raw_frame(client.rfile))
                 h2_conn.receive_data(raw)
             except HttpException:
                 print(traceback.format_exc())
@@ -863,7 +874,7 @@ class TestMaxConcurrentStreams(_Http2Test):
         ended_streams = 0
         while ended_streams != len(new_streams):
             try:
-                header, body = framereader.http2_read_raw_frame(client.rfile)
+                header, body = http2.read_raw_frame(client.rfile)
                 events = h2_conn.receive_data(b''.join([header, body]))
             except:
                 break
@@ -909,7 +920,7 @@ class TestConnectionTerminated(_Http2Test):
         connection_terminated_event = None
         while not done:
             try:
-                raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+                raw = b''.join(http2.read_raw_frame(client.rfile))
                 events = h2_conn.receive_data(raw)
                 for event in events:
                     if isinstance(event, h2.events.ConnectionTerminated):
diff --git a/test/pathod/test_protocols_http2.py b/test/pathod/test_protocols_http2.py
index 8d7efc82..7f65c0eb 100644
--- a/test/pathod/test_protocols_http2.py
+++ b/test/pathod/test_protocols_http2.py
@@ -5,7 +5,7 @@ import hyperframe
 from netlib import tcp, http
 from netlib.tutils import raises
 from netlib.exceptions import TcpDisconnect
-from netlib.http.http2 import framereader
+from netlib.http import http2
 
 from ..netlib import tservers as netlib_tservers
 
@@ -112,11 +112,11 @@ class TestPerformServerConnectionPreface(netlib_tservers.ServerTestBase):
             self.wfile.flush()
 
             # check empty settings frame
-            raw = framereader.http2_read_raw_frame(self.rfile)
+            raw = http2.read_raw_frame(self.rfile)
             assert raw == codecs.decode('00000c040000000000000200000000000300000001', 'hex_codec')
 
             # check settings acknowledgement
-            raw = framereader.http2_read_raw_frame(self.rfile)
+            raw = http2.read_raw_frame(self.rfile)
             assert raw == codecs.decode('000000040100000000', 'hex_codec')
 
             # send settings acknowledgement