From e9eed5e4c206e507665f5a4ff92654e969f4dd89 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Tue, 29 Sep 2015 16:23:55 +0200
Subject: --wip--

---
 libmproxy/models/http.py   |   4 -
 libmproxy/protocol/http.py | 420 +++++++++++++++++++++++++++++++--------------
 requirements.txt           |   2 +-
 setup.py                   |   1 +
 4 files changed, 296 insertions(+), 131 deletions(-)

diff --git a/libmproxy/models/http.py b/libmproxy/models/http.py
index e07dff69..3914440e 100644
--- a/libmproxy/models/http.py
+++ b/libmproxy/models/http.py
@@ -241,8 +241,6 @@ class HTTPRequest(MessageMixin, Request):
             timestamp_end=request.timestamp_end,
             form_out=(request.form_out if hasattr(request, 'form_out') else None),
         )
-        if hasattr(request, 'stream_id'):
-            req.stream_id = request.stream_id
         return req
 
     def __hash__(self):
@@ -347,8 +345,6 @@ class HTTPResponse(MessageMixin, Response):
             timestamp_start=response.timestamp_start,
             timestamp_end=response.timestamp_end,
         )
-        if hasattr(response, 'stream_id'):
-            resp.stream_id = response.stream_id
         return resp
 
     def _refresh_cookie(self, c, delta):
diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 12d09e71..fc045e0d 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -1,27 +1,37 @@
 from __future__ import (absolute_import, print_function, division)
+
 import sys
 import traceback
-
 import six
+import struct
+import threading
+import Queue
 
 from netlib import tcp
 from netlib.exceptions import HttpException, HttpReadDisconnect, NetlibException
-from netlib.http import http1, Headers
-from netlib.http import CONTENT_MISSING
-from netlib.tcp import Address
-from netlib.http.http2.connections import HTTP2Protocol
-from netlib.http.http2.frame import GoAwayFrame, PriorityFrame, WindowUpdateFrame
+from netlib.http import http1, Headers, CONTENT_MISSING
+from netlib.tcp import Address, ssl_read_select
+
+import h2
+from h2.connection import H2Connection
+from h2.events import *
+from hyperframe import frame
+
+from .base import Layer, Kill
 from .. import utils
 from ..exceptions import HttpProtocolException, ProtocolException
 from ..models import (
-    HTTPFlow, HTTPRequest, HTTPResponse, make_error_response, make_connect_response, Error, expect_continue_response
+    HTTPFlow,
+    HTTPRequest,
+    HTTPResponse,
+    make_error_response,
+    make_connect_response,
+    Error,
+    expect_continue_response
 )
-from .base import Layer, Kill
 
 
 class _HttpLayer(Layer):
-    supports_streaming = False
-
     def read_request(self):
         raise NotImplementedError()
 
@@ -31,26 +41,6 @@ class _HttpLayer(Layer):
     def send_request(self, request):
         raise NotImplementedError()
 
-    def read_response(self, request):
-        raise NotImplementedError()
-
-    def send_response(self, response):
-        raise NotImplementedError()
-
-    def check_close_connection(self, flow):
-        raise NotImplementedError()
-
-
-class _StreamingHttpLayer(_HttpLayer):
-    supports_streaming = True
-
-    def read_response_headers(self):
-        raise NotImplementedError
-
-    def read_response_body(self, request, response):
-        raise NotImplementedError()
-        yield "this is a generator"  # pragma: no cover
-
     def read_response(self, request):
         response = self.read_response_headers()
         response.data.content = b"".join(
@@ -58,11 +48,12 @@ class _StreamingHttpLayer(_HttpLayer):
         )
         return response
 
-    def send_response_headers(self, response):
-        raise NotImplementedError
+    def read_response_headers(self):
+        raise NotImplementedError()
 
-    def send_response_body(self, response, chunks):
+    def read_response_body(self, request, response):
         raise NotImplementedError()
+        yield "this is a generator"  # pragma: no cover
 
     def send_response(self, response):
         if response.content == CONTENT_MISSING:
@@ -70,9 +61,17 @@ class _StreamingHttpLayer(_HttpLayer):
         self.send_response_headers(response)
         self.send_response_body(response, [response.content])
 
+    def send_response_headers(self, response):
+        raise NotImplementedError()
+
+    def send_response_body(self, response, chunks):
+        raise NotImplementedError()
+
+    def check_close_connection(self, flow):
+        raise NotImplementedError()
 
-class Http1Layer(_StreamingHttpLayer):
 
+class Http1Layer(_HttpLayer):
     def __init__(self, ctx, mode):
         super(Http1Layer, self).__init__(ctx)
         self.mode = mode
@@ -130,104 +129,277 @@ class Http1Layer(_StreamingHttpLayer):
         layer = HttpLayer(self, self.mode)
         layer()
 
-
-# TODO: The HTTP2 layer is missing multiplexing, which requires a major rewrite.
-class Http2Layer(_HttpLayer):
-
+class SafeH2Connection(H2Connection):
+    def __init__(self, conn, *args, **kwargs):
+        super(SafeH2Connection, self).__init__(*args, **kwargs)
+        self.conn = conn
+        self.lock = threading.RLock()
+
+    def safe_increment_flow_control(self, stream_id, length):
+        with self.lock:
+            self.increment_flow_control_window(length)
+            self.conn.send(self.data_to_send())
+        with self.lock:
+            if stream_id in self.streams and not self.streams[stream_id].closed:
+                self.increment_flow_control_window(length, stream_id=stream_id)
+                self.conn.send(self.data_to_send())
+
+    def safe_reset_stream(self, stream_id, error_code):
+        with self.lock:
+            self.reset_stream(stream_id, error_code)
+            self.conn.send(self.h2.data_to_send())
+
+    def safe_acknowledge_settings(self, event):
+        with self.conn.h2.lock:
+            self.conn.h2.acknowledge_settings(event)
+            self.conn.send(self.data_to_send())
+
+    def safe_update_settings(self, new_settings):
+        with self.conn.h2.lock:
+            self.update_settings(new_settings)
+            self.conn.send(self.data_to_send())
+
+    def safe_send_headers(self, stream_id, headers):
+        with self.lock:
+            self.send_headers(stream_id, headers)
+            self.conn.send(self.data_to_send())
+
+    def safe_send_body(self, stream_id, chunks):
+        for chunk in chunks:
+            max_outbound_frame_size = self.max_outbound_frame_size
+            for i in xrange(0, len(chunk), max_outbound_frame_size):
+                frame_chunk = chunk[i:i+max_outbound_frame_size]
+                with self.lock:
+                    self.send_data(stream_id, frame_chunk)
+                    self.conn.send(self.data_to_send())
+        with self.lock:
+            self.end_stream(stream_id)
+            self.conn.send(self.data_to_send())
+
+class Http2Layer(Layer):
     def __init__(self, ctx, mode):
         super(Http2Layer, self).__init__(ctx)
         self.mode = mode
-        self.client_protocol = HTTP2Protocol(self.client_conn, is_server=True,
-                                             unhandled_frame_cb=self.handle_unexpected_frame_from_client)
-        self.server_protocol = HTTP2Protocol(self.server_conn, is_server=False,
-                                             unhandled_frame_cb=self.handle_unexpected_frame_from_server)
+        self.streams = dict()
+        self.server_to_client_stream_ids = dict([(0, 0)])
+        self.client_conn.h2 = SafeH2Connection(self.client_conn, client_side=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]
+
+        if self.server_conn:
+            self._initiate_server_conn()
+
+    def _initiate_server_conn(self):
+        self.server_conn.h2 = SafeH2Connection(self.server_conn, client_side=True)
+        self.server_conn.h2.initiate_connection()
+        self.server_conn.h2.update_settings({frame.SettingsFrame.ENABLE_PUSH: False})
+        self.server_conn.send(self.server_conn.h2.data_to_send())
+        self.active_conns.append(self.server_conn.connection)
+
+    def connect(self):
+        self.ctx.connect()
+        self.server_conn.connect()
+        self._initiate_server_conn()
+
+    def set_server(self):
+        raise NotImplementedError("Cannot change server for HTTP2 connections.")
+
+    def disconnect(self):
+        raise NotImplementedError("Cannot dis- or reconnect in HTTP2 connections.")
+
+    def __call__(self):
+        preamble = self.client_conn.rfile.read(24)
+        self.client_conn.h2.initiate_connection()
+        self.client_conn.h2.update_settings({frame.SettingsFrame.ENABLE_PUSH: False})
+        self.client_conn.h2.receive_data(preamble)
+        self.client_conn.send(self.client_conn.h2.data_to_send())
+
+        while True:
+            r = ssl_read_select(self.active_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
+                is_server = (conn == self.server_conn.connection)
+
+                fields = struct.unpack("!HB", source_conn.rfile.peek(3))
+                length = (fields[0] << 8) + fields[1]
+                raw_frame = source_conn.rfile.safe_read(9 + length)
+
+                with source_conn.h2.lock:
+                    events = source_conn.h2.receive_data(raw_frame)
+                    source_conn.send(source_conn.h2.data_to_send())
+
+                for event in events:
+                    if hasattr(event, 'stream_id'):
+                        if is_server:
+                            eid = self.server_to_client_stream_ids[event.stream_id]
+                        else:
+                            eid = event.stream_id
+
+                    if isinstance(event, RequestReceived):
+                        headers = Headers([[str(k), str(v)] for k, v in event.headers])
+                        self.streams[eid] = Http2SingleStreamLayer(self, eid, headers)
+                        self.streams[eid].start()
+                    elif isinstance(event, ResponseReceived):
+                        headers = Headers([[str(k), str(v)] for k, v in event.headers])
+                        self.streams[eid].response_headers = headers
+                        self.streams[eid].response_arrived.set()
+                    elif isinstance(event, DataReceived):
+                        self.streams[eid].data_queue.put(event.data)
+                        source_conn.h2.safe_increment_flow_control(event.stream_id, len(event.data))
+                    elif isinstance(event, StreamEnded):
+                        self.streams[eid].data_finished.set()
+                    elif isinstance(event, StreamReset):
+                        self.streams[eid].zombie = True
+                        if eid in self.streams and event.error_code == 0x8:
+                            if is_server:
+                                other_stream_id = self.streams[eid].client_stream_id
+                            else:
+                                other_stream_id = self.streams[eid].server_stream_id
+                            other_conn.h2.safe_reset_stream(other_stream_id, event.error_code)
+                    elif isinstance(event, RemoteSettingsChanged):
+                        source_conn.h2.safe_acknowledge_settings(event)
+                        new_settings = dict([(id, cs.new_value) for (id, cs) in event.changed_settings.iteritems()])
+                        other_conn.h2.safe_update_settings(new_settings)
+
+            # TODO: cleanup resources once we are sure nobody needs them
+            # for stream_id in self.streams.keys():
+            #     if self.streams[stream_id].zombie:
+            #         self.streams.pop(stream_id, None)
+
+
+class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
+    def __init__(self, ctx, stream_id, request_headers):
+        super(Http2SingleStreamLayer, self).__init__(ctx)
+        self.zombie = False
+        self.client_stream_id = stream_id
+        self.server_stream_id = None
+        self.request_headers = request_headers
+        self.response_headers = None
+        self.data_queue = Queue.Queue()
+
+        self.response_arrived = threading.Event()
+        self.data_finished = threading.Event()
 
     def read_request(self):
-        request = HTTPRequest.from_protocol(
-            self.client_protocol,
-            body_size_limit=self.config.body_size_limit
+        self.data_finished.wait()
+        self.data_finished.clear()
+
+        authority = self.request_headers.get(':authority', '')
+        method = self.request_headers.get(':method', 'GET')
+        scheme = self.request_headers.get(':scheme', 'https')
+        path = self.request_headers.get(':path', '/')
+        host = None
+        port = None
+
+        if path == '*' or path.startswith("/"):
+            form_in = "relative"
+        elif method == 'CONNECT':
+            form_in = "authority"
+            if ":" in authority:
+                host, port = authority.split(":", 1)
+            else:
+                host = authority
+        else:
+            form_in = "absolute"
+            # FIXME: verify if path or :host contains what we need
+            scheme, host, port, _ = utils.parse_url(path)
+
+        if host is None:
+            host = 'localhost'
+        if port is None:
+            port = 80 if scheme == 'http' else 443
+        port = int(port)
+
+        data = []
+        while self.data_queue.qsize() > 0:
+            data.append(self.data_queue.get())
+
+        return HTTPRequest(
+            form_in,
+            method,
+            scheme,
+            host,
+            port,
+            path,
+            (2, 0),
+            self.request_headers,
+            data,
+            # TODO: timestamp_start=None,
+            # TODO: timestamp_end=None,
+            form_out=None, # TODO: (request.form_out if hasattr(request, 'form_out') else None),
         )
-        self._stream_id = request.stream_id
-        return request
 
     def send_request(self, message):
-        # TODO: implement flow control and WINDOW_UPDATE frames
-        self.server_conn.send(self.server_protocol.assemble(message))
+        with self.server_conn.h2.lock:
+            self.server_stream_id = self.server_conn.h2.get_next_available_stream_id()
+            self.server_to_client_stream_ids[self.server_stream_id] = self.client_stream_id
 
-    def read_response(self, request):
-        return HTTPResponse.from_protocol(
-            self.server_protocol,
-            request_method=request.method,
-            body_size_limit=self.config.body_size_limit,
-            include_body=True,
-            stream_id=self._stream_id
+            self.server_conn.h2.safe_send_headers(
+                self.server_stream_id,
+                message.headers
+            )
+            self.server_conn.h2.safe_send_body(
+                self.server_stream_id,
+                message.body
+            )
+
+    def read_response_headers(self):
+        self.response_arrived.wait()
+
+        status_code = int(self.response_headers.get(':status', 502))
+
+        return HTTPResponse(
+            http_version=(2, 0),
+            status_code=status_code,
+            reason='',
+            headers=self.response_headers,
+            content=None,
+            # TODO: timestamp_start=response.timestamp_start,
+            # TODO: timestamp_end=response.timestamp_end,
         )
 
-    def send_response(self, message):
-        # TODO: implement flow control to prevent client buffer filling up
-        # maintain a send buffer size, and read WindowUpdateFrames from client to increase the send buffer
-        self.client_conn.send(self.client_protocol.assemble(message))
+    def read_response_body(self, request, response):
+        while True:
+            try:
+                yield self.data_queue.get(timeout=1)
+            except Queue.Empty:
+                pass
+            if self.data_finished.is_set():
+                while self.data_queue.qsize() > 0:
+                    yield self.data_queue.get()
+                return
+
+    def send_response_headers(self, response):
+        self.client_conn.h2.safe_send_headers(
+            self.client_stream_id,
+            response.headers
+        )
+
+    def send_response_body(self, _response, chunks):
+        self.client_conn.h2.safe_send_body(
+            self.client_stream_id,
+            chunks
+        )
 
     def check_close_connection(self, flow):
-        # TODO: add a timer to disconnect after a 10 second timeout
-        return False
+        # This layer only handles a single stream.
+        # RFC 7540 8.1: An HTTP request/response exchange fully consumes a single stream.
+        return True
 
     def connect(self):
-        self.ctx.connect()
-        self.server_protocol = HTTP2Protocol(self.server_conn, is_server=False,
-                                             unhandled_frame_cb=self.handle_unexpected_frame_from_server)
-        self.server_protocol.perform_connection_preface()
+        raise ValueError("CONNECT inside an HTTP2 stream is not supported.")
 
     def set_server(self, *args, **kwargs):
-        self.ctx.set_server(*args, **kwargs)
-        self.server_protocol = HTTP2Protocol(self.server_conn, is_server=False,
-                                             unhandled_frame_cb=self.handle_unexpected_frame_from_server)
-        self.server_protocol.perform_connection_preface()
+        # do not mess with the server connection - all streams share it.
+        pass
 
-    def __call__(self):
-        self.server_protocol.perform_connection_preface()
+    def run(self):
         layer = HttpLayer(self, self.mode)
         layer()
-
-        # terminate the connection
-        self.client_conn.send(GoAwayFrame().to_bytes())
-
-    def handle_unexpected_frame_from_client(self, frame):
-        if isinstance(frame, WindowUpdateFrame):
-            # Clients are sending WindowUpdate frames depending on their flow control algorithm.
-            # Since we cannot predict these frames, and we do not need to respond to them,
-            # simply accept them, and hide them from the log.
-            # Ideally we should keep track of our own flow control window and
-            # stall transmission if the outgoing flow control buffer is full.
-            return
-        if isinstance(frame, PriorityFrame):
-            # Clients are sending Priority frames depending on their implementation.
-            # The RFC does not clearly state when or which priority preferences should be set.
-            # Since we cannot predict these frames, and we do not need to respond to them,
-            # simply accept them, and hide them from the log.
-            # Ideally we should forward them to the server.
-            return
-        if isinstance(frame, GoAwayFrame):
-            # Client wants to terminate the connection,
-            # relay it to the server.
-            self.server_conn.send(frame.to_bytes())
-            return
-        self.log("Unexpected HTTP2 frame from client: %s" % frame.human_readable(), "info")
-
-    def handle_unexpected_frame_from_server(self, frame):
-        if isinstance(frame, WindowUpdateFrame):
-            # Servers are sending WindowUpdate frames depending on their flow control algorithm.
-            # Since we cannot predict these frames, and we do not need to respond to them,
-            # simply accept them, and hide them from the log.
-            # Ideally we should keep track of our own flow control window and
-            # stall transmission if the outgoing flow control buffer is full.
-            return
-        if isinstance(frame, GoAwayFrame):
-            # Server wants to terminate the connection,
-            # relay it to the client.
-            self.client_conn.send(frame.to_bytes())
-            return
-        self.log("Unexpected HTTP2 frame from server: %s" % frame.human_readable(), "info")
+        self.zombie = True
 
 
 class ConnectServerConnection(object):
@@ -420,7 +592,7 @@ class HttpLayer(Layer):
         layer()
 
     def send_response_to_client(self, flow):
-        if not (self.supports_streaming and flow.response.stream):
+        if not flow.response.stream:
             # no streaming:
             # we already received the full response from the server and can
             # send it to the client straight away.
@@ -441,10 +613,7 @@ class HttpLayer(Layer):
     def get_response_from_server(self, flow):
         def get_response():
             self.send_request(flow.request)
-            if self.supports_streaming:
-                flow.response = self.read_response_headers()
-            else:
-                flow.response = self.read_response(flow.request)
+            flow.response = self.read_response_headers()
 
         try:
             get_response()
@@ -474,15 +643,14 @@ class HttpLayer(Layer):
         if flow == Kill:
             raise Kill()
 
-        if self.supports_streaming:
-            if flow.response.stream:
-                flow.response.data.content = CONTENT_MISSING
-            else:
-                flow.response.data.content = b"".join(self.read_response_body(
-                    flow.request,
-                    flow.response
-                ))
-            flow.response.timestamp_end = utils.timestamp()
+        if flow.response.stream:
+            flow.response.data.content = CONTENT_MISSING
+        else:
+            flow.response.data.content = b"".join(self.read_response_body(
+                flow.request,
+                flow.response
+            ))
+        flow.response.timestamp_end = utils.timestamp()
 
         # no further manipulation of self.server_conn beyond this point
         # we can safely set it as the final attribute value here.
diff --git a/requirements.txt b/requirements.txt
index 3832c953..49f86b9a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,3 +1,3 @@
 -e git+https://github.com/mitmproxy/netlib.git#egg=netlib
 -e git+https://github.com/mitmproxy/pathod.git#egg=pathod
--e .[dev,examples,contentviews]
\ No newline at end of file
+-e .[dev,examples,contentviews]
diff --git a/setup.py b/setup.py
index 29ac4753..c7686c6e 100644
--- a/setup.py
+++ b/setup.py
@@ -17,6 +17,7 @@ with open(os.path.join(here, 'README.rst'), encoding='utf-8') as f:
 # This will break `pip install` on systems with old setuptools versions.
 deps = {
     "netlib>=%s, <%s" % (version.MINORVERSION, version.NEXT_MINORVERSION),
+    "h2>=2.0.0",
     "tornado>=4.3.0, <4.4",
     "configargparse>=0.10.0, <0.11",
     "pyperclip>=1.5.22, <1.6",
-- 
cgit v1.2.3


From b44c3ac6e0f54413a067294bbcb0fe019fade3f3 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 14 Jan 2016 19:05:15 +0100
Subject: propagate GoAway to the other side

---
 libmproxy/protocol/http.py | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index fc045e0d..8315aed0 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -135,6 +135,11 @@ class SafeH2Connection(H2Connection):
         self.conn = conn
         self.lock = threading.RLock()
 
+    def safe_close_connection(self, error_code):
+        with self.lock:
+            self.close_connection(error_code)
+            self.conn.send(self.data_to_send())
+
     def safe_increment_flow_control(self, stream_id, length):
         with self.lock:
             self.increment_flow_control_window(length)
@@ -263,6 +268,9 @@ class Http2Layer(Layer):
                         source_conn.h2.safe_acknowledge_settings(event)
                         new_settings = dict([(id, cs.new_value) for (id, cs) in event.changed_settings.iteritems()])
                         other_conn.h2.safe_update_settings(new_settings)
+                    elif isinstance(event, ConnectionTerminated):
+                        other_conn.h2.safe_close_connection(event.error_code)
+                        return
 
             # TODO: cleanup resources once we are sure nobody needs them
             # for stream_id in self.streams.keys():
-- 
cgit v1.2.3


From 9e619742887f686aec1059283316ed389443cdf3 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 14 Jan 2016 19:05:37 +0100
Subject: improve flow control

---
 libmproxy/protocol/http.py | 19 +++++++++++++++----
 1 file changed, 15 insertions(+), 4 deletions(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 8315aed0..7ea5f57d 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -141,6 +141,9 @@ class SafeH2Connection(H2Connection):
             self.conn.send(self.data_to_send())
 
     def safe_increment_flow_control(self, stream_id, length):
+        if length == 0:
+            return
+
         with self.lock:
             self.increment_flow_control_window(length)
             self.conn.send(self.data_to_send())
@@ -152,7 +155,7 @@ class SafeH2Connection(H2Connection):
     def safe_reset_stream(self, stream_id, error_code):
         with self.lock:
             self.reset_stream(stream_id, error_code)
-            self.conn.send(self.h2.data_to_send())
+            self.conn.send(self.data_to_send())
 
     def safe_acknowledge_settings(self, event):
         with self.conn.h2.lock:
@@ -174,9 +177,17 @@ class SafeH2Connection(H2Connection):
             max_outbound_frame_size = self.max_outbound_frame_size
             for i in xrange(0, len(chunk), max_outbound_frame_size):
                 frame_chunk = chunk[i:i+max_outbound_frame_size]
-                with self.lock:
-                    self.send_data(stream_id, frame_chunk)
-                    self.conn.send(self.data_to_send())
+
+                self.lock.acquire()
+                while True:
+                    if self.local_flow_control_window(stream_id) < len(frame_chunk):
+                        self.lock.release()
+                        time.sleep(0)
+                    else:
+                        break
+                self.send_data(stream_id, frame_chunk)
+                self.conn.send(self.data_to_send())
+                self.lock.release()
         with self.lock:
             self.end_stream(stream_id)
             self.conn.send(self.data_to_send())
-- 
cgit v1.2.3


From 986e30fb19dbe97a72540b2849312032a92976f0 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 14 Jan 2016 19:14:05 +0100
Subject: add todo note

---
 libmproxy/protocol/http.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 7ea5f57d..17ddd5dc 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -173,6 +173,8 @@ class SafeH2Connection(H2Connection):
             self.conn.send(self.data_to_send())
 
     def safe_send_body(self, stream_id, chunks):
+        # TODO: this assumes the MAX_FRAME_SIZE does not change in the middle
+        # of a transfer - it could though. Then we need to re-chunk everything.
         for chunk in chunks:
             max_outbound_frame_size = self.max_outbound_frame_size
             for i in xrange(0, len(chunk), max_outbound_frame_size):
-- 
cgit v1.2.3


From 3f44eff1430b092708b127e2647b2f747c306d4a Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 14 Jan 2016 19:47:36 +0100
Subject: --wip--

---
 libmproxy/protocol/http.py | 45 +++++++++++++++++++++++++++++++++------------
 1 file changed, 33 insertions(+), 12 deletions(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 17ddd5dc..75ca520d 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -5,6 +5,7 @@ import traceback
 import six
 import struct
 import threading
+import time
 import Queue
 
 from netlib import tcp
@@ -154,7 +155,11 @@ class SafeH2Connection(H2Connection):
 
     def safe_reset_stream(self, stream_id, error_code):
         with self.lock:
-            self.reset_stream(stream_id, error_code)
+            try:
+                self.reset_stream(stream_id, error_code)
+            except StreamClosedError:
+                # stream is already closed - good
+                pass
             self.conn.send(self.data_to_send())
 
     def safe_acknowledge_settings(self, event):
@@ -270,7 +275,7 @@ class Http2Layer(Layer):
                     elif isinstance(event, StreamEnded):
                         self.streams[eid].data_finished.set()
                     elif isinstance(event, StreamReset):
-                        self.streams[eid].zombie = True
+                        self.streams[eid].zombie = time.time()
                         if eid in self.streams and event.error_code == 0x8:
                             if is_server:
                                 other_stream_id = self.streams[eid].client_stream_id
@@ -284,17 +289,22 @@ class Http2Layer(Layer):
                     elif isinstance(event, ConnectionTerminated):
                         other_conn.h2.safe_close_connection(event.error_code)
                         return
+                    elif isinstance(event, TrailersReceived):
+                        raise NotImplementedError()
+                    elif isinstance(event, PushedStreamReceived):
+                        raise NotImplementedError()
 
-            # TODO: cleanup resources once we are sure nobody needs them
-            # for stream_id in self.streams.keys():
-            #     if self.streams[stream_id].zombie:
-            #         self.streams.pop(stream_id, None)
+            death_time = time.time() - 10
+            for stream_id in self.streams.keys():
+                zombie = self.streams[stream_id].zombie
+                if zombie and zombie <= death_time:
+                    self.streams.pop(stream_id, None)
 
 
 class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
     def __init__(self, ctx, stream_id, request_headers):
         super(Http2SingleStreamLayer, self).__init__(ctx)
-        self.zombie = False
+        self.zombie = None
         self.client_stream_id = stream_id
         self.server_stream_id = None
         self.request_headers = request_headers
@@ -354,6 +364,9 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
         )
 
     def send_request(self, message):
+        if self.zombie:
+            return
+
         with self.server_conn.h2.lock:
             self.server_stream_id = self.server_conn.h2.get_next_available_stream_id()
             self.server_to_client_stream_ids[self.server_stream_id] = self.client_stream_id
@@ -362,10 +375,10 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
                 self.server_stream_id,
                 message.headers
             )
-            self.server_conn.h2.safe_send_body(
-                self.server_stream_id,
-                message.body
-            )
+        self.server_conn.h2.safe_send_body(
+            self.server_stream_id,
+            message.body
+        )
 
     def read_response_headers(self):
         self.response_arrived.wait()
@@ -392,14 +405,22 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
                 while self.data_queue.qsize() > 0:
                     yield self.data_queue.get()
                 return
+            if self.zombie:
+                return
 
     def send_response_headers(self, response):
+        if self.zombie:
+            return
+
         self.client_conn.h2.safe_send_headers(
             self.client_stream_id,
             response.headers
         )
 
     def send_response_body(self, _response, chunks):
+        if self.zombie:
+            return
+
         self.client_conn.h2.safe_send_body(
             self.client_stream_id,
             chunks
@@ -420,7 +441,7 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
     def run(self):
         layer = HttpLayer(self, self.mode)
         layer()
-        self.zombie = True
+        self.zombie = time.time()
 
 
 class ConnectServerConnection(object):
-- 
cgit v1.2.3


From 947f79eb6c173d445c57203d8f301a033176272b Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Sat, 16 Jan 2016 11:31:43 +0100
Subject: improved zombie detection

---
 libmproxy/protocol/http.py | 132 ++++++++++++++++++++++++---------------------
 1 file changed, 71 insertions(+), 61 deletions(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 75ca520d..ea069bcb 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -157,7 +157,7 @@ class SafeH2Connection(H2Connection):
         with self.lock:
             try:
                 self.reset_stream(stream_id, error_code)
-            except StreamClosedError:
+            except h2.exceptions.ProtocolError:
                 # stream is already closed - good
                 pass
             self.conn.send(self.data_to_send())
@@ -172,30 +172,33 @@ class SafeH2Connection(H2Connection):
             self.update_settings(new_settings)
             self.conn.send(self.data_to_send())
 
-    def safe_send_headers(self, stream_id, headers):
+    def safe_send_headers(self, is_zombie, stream_id, headers):
         with self.lock:
+            if is_zombie(self, stream_id):
+                return
             self.send_headers(stream_id, headers)
             self.conn.send(self.data_to_send())
 
-    def safe_send_body(self, stream_id, chunks):
-        # TODO: this assumes the MAX_FRAME_SIZE does not change in the middle
-        # of a transfer - it could though. Then we need to re-chunk everything.
+    def safe_send_body(self, is_zombie, stream_id, chunks):
         for chunk in chunks:
-            max_outbound_frame_size = self.max_outbound_frame_size
-            for i in xrange(0, len(chunk), max_outbound_frame_size):
-                frame_chunk = chunk[i:i+max_outbound_frame_size]
-
+            position = 0
+            while position < len(chunk):
                 self.lock.acquire()
-                while True:
-                    if self.local_flow_control_window(stream_id) < len(frame_chunk):
-                        self.lock.release()
-                        time.sleep(0)
-                    else:
-                        break
+                max_outbound_frame_size = self.max_outbound_frame_size
+                frame_chunk = chunk[position:position+max_outbound_frame_size]
+                if self.local_flow_control_window(stream_id) < len(frame_chunk):
+                    self.lock.release()
+                    time.sleep(0)
+                    continue
+                if is_zombie(self, stream_id):
+                    return
                 self.send_data(stream_id, frame_chunk)
                 self.conn.send(self.data_to_send())
                 self.lock.release()
+                position += max_outbound_frame_size
         with self.lock:
+            if is_zombie(self, stream_id):
+                return
             self.end_stream(stream_id)
             self.conn.send(self.data_to_send())
 
@@ -254,45 +257,45 @@ class Http2Layer(Layer):
                     events = source_conn.h2.receive_data(raw_frame)
                     source_conn.send(source_conn.h2.data_to_send())
 
-                for event in events:
-                    if hasattr(event, 'stream_id'):
-                        if is_server:
-                            eid = self.server_to_client_stream_ids[event.stream_id]
-                        else:
-                            eid = event.stream_id
-
-                    if isinstance(event, RequestReceived):
-                        headers = Headers([[str(k), str(v)] for k, v in event.headers])
-                        self.streams[eid] = Http2SingleStreamLayer(self, eid, headers)
-                        self.streams[eid].start()
-                    elif isinstance(event, ResponseReceived):
-                        headers = Headers([[str(k), str(v)] for k, v in event.headers])
-                        self.streams[eid].response_headers = headers
-                        self.streams[eid].response_arrived.set()
-                    elif isinstance(event, DataReceived):
-                        self.streams[eid].data_queue.put(event.data)
-                        source_conn.h2.safe_increment_flow_control(event.stream_id, len(event.data))
-                    elif isinstance(event, StreamEnded):
-                        self.streams[eid].data_finished.set()
-                    elif isinstance(event, StreamReset):
-                        self.streams[eid].zombie = time.time()
-                        if eid in self.streams and event.error_code == 0x8:
+                    for event in events:
+                        if hasattr(event, 'stream_id'):
                             if is_server:
-                                other_stream_id = self.streams[eid].client_stream_id
+                                eid = self.server_to_client_stream_ids[event.stream_id]
                             else:
-                                other_stream_id = self.streams[eid].server_stream_id
-                            other_conn.h2.safe_reset_stream(other_stream_id, event.error_code)
-                    elif isinstance(event, RemoteSettingsChanged):
-                        source_conn.h2.safe_acknowledge_settings(event)
-                        new_settings = dict([(id, cs.new_value) for (id, cs) in event.changed_settings.iteritems()])
-                        other_conn.h2.safe_update_settings(new_settings)
-                    elif isinstance(event, ConnectionTerminated):
-                        other_conn.h2.safe_close_connection(event.error_code)
-                        return
-                    elif isinstance(event, TrailersReceived):
-                        raise NotImplementedError()
-                    elif isinstance(event, PushedStreamReceived):
-                        raise NotImplementedError()
+                                eid = event.stream_id
+
+                        if isinstance(event, RequestReceived):
+                            headers = Headers([[str(k), str(v)] for k, v in event.headers])
+                            self.streams[eid] = Http2SingleStreamLayer(self, eid, headers)
+                            self.streams[eid].start()
+                        elif isinstance(event, ResponseReceived):
+                            headers = Headers([[str(k), str(v)] for k, v in event.headers])
+                            self.streams[eid].response_headers = headers
+                            self.streams[eid].response_arrived.set()
+                        elif isinstance(event, DataReceived):
+                            self.streams[eid].data_queue.put(event.data)
+                            source_conn.h2.safe_increment_flow_control(event.stream_id, len(event.data))
+                        elif isinstance(event, StreamEnded):
+                            self.streams[eid].data_finished.set()
+                        elif isinstance(event, StreamReset):
+                            self.streams[eid].zombie = time.time()
+                            if eid in self.streams and event.error_code == 0x8:
+                                if is_server:
+                                    other_stream_id = self.streams[eid].client_stream_id
+                                else:
+                                    other_stream_id = self.streams[eid].server_stream_id
+                                other_conn.h2.safe_reset_stream(other_stream_id, event.error_code)
+                        elif isinstance(event, RemoteSettingsChanged):
+                            source_conn.h2.safe_acknowledge_settings(event)
+                            new_settings = dict([(id, cs.new_value) for (id, cs) in event.changed_settings.iteritems()])
+                            other_conn.h2.safe_update_settings(new_settings)
+                        elif isinstance(event, ConnectionTerminated):
+                            other_conn.h2.safe_close_connection(event.error_code)
+                            return
+                        elif isinstance(event, TrailersReceived):
+                            raise NotImplementedError()
+                        elif isinstance(event, PushedStreamReceived):
+                            raise NotImplementedError()
 
             death_time = time.time() - 10
             for stream_id in self.streams.keys():
@@ -314,6 +317,18 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
         self.response_arrived = threading.Event()
         self.data_finished = threading.Event()
 
+    def is_zombie(self, h2_conn, stream_id):
+        if self.zombie:
+            return True
+
+        try:
+            h2_conn._get_stream_by_id(stream_id)
+        except Exception as e:
+            if isinstance(e, h2.exceptions.StreamClosedError):
+                return true
+
+        return False
+
     def read_request(self):
         self.data_finished.wait()
         self.data_finished.clear()
@@ -364,18 +379,17 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
         )
 
     def send_request(self, message):
-        if self.zombie:
-            return
-
         with self.server_conn.h2.lock:
             self.server_stream_id = self.server_conn.h2.get_next_available_stream_id()
             self.server_to_client_stream_ids[self.server_stream_id] = self.client_stream_id
 
             self.server_conn.h2.safe_send_headers(
+                self.is_zombie,
                 self.server_stream_id,
                 message.headers
             )
         self.server_conn.h2.safe_send_body(
+            self.is_zombie,
             self.server_stream_id,
             message.body
         )
@@ -409,19 +423,15 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
                 return
 
     def send_response_headers(self, response):
-        if self.zombie:
-            return
-
         self.client_conn.h2.safe_send_headers(
+            self.is_zombie,
             self.client_stream_id,
             response.headers
         )
 
     def send_response_body(self, _response, chunks):
-        if self.zombie:
-            return
-
         self.client_conn.h2.safe_send_body(
+            self.is_zombie,
             self.client_stream_id,
             chunks
         )
-- 
cgit v1.2.3


From de1b637a47c031840835fb9927b9289c04d777dc Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Sat, 16 Jan 2016 11:48:27 +0100
Subject: --wip--

---
 libmproxy/protocol/http.py | 31 +++++++++++++++++++++++--------
 1 file changed, 23 insertions(+), 8 deletions(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index ea069bcb..4426a541 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -130,6 +130,7 @@ class Http1Layer(_HttpLayer):
         layer = HttpLayer(self, self.mode)
         layer()
 
+
 class SafeH2Connection(H2Connection):
     def __init__(self, conn, *args, **kwargs):
         super(SafeH2Connection, self).__init__(*args, **kwargs)
@@ -202,6 +203,7 @@ class SafeH2Connection(H2Connection):
             self.end_stream(stream_id)
             self.conn.send(self.data_to_send())
 
+
 class Http2Layer(Layer):
     def __init__(self, ctx, mode):
         super(Http2Layer, self).__init__(ctx)
@@ -214,9 +216,6 @@ class Http2Layer(Layer):
         # because otherwise ssl_read_select fails!
         self.active_conns = [self.client_conn.connection]
 
-        if self.server_conn:
-            self._initiate_server_conn()
-
     def _initiate_server_conn(self):
         self.server_conn.h2 = SafeH2Connection(self.server_conn, client_side=True)
         self.server_conn.h2.initiate_connection()
@@ -235,7 +234,15 @@ class Http2Layer(Layer):
     def disconnect(self):
         raise NotImplementedError("Cannot dis- or reconnect in HTTP2 connections.")
 
+    def next_layer(self):
+        # WebSockets over HTTP/2?
+        # CONNECT for proxying?
+        raise NotImplementedError()
+
     def __call__(self):
+        if self.server_conn:
+            self._initiate_server_conn()
+
         preamble = self.client_conn.rfile.read(24)
         self.client_conn.h2.initiate_connection()
         self.client_conn.h2.update_settings({frame.SettingsFrame.ENABLE_PUSH: False})
@@ -267,15 +274,18 @@ class Http2Layer(Layer):
                         if isinstance(event, RequestReceived):
                             headers = Headers([[str(k), str(v)] for k, v in event.headers])
                             self.streams[eid] = Http2SingleStreamLayer(self, eid, headers)
+                            self.streams[eid].timestamp_start = time.time()
                             self.streams[eid].start()
                         elif isinstance(event, ResponseReceived):
                             headers = Headers([[str(k), str(v)] for k, v in event.headers])
+                            self.streams[eid].timestamp_start = time.time()
                             self.streams[eid].response_headers = headers
                             self.streams[eid].response_arrived.set()
                         elif isinstance(event, DataReceived):
                             self.streams[eid].data_queue.put(event.data)
                             source_conn.h2.safe_increment_flow_control(event.stream_id, len(event.data))
                         elif isinstance(event, StreamEnded):
+                            self.streams[eid].timestamp_end = time.time()
                             self.streams[eid].data_finished.set()
                         elif isinstance(event, StreamReset):
                             self.streams[eid].zombie = time.time()
@@ -322,6 +332,7 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
             return True
 
         try:
+            # TODO: replace private API call
             h2_conn._get_stream_by_id(stream_id)
         except Exception as e:
             if isinstance(e, h2.exceptions.StreamClosedError):
@@ -373,8 +384,8 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
             (2, 0),
             self.request_headers,
             data,
-            # TODO: timestamp_start=None,
-            # TODO: timestamp_end=None,
+            timestamp_start=self.timestamp_start,
+            timestamp_end=self.timestamp_end,
             form_out=None, # TODO: (request.form_out if hasattr(request, 'form_out') else None),
         )
 
@@ -405,8 +416,8 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
             reason='',
             headers=self.response_headers,
             content=None,
-            # TODO: timestamp_start=response.timestamp_start,
-            # TODO: timestamp_end=response.timestamp_end,
+            timestamp_start=self.timestamp_start,
+            timestamp_end=self.timestamp_end,
         )
 
     def read_response_body(self, request, response):
@@ -624,7 +635,11 @@ class HttpLayer(Layer):
         try:
             response = make_error_response(code, message)
             self.send_response(response)
-        except NetlibException:
+        except Exception as e:
+            self.log(
+                "error: %s" % repr(e),
+                level="debug"
+            )
             pass
 
     def change_upstream_proxy_server(self, address):
-- 
cgit v1.2.3


From c44a8949f7e0bc8f1725ee1f51633619a8643e0f Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Sun, 17 Jan 2016 17:41:39 +0100
Subject: use proper exception classes

---
 libmproxy/protocol/http.py | 6 +-----
 1 file changed, 1 insertion(+), 5 deletions(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 4426a541..aeb12656 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -635,11 +635,7 @@ class HttpLayer(Layer):
         try:
             response = make_error_response(code, message)
             self.send_response(response)
-        except Exception as e:
-            self.log(
-                "error: %s" % repr(e),
-                level="debug"
-            )
+        except NetlibException, h2.exceptions.H2Error:
             pass
 
     def change_upstream_proxy_server(self, address):
-- 
cgit v1.2.3


From 2cd71091adfa68ef6467a59932e63327608c02eb Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Sun, 17 Jan 2016 17:48:54 +0100
Subject: remove form_out todo

---
 libmproxy/protocol/http.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index aeb12656..75658661 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -386,7 +386,6 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
             data,
             timestamp_start=self.timestamp_start,
             timestamp_end=self.timestamp_end,
-            form_out=None, # TODO: (request.form_out if hasattr(request, 'form_out') else None),
         )
 
     def send_request(self, message):
-- 
cgit v1.2.3


From 24641d8561e6765b2aafada24fbefec2407eeeb3 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Sun, 17 Jan 2016 20:18:53 +0100
Subject: cleanup code

---
 libmproxy/protocol/http.py | 106 ++++++++++++++++++++++++---------------------
 1 file changed, 56 insertions(+), 50 deletions(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 75658661..74bb0c19 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -239,6 +239,57 @@ class Http2Layer(Layer):
         # CONNECT for proxying?
         raise NotImplementedError()
 
+    def _handle_event(self, event, source_conn, other_conn, is_server):
+        is_server = (conn == self.server_conn.connection)
+        if hasattr(event, 'stream_id'):
+            if is_server:
+                eid = self.server_to_client_stream_ids[event.stream_id]
+            else:
+                eid = event.stream_id
+
+        if isinstance(event, RequestReceived):
+            headers = Headers([[str(k), str(v)] for k, v in event.headers])
+            self.streams[eid] = Http2SingleStreamLayer(self, eid, headers)
+            self.streams[eid].timestamp_start = time.time()
+            self.streams[eid].start()
+        elif isinstance(event, ResponseReceived):
+            headers = Headers([[str(k), str(v)] for k, v in event.headers])
+            self.streams[eid].timestamp_start = time.time()
+            self.streams[eid].response_headers = headers
+            self.streams[eid].response_arrived.set()
+        elif isinstance(event, DataReceived):
+            self.streams[eid].data_queue.put(event.data)
+            source_conn.h2.safe_increment_flow_control(event.stream_id, len(event.data))
+        elif isinstance(event, StreamEnded):
+            self.streams[eid].timestamp_end = time.time()
+            self.streams[eid].data_finished.set()
+        elif isinstance(event, StreamReset):
+            self.streams[eid].zombie = time.time()
+            if eid in self.streams and event.error_code == 0x8:
+                if is_server:
+                    other_stream_id = self.streams[eid].client_stream_id
+                else:
+                    other_stream_id = self.streams[eid].server_stream_id
+                other_conn.h2.safe_reset_stream(other_stream_id, event.error_code)
+        elif isinstance(event, RemoteSettingsChanged):
+            source_conn.h2.safe_acknowledge_settings(event)
+            new_settings = dict([(id, cs.new_value) for (id, cs) in event.changed_settings.iteritems()])
+            other_conn.h2.safe_update_settings(new_settings)
+        elif isinstance(event, ConnectionTerminated):
+            other_conn.h2.safe_close_connection(event.error_code)
+            return
+        elif isinstance(event, TrailersReceived):
+            raise NotImplementedError()
+        elif isinstance(event, PushedStreamReceived):
+            raise NotImplementedError()
+
+    def _cleanup_streams(self):
+        death_time = time.time() - 10
+        for stream_id in self.streams.keys():
+            zombie = self.streams[stream_id].zombie
+            if zombie and zombie <= death_time:
+                self.streams.pop(stream_id, None)
+
     def __call__(self):
         if self.server_conn:
             self._initiate_server_conn()
@@ -254,10 +305,9 @@ class Http2Layer(Layer):
             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
-                is_server = (conn == self.server_conn.connection)
 
-                fields = struct.unpack("!HB", source_conn.rfile.peek(3))
-                length = (fields[0] << 8) + fields[1]
+                field = source_conn.rfile.peek(3)
+                length = (field[0] << 16) + (field[1] << 8) + field[2]
                 raw_frame = source_conn.rfile.safe_read(9 + length)
 
                 with source_conn.h2.lock:
@@ -265,53 +315,9 @@ class Http2Layer(Layer):
                     source_conn.send(source_conn.h2.data_to_send())
 
                     for event in events:
-                        if hasattr(event, 'stream_id'):
-                            if is_server:
-                                eid = self.server_to_client_stream_ids[event.stream_id]
-                            else:
-                                eid = event.stream_id
-
-                        if isinstance(event, RequestReceived):
-                            headers = Headers([[str(k), str(v)] for k, v in event.headers])
-                            self.streams[eid] = Http2SingleStreamLayer(self, eid, headers)
-                            self.streams[eid].timestamp_start = time.time()
-                            self.streams[eid].start()
-                        elif isinstance(event, ResponseReceived):
-                            headers = Headers([[str(k), str(v)] for k, v in event.headers])
-                            self.streams[eid].timestamp_start = time.time()
-                            self.streams[eid].response_headers = headers
-                            self.streams[eid].response_arrived.set()
-                        elif isinstance(event, DataReceived):
-                            self.streams[eid].data_queue.put(event.data)
-                            source_conn.h2.safe_increment_flow_control(event.stream_id, len(event.data))
-                        elif isinstance(event, StreamEnded):
-                            self.streams[eid].timestamp_end = time.time()
-                            self.streams[eid].data_finished.set()
-                        elif isinstance(event, StreamReset):
-                            self.streams[eid].zombie = time.time()
-                            if eid in self.streams and event.error_code == 0x8:
-                                if is_server:
-                                    other_stream_id = self.streams[eid].client_stream_id
-                                else:
-                                    other_stream_id = self.streams[eid].server_stream_id
-                                other_conn.h2.safe_reset_stream(other_stream_id, event.error_code)
-                        elif isinstance(event, RemoteSettingsChanged):
-                            source_conn.h2.safe_acknowledge_settings(event)
-                            new_settings = dict([(id, cs.new_value) for (id, cs) in event.changed_settings.iteritems()])
-                            other_conn.h2.safe_update_settings(new_settings)
-                        elif isinstance(event, ConnectionTerminated):
-                            other_conn.h2.safe_close_connection(event.error_code)
-                            return
-                        elif isinstance(event, TrailersReceived):
-                            raise NotImplementedError()
-                        elif isinstance(event, PushedStreamReceived):
-                            raise NotImplementedError()
-
-            death_time = time.time() - 10
-            for stream_id in self.streams.keys():
-                zombie = self.streams[stream_id].zombie
-                if zombie and zombie <= death_time:
-                    self.streams.pop(stream_id, None)
+                        self.handle_event(event, source_conn, other_conn)
+
+            self.cleanup_streams()
 
 
 class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
-- 
cgit v1.2.3


From db38e5a1cc61d499839d3463b79bb5f21f330193 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 18 Jan 2016 21:44:27 +0100
Subject: update hyper-h2 exception handling

---
 libmproxy/protocol/http.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 74bb0c19..a7160e43 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -158,7 +158,7 @@ class SafeH2Connection(H2Connection):
         with self.lock:
             try:
                 self.reset_stream(stream_id, error_code)
-            except h2.exceptions.ProtocolError:
+            except h2.exceptions.StreamClosedError:
                 # stream is already closed - good
                 pass
             self.conn.send(self.data_to_send())
-- 
cgit v1.2.3


From 4468fc7c2d2df795ac9cfc93cab96861a785c05e Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 18 Jan 2016 22:20:47 +0100
Subject: fix private API and RstStream issues

---
 libmproxy/protocol/http.py | 25 ++++++++++++++-----------
 1 file changed, 14 insertions(+), 11 deletions(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index a7160e43..597cad65 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -240,7 +240,6 @@ class Http2Layer(Layer):
         raise NotImplementedError()
 
     def _handle_event(self, event, source_conn, other_conn, is_server):
-        is_server = (conn == self.server_conn.connection)
         if hasattr(event, 'stream_id'):
             if is_server:
                 eid = self.server_to_client_stream_ids[event.stream_id]
@@ -277,12 +276,14 @@ class Http2Layer(Layer):
             other_conn.h2.safe_update_settings(new_settings)
         elif isinstance(event, ConnectionTerminated):
             other_conn.h2.safe_close_connection(event.error_code)
-            return
+            return False
         elif isinstance(event, TrailersReceived):
             raise NotImplementedError()
         elif isinstance(event, PushedStreamReceived):
             raise NotImplementedError()
 
+        return True
+
     def _cleanup_streams(self):
         death_time = time.time() - 10
         for stream_id in self.streams.keys():
@@ -305,9 +306,10 @@ class Http2Layer(Layer):
             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
+                is_server = (conn == self.server_conn.connection)
 
                 field = source_conn.rfile.peek(3)
-                length = (field[0] << 16) + (field[1] << 8) + field[2]
+                length = int(field.encode('hex'), 16)
                 raw_frame = source_conn.rfile.safe_read(9 + length)
 
                 with source_conn.h2.lock:
@@ -315,9 +317,10 @@ class Http2Layer(Layer):
                     source_conn.send(source_conn.h2.data_to_send())
 
                     for event in events:
-                        self.handle_event(event, source_conn, other_conn)
+                        if not self._handle_event(event, source_conn, other_conn, is_server):
+                            return
 
-            self.cleanup_streams()
+            self._cleanup_streams()
 
 
 class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
@@ -337,12 +340,12 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
         if self.zombie:
             return True
 
-        try:
-            # TODO: replace private API call
-            h2_conn._get_stream_by_id(stream_id)
-        except Exception as e:
-            if isinstance(e, h2.exceptions.StreamClosedError):
-                return true
+        # try:
+        #     # TODO: replace private API call
+        #     h2_conn._get_stream_by_id(stream_id)
+        # except Exception as e:
+        #     if isinstance(e, h2.exceptions.StreamClosedError):
+        #         return true
 
         return False
 
-- 
cgit v1.2.3


From 3f5e7987430f94d885e61b0d7c035f4318374990 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Tue, 19 Jan 2016 22:10:19 +0100
Subject: fix errors in http body parsing

---
 libmproxy/protocol/http.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 597cad65..276af132 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -253,11 +253,15 @@ class Http2Layer(Layer):
             self.streams[eid].start()
         elif isinstance(event, ResponseReceived):
             headers = Headers([[str(k), str(v)] for k, v in event.headers])
+            self.streams[eid].queued_data_length = 0
             self.streams[eid].timestamp_start = time.time()
             self.streams[eid].response_headers = headers
             self.streams[eid].response_arrived.set()
         elif isinstance(event, DataReceived):
+            if self.config.body_size_limit and self.streams[eid].queued_data_length > self.config.body_size_limit:
+                raise HttpException("HTTP body too large. Limit is {}.".format(self.config.body_size_limit))
             self.streams[eid].data_queue.put(event.data)
+            self.streams[eid].queued_data_length += len(event.data)
             source_conn.h2.safe_increment_flow_control(event.stream_id, len(event.data))
         elif isinstance(event, StreamEnded):
             self.streams[eid].timestamp_end = time.time()
@@ -332,6 +336,7 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
         self.request_headers = request_headers
         self.response_headers = None
         self.data_queue = Queue.Queue()
+        self.queued_data_length = 0
 
         self.response_arrived = threading.Event()
         self.data_finished = threading.Event()
@@ -382,6 +387,7 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
         data = []
         while self.data_queue.qsize() > 0:
             data.append(self.data_queue.get())
+        data = b"".join(data)
 
         return HTTPRequest(
             form_in,
-- 
cgit v1.2.3


From 83a443948508008c0bf2e03a28637312b1911b6a Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Tue, 19 Jan 2016 22:22:36 +0100
Subject: fix flow control on closed streams

---
 libmproxy/protocol/http.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 276af132..3049e34c 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -185,14 +185,14 @@ class SafeH2Connection(H2Connection):
             position = 0
             while position < len(chunk):
                 self.lock.acquire()
+                if is_zombie(self, stream_id):
+                    return
                 max_outbound_frame_size = self.max_outbound_frame_size
                 frame_chunk = chunk[position:position+max_outbound_frame_size]
                 if self.local_flow_control_window(stream_id) < len(frame_chunk):
                     self.lock.release()
                     time.sleep(0)
                     continue
-                if is_zombie(self, stream_id):
-                    return
                 self.send_data(stream_id, frame_chunk)
                 self.conn.send(self.data_to_send())
                 self.lock.release()
-- 
cgit v1.2.3


From 26b7ff95251885dd321fc7ed4bee90b229f409b8 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Wed, 20 Jan 2016 10:28:04 +0100
Subject: implemented push promise

---
 libmproxy/protocol/http.py | 22 +++++++++++++++++-----
 1 file changed, 17 insertions(+), 5 deletions(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 3049e34c..d770c937 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -219,7 +219,6 @@ class Http2Layer(Layer):
     def _initiate_server_conn(self):
         self.server_conn.h2 = SafeH2Connection(self.server_conn, client_side=True)
         self.server_conn.h2.initiate_connection()
-        self.server_conn.h2.update_settings({frame.SettingsFrame.ENABLE_PUSH: False})
         self.server_conn.send(self.server_conn.h2.data_to_send())
         self.active_conns.append(self.server_conn.connection)
 
@@ -241,7 +240,7 @@ class Http2Layer(Layer):
 
     def _handle_event(self, event, source_conn, other_conn, is_server):
         if hasattr(event, 'stream_id'):
-            if is_server:
+            if is_server and event.stream_id % 2 == 1:
                 eid = self.server_to_client_stream_ids[event.stream_id]
             else:
                 eid = event.stream_id
@@ -281,9 +280,23 @@ class Http2Layer(Layer):
         elif isinstance(event, ConnectionTerminated):
             other_conn.h2.safe_close_connection(event.error_code)
             return False
-        elif isinstance(event, TrailersReceived):
-            raise NotImplementedError()
         elif isinstance(event, PushedStreamReceived):
+            # pushed stream ids should be uniq 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]
+            with self.client_conn.h2.lock:
+                self.client_conn.h2.push_stream(parent_eid, event.pushed_stream_id, event.headers)
+
+            headers = Headers([[str(k), str(v)] for k, v in event.headers])
+            headers['x-mitmproxy-pushed'] = 'true'
+            self.streams[event.pushed_stream_id] = Http2SingleStreamLayer(self, 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
+            self.streams[event.pushed_stream_id].timestamp_end = time.time()
+            self.streams[event.pushed_stream_id].data_finished.set()
+            self.streams[event.pushed_stream_id].start()
+        elif isinstance(event, TrailersReceived):
             raise NotImplementedError()
 
         return True
@@ -301,7 +314,6 @@ class Http2Layer(Layer):
 
         preamble = self.client_conn.rfile.read(24)
         self.client_conn.h2.initiate_connection()
-        self.client_conn.h2.update_settings({frame.SettingsFrame.ENABLE_PUSH: False})
         self.client_conn.h2.receive_data(preamble)
         self.client_conn.send(self.client_conn.h2.data_to_send())
 
-- 
cgit v1.2.3


From 94977e0e3ddc16c70386b1f4d443984ff84320ac Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Wed, 20 Jan 2016 19:13:36 +0100
Subject: remove manual settings acknowledge

---
 libmproxy/protocol/http.py | 15 ---------------
 1 file changed, 15 deletions(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index d770c937..dad97fc3 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -16,7 +16,6 @@ from netlib.tcp import Address, ssl_read_select
 import h2
 from h2.connection import H2Connection
 from h2.events import *
-from hyperframe import frame
 
 from .base import Layer, Kill
 from .. import utils
@@ -163,11 +162,6 @@ class SafeH2Connection(H2Connection):
                 pass
             self.conn.send(self.data_to_send())
 
-    def safe_acknowledge_settings(self, event):
-        with self.conn.h2.lock:
-            self.conn.h2.acknowledge_settings(event)
-            self.conn.send(self.data_to_send())
-
     def safe_update_settings(self, new_settings):
         with self.conn.h2.lock:
             self.update_settings(new_settings)
@@ -274,7 +268,6 @@ class Http2Layer(Layer):
                     other_stream_id = self.streams[eid].server_stream_id
                 other_conn.h2.safe_reset_stream(other_stream_id, event.error_code)
         elif isinstance(event, RemoteSettingsChanged):
-            source_conn.h2.safe_acknowledge_settings(event)
             new_settings = dict([(id, cs.new_value) for (id, cs) in event.changed_settings.iteritems()])
             other_conn.h2.safe_update_settings(new_settings)
         elif isinstance(event, ConnectionTerminated):
@@ -356,14 +349,6 @@ class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
     def is_zombie(self, h2_conn, stream_id):
         if self.zombie:
             return True
-
-        # try:
-        #     # TODO: replace private API call
-        #     h2_conn._get_stream_by_id(stream_id)
-        # except Exception as e:
-        #     if isinstance(e, h2.exceptions.StreamClosedError):
-        #         return true
-
         return False
 
     def read_request(self):
-- 
cgit v1.2.3


From a05a961e7fa44259fe2768f43e53dde8888860ba Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Wed, 20 Jan 2016 19:31:37 +0100
Subject: cleanup lock usage

---
 libmproxy/protocol/http.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index dad97fc3..a8626af8 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -163,7 +163,7 @@ class SafeH2Connection(H2Connection):
             self.conn.send(self.data_to_send())
 
     def safe_update_settings(self, new_settings):
-        with self.conn.h2.lock:
+        with self.lock:
             self.update_settings(new_settings)
             self.conn.send(self.data_to_send())
 
-- 
cgit v1.2.3


From 936422cd735ac7d2bf19633c24dfeb1175cb3377 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Wed, 20 Jan 2016 19:58:24 +0100
Subject: split files into http, http1, and http2

---
 libmproxy/protocol/__init__.py |   8 +-
 libmproxy/protocol/http.py     | 425 +----------------------------------------
 libmproxy/protocol/http1.py    |  75 ++++++++
 libmproxy/protocol/http2.py    | 365 +++++++++++++++++++++++++++++++++++
 4 files changed, 452 insertions(+), 421 deletions(-)
 create mode 100644 libmproxy/protocol/http1.py
 create mode 100644 libmproxy/protocol/http2.py

diff --git a/libmproxy/protocol/__init__.py b/libmproxy/protocol/__init__.py
index d46f16f5..ea958d06 100644
--- a/libmproxy/protocol/__init__.py
+++ b/libmproxy/protocol/__init__.py
@@ -27,15 +27,19 @@ as late as possible; this makes server replay without any outgoing connections p
 
 from __future__ import (absolute_import, print_function, division)
 from .base import Layer, ServerConnectionMixin, Kill
-from .http import Http1Layer, UpstreamConnectLayer, Http2Layer
 from .tls import TlsLayer
 from .tls import is_tls_record_magic
 from .tls import TlsClientHello
+from .http import UpstreamConnectLayer
+from .http1 import Http1Layer
+from .http2 import Http2Layer
 from .rawtcp import RawTCPLayer
 
 __all__ = [
     "Layer", "ServerConnectionMixin", "Kill",
-    "Http1Layer", "UpstreamConnectLayer", "Http2Layer",
     "TlsLayer", "is_tls_record_magic", "TlsClientHello",
+    "UpstreamConnectLayer",
+    "Http1Layer",
+    "Http2Layer",
     "RawTCPLayer",
 ]
diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index a8626af8..ee9d2d46 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -3,26 +3,17 @@ from __future__ import (absolute_import, print_function, division)
 import sys
 import traceback
 import six
-import struct
-import threading
-import time
-import Queue
 
 from netlib import tcp
 from netlib.exceptions import HttpException, HttpReadDisconnect, NetlibException
-from netlib.http import http1, Headers, CONTENT_MISSING
-from netlib.tcp import Address, ssl_read_select
+from netlib.http import Headers, CONTENT_MISSING
 
-import h2
-from h2.connection import H2Connection
-from h2.events import *
+from h2.exceptions import H2Error
 
-from .base import Layer, Kill
 from .. import utils
 from ..exceptions import HttpProtocolException, ProtocolException
 from ..models import (
     HTTPFlow,
-    HTTPRequest,
     HTTPResponse,
     make_error_response,
     make_connect_response,
@@ -30,8 +21,9 @@ from ..models import (
     expect_continue_response
 )
 
+from .base import Layer, Kill
 
-class _HttpLayer(Layer):
+class _HttpTransmissionLayer(Layer):
     def read_request(self):
         raise NotImplementedError()
 
@@ -71,411 +63,6 @@ class _HttpLayer(Layer):
         raise NotImplementedError()
 
 
-class Http1Layer(_HttpLayer):
-    def __init__(self, ctx, mode):
-        super(Http1Layer, self).__init__(ctx)
-        self.mode = mode
-
-    def read_request(self):
-        req = http1.read_request(self.client_conn.rfile, body_size_limit=self.config.body_size_limit)
-        return HTTPRequest.wrap(req)
-
-    def read_request_body(self, request):
-        expected_size = http1.expected_http_body_size(request)
-        return http1.read_body(self.client_conn.rfile, expected_size, self.config.body_size_limit)
-
-    def send_request(self, request):
-        self.server_conn.wfile.write(http1.assemble_request(request))
-        self.server_conn.wfile.flush()
-
-    def read_response_headers(self):
-        resp = http1.read_response_head(self.server_conn.rfile)
-        return HTTPResponse.wrap(resp)
-
-    def read_response_body(self, request, response):
-        expected_size = http1.expected_http_body_size(request, response)
-        return http1.read_body(self.server_conn.rfile, expected_size, self.config.body_size_limit)
-
-    def send_response_headers(self, response):
-        raw = http1.assemble_response_head(response)
-        self.client_conn.wfile.write(raw)
-        self.client_conn.wfile.flush()
-
-    def send_response_body(self, response, chunks):
-        for chunk in http1.assemble_body(response.headers, chunks):
-            self.client_conn.wfile.write(chunk)
-            self.client_conn.wfile.flush()
-
-    def check_close_connection(self, flow):
-        request_close = http1.connection_close(
-            flow.request.http_version,
-            flow.request.headers
-        )
-        response_close = http1.connection_close(
-            flow.response.http_version,
-            flow.response.headers
-        )
-        read_until_eof = http1.expected_http_body_size(flow.request, flow.response) == -1
-        close_connection = request_close or response_close or read_until_eof
-        if flow.request.form_in == "authority" and flow.response.status_code == 200:
-            # Workaround for https://github.com/mitmproxy/mitmproxy/issues/313:
-            # Charles Proxy sends a CONNECT response with HTTP/1.0
-            # and no Content-Length header
-
-            return False
-        return close_connection
-
-    def __call__(self):
-        layer = HttpLayer(self, self.mode)
-        layer()
-
-
-class SafeH2Connection(H2Connection):
-    def __init__(self, conn, *args, **kwargs):
-        super(SafeH2Connection, self).__init__(*args, **kwargs)
-        self.conn = conn
-        self.lock = threading.RLock()
-
-    def safe_close_connection(self, error_code):
-        with self.lock:
-            self.close_connection(error_code)
-            self.conn.send(self.data_to_send())
-
-    def safe_increment_flow_control(self, stream_id, length):
-        if length == 0:
-            return
-
-        with self.lock:
-            self.increment_flow_control_window(length)
-            self.conn.send(self.data_to_send())
-        with self.lock:
-            if stream_id in self.streams and not self.streams[stream_id].closed:
-                self.increment_flow_control_window(length, stream_id=stream_id)
-                self.conn.send(self.data_to_send())
-
-    def safe_reset_stream(self, stream_id, error_code):
-        with self.lock:
-            try:
-                self.reset_stream(stream_id, error_code)
-            except h2.exceptions.StreamClosedError:
-                # stream is already closed - good
-                pass
-            self.conn.send(self.data_to_send())
-
-    def safe_update_settings(self, new_settings):
-        with self.lock:
-            self.update_settings(new_settings)
-            self.conn.send(self.data_to_send())
-
-    def safe_send_headers(self, is_zombie, stream_id, headers):
-        with self.lock:
-            if is_zombie(self, stream_id):
-                return
-            self.send_headers(stream_id, headers)
-            self.conn.send(self.data_to_send())
-
-    def safe_send_body(self, is_zombie, stream_id, chunks):
-        for chunk in chunks:
-            position = 0
-            while position < len(chunk):
-                self.lock.acquire()
-                if is_zombie(self, stream_id):
-                    return
-                max_outbound_frame_size = self.max_outbound_frame_size
-                frame_chunk = chunk[position:position+max_outbound_frame_size]
-                if self.local_flow_control_window(stream_id) < len(frame_chunk):
-                    self.lock.release()
-                    time.sleep(0)
-                    continue
-                self.send_data(stream_id, frame_chunk)
-                self.conn.send(self.data_to_send())
-                self.lock.release()
-                position += max_outbound_frame_size
-        with self.lock:
-            if is_zombie(self, stream_id):
-                return
-            self.end_stream(stream_id)
-            self.conn.send(self.data_to_send())
-
-
-class Http2Layer(Layer):
-    def __init__(self, ctx, mode):
-        super(Http2Layer, self).__init__(ctx)
-        self.mode = mode
-        self.streams = dict()
-        self.server_to_client_stream_ids = dict([(0, 0)])
-        self.client_conn.h2 = SafeH2Connection(self.client_conn, client_side=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)
-        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)
-
-    def connect(self):
-        self.ctx.connect()
-        self.server_conn.connect()
-        self._initiate_server_conn()
-
-    def set_server(self):
-        raise NotImplementedError("Cannot change server for HTTP2 connections.")
-
-    def disconnect(self):
-        raise NotImplementedError("Cannot dis- or reconnect in HTTP2 connections.")
-
-    def next_layer(self):
-        # WebSockets over HTTP/2?
-        # CONNECT for proxying?
-        raise NotImplementedError()
-
-    def _handle_event(self, event, source_conn, other_conn, is_server):
-        if hasattr(event, 'stream_id'):
-            if is_server and event.stream_id % 2 == 1:
-                eid = self.server_to_client_stream_ids[event.stream_id]
-            else:
-                eid = event.stream_id
-
-        if isinstance(event, RequestReceived):
-            headers = Headers([[str(k), str(v)] for k, v in event.headers])
-            self.streams[eid] = Http2SingleStreamLayer(self, eid, headers)
-            self.streams[eid].timestamp_start = time.time()
-            self.streams[eid].start()
-        elif isinstance(event, ResponseReceived):
-            headers = Headers([[str(k), str(v)] for k, v in event.headers])
-            self.streams[eid].queued_data_length = 0
-            self.streams[eid].timestamp_start = time.time()
-            self.streams[eid].response_headers = headers
-            self.streams[eid].response_arrived.set()
-        elif isinstance(event, DataReceived):
-            if self.config.body_size_limit and self.streams[eid].queued_data_length > self.config.body_size_limit:
-                raise HttpException("HTTP body too large. Limit is {}.".format(self.config.body_size_limit))
-            self.streams[eid].data_queue.put(event.data)
-            self.streams[eid].queued_data_length += len(event.data)
-            source_conn.h2.safe_increment_flow_control(event.stream_id, len(event.data))
-        elif isinstance(event, StreamEnded):
-            self.streams[eid].timestamp_end = time.time()
-            self.streams[eid].data_finished.set()
-        elif isinstance(event, StreamReset):
-            self.streams[eid].zombie = time.time()
-            if eid in self.streams and event.error_code == 0x8:
-                if is_server:
-                    other_stream_id = self.streams[eid].client_stream_id
-                else:
-                    other_stream_id = self.streams[eid].server_stream_id
-                other_conn.h2.safe_reset_stream(other_stream_id, event.error_code)
-        elif isinstance(event, RemoteSettingsChanged):
-            new_settings = dict([(id, cs.new_value) for (id, cs) in event.changed_settings.iteritems()])
-            other_conn.h2.safe_update_settings(new_settings)
-        elif isinstance(event, ConnectionTerminated):
-            other_conn.h2.safe_close_connection(event.error_code)
-            return False
-        elif isinstance(event, PushedStreamReceived):
-            # pushed stream ids should be uniq 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]
-            with self.client_conn.h2.lock:
-                self.client_conn.h2.push_stream(parent_eid, event.pushed_stream_id, event.headers)
-
-            headers = Headers([[str(k), str(v)] for k, v in event.headers])
-            headers['x-mitmproxy-pushed'] = 'true'
-            self.streams[event.pushed_stream_id] = Http2SingleStreamLayer(self, 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
-            self.streams[event.pushed_stream_id].timestamp_end = time.time()
-            self.streams[event.pushed_stream_id].data_finished.set()
-            self.streams[event.pushed_stream_id].start()
-        elif isinstance(event, TrailersReceived):
-            raise NotImplementedError()
-
-        return True
-
-    def _cleanup_streams(self):
-        death_time = time.time() - 10
-        for stream_id in self.streams.keys():
-            zombie = self.streams[stream_id].zombie
-            if zombie and zombie <= death_time:
-                self.streams.pop(stream_id, None)
-
-    def __call__(self):
-        if self.server_conn:
-            self._initiate_server_conn()
-
-        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())
-
-        while True:
-            r = ssl_read_select(self.active_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
-                is_server = (conn == self.server_conn.connection)
-
-                field = source_conn.rfile.peek(3)
-                length = int(field.encode('hex'), 16)
-                raw_frame = source_conn.rfile.safe_read(9 + length)
-
-                with source_conn.h2.lock:
-                    events = source_conn.h2.receive_data(raw_frame)
-                    source_conn.send(source_conn.h2.data_to_send())
-
-                    for event in events:
-                        if not self._handle_event(event, source_conn, other_conn, is_server):
-                            return
-
-            self._cleanup_streams()
-
-
-class Http2SingleStreamLayer(_HttpLayer, threading.Thread):
-    def __init__(self, ctx, stream_id, request_headers):
-        super(Http2SingleStreamLayer, self).__init__(ctx)
-        self.zombie = None
-        self.client_stream_id = stream_id
-        self.server_stream_id = None
-        self.request_headers = request_headers
-        self.response_headers = None
-        self.data_queue = Queue.Queue()
-        self.queued_data_length = 0
-
-        self.response_arrived = threading.Event()
-        self.data_finished = threading.Event()
-
-    def is_zombie(self, h2_conn, stream_id):
-        if self.zombie:
-            return True
-        return False
-
-    def read_request(self):
-        self.data_finished.wait()
-        self.data_finished.clear()
-
-        authority = self.request_headers.get(':authority', '')
-        method = self.request_headers.get(':method', 'GET')
-        scheme = self.request_headers.get(':scheme', 'https')
-        path = self.request_headers.get(':path', '/')
-        host = None
-        port = None
-
-        if path == '*' or path.startswith("/"):
-            form_in = "relative"
-        elif method == 'CONNECT':
-            form_in = "authority"
-            if ":" in authority:
-                host, port = authority.split(":", 1)
-            else:
-                host = authority
-        else:
-            form_in = "absolute"
-            # FIXME: verify if path or :host contains what we need
-            scheme, host, port, _ = utils.parse_url(path)
-
-        if host is None:
-            host = 'localhost'
-        if port is None:
-            port = 80 if scheme == 'http' else 443
-        port = int(port)
-
-        data = []
-        while self.data_queue.qsize() > 0:
-            data.append(self.data_queue.get())
-        data = b"".join(data)
-
-        return HTTPRequest(
-            form_in,
-            method,
-            scheme,
-            host,
-            port,
-            path,
-            (2, 0),
-            self.request_headers,
-            data,
-            timestamp_start=self.timestamp_start,
-            timestamp_end=self.timestamp_end,
-        )
-
-    def send_request(self, message):
-        with self.server_conn.h2.lock:
-            self.server_stream_id = self.server_conn.h2.get_next_available_stream_id()
-            self.server_to_client_stream_ids[self.server_stream_id] = self.client_stream_id
-
-            self.server_conn.h2.safe_send_headers(
-                self.is_zombie,
-                self.server_stream_id,
-                message.headers
-            )
-        self.server_conn.h2.safe_send_body(
-            self.is_zombie,
-            self.server_stream_id,
-            message.body
-        )
-
-    def read_response_headers(self):
-        self.response_arrived.wait()
-
-        status_code = int(self.response_headers.get(':status', 502))
-
-        return HTTPResponse(
-            http_version=(2, 0),
-            status_code=status_code,
-            reason='',
-            headers=self.response_headers,
-            content=None,
-            timestamp_start=self.timestamp_start,
-            timestamp_end=self.timestamp_end,
-        )
-
-    def read_response_body(self, request, response):
-        while True:
-            try:
-                yield self.data_queue.get(timeout=1)
-            except Queue.Empty:
-                pass
-            if self.data_finished.is_set():
-                while self.data_queue.qsize() > 0:
-                    yield self.data_queue.get()
-                return
-            if self.zombie:
-                return
-
-    def send_response_headers(self, response):
-        self.client_conn.h2.safe_send_headers(
-            self.is_zombie,
-            self.client_stream_id,
-            response.headers
-        )
-
-    def send_response_body(self, _response, chunks):
-        self.client_conn.h2.safe_send_body(
-            self.is_zombie,
-            self.client_stream_id,
-            chunks
-        )
-
-    def check_close_connection(self, flow):
-        # This layer only handles a single stream.
-        # RFC 7540 8.1: An HTTP request/response exchange fully consumes a single stream.
-        return True
-
-    def connect(self):
-        raise ValueError("CONNECT inside an HTTP2 stream is not supported.")
-
-    def set_server(self, *args, **kwargs):
-        # do not mess with the server connection - all streams share it.
-        pass
-
-    def run(self):
-        layer = HttpLayer(self, self.mode)
-        layer()
-        self.zombie = time.time()
-
-
 class ConnectServerConnection(object):
 
     """
@@ -531,7 +118,7 @@ class UpstreamConnectLayer(Layer):
     def set_server(self, address, server_tls=None, sni=None):
         if self.ctx.server_conn:
             self.ctx.disconnect()
-        address = Address.wrap(address)
+        address = tcp.Address.wrap(address)
         self.connect_request.host = address.host
         self.connect_request.port = address.port
         self.server_conn.address = address
@@ -646,7 +233,7 @@ class HttpLayer(Layer):
         try:
             response = make_error_response(code, message)
             self.send_response(response)
-        except NetlibException, h2.exceptions.H2Error:
+        except NetlibException, H2Error:
             pass
 
     def change_upstream_proxy_server(self, address):
diff --git a/libmproxy/protocol/http1.py b/libmproxy/protocol/http1.py
new file mode 100644
index 00000000..0d6095df
--- /dev/null
+++ b/libmproxy/protocol/http1.py
@@ -0,0 +1,75 @@
+from __future__ import (absolute_import, print_function, division)
+
+import sys
+import traceback
+import six
+import struct
+import threading
+import time
+import Queue
+
+from netlib import tcp
+from netlib.http import http1
+
+from .http import _HttpTransmissionLayer, HttpLayer
+from .. import utils
+from ..models import HTTPRequest, HTTPResponse
+
+
+class Http1Layer(_HttpTransmissionLayer):
+    def __init__(self, ctx, mode):
+        super(Http1Layer, self).__init__(ctx)
+        self.mode = mode
+
+    def read_request(self):
+        req = http1.read_request(self.client_conn.rfile, body_size_limit=self.config.body_size_limit)
+        return HTTPRequest.wrap(req)
+
+    def read_request_body(self, request):
+        expected_size = http1.expected_http_body_size(request)
+        return http1.read_body(self.client_conn.rfile, expected_size, self.config.body_size_limit)
+
+    def send_request(self, request):
+        self.server_conn.wfile.write(http1.assemble_request(request))
+        self.server_conn.wfile.flush()
+
+    def read_response_headers(self):
+        resp = http1.read_response_head(self.server_conn.rfile)
+        return HTTPResponse.wrap(resp)
+
+    def read_response_body(self, request, response):
+        expected_size = http1.expected_http_body_size(request, response)
+        return http1.read_body(self.server_conn.rfile, expected_size, self.config.body_size_limit)
+
+    def send_response_headers(self, response):
+        raw = http1.assemble_response_head(response)
+        self.client_conn.wfile.write(raw)
+        self.client_conn.wfile.flush()
+
+    def send_response_body(self, response, chunks):
+        for chunk in http1.assemble_body(response.headers, chunks):
+            self.client_conn.wfile.write(chunk)
+            self.client_conn.wfile.flush()
+
+    def check_close_connection(self, flow):
+        request_close = http1.connection_close(
+            flow.request.http_version,
+            flow.request.headers
+        )
+        response_close = http1.connection_close(
+            flow.response.http_version,
+            flow.response.headers
+        )
+        read_until_eof = http1.expected_http_body_size(flow.request, flow.response) == -1
+        close_connection = request_close or response_close or read_until_eof
+        if flow.request.form_in == "authority" and flow.response.status_code == 200:
+            # Workaround for https://github.com/mitmproxy/mitmproxy/issues/313:
+            # Charles Proxy sends a CONNECT response with HTTP/1.0
+            # and no Content-Length header
+
+            return False
+        return close_connection
+
+    def __call__(self):
+        layer = HttpLayer(self, self.mode)
+        layer()
diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
new file mode 100644
index 00000000..a6d1b73f
--- /dev/null
+++ b/libmproxy/protocol/http2.py
@@ -0,0 +1,365 @@
+from __future__ import (absolute_import, print_function, division)
+
+import struct
+import threading
+import time
+import Queue
+
+from netlib.tcp import ssl_read_select
+from netlib.exceptions import HttpException
+from netlib.http import Headers
+
+import h2
+from h2.connection import H2Connection
+from h2.events import *
+
+from .base import Layer
+from .http import _HttpTransmissionLayer, HttpLayer
+from .. import utils
+from ..models import HTTPRequest, HTTPResponse
+
+
+class SafeH2Connection(H2Connection):
+    def __init__(self, conn, *args, **kwargs):
+        super(SafeH2Connection, self).__init__(*args, **kwargs)
+        self.conn = conn
+        self.lock = threading.RLock()
+
+    def safe_close_connection(self, error_code):
+        with self.lock:
+            self.close_connection(error_code)
+            self.conn.send(self.data_to_send())
+
+    def safe_increment_flow_control(self, stream_id, length):
+        if length == 0:
+            return
+
+        with self.lock:
+            self.increment_flow_control_window(length)
+            self.conn.send(self.data_to_send())
+        with self.lock:
+            if stream_id in self.streams and not self.streams[stream_id].closed:
+                self.increment_flow_control_window(length, stream_id=stream_id)
+                self.conn.send(self.data_to_send())
+
+    def safe_reset_stream(self, stream_id, error_code):
+        with self.lock:
+            try:
+                self.reset_stream(stream_id, error_code)
+            except h2.exceptions.StreamClosedError:
+                # stream is already closed - good
+                pass
+            self.conn.send(self.data_to_send())
+
+    def safe_update_settings(self, new_settings):
+        with self.lock:
+            self.update_settings(new_settings)
+            self.conn.send(self.data_to_send())
+
+    def safe_send_headers(self, is_zombie, stream_id, headers):
+        with self.lock:
+            if is_zombie(self, stream_id):
+                return
+            self.send_headers(stream_id, headers)
+            self.conn.send(self.data_to_send())
+
+    def safe_send_body(self, is_zombie, stream_id, chunks):
+        for chunk in chunks:
+            position = 0
+            while position < len(chunk):
+                self.lock.acquire()
+                if is_zombie(self, stream_id):
+                    return
+                max_outbound_frame_size = self.max_outbound_frame_size
+                frame_chunk = chunk[position:position+max_outbound_frame_size]
+                if self.local_flow_control_window(stream_id) < len(frame_chunk):
+                    self.lock.release()
+                    time.sleep(0)
+                    continue
+                self.send_data(stream_id, frame_chunk)
+                self.conn.send(self.data_to_send())
+                self.lock.release()
+                position += max_outbound_frame_size
+        with self.lock:
+            if is_zombie(self, stream_id):
+                return
+            self.end_stream(stream_id)
+            self.conn.send(self.data_to_send())
+
+
+class Http2Layer(Layer):
+    def __init__(self, ctx, mode):
+        super(Http2Layer, self).__init__(ctx)
+        self.mode = mode
+        self.streams = dict()
+        self.server_to_client_stream_ids = dict([(0, 0)])
+        self.client_conn.h2 = SafeH2Connection(self.client_conn, client_side=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)
+        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)
+
+    def connect(self):
+        self.ctx.connect()
+        self.server_conn.connect()
+        self._initiate_server_conn()
+
+    def set_server(self):
+        raise NotImplementedError("Cannot change server for HTTP2 connections.")
+
+    def disconnect(self):
+        raise NotImplementedError("Cannot dis- or reconnect in HTTP2 connections.")
+
+    def next_layer(self):
+        # WebSockets over HTTP/2?
+        # CONNECT for proxying?
+        raise NotImplementedError()
+
+    def _handle_event(self, event, source_conn, other_conn, is_server):
+        if hasattr(event, 'stream_id'):
+            if is_server and event.stream_id % 2 == 1:
+                eid = self.server_to_client_stream_ids[event.stream_id]
+            else:
+                eid = event.stream_id
+
+        if isinstance(event, RequestReceived):
+            headers = Headers([[str(k), str(v)] for k, v in event.headers])
+            self.streams[eid] = Http2SingleStreamLayer(self, eid, headers)
+            self.streams[eid].timestamp_start = time.time()
+            self.streams[eid].start()
+        elif isinstance(event, ResponseReceived):
+            headers = Headers([[str(k), str(v)] for k, v in event.headers])
+            self.streams[eid].queued_data_length = 0
+            self.streams[eid].timestamp_start = time.time()
+            self.streams[eid].response_headers = headers
+            self.streams[eid].response_arrived.set()
+        elif isinstance(event, DataReceived):
+            if self.config.body_size_limit and self.streams[eid].queued_data_length > self.config.body_size_limit:
+                raise HttpException("HTTP body too large. Limit is {}.".format(self.config.body_size_limit))
+            self.streams[eid].data_queue.put(event.data)
+            self.streams[eid].queued_data_length += len(event.data)
+            source_conn.h2.safe_increment_flow_control(event.stream_id, len(event.data))
+        elif isinstance(event, StreamEnded):
+            self.streams[eid].timestamp_end = time.time()
+            self.streams[eid].data_finished.set()
+        elif isinstance(event, StreamReset):
+            self.streams[eid].zombie = time.time()
+            if eid in self.streams and event.error_code == 0x8:
+                if is_server:
+                    other_stream_id = self.streams[eid].client_stream_id
+                else:
+                    other_stream_id = self.streams[eid].server_stream_id
+                other_conn.h2.safe_reset_stream(other_stream_id, event.error_code)
+        elif isinstance(event, RemoteSettingsChanged):
+            new_settings = dict([(id, cs.new_value) for (id, cs) in event.changed_settings.iteritems()])
+            other_conn.h2.safe_update_settings(new_settings)
+        elif isinstance(event, ConnectionTerminated):
+            other_conn.h2.safe_close_connection(event.error_code)
+            return False
+        elif isinstance(event, PushedStreamReceived):
+            # pushed stream ids should be uniq 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]
+            with self.client_conn.h2.lock:
+                self.client_conn.h2.push_stream(parent_eid, event.pushed_stream_id, event.headers)
+
+            headers = Headers([[str(k), str(v)] for k, v in event.headers])
+            headers['x-mitmproxy-pushed'] = 'true'
+            self.streams[event.pushed_stream_id] = Http2SingleStreamLayer(self, 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
+            self.streams[event.pushed_stream_id].timestamp_end = time.time()
+            self.streams[event.pushed_stream_id].data_finished.set()
+            self.streams[event.pushed_stream_id].start()
+        elif isinstance(event, TrailersReceived):
+            raise NotImplementedError()
+
+        return True
+
+    def _cleanup_streams(self):
+        death_time = time.time() - 10
+        for stream_id in self.streams.keys():
+            zombie = self.streams[stream_id].zombie
+            if zombie and zombie <= death_time:
+                self.streams.pop(stream_id, None)
+
+    def __call__(self):
+        if self.server_conn:
+            self._initiate_server_conn()
+
+        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())
+
+        while True:
+            r = ssl_read_select(self.active_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
+                is_server = (conn == self.server_conn.connection)
+
+                field = source_conn.rfile.peek(3)
+                length = int(field.encode('hex'), 16)
+                raw_frame = source_conn.rfile.safe_read(9 + length)
+
+                with source_conn.h2.lock:
+                    events = source_conn.h2.receive_data(raw_frame)
+                    source_conn.send(source_conn.h2.data_to_send())
+
+                    for event in events:
+                        if not self._handle_event(event, source_conn, other_conn, is_server):
+                            return
+
+            self._cleanup_streams()
+
+
+class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
+    def __init__(self, ctx, stream_id, request_headers):
+        super(Http2SingleStreamLayer, self).__init__(ctx)
+        self.zombie = None
+        self.client_stream_id = stream_id
+        self.server_stream_id = None
+        self.request_headers = request_headers
+        self.response_headers = None
+        self.data_queue = Queue.Queue()
+        self.queued_data_length = 0
+
+        self.response_arrived = threading.Event()
+        self.data_finished = threading.Event()
+
+    def is_zombie(self, h2_conn, stream_id):
+        if self.zombie:
+            return True
+        return False
+
+    def read_request(self):
+        self.data_finished.wait()
+        self.data_finished.clear()
+
+        authority = self.request_headers.get(':authority', '')
+        method = self.request_headers.get(':method', 'GET')
+        scheme = self.request_headers.get(':scheme', 'https')
+        path = self.request_headers.get(':path', '/')
+        host = None
+        port = None
+
+        if path == '*' or path.startswith("/"):
+            form_in = "relative"
+        elif method == 'CONNECT':
+            form_in = "authority"
+            if ":" in authority:
+                host, port = authority.split(":", 1)
+            else:
+                host = authority
+        else:
+            form_in = "absolute"
+            # FIXME: verify if path or :host contains what we need
+            scheme, host, port, _ = utils.parse_url(path)
+
+        if host is None:
+            host = 'localhost'
+        if port is None:
+            port = 80 if scheme == 'http' else 443
+        port = int(port)
+
+        data = []
+        while self.data_queue.qsize() > 0:
+            data.append(self.data_queue.get())
+        data = b"".join(data)
+
+        return HTTPRequest(
+            form_in,
+            method,
+            scheme,
+            host,
+            port,
+            path,
+            (2, 0),
+            self.request_headers,
+            data,
+            timestamp_start=self.timestamp_start,
+            timestamp_end=self.timestamp_end,
+        )
+
+    def send_request(self, message):
+        with self.server_conn.h2.lock:
+            self.server_stream_id = self.server_conn.h2.get_next_available_stream_id()
+            self.server_to_client_stream_ids[self.server_stream_id] = self.client_stream_id
+
+            self.server_conn.h2.safe_send_headers(
+                self.is_zombie,
+                self.server_stream_id,
+                message.headers
+            )
+        self.server_conn.h2.safe_send_body(
+            self.is_zombie,
+            self.server_stream_id,
+            message.body
+        )
+
+    def read_response_headers(self):
+        self.response_arrived.wait()
+
+        status_code = int(self.response_headers.get(':status', 502))
+
+        return HTTPResponse(
+            http_version=(2, 0),
+            status_code=status_code,
+            reason='',
+            headers=self.response_headers,
+            content=None,
+            timestamp_start=self.timestamp_start,
+            timestamp_end=self.timestamp_end,
+        )
+
+    def read_response_body(self, request, response):
+        while True:
+            try:
+                yield self.data_queue.get(timeout=1)
+            except Queue.Empty:
+                pass
+            if self.data_finished.is_set():
+                while self.data_queue.qsize() > 0:
+                    yield self.data_queue.get()
+                return
+            if self.zombie:
+                return
+
+    def send_response_headers(self, response):
+        self.client_conn.h2.safe_send_headers(
+            self.is_zombie,
+            self.client_stream_id,
+            response.headers
+        )
+
+    def send_response_body(self, _response, chunks):
+        self.client_conn.h2.safe_send_body(
+            self.is_zombie,
+            self.client_stream_id,
+            chunks
+        )
+
+    def check_close_connection(self, flow):
+        # This layer only handles a single stream.
+        # RFC 7540 8.1: An HTTP request/response exchange fully consumes a single stream.
+        return True
+
+    def connect(self):
+        raise ValueError("CONNECT inside an HTTP2 stream is not supported.")
+
+    def set_server(self, *args, **kwargs):
+        # do not mess with the server connection - all streams share it.
+        pass
+
+    def run(self):
+        layer = HttpLayer(self, self.mode)
+        layer()
+        self.zombie = time.time()
-- 
cgit v1.2.3


From 2964a607ad927c01abb7596238f599ce7d546de8 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Wed, 20 Jan 2016 19:58:33 +0100
Subject: fix import in tests

---
 test/test_filt.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/test/test_filt.py b/test/test_filt.py
index b1fd2ad9..b1f3a21f 100644
--- a/test/test_filt.py
+++ b/test/test_filt.py
@@ -1,7 +1,7 @@
 import cStringIO
 from libmproxy import filt
-from libmproxy.protocol import http
 from libmproxy.models import Error
+from libmproxy.models import http
 from netlib.http import Headers
 from . import tutils
 
-- 
cgit v1.2.3


From 4de9cbb61ee0bedd882518e72a1605ca999ccf97 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Sun, 24 Jan 2016 23:15:42 +0100
Subject: rename test file

---
 test/test_protocol_http.py  | 92 ---------------------------------------------
 test/test_protocol_http1.py | 65 ++++++++++++++++++++++++++++++++
 2 files changed, 65 insertions(+), 92 deletions(-)
 delete mode 100644 test/test_protocol_http.py
 create mode 100644 test/test_protocol_http1.py

diff --git a/test/test_protocol_http.py b/test/test_protocol_http.py
deleted file mode 100644
index 489be3f9..00000000
--- a/test/test_protocol_http.py
+++ /dev/null
@@ -1,92 +0,0 @@
-from io import BytesIO
-from netlib.exceptions import HttpSyntaxException
-from netlib.http import http1
-from netlib.tcp import TCPClient
-from netlib.tutils import treq, raises
-from . import tutils, tservers
-
-
-class TestHTTPResponse:
-
-    def test_read_from_stringio(self):
-        s = (
-            b"HTTP/1.1 200 OK\r\n"
-            b"Content-Length: 7\r\n"
-            b"\r\n"
-            b"content\r\n"
-            b"HTTP/1.1 204 OK\r\n"
-            b"\r\n"
-        )
-        rfile = BytesIO(s)
-        r = http1.read_response(rfile, treq())
-        assert r.status_code == 200
-        assert r.content == b"content"
-        assert http1.read_response(rfile, treq()).status_code == 204
-
-        rfile = BytesIO(s)
-        # HEAD must not have content by spec. We should leave it on the pipe.
-        r = http1.read_response(rfile, treq(method=b"HEAD"))
-        assert r.status_code == 200
-        assert r.content == b""
-
-        with raises(HttpSyntaxException):
-            http1.read_response(rfile, treq())
-
-
-class TestHTTPFlow(object):
-
-    def test_repr(self):
-        f = tutils.tflow(resp=True, err=True)
-        assert repr(f)
-
-
-class TestInvalidRequests(tservers.HTTPProxTest):
-    ssl = True
-
-    def test_double_connect(self):
-        p = self.pathoc()
-        r = p.request("connect:'%s:%s'" % ("127.0.0.1", self.server2.port))
-        assert r.status_code == 400
-        assert "Invalid HTTP request form" in r.content
-
-    def test_relative_request(self):
-        p = self.pathoc_raw()
-        p.connect()
-        r = p.request("get:/p/200")
-        assert r.status_code == 400
-        assert "Invalid HTTP request form" in r.content
-
-
-class TestExpectHeader(tservers.HTTPProxTest):
-
-    def test_simple(self):
-        client = TCPClient(("127.0.0.1", self.proxy.port))
-        client.connect()
-
-        # call pathod server, wait a second to complete the request
-        client.wfile.write(
-            b"POST http://localhost:%d/p/200 HTTP/1.1\r\n"
-            b"Expect: 100-continue\r\n"
-            b"Content-Length: 16\r\n"
-            b"\r\n" % self.server.port
-        )
-        client.wfile.flush()
-
-        assert client.rfile.readline() == "HTTP/1.1 100 Continue\r\n"
-        assert client.rfile.readline() == "\r\n"
-
-        client.wfile.write(b"0123456789abcdef\r\n")
-        client.wfile.flush()
-
-        resp = http1.read_response(client.rfile, treq())
-        assert resp.status_code == 200
-
-        client.finish()
-
-
-class TestHeadContentLength(tservers.HTTPProxTest):
-
-    def test_head_content_length(self):
-        p = self.pathoc()
-        resp = p.request("""head:'%s/p/200:h"Content-Length"="42"'""" % self.server.urlbase)
-        assert resp.headers["Content-Length"] == "42"
diff --git a/test/test_protocol_http1.py b/test/test_protocol_http1.py
new file mode 100644
index 00000000..f5fe93a8
--- /dev/null
+++ b/test/test_protocol_http1.py
@@ -0,0 +1,65 @@
+from io import BytesIO
+from netlib.exceptions import HttpSyntaxException
+from netlib.http import http1
+from netlib.tcp import TCPClient
+from netlib.tutils import treq, raises
+from . import tutils, tservers
+
+
+class TestHTTPFlow(object):
+
+    def test_repr(self):
+        f = tutils.tflow(resp=True, err=True)
+        assert repr(f)
+
+
+class TestInvalidRequests(tservers.HTTPProxTest):
+    ssl = True
+
+    def test_double_connect(self):
+        p = self.pathoc()
+        r = p.request("connect:'%s:%s'" % ("127.0.0.1", self.server2.port))
+        assert r.status_code == 400
+        assert "Invalid HTTP request form" in r.content
+
+    def test_relative_request(self):
+        p = self.pathoc_raw()
+        p.connect()
+        r = p.request("get:/p/200")
+        assert r.status_code == 400
+        assert "Invalid HTTP request form" in r.content
+
+
+class TestExpectHeader(tservers.HTTPProxTest):
+
+    def test_simple(self):
+        client = TCPClient(("127.0.0.1", self.proxy.port))
+        client.connect()
+
+        # call pathod server, wait a second to complete the request
+        client.wfile.write(
+            b"POST http://localhost:%d/p/200 HTTP/1.1\r\n"
+            b"Expect: 100-continue\r\n"
+            b"Content-Length: 16\r\n"
+            b"\r\n" % self.server.port
+        )
+        client.wfile.flush()
+
+        assert client.rfile.readline() == "HTTP/1.1 100 Continue\r\n"
+        assert client.rfile.readline() == "\r\n"
+
+        client.wfile.write(b"0123456789abcdef\r\n")
+        client.wfile.flush()
+
+        resp = http1.read_response(client.rfile, treq())
+        assert resp.status_code == 200
+
+        client.finish()
+
+
+class TestHeadContentLength(tservers.HTTPProxTest):
+
+    def test_head_content_length(self):
+        p = self.pathoc()
+        resp = p.request("""head:'%s/p/200:h"Content-Length"="42"'""" % self.server.urlbase)
+        assert resp.headers["Content-Length"] == "42"
-- 
cgit v1.2.3


From a99ef584ad184658199d6ba83c73ed173f8ac0d0 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Sun, 24 Jan 2016 23:16:50 +0100
Subject: reuse frame reading snippet

---
 libmproxy/protocol/http2.py | 6 +-----
 libmproxy/utils.py          | 6 ++++++
 2 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index a6d1b73f..ff1726ae 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -206,12 +206,8 @@ class Http2Layer(Layer):
                 other_conn = self.server_conn if conn == self.client_conn.connection else self.client_conn
                 is_server = (conn == self.server_conn.connection)
 
-                field = source_conn.rfile.peek(3)
-                length = int(field.encode('hex'), 16)
-                raw_frame = source_conn.rfile.safe_read(9 + length)
-
                 with source_conn.h2.lock:
-                    events = source_conn.h2.receive_data(raw_frame)
+                    events = source_conn.h2.receive_data(utils.http2_read_frame(source_conn.rfile))
                     source_conn.send(source_conn.h2.data_to_send())
 
                     for event in events:
diff --git a/libmproxy/utils.py b/libmproxy/utils.py
index a697a637..94dbbca8 100644
--- a/libmproxy/utils.py
+++ b/libmproxy/utils.py
@@ -173,3 +173,9 @@ def safe_subn(pattern, repl, target, *args, **kwargs):
         need a better solution that is aware of the actual content ecoding.
     """
     return re.subn(str(pattern), str(repl), target, *args, **kwargs)
+
+def http2_read_frame(rfile):
+    field = rfile.peek(3)
+    length = int(field.encode('hex'), 16)
+    raw_frame = rfile.safe_read(9 + length)
+    return raw_frame
-- 
cgit v1.2.3


From f49c1cd1c5771aea8ab64bd93619e3b8f0729253 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Sun, 24 Jan 2016 23:53:23 +0100
Subject: improve http2 header parsing

---
 libmproxy/protocol/http2.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index ff1726ae..03408142 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -260,6 +260,9 @@ class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
             # FIXME: verify if path or :host contains what we need
             scheme, host, port, _ = utils.parse_url(path)
 
+        if authority:
+            host, port = authority.split(':')
+
         if host is None:
             host = 'localhost'
         if port is None:
-- 
cgit v1.2.3


From 4501c8a0a158f15386d9dfe4884a9ffacb8ffd0b Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Sun, 24 Jan 2016 23:53:32 +0100
Subject: add http2 full-stack test

---
 test/test_protocol_http2.py | 117 ++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 117 insertions(+)
 create mode 100644 test/test_protocol_http2.py

diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
new file mode 100644
index 00000000..17b8506d
--- /dev/null
+++ b/test/test_protocol_http2.py
@@ -0,0 +1,117 @@
+from __future__ import (absolute_import, print_function, division)
+
+import inspect
+import socket
+from io import BytesIO
+
+import logging
+logging.getLogger("hyper.packages.hpack.hpack").setLevel(logging.WARNING)
+
+import netlib
+from netlib import tservers as netlib_tservers
+
+import h2
+
+from libmproxy import utils
+from . import tservers
+
+class SimpleHttp2Server(netlib_tservers.ServerTestBase):
+    ssl = dict(
+        alpn_select=b'h2',
+    )
+
+    class handler(netlib.tcp.BaseHandler):
+        def handle(self):
+            h2_conn = h2.connection.H2Connection(client_side=False)
+
+            preamble = self.rfile.read(24)
+            h2_conn.initiate_connection()
+            h2_conn.receive_data(preamble)
+            self.wfile.write(h2_conn.data_to_send())
+            self.wfile.flush()
+
+            while True:
+                events = h2_conn.receive_data(utils.http2_read_frame(self.rfile))
+                self.wfile.write(h2_conn.data_to_send())
+                self.wfile.flush()
+
+                for event in events:
+                    if isinstance(event, h2.events.RequestReceived):
+                        h2_conn.send_headers(1, [
+                            (':status', '200'),
+                            ('foo', 'bar'),
+                        ])
+                        h2_conn.send_data(1, b'foobar')
+                        h2_conn.end_stream(1)
+                        self.wfile.write(h2_conn.data_to_send())
+                        self.wfile.flush()
+                    elif isinstance(event, h2.events.ConnectionTerminated):
+                        return
+
+
+class TestHttp2(tservers.ProxTestBase):
+    def _setup_connection(self):
+        self.config.http2 = True
+
+        client = netlib.tcp.TCPClient(("127.0.0.1", self.proxy.port))
+        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.port, self.server.port)
+        )
+        client.wfile.flush()
+
+        # read CONNECT response
+        while client.rfile.readline() != "\r\n":
+            pass
+
+        client.convert_to_ssl(alpn_protos=[b'h2'])
+
+        h2_conn = h2.connection.H2Connection(client_side=True)
+        h2_conn.initiate_connection()
+        client.wfile.write(h2_conn.data_to_send())
+        client.wfile.flush()
+
+        return client, h2_conn
+
+    def _send_request(self, wfile, h2_conn, stream_id=1, headers=[], end_stream=True):
+        h2_conn.send_headers(
+            stream_id=stream_id,
+            headers=headers,
+            end_stream=end_stream,
+        )
+        wfile.write(h2_conn.data_to_send())
+        wfile.flush()
+
+    def test_simple(self):
+        self.server = SimpleHttp2Server()
+        self.server.setup_class()
+
+        client, h2_conn = self._setup_connection()
+
+        self._send_request(client.wfile, h2_conn, headers=[
+            (':authority', "127.0.0.1:%s" % self.server.port),
+            (':method', 'GET'),
+            (':scheme', 'https'),
+            (':path', '/'),
+        ])
+
+        done = False
+        while not done:
+            events = h2_conn.receive_data(utils.http2_read_frame(client.rfile))
+            client.wfile.write(h2_conn.data_to_send())
+            client.wfile.flush()
+
+            for event in events:
+                if isinstance(event, h2.events.StreamEnded):
+                    done = True
+
+        self.server.teardown_class()
+
+        assert len(self.master.state.flows) == 1
+        assert self.master.state.flows[0].response.status_code == 200
+        assert self.master.state.flows[0].response.headers['foo'] == 'bar'
+        assert self.master.state.flows[0].response.body == b'foobar'
-- 
cgit v1.2.3


From bfc7d3967c0978b22faeaedc653a695e65156b34 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 25 Jan 2016 19:09:14 +0100
Subject: exclude tests if no alpn support present

---
 test/test_protocol_http2.py | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index 17b8506d..d3725e81 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -2,6 +2,8 @@ from __future__ import (absolute_import, print_function, division)
 
 import inspect
 import socket
+import OpenSSL
+import pytest
 from io import BytesIO
 
 import logging
@@ -15,6 +17,11 @@ import h2
 from libmproxy import utils
 from . import tservers
 
+requires_alpn = pytest.mark.skipif(
+    not OpenSSL._util.lib.Cryptography_HAS_ALPN,
+    reason="requires OpenSSL with ALPN support")
+
+
 class SimpleHttp2Server(netlib_tservers.ServerTestBase):
     ssl = dict(
         alpn_select=b'h2',
@@ -49,6 +56,7 @@ class SimpleHttp2Server(netlib_tservers.ServerTestBase):
                         return
 
 
+@requires_alpn
 class TestHttp2(tservers.ProxTestBase):
     def _setup_connection(self):
         self.config.http2 = True
-- 
cgit v1.2.3


From 47cf27c01146e87b8bbf315f70e9752a38ba8edd Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 25 Jan 2016 19:20:44 +0100
Subject: silence 3rd party module loggers

---
 test/test_protocol_http2.py | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index d3725e81..a7e8978a 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -8,6 +8,11 @@ from io import BytesIO
 
 import logging
 logging.getLogger("hyper.packages.hpack.hpack").setLevel(logging.WARNING)
+logging.getLogger("requests.packages.urllib3.connectionpool").setLevel(logging.WARNING)
+logging.getLogger("passlib.utils.compat").setLevel(logging.WARNING)
+logging.getLogger("passlib.registry").setLevel(logging.WARNING)
+logging.getLogger("PIL.Image").setLevel(logging.WARNING)
+logging.getLogger("PIL.PngImagePlugin").setLevel(logging.WARNING)
 
 import netlib
 from netlib import tservers as netlib_tservers
-- 
cgit v1.2.3


From 4e9579e93eee4a156fa60c1d58dcc9b8c113d367 Mon Sep 17 00:00:00 2001
From: Maximilian Hils <git@maximilianhils.com>
Date: Mon, 11 Jan 2016 15:40:09 +0100
Subject: try to fix travis

---
 .travis.yml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/.travis.yml b/.travis.yml
index ed5a7ad5..e86a4117 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -45,6 +45,7 @@ install:
       brew outdated openssl || brew upgrade openssl
       brew install python
     fi
+  - "pip install -U pip setuptools"
   - "pip install --src .. -r requirements.txt"
 
 before_script:
-- 
cgit v1.2.3


From 735c79a2edb3b31a35ed3e484744807bb626ab77 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 25 Jan 2016 19:41:22 +0100
Subject: increase coverage

---
 .travis.yml                 |  2 +-
 libmproxy/contentviews.py   |  4 ++--
 libmproxy/controller.py     |  2 +-
 libmproxy/main.py           |  6 +++---
 libmproxy/protocol/base.py  |  2 +-
 libmproxy/protocol/http2.py | 13 +++++++------
 libmproxy/protocol/tls.py   |  2 +-
 libmproxy/proxy/server.py   |  4 ++--
 test/test_protocol_http2.py |  4 ++++
 9 files changed, 22 insertions(+), 17 deletions(-)

diff --git a/.travis.yml b/.travis.yml
index e86a4117..ed6d8e11 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -52,7 +52,7 @@ before_script:
   - "openssl version -a"
 
 script:
-  - "py.test -n 4 --cov libmproxy"
+  - "py.test -n 4 -s --cov libmproxy"
 
 after_success:
   - coveralls
diff --git a/libmproxy/contentviews.py b/libmproxy/contentviews.py
index 80955b0f..c0652c18 100644
--- a/libmproxy/contentviews.py
+++ b/libmproxy/contentviews.py
@@ -35,12 +35,12 @@ from .contrib.wbxml.ASCommandResponse import ASCommandResponse
 try:
     import pyamf
     from pyamf import remoting, flex
-except ImportError:  # pragma nocover
+except ImportError:  # pragma no cover
     pyamf = None
 
 try:
     import cssutils
-except ImportError:  # pragma nocover
+except ImportError:  # pragma no cover
     cssutils = None
 else:
     cssutils.log.setLevel(logging.CRITICAL)
diff --git a/libmproxy/controller.py b/libmproxy/controller.py
index 712ab1d2..9a059856 100644
--- a/libmproxy/controller.py
+++ b/libmproxy/controller.py
@@ -56,7 +56,7 @@ class Channel:
             try:
                 # The timeout is here so we can handle a should_exit event.
                 g = m.reply.q.get(timeout=0.5)
-            except Queue.Empty:  # pragma: nocover
+            except Queue.Empty:  # pragma: no cover
                 continue
             return g
 
diff --git a/libmproxy/main.py b/libmproxy/main.py
index 655d573d..1c3cbf78 100644
--- a/libmproxy/main.py
+++ b/libmproxy/main.py
@@ -37,7 +37,7 @@ def get_server(dummy_server, options):
             sys.exit(1)
 
 
-def mitmproxy(args=None):  # pragma: nocover
+def mitmproxy(args=None):  # pragma: no cover
     from . import console
 
     check_pyopenssl_version()
@@ -68,7 +68,7 @@ def mitmproxy(args=None):  # pragma: nocover
         pass
 
 
-def mitmdump(args=None):  # pragma: nocover
+def mitmdump(args=None):  # pragma: no cover
     from . import dump
 
     check_pyopenssl_version()
@@ -103,7 +103,7 @@ def mitmdump(args=None):  # pragma: nocover
         pass
 
 
-def mitmweb(args=None):  # pragma: nocover
+def mitmweb(args=None):  # pragma: no cover
     from . import web
 
     check_pyopenssl_version()
diff --git a/libmproxy/protocol/base.py b/libmproxy/protocol/base.py
index 4eb034c0..40fcaf65 100644
--- a/libmproxy/protocol/base.py
+++ b/libmproxy/protocol/base.py
@@ -14,7 +14,7 @@ class _LayerCodeCompletion(object):
     Dummy class that provides type hinting in PyCharm, which simplifies development a lot.
     """
 
-    def __init__(self, **mixin_args):  # pragma: nocover
+    def __init__(self, **mixin_args):  # pragma: no cover
         super(_LayerCodeCompletion, self).__init__(**mixin_args)
         if True:
             return
diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index 03408142..54e7572e 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -249,12 +249,13 @@ class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
 
         if path == '*' or path.startswith("/"):
             form_in = "relative"
-        elif method == 'CONNECT':
-            form_in = "authority"
-            if ":" in authority:
-                host, port = authority.split(":", 1)
-            else:
-                host = authority
+        elif method == 'CONNECT':  # pragma: no cover
+            # form_in = "authority"
+            # if ":" in authority:
+            #     host, port = authority.split(":", 1)
+            # else:
+            #     host = authority
+            raise NotImplementedError("CONNECT over HTTP/2 is not implemented.")
         else:
             form_in = "absolute"
             # FIXME: verify if path or :host contains what we need
diff --git a/libmproxy/protocol/tls.py b/libmproxy/protocol/tls.py
index af1a6055..ccae1661 100644
--- a/libmproxy/protocol/tls.py
+++ b/libmproxy/protocol/tls.py
@@ -349,7 +349,7 @@ class TlsLayer(Layer):
         layer = self.ctx.next_layer(self)
         layer()
 
-    def __repr__(self):
+    def __repr__(self):  # pragma: no cover
         if self._client_tls and self._server_tls:
             return "TlsLayer(client and server)"
         elif self._client_tls:
diff --git a/libmproxy/proxy/server.py b/libmproxy/proxy/server.py
index 750cb1a4..d208cff5 100644
--- a/libmproxy/proxy/server.py
+++ b/libmproxy/proxy/server.py
@@ -103,9 +103,9 @@ class ConnectionHandler(object):
             return Socks5Proxy(root_context)
         elif mode == "regular":
             return HttpProxy(root_context)
-        elif callable(mode):  # pragma: nocover
+        elif callable(mode):  # pragma: no cover
             return mode(root_context)
-        else:  # pragma: nocover
+        else:  # pragma: no cover
             raise ValueError("Unknown proxy mode: %s" % mode)
 
     def handle(self):
diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index a7e8978a..e72113c4 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -122,6 +122,10 @@ class TestHttp2(tservers.ProxTestBase):
                 if isinstance(event, h2.events.StreamEnded):
                     done = True
 
+        h2_conn.close_connection()
+        client.wfile.write(h2_conn.data_to_send())
+        client.wfile.flush()
+
         self.server.teardown_class()
 
         assert len(self.master.state.flows) == 1
-- 
cgit v1.2.3


From 8d14dd33d083eafb4f12cd88269f1b11b1a65ee8 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 25 Jan 2016 19:55:59 +0100
Subject: try to show weird test output

---
 .travis.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.travis.yml b/.travis.yml
index ed6d8e11..2233f67b 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -52,7 +52,7 @@ before_script:
   - "openssl version -a"
 
 script:
-  - "py.test -n 4 -s --cov libmproxy"
+  - "py.test -s --cov libmproxy"
 
 after_success:
   - coveralls
-- 
cgit v1.2.3


From ef7b4f56aff3251fd5ae69f5406c17816f5fd558 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Wed, 27 Jan 2016 10:57:07 +0100
Subject: update CI integration

---
 .appveyor.yml | 11 +++++++----
 .travis.yml   | 22 ++++++++++++++++++----
 2 files changed, 25 insertions(+), 8 deletions(-)

diff --git a/.appveyor.yml b/.appveyor.yml
index 914e75eb..55aed6aa 100644
--- a/.appveyor.yml
+++ b/.appveyor.yml
@@ -1,16 +1,19 @@
 version: '{build}'
 shallow_clone: true
+build: off  # Not a C# project
 environment:
   matrix:
     - PYTHON: "C:\\Python27"
-  PATH: "C:\\Python27;C:\\Python27\\Scripts;%PATH%"
+  PATH: "%APPDATA%\\Python\\Scripts;C:\\Python27;C:\\Python27\\Scripts;%PATH%"
   PYINSTALLER_VERSION: "git+https://github.com/pyinstaller/pyinstaller.git"
 install:
-  - "pip install --src .. -r requirements.txt"
+  - "pip install --user -U pip setuptools"
+  - "pip install --user --src .. -r requirements.txt"
   - "python -c \"from OpenSSL import SSL; print(SSL.SSLeay_version(SSL.SSLEAY_VERSION))\""
-build: off  # Not a C# project
 test_script:
-  - "py.test -n 4"
+  - "py.test -s --cov libmproxy"
+cache:
+  - C:\Users\appveyor\AppData\Local\pip\cache
 after_test:
   - |
     git clone https://github.com/mitmproxy/release.git ..\release
diff --git a/.travis.yml b/.travis.yml
index 2233f67b..9091887a 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -45,6 +45,19 @@ install:
       brew outdated openssl || brew upgrade openssl
       brew install python
     fi
+  - |
+      if [ "$TRAVIS_PYTHON_VERSION" = "pypy" ]; then
+        export PYENV_ROOT="$HOME/.pyenv"
+        if [ -f "$PYENV_ROOT/bin/pyenv" ]; then
+          pushd "$PYENV_ROOT" && git pull && popd
+        else
+          rm -rf "$PYENV_ROOT" && git clone --depth 1 https://github.com/yyuu/pyenv.git "$PYENV_ROOT"
+        fi
+        export PYPY_VERSION="4.0.1"
+        "$PYENV_ROOT/bin/pyenv" install --skip-existing "pypy-$PYPY_VERSION"
+        virtualenv --python="$PYENV_ROOT/versions/pypy-$PYPY_VERSION/bin/python" "$HOME/virtualenvs/pypy-$PYPY_VERSION"
+        source "$HOME/virtualenvs/pypy-$PYPY_VERSION/bin/activate"
+      fi
   - "pip install -U pip setuptools"
   - "pip install --src .. -r requirements.txt"
 
@@ -90,7 +103,8 @@ before_cache:
 cache:
   directories:
     - $HOME/.cache/pip
-    - /home/travis/virtualenv/python2.7.9/lib/python2.7/site-packages
-    - /home/travis/virtualenv/python2.7.9/bin
-    - /home/travis/virtualenv/pypy-2.5.0/site-packages
-    - /home/travis/virtualenv/pypy-2.5.0/bin
+    - $HOME/virtualenv/python2.7.9/lib/python2.7/site-packages
+    - $HOME/virtualenv/python2.7.9/bin
+    - $HOME/.pyenv
+    - $HOME/virtualenv/pypy-2.5.0/site-packages
+    - $HOME/virtualenv/pypy-2.5.0/bin
-- 
cgit v1.2.3


From 97c2530f90e8ddf0b36539408372f21f5964e9bb Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Sat, 30 Jan 2016 22:05:37 +0100
Subject: allow pypy on travis

---
 .travis.yml | 16 ++++++----------
 1 file changed, 6 insertions(+), 10 deletions(-)

diff --git a/.travis.yml b/.travis.yml
index 9091887a..2545899c 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -5,9 +5,6 @@ matrix:
   fast_finish: true
   include:
     - python: 2.7
-    - language: generic
-      os: osx
-      osx_image: xcode7.1
     - python: 2.7
       env: OPENSSL=1.0.2
       addons:
@@ -18,9 +15,6 @@ matrix:
             - debian-sid
           packages:
             - libssl-dev
-    - python: 2.7
-      env: DOCS=1
-      script: 'cd docs && make html'
     - python: pypy
     - python: pypy
       env: OPENSSL=1.0.2
@@ -32,10 +26,12 @@ matrix:
             - debian-sid
           packages:
             - libssl-dev
-  allow_failures:
-    # We allow pypy to fail until Travis fixes their infrastructure to a pypy
-    # with a recent enought CFFI library to run cryptography 1.0+.
-    - python: pypy
+    - language: generic
+      os: osx
+      osx_image: xcode7.1
+    - python: 2.7
+      env: DOCS=1
+      script: 'cd docs && make html'
 
 install:
   - |
-- 
cgit v1.2.3


From 41f4197a0dd73a2b00ea8485608ba9b05a605dd4 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 25 Jan 2016 21:14:58 +0100
Subject: test PushPromise support

---
 libmproxy/protocol/http2.py | 11 +++++-
 test/test_protocol_http2.py | 91 +++++++++++++++++++++++++++++++++++++++++++--
 2 files changed, 97 insertions(+), 5 deletions(-)

diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index 54e7572e..71423bf7 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -17,7 +17,7 @@ from .base import Layer
 from .http import _HttpTransmissionLayer, HttpLayer
 from .. import utils
 from ..models import HTTPRequest, HTTPResponse
-
+from ..exceptions import HttpProtocolException, ProtocolException
 
 class SafeH2Connection(H2Connection):
     def __init__(self, conn, *args, **kwargs):
@@ -207,7 +207,14 @@ class Http2Layer(Layer):
                 is_server = (conn == self.server_conn.connection)
 
                 with source_conn.h2.lock:
-                    events = source_conn.h2.receive_data(utils.http2_read_frame(source_conn.rfile))
+                    try:
+                        raw_frame = utils.http2_read_frame(source_conn.rfile)
+                    except:
+                        for stream in self.streams.values():
+                            stream.zombie = time.time()
+                        return
+
+                    events = source_conn.h2.receive_data(raw_frame)
                     source_conn.send(source_conn.h2.data_to_send())
 
                     for event in events:
diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index e72113c4..4fa2c701 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -28,9 +28,7 @@ requires_alpn = pytest.mark.skipif(
 
 
 class SimpleHttp2Server(netlib_tservers.ServerTestBase):
-    ssl = dict(
-        alpn_select=b'h2',
-    )
+    ssl = dict(alpn_select=b'h2')
 
     class handler(netlib.tcp.BaseHandler):
         def handle(self):
@@ -61,6 +59,59 @@ class SimpleHttp2Server(netlib_tservers.ServerTestBase):
                         return
 
 
+class PushHttp2Server(netlib_tservers.ServerTestBase):
+    ssl = dict(alpn_select=b'h2')
+
+    class handler(netlib.tcp.BaseHandler):
+        def handle(self):
+            h2_conn = h2.connection.H2Connection(client_side=False)
+
+            preamble = self.rfile.read(24)
+            h2_conn.initiate_connection()
+            h2_conn.receive_data(preamble)
+            self.wfile.write(h2_conn.data_to_send())
+            self.wfile.flush()
+
+            while True:
+                events = h2_conn.receive_data(utils.http2_read_frame(self.rfile))
+                self.wfile.write(h2_conn.data_to_send())
+                self.wfile.flush()
+
+                for event in events:
+                    if isinstance(event, h2.events.RequestReceived):
+                        h2_conn.send_headers(1, [(':status', '200')])
+                        h2_conn.push_stream(1, 2, [
+                            (':authority', "127.0.0.1:%s" % self.address.port),
+                            (':method', 'GET'),
+                            (':scheme', 'https'),
+                            (':path', '/pushed_stream_foo'),
+                            ('foo', 'bar')
+                        ])
+                        h2_conn.push_stream(1, 4, [
+                            (':authority', "127.0.0.1:%s" % self.address.port),
+                            (':method', 'GET'),
+                            (':scheme', 'https'),
+                            (':path', '/pushed_stream_bar'),
+                            ('foo', 'bar')
+                        ])
+                        self.wfile.write(h2_conn.data_to_send())
+                        self.wfile.flush()
+
+                        h2_conn.send_headers(2, [(':status', '202')])
+                        h2_conn.send_headers(4, [(':status', '204')])
+                        h2_conn.send_data(1, b'regular_stream')
+                        h2_conn.send_data(2, b'pushed_stream_foo')
+                        h2_conn.send_data(4, b'pushed_stream_bar')
+                        h2_conn.end_stream(1)
+                        h2_conn.end_stream(2)
+                        h2_conn.end_stream(4)
+                        self.wfile.write(h2_conn.data_to_send())
+                        self.wfile.flush()
+                        print("HERE")
+                    elif isinstance(event, h2.events.ConnectionTerminated):
+                        return
+
+
 @requires_alpn
 class TestHttp2(tservers.ProxTestBase):
     def _setup_connection(self):
@@ -132,3 +183,37 @@ class TestHttp2(tservers.ProxTestBase):
         assert self.master.state.flows[0].response.status_code == 200
         assert self.master.state.flows[0].response.headers['foo'] == 'bar'
         assert self.master.state.flows[0].response.body == b'foobar'
+
+    def test_pushed_streams(self):
+        self.server = PushHttp2Server()
+        self.server.setup_class()
+
+        client, h2_conn = self._setup_connection()
+
+        self._send_request(client.wfile, h2_conn, headers=[
+            (':authority', "127.0.0.1:%s" % self.server.port),
+            (':method', 'GET'),
+            (':scheme', 'https'),
+            (':path', '/'),
+            ('foo', 'bar')
+        ])
+
+        ended_streams = 0
+        while ended_streams != 3:
+            try:
+                events = h2_conn.receive_data(utils.http2_read_frame(client.rfile))
+            except:
+                break
+            client.wfile.write(h2_conn.data_to_send())
+            client.wfile.flush()
+
+            for event in events:
+                if isinstance(event, h2.events.StreamEnded):
+                    ended_streams += 1
+
+        self.server.teardown_class()
+
+        assert len(self.master.state.flows) == 3
+        assert self.master.state.flows[0].response.body == b'regular_stream'
+        assert self.master.state.flows[1].response.body == b'pushed_stream_foo'
+        assert self.master.state.flows[2].response.body == b'pushed_stream_bar'
-- 
cgit v1.2.3


From 187691e65bf4a18de3567d6d801d78aa721b9fa5 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 25 Jan 2016 23:04:15 +0100
Subject: remove print

---
 libmproxy/utils.py          | 4 ++++
 test/test_protocol_http2.py | 1 -
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/libmproxy/utils.py b/libmproxy/utils.py
index 94dbbca8..299e8749 100644
--- a/libmproxy/utils.py
+++ b/libmproxy/utils.py
@@ -177,5 +177,9 @@ def safe_subn(pattern, repl, target, *args, **kwargs):
 def http2_read_frame(rfile):
     field = rfile.peek(3)
     length = int(field.encode('hex'), 16)
+
+    if length == 4740180:
+        raise ValueError("Probably not the correct length bytes: %s" % rfile.peek(20))
+
     raw_frame = rfile.safe_read(9 + length)
     return raw_frame
diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index 4fa2c701..17687b45 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -107,7 +107,6 @@ class PushHttp2Server(netlib_tservers.ServerTestBase):
                         h2_conn.end_stream(4)
                         self.wfile.write(h2_conn.data_to_send())
                         self.wfile.flush()
-                        print("HERE")
                     elif isinstance(event, h2.events.ConnectionTerminated):
                         return
 
-- 
cgit v1.2.3


From 276817e40e99dbb2ddc7638839bd74e944fd704e Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Tue, 26 Jan 2016 13:15:20 +0100
Subject: refactor http2 tests

---
 test/test_protocol_http2.py | 234 ++++++++++++++++++++++++++++----------------
 1 file changed, 149 insertions(+), 85 deletions(-)

diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index 17687b45..b42b86cb 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -4,8 +4,16 @@ import inspect
 import socket
 import OpenSSL
 import pytest
+import traceback
+import os
+import tempfile
+
 from io import BytesIO
 
+from libmproxy.proxy.config import ProxyConfig
+from libmproxy.proxy.server import ProxyServer
+from libmproxy.cmdline import APP_HOST, APP_PORT
+
 import logging
 logging.getLogger("hyper.packages.hpack.hpack").setLevel(logging.WARNING)
 logging.getLogger("requests.packages.urllib3.connectionpool").setLevel(logging.WARNING)
@@ -18,6 +26,7 @@ import netlib
 from netlib import tservers as netlib_tservers
 
 import h2
+from hyperframe.frame import Frame
 
 from libmproxy import utils
 from . import tservers
@@ -26,8 +35,7 @@ requires_alpn = pytest.mark.skipif(
     not OpenSSL._util.lib.Cryptography_HAS_ALPN,
     reason="requires OpenSSL with ALPN support")
 
-
-class SimpleHttp2Server(netlib_tservers.ServerTestBase):
+class _Http2ServerBase(netlib_tservers.ServerTestBase):
     ssl = dict(alpn_select=b'h2')
 
     class handler(netlib.tcp.BaseHandler):
@@ -41,78 +49,56 @@ class SimpleHttp2Server(netlib_tservers.ServerTestBase):
             self.wfile.flush()
 
             while True:
-                events = h2_conn.receive_data(utils.http2_read_frame(self.rfile))
+                raw_frame = utils.http2_read_frame(self.rfile)
+                events = h2_conn.receive_data(raw_frame)
                 self.wfile.write(h2_conn.data_to_send())
                 self.wfile.flush()
 
                 for event in events:
-                    if isinstance(event, h2.events.RequestReceived):
-                        h2_conn.send_headers(1, [
-                            (':status', '200'),
-                            ('foo', 'bar'),
-                        ])
-                        h2_conn.send_data(1, b'foobar')
-                        h2_conn.end_stream(1)
-                        self.wfile.write(h2_conn.data_to_send())
-                        self.wfile.flush()
-                    elif isinstance(event, h2.events.ConnectionTerminated):
-                        return
-
-
-class PushHttp2Server(netlib_tservers.ServerTestBase):
-    ssl = dict(alpn_select=b'h2')
-
-    class handler(netlib.tcp.BaseHandler):
-        def handle(self):
-            h2_conn = h2.connection.H2Connection(client_side=False)
-
-            preamble = self.rfile.read(24)
-            h2_conn.initiate_connection()
-            h2_conn.receive_data(preamble)
-            self.wfile.write(h2_conn.data_to_send())
-            self.wfile.flush()
-
-            while True:
-                events = h2_conn.receive_data(utils.http2_read_frame(self.rfile))
-                self.wfile.write(h2_conn.data_to_send())
-                self.wfile.flush()
-
-                for event in events:
-                    if isinstance(event, h2.events.RequestReceived):
-                        h2_conn.send_headers(1, [(':status', '200')])
-                        h2_conn.push_stream(1, 2, [
-                            (':authority', "127.0.0.1:%s" % self.address.port),
-                            (':method', 'GET'),
-                            (':scheme', 'https'),
-                            (':path', '/pushed_stream_foo'),
-                            ('foo', 'bar')
-                        ])
-                        h2_conn.push_stream(1, 4, [
-                            (':authority', "127.0.0.1:%s" % self.address.port),
-                            (':method', 'GET'),
-                            (':scheme', 'https'),
-                            (':path', '/pushed_stream_bar'),
-                            ('foo', 'bar')
-                        ])
-                        self.wfile.write(h2_conn.data_to_send())
-                        self.wfile.flush()
-
-                        h2_conn.send_headers(2, [(':status', '202')])
-                        h2_conn.send_headers(4, [(':status', '204')])
-                        h2_conn.send_data(1, b'regular_stream')
-                        h2_conn.send_data(2, b'pushed_stream_foo')
-                        h2_conn.send_data(4, b'pushed_stream_bar')
-                        h2_conn.end_stream(1)
-                        h2_conn.end_stream(2)
-                        h2_conn.end_stream(4)
-                        self.wfile.write(h2_conn.data_to_send())
-                        self.wfile.flush()
-                    elif isinstance(event, h2.events.ConnectionTerminated):
-                        return
+                    try:
+                        if not self.server.handle_server_event(event, h2_conn, self.rfile, self.wfile):
+                            break
+                    except Exception as e:
+                        print(repr(e))
+                        print(traceback.format_exc())
+                        break
+
+    def handle_server_event(self, h2_conn, rfile, wfile):
+        raise NotImplementedError()
+
+
+class _Http2TestBase(object):
+    @classmethod
+    def setup_class(self):
+        self.config = ProxyConfig(**self.get_proxy_config())
+
+        tmaster = tservers.TestMaster(self.config)
+        tmaster.start_app(APP_HOST, APP_PORT)
+        self.proxy = tservers.ProxyThread(tmaster)
+        self.proxy.start()
+
+    @classmethod
+    def teardown_class(cls):
+        cls.proxy.shutdown()
+
+    @property
+    def master(self):
+        return self.proxy.tmaster
+
+    @classmethod
+    def get_proxy_config(cls):
+        cls.cadir = os.path.join(tempfile.gettempdir(), "mitmproxy")
+        return dict(
+            no_upstream_cert = False,
+            cadir = cls.cadir,
+            authenticator = None,
+        )
 
+    def setup(self):
+        self.master.clear_log()
+        self.master.state.clear()
+        self.server.server.handle_server_event = self.handle_server_event
 
-@requires_alpn
-class TestHttp2(tservers.ProxTestBase):
     def _setup_connection(self):
         self.config.http2 = True
 
@@ -123,7 +109,7 @@ class TestHttp2(tservers.ProxTestBase):
         client.wfile.write(
             b"CONNECT localhost:%d HTTP/1.1\r\n"
             b"Host: localhost:%d\r\n"
-            b"\r\n" % (self.server.port, self.server.port)
+            b"\r\n" % (self.server.server.address.port, self.server.server.address.port)
         )
         client.wfile.flush()
 
@@ -149,14 +135,40 @@ class TestHttp2(tservers.ProxTestBase):
         wfile.write(h2_conn.data_to_send())
         wfile.flush()
 
-    def test_simple(self):
-        self.server = SimpleHttp2Server()
-        self.server.setup_class()
 
+@requires_alpn
+class TestSimple(_Http2TestBase, _Http2ServerBase):
+    @classmethod
+    def setup_class(self):
+        _Http2TestBase.setup_class()
+        _Http2ServerBase.setup_class()
+
+    @classmethod
+    def teardown_class(self):
+        _Http2TestBase.teardown_class()
+        _Http2ServerBase.teardown_class()
+
+    @classmethod
+    def handle_server_event(self, event, h2_conn, rfile, wfile):
+        if isinstance(event, h2.events.ConnectionTerminated):
+            return False
+        elif isinstance(event, h2.events.RequestReceived):
+            h2_conn.send_headers(1, [
+                (':status', '200'),
+                ('foo', 'bar'),
+            ])
+            h2_conn.send_data(1, b'foobar')
+            h2_conn.end_stream(1)
+            wfile.write(h2_conn.data_to_send())
+            wfile.flush()
+
+        return True
+
+    def test_simple(self):
         client, h2_conn = self._setup_connection()
 
         self._send_request(client.wfile, h2_conn, headers=[
-            (':authority', "127.0.0.1:%s" % self.server.port),
+            (':authority', "127.0.0.1:%s" % self.server.server.address.port),
             (':method', 'GET'),
             (':scheme', 'https'),
             (':path', '/'),
@@ -176,21 +188,69 @@ class TestHttp2(tservers.ProxTestBase):
         client.wfile.write(h2_conn.data_to_send())
         client.wfile.flush()
 
-        self.server.teardown_class()
-
         assert len(self.master.state.flows) == 1
         assert self.master.state.flows[0].response.status_code == 200
         assert self.master.state.flows[0].response.headers['foo'] == 'bar'
         assert self.master.state.flows[0].response.body == b'foobar'
 
-    def test_pushed_streams(self):
-        self.server = PushHttp2Server()
-        self.server.setup_class()
 
+@requires_alpn
+class TestPushPromise(_Http2TestBase, _Http2ServerBase):
+    @classmethod
+    def setup_class(self):
+        _Http2TestBase.setup_class()
+        _Http2ServerBase.setup_class()
+
+    @classmethod
+    def teardown_class(self):
+        _Http2TestBase.teardown_class()
+        _Http2ServerBase.teardown_class()
+
+    @classmethod
+    def handle_server_event(self, event, h2_conn, rfile, wfile):
+        if isinstance(event, h2.events.ConnectionTerminated):
+            return False
+        elif isinstance(event, h2.events.RequestReceived):
+            if event.stream_id != 1:
+                # ignore requests initiated by push promises
+                return True
+
+            h2_conn.send_headers(1, [(':status', '200')])
+            h2_conn.push_stream(1, 2, [
+                (':authority', "127.0.0.1:%s" % self.port),
+                (':method', 'GET'),
+                (':scheme', 'https'),
+                (':path', '/pushed_stream_foo'),
+                ('foo', 'bar')
+            ])
+            h2_conn.push_stream(1, 4, [
+                (':authority', "127.0.0.1:%s" % self.port),
+                (':method', 'GET'),
+                (':scheme', 'https'),
+                (':path', '/pushed_stream_bar'),
+                ('foo', 'bar')
+            ])
+            wfile.write(h2_conn.data_to_send())
+            wfile.flush()
+
+            h2_conn.send_headers(2, [(':status', '202')])
+            h2_conn.send_headers(4, [(':status', '204')])
+            h2_conn.send_data(1, b'regular_stream')
+            h2_conn.send_data(2, b'pushed_stream_foo')
+            h2_conn.send_data(4, b'pushed_stream_bar')
+            h2_conn.end_stream(1)
+            h2_conn.end_stream(2)
+            h2_conn.end_stream(4)
+            wfile.write(h2_conn.data_to_send())
+            wfile.flush()
+
+        return True
+
+    def test_push_promise(self):
         client, h2_conn = self._setup_connection()
 
-        self._send_request(client.wfile, h2_conn, headers=[
-            (':authority', "127.0.0.1:%s" % self.server.port),
+        self._send_request(client.wfile, h2_conn, stream_id=1, headers=[
+            (':authority', "127.0.0.1:%s" % self.server.server.address.port),
             (':method', 'GET'),
             (':scheme', 'https'),
             (':path', '/'),
@@ -198,6 +258,7 @@ class TestHttp2(tservers.ProxTestBase):
         ])
 
         ended_streams = 0
+        pushed_streams = 0
         while ended_streams != 3:
             try:
                 events = h2_conn.receive_data(utils.http2_read_frame(client.rfile))
@@ -209,10 +270,13 @@ class TestHttp2(tservers.ProxTestBase):
             for event in events:
                 if isinstance(event, h2.events.StreamEnded):
                     ended_streams += 1
+                elif isinstance(event, h2.events.PushedStreamReceived):
+                    pushed_streams += 1
 
-        self.server.teardown_class()
+        assert pushed_streams == 2
 
-        assert len(self.master.state.flows) == 3
-        assert self.master.state.flows[0].response.body == b'regular_stream'
-        assert self.master.state.flows[1].response.body == b'pushed_stream_foo'
-        assert self.master.state.flows[2].response.body == b'pushed_stream_bar'
+        bodies = [flow.response.body for flow in self.master.state.flows]
+        assert len(bodies) == 3
+        assert b'regular_stream' in bodies
+        assert b'pushed_stream_foo' in bodies
+        assert b'pushed_stream_bar' in bodies
-- 
cgit v1.2.3


From cd2b4ea058e82ef9e2dc824a379309d68f08cec4 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Tue, 26 Jan 2016 19:38:17 +0100
Subject: bump h2 dependency and use latest API

---
 libmproxy/protocol/http2.py | 2 +-
 setup.py                    | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index 71423bf7..23004497 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -144,7 +144,7 @@ class Http2Layer(Layer):
                 raise HttpException("HTTP body too large. Limit is {}.".format(self.config.body_size_limit))
             self.streams[eid].data_queue.put(event.data)
             self.streams[eid].queued_data_length += len(event.data)
-            source_conn.h2.safe_increment_flow_control(event.stream_id, len(event.data))
+            source_conn.h2.safe_increment_flow_control(event.stream_id, event.flow_controlled_length)
         elif isinstance(event, StreamEnded):
             self.streams[eid].timestamp_end = time.time()
             self.streams[eid].data_finished.set()
diff --git a/setup.py b/setup.py
index c7686c6e..522857fd 100644
--- a/setup.py
+++ b/setup.py
@@ -17,7 +17,7 @@ with open(os.path.join(here, 'README.rst'), encoding='utf-8') as f:
 # This will break `pip install` on systems with old setuptools versions.
 deps = {
     "netlib>=%s, <%s" % (version.MINORVERSION, version.NEXT_MINORVERSION),
-    "h2>=2.0.0",
+    "h2>=2.1.0, <3.0",
     "tornado>=4.3.0, <4.4",
     "configargparse>=0.10.0, <0.11",
     "pyperclip>=1.5.22, <1.6",
-- 
cgit v1.2.3


From bd1d9e28e4aa942b6fdaac0d327d9147ee911ab1 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Tue, 26 Jan 2016 19:38:29 +0100
Subject: test stream resets in push promise

---
 test/test_protocol_http2.py | 37 +++++++++++++++++++++++++++++++++++++
 1 file changed, 37 insertions(+)

diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index b42b86cb..b1c88b27 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -235,6 +235,9 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
 
             h2_conn.send_headers(2, [(':status', '202')])
             h2_conn.send_headers(4, [(':status', '204')])
+            wfile.write(h2_conn.data_to_send())
+            wfile.flush()
+
             h2_conn.send_data(1, b'regular_stream')
             h2_conn.send_data(2, b'pushed_stream_foo')
             h2_conn.send_data(4, b'pushed_stream_bar')
@@ -280,3 +283,37 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
         assert b'regular_stream' in bodies
         assert b'pushed_stream_foo' in bodies
         assert b'pushed_stream_bar' in bodies
+
+    def test_push_promise_reset(self):
+        client, h2_conn = self._setup_connection()
+
+        self._send_request(client.wfile, h2_conn, stream_id=1, headers=[
+            (':authority', "127.0.0.1:%s" % self.server.server.address.port),
+            (':method', 'GET'),
+            (':scheme', 'https'),
+            (':path', '/'),
+            ('foo', 'bar')
+        ])
+
+        done = False
+        while not done:
+            try:
+                events = h2_conn.receive_data(utils.http2_read_frame(client.rfile))
+            except:
+                break
+            client.wfile.write(h2_conn.data_to_send())
+            client.wfile.flush()
+
+            for event in events:
+                if isinstance(event, h2.events.StreamEnded) and event.stream_id == 1:
+                    done = True
+                elif isinstance(event, h2.events.PushedStreamReceived):
+                    h2_conn.reset_stream(event.pushed_stream_id)
+                    client.wfile.write(h2_conn.data_to_send())
+                    client.wfile.flush()
+
+        bodies = [flow.response.body for flow in self.master.state.flows]
+        assert len(bodies) == 3
+        assert b'regular_stream' in bodies
+        assert b'pushed_stream_foo' in bodies
+        assert b'pushed_stream_bar' in bodies
-- 
cgit v1.2.3


From 44f83b594701f9756418cf8208c30a9ba5ac4aad Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Tue, 26 Jan 2016 20:44:53 +0100
Subject: add more tests, improve coverage

---
 libmproxy/protocol/http2.py | 26 +++++++--------
 test/test_protocol_http2.py | 79 +++++++++++++++++++++++++++++++++++++++++----
 2 files changed, 83 insertions(+), 22 deletions(-)

diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index 23004497..20321d53 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -105,18 +105,19 @@ class Http2Layer(Layer):
         self.server_conn.send(self.server_conn.h2.data_to_send())
         self.active_conns.append(self.server_conn.connection)
 
-    def connect(self):
-        self.ctx.connect()
-        self.server_conn.connect()
-        self._initiate_server_conn()
+    def connect(self):  # pragma: no cover
+        raise ValueError("CONNECT inside an HTTP2 stream is not supported.")
+        # self.ctx.connect()
+        # self.server_conn.connect()
+        # self._initiate_server_conn()
 
-    def set_server(self):
+    def set_server(self):  # pragma: no cover
         raise NotImplementedError("Cannot change server for HTTP2 connections.")
 
-    def disconnect(self):
+    def disconnect(self):  # pragma: no cover
         raise NotImplementedError("Cannot dis- or reconnect in HTTP2 connections.")
 
-    def next_layer(self):
+    def next_layer(self):  # pragma: no cover
         # WebSockets over HTTP/2?
         # CONNECT for proxying?
         raise NotImplementedError()
@@ -257,13 +258,8 @@ class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
         if path == '*' or path.startswith("/"):
             form_in = "relative"
         elif method == 'CONNECT':  # pragma: no cover
-            # form_in = "authority"
-            # if ":" in authority:
-            #     host, port = authority.split(":", 1)
-            # else:
-            #     host = authority
             raise NotImplementedError("CONNECT over HTTP/2 is not implemented.")
-        else:
+        else:  # pragma: no cover
             form_in = "absolute"
             # FIXME: verify if path or :host contains what we need
             scheme, host, port, _ = utils.parse_url(path)
@@ -359,10 +355,10 @@ class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
         # RFC 7540 8.1: An HTTP request/response exchange fully consumes a single stream.
         return True
 
-    def connect(self):
+    def connect(self):  # pragma: no cover
         raise ValueError("CONNECT inside an HTTP2 stream is not supported.")
 
-    def set_server(self, *args, **kwargs):
+    def set_server(self, *args, **kwargs):  # pragma: no cover
         # do not mess with the server connection - all streams share it.
         pass
 
diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index b1c88b27..3554fa4d 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -7,7 +7,7 @@ import pytest
 import traceback
 import os
 import tempfile
-
+import time
 from io import BytesIO
 
 from libmproxy.proxy.config import ProxyConfig
@@ -126,12 +126,15 @@ class _Http2TestBase(object):
 
         return client, h2_conn
 
-    def _send_request(self, wfile, h2_conn, stream_id=1, headers=[], end_stream=True):
+    def _send_request(self, wfile, h2_conn, stream_id=1, headers=[], body=b''):
         h2_conn.send_headers(
             stream_id=stream_id,
             headers=headers,
-            end_stream=end_stream,
+            end_stream=(len(body) == 0),
         )
+        if body:
+            h2_conn.send_data(stream_id, body)
+            h2_conn.end_stream(stream_id)
         wfile.write(h2_conn.data_to_send())
         wfile.flush()
 
@@ -172,7 +175,7 @@ class TestSimple(_Http2TestBase, _Http2ServerBase):
             (':method', 'GET'),
             (':scheme', 'https'),
             (':path', '/'),
-        ])
+        ], body='my request body echoed back to me')
 
         done = False
         while not done:
@@ -194,6 +197,69 @@ class TestSimple(_Http2TestBase, _Http2ServerBase):
         assert self.master.state.flows[0].response.body == b'foobar'
 
 
+@requires_alpn
+class TestWithBodies(_Http2TestBase, _Http2ServerBase):
+    tmp_data_buffer_foobar = b''
+
+    @classmethod
+    def setup_class(self):
+        _Http2TestBase.setup_class()
+        _Http2ServerBase.setup_class()
+
+    @classmethod
+    def teardown_class(self):
+        _Http2TestBase.teardown_class()
+        _Http2ServerBase.teardown_class()
+
+    @classmethod
+    def handle_server_event(self, event, h2_conn, rfile, wfile):
+        if isinstance(event, h2.events.ConnectionTerminated):
+            return False
+        if isinstance(event, h2.events.DataReceived):
+            self.tmp_data_buffer_foobar += event.data
+        elif isinstance(event, h2.events.StreamEnded):
+            h2_conn.send_headers(1, [
+                (':status', '200'),
+            ])
+            h2_conn.send_data(1, self.tmp_data_buffer_foobar)
+            h2_conn.end_stream(1)
+            wfile.write(h2_conn.data_to_send())
+            wfile.flush()
+
+        return True
+
+    def test_with_bodies(self):
+        client, h2_conn = self._setup_connection()
+
+        self._send_request(
+            client.wfile,
+            h2_conn,
+            headers=[
+                (':authority', "127.0.0.1:%s" % self.server.server.address.port),
+                (':method', 'GET'),
+                (':scheme', 'https'),
+                (':path', '/'),
+            ],
+            body='foobar with request body',
+        )
+
+        done = False
+        while not done:
+            events = h2_conn.receive_data(utils.http2_read_frame(client.rfile))
+            client.wfile.write(h2_conn.data_to_send())
+            client.wfile.flush()
+
+            for event in events:
+                if isinstance(event, h2.events.StreamEnded):
+                    done = True
+
+        h2_conn.close_connection()
+        client.wfile.write(h2_conn.data_to_send())
+        client.wfile.flush()
+
+        assert self.master.state.flows[0].response.body == b'foobar with request body'
+
+
 @requires_alpn
 class TestPushPromise(_Http2TestBase, _Http2ServerBase):
     @classmethod
@@ -308,12 +374,11 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
                 if isinstance(event, h2.events.StreamEnded) and event.stream_id == 1:
                     done = True
                 elif isinstance(event, h2.events.PushedStreamReceived):
-                    h2_conn.reset_stream(event.pushed_stream_id)
+                    h2_conn.reset_stream(event.pushed_stream_id, error_code=0x8)
                     client.wfile.write(h2_conn.data_to_send())
                     client.wfile.flush()
 
         bodies = [flow.response.body for flow in self.master.state.flows]
         assert len(bodies) == 3
         assert b'regular_stream' in bodies
-        assert b'pushed_stream_foo' in bodies
-        assert b'pushed_stream_bar' in bodies
+        # the other two bodies might not be transmitted before the reset
-- 
cgit v1.2.3


From 64978968f2383988511dc7021ecd9892f098f723 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Tue, 26 Jan 2016 21:04:40 +0100
Subject: fix authority handling

---
 libmproxy/protocol/http2.py | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index 20321d53..423c5caa 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -265,12 +265,12 @@ class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
             scheme, host, port, _ = utils.parse_url(path)
 
         if authority:
-            host, port = authority.split(':')
+            host, _, port = authority.partition(':')
 
-        if host is None:
+        if not host:
             host = 'localhost'
-        if port is None:
-            port = 80 if scheme == 'http' else 443
+        if not port:
+            port = 443 if scheme == 'https' else 80
         port = int(port)
 
         data = []
-- 
cgit v1.2.3


From 6d3b3994e27b3cd97c74612bc64c7e0457aeb448 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Wed, 27 Jan 2016 12:52:18 +0100
Subject: code formatting

---
 libmproxy/protocol/http.py  |  4 +++-
 libmproxy/protocol/http1.py |  7 +------
 libmproxy/protocol/http2.py | 10 +++++++---
 libmproxy/utils.py          |  1 +
 test/test_protocol_http1.py |  1 -
 test/test_protocol_http2.py |  9 +++++----
 6 files changed, 17 insertions(+), 15 deletions(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index ee9d2d46..6a669ae1 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -23,7 +23,9 @@ from ..models import (
 
 from .base import Layer, Kill
 
+
 class _HttpTransmissionLayer(Layer):
+
     def read_request(self):
         raise NotImplementedError()
 
@@ -233,7 +235,7 @@ class HttpLayer(Layer):
         try:
             response = make_error_response(code, message)
             self.send_response(response)
-        except NetlibException, H2Error:
+        except NetlibException as H2Error:
             pass
 
     def change_upstream_proxy_server(self, address):
diff --git a/libmproxy/protocol/http1.py b/libmproxy/protocol/http1.py
index 0d6095df..fc2cf07a 100644
--- a/libmproxy/protocol/http1.py
+++ b/libmproxy/protocol/http1.py
@@ -1,12 +1,6 @@
 from __future__ import (absolute_import, print_function, division)
 
-import sys
-import traceback
 import six
-import struct
-import threading
-import time
-import Queue
 
 from netlib import tcp
 from netlib.http import http1
@@ -17,6 +11,7 @@ from ..models import HTTPRequest, HTTPResponse
 
 
 class Http1Layer(_HttpTransmissionLayer):
+
     def __init__(self, ctx, mode):
         super(Http1Layer, self).__init__(ctx)
         self.mode = mode
diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index 423c5caa..5c4586de 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -1,6 +1,5 @@
 from __future__ import (absolute_import, print_function, division)
 
-import struct
 import threading
 import time
 import Queue
@@ -17,9 +16,12 @@ from .base import Layer
 from .http import _HttpTransmissionLayer, HttpLayer
 from .. import utils
 from ..models import HTTPRequest, HTTPResponse
-from ..exceptions import HttpProtocolException, ProtocolException
+from ..exceptions import HttpProtocolException
+from ..exceptions import ProtocolException
+
 
 class SafeH2Connection(H2Connection):
+
     def __init__(self, conn, *args, **kwargs):
         super(SafeH2Connection, self).__init__(*args, **kwargs)
         self.conn = conn
@@ -71,7 +73,7 @@ class SafeH2Connection(H2Connection):
                 if is_zombie(self, stream_id):
                     return
                 max_outbound_frame_size = self.max_outbound_frame_size
-                frame_chunk = chunk[position:position+max_outbound_frame_size]
+                frame_chunk = chunk[position:position + max_outbound_frame_size]
                 if self.local_flow_control_window(stream_id) < len(frame_chunk):
                     self.lock.release()
                     time.sleep(0)
@@ -88,6 +90,7 @@ class SafeH2Connection(H2Connection):
 
 
 class Http2Layer(Layer):
+
     def __init__(self, ctx, mode):
         super(Http2Layer, self).__init__(ctx)
         self.mode = mode
@@ -226,6 +229,7 @@ class Http2Layer(Layer):
 
 
 class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
+
     def __init__(self, ctx, stream_id, request_headers):
         super(Http2SingleStreamLayer, self).__init__(ctx)
         self.zombie = None
diff --git a/libmproxy/utils.py b/libmproxy/utils.py
index 299e8749..5b1c41f1 100644
--- a/libmproxy/utils.py
+++ b/libmproxy/utils.py
@@ -174,6 +174,7 @@ def safe_subn(pattern, repl, target, *args, **kwargs):
     """
     return re.subn(str(pattern), str(repl), target, *args, **kwargs)
 
+
 def http2_read_frame(rfile):
     field = rfile.peek(3)
     length = int(field.encode('hex'), 16)
diff --git a/test/test_protocol_http1.py b/test/test_protocol_http1.py
index f5fe93a8..a1485f1b 100644
--- a/test/test_protocol_http1.py
+++ b/test/test_protocol_http1.py
@@ -1,4 +1,3 @@
-from io import BytesIO
 from netlib.exceptions import HttpSyntaxException
 from netlib.http import http1
 from netlib.tcp import TCPClient
diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index 3554fa4d..6da8cd31 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -1,14 +1,10 @@
 from __future__ import (absolute_import, print_function, division)
 
-import inspect
-import socket
 import OpenSSL
 import pytest
 import traceback
 import os
 import tempfile
-import time
-from io import BytesIO
 
 from libmproxy.proxy.config import ProxyConfig
 from libmproxy.proxy.server import ProxyServer
@@ -35,10 +31,12 @@ requires_alpn = pytest.mark.skipif(
     not OpenSSL._util.lib.Cryptography_HAS_ALPN,
     reason="requires OpenSSL with ALPN support")
 
+
 class _Http2ServerBase(netlib_tservers.ServerTestBase):
     ssl = dict(alpn_select=b'h2')
 
     class handler(netlib.tcp.BaseHandler):
+
         def handle(self):
             h2_conn = h2.connection.H2Connection(client_side=False)
 
@@ -68,6 +66,7 @@ class _Http2ServerBase(netlib_tservers.ServerTestBase):
 
 
 class _Http2TestBase(object):
+
     @classmethod
     def setup_class(self):
         self.config = ProxyConfig(**self.get_proxy_config())
@@ -141,6 +140,7 @@ class _Http2TestBase(object):
 
 @requires_alpn
 class TestSimple(_Http2TestBase, _Http2ServerBase):
+
     @classmethod
     def setup_class(self):
         _Http2TestBase.setup_class()
@@ -262,6 +262,7 @@ class TestWithBodies(_Http2TestBase, _Http2ServerBase):
 
 @requires_alpn
 class TestPushPromise(_Http2TestBase, _Http2ServerBase):
+
     @classmethod
     def setup_class(self):
         _Http2TestBase.setup_class()
-- 
cgit v1.2.3


From f2097b47ce6c88e2bef0889fb7e7d52b63158082 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Sat, 30 Jan 2016 22:05:23 +0100
Subject: fix race condition during state loading

PyPy and Python2.7 might process the state attributes in different order.
---
 libmproxy/models/http.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/libmproxy/models/http.py b/libmproxy/models/http.py
index 3914440e..a52597b4 100644
--- a/libmproxy/models/http.py
+++ b/libmproxy/models/http.py
@@ -11,9 +11,10 @@ from netlib.tcp import Address
 from .. import version, stateobject
 from .flow import Flow
 
+from collections import OrderedDict
 
 class MessageMixin(stateobject.StateObject):
-    _stateobject_attributes = dict(
+    _stateobject_attributes = OrderedDict(
         http_version=bytes,
         headers=Headers,
         timestamp_start=float,
-- 
cgit v1.2.3


From d8ae7c3e29bcf117eb051f9e74b76fea733c1c64 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 1 Feb 2016 23:03:15 +0100
Subject: fix tests and use netlib utils

---
 libmproxy/protocol/http2.py |  3 ++-
 libmproxy/utils.py          | 11 -----------
 test/test_protocol_http2.py | 14 +++++++-------
 3 files changed, 9 insertions(+), 19 deletions(-)

diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index 5c4586de..4b582f51 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -7,6 +7,7 @@ import Queue
 from netlib.tcp import ssl_read_select
 from netlib.exceptions import HttpException
 from netlib.http import Headers
+from netlib.utils import http2_read_raw_frame
 
 import h2
 from h2.connection import H2Connection
@@ -212,7 +213,7 @@ class Http2Layer(Layer):
 
                 with source_conn.h2.lock:
                     try:
-                        raw_frame = utils.http2_read_frame(source_conn.rfile)
+                        raw_frame = b''.join(http2_read_raw_frame(source_conn.rfile))
                     except:
                         for stream in self.streams.values():
                             stream.zombie = time.time()
diff --git a/libmproxy/utils.py b/libmproxy/utils.py
index 5b1c41f1..a697a637 100644
--- a/libmproxy/utils.py
+++ b/libmproxy/utils.py
@@ -173,14 +173,3 @@ def safe_subn(pattern, repl, target, *args, **kwargs):
         need a better solution that is aware of the actual content ecoding.
     """
     return re.subn(str(pattern), str(repl), target, *args, **kwargs)
-
-
-def http2_read_frame(rfile):
-    field = rfile.peek(3)
-    length = int(field.encode('hex'), 16)
-
-    if length == 4740180:
-        raise ValueError("Probably not the correct length bytes: %s" % rfile.peek(20))
-
-    raw_frame = rfile.safe_read(9 + length)
-    return raw_frame
diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index 6da8cd31..831f70ab 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -20,6 +20,7 @@ logging.getLogger("PIL.PngImagePlugin").setLevel(logging.WARNING)
 
 import netlib
 from netlib import tservers as netlib_tservers
+from netlib.utils import http2_read_raw_frame
 
 import h2
 from hyperframe.frame import Frame
@@ -47,8 +48,7 @@ class _Http2ServerBase(netlib_tservers.ServerTestBase):
             self.wfile.flush()
 
             while True:
-                raw_frame = utils.http2_read_frame(self.rfile)
-                events = h2_conn.receive_data(raw_frame)
+                events = h2_conn.receive_data(b''.join(http2_read_raw_frame(self.rfile)))
                 self.wfile.write(h2_conn.data_to_send())
                 self.wfile.flush()
 
@@ -179,7 +179,7 @@ class TestSimple(_Http2TestBase, _Http2ServerBase):
 
         done = False
         while not done:
-            events = h2_conn.receive_data(utils.http2_read_frame(client.rfile))
+            events = h2_conn.receive_data(b''.join(http2_read_raw_frame(client.rfile)))
             client.wfile.write(h2_conn.data_to_send())
             client.wfile.flush()
 
@@ -245,7 +245,7 @@ class TestWithBodies(_Http2TestBase, _Http2ServerBase):
 
         done = False
         while not done:
-            events = h2_conn.receive_data(utils.http2_read_frame(client.rfile))
+            events = h2_conn.receive_data(b''.join(http2_read_raw_frame(client.rfile)))
             client.wfile.write(h2_conn.data_to_send())
             client.wfile.flush()
 
@@ -331,7 +331,7 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
         pushed_streams = 0
         while ended_streams != 3:
             try:
-                events = h2_conn.receive_data(utils.http2_read_frame(client.rfile))
+                events = h2_conn.receive_data(b''.join(http2_read_raw_frame(client.rfile)))
             except:
                 break
             client.wfile.write(h2_conn.data_to_send())
@@ -365,7 +365,7 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
         done = False
         while not done:
             try:
-                events = h2_conn.receive_data(utils.http2_read_frame(client.rfile))
+                events = h2_conn.receive_data(b''.join(http2_read_raw_frame(client.rfile)))
             except:
                 break
             client.wfile.write(h2_conn.data_to_send())
@@ -379,7 +379,7 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
                     client.wfile.write(h2_conn.data_to_send())
                     client.wfile.flush()
 
-        bodies = [flow.response.body for flow in self.master.state.flows]
+        bodies = [flow.response.body for flow in self.master.state.flows if flow.response]
         assert len(bodies) == 3
         assert b'regular_stream' in bodies
         # the other two bodies might not be transmitted before the reset
-- 
cgit v1.2.3


From ab3543ba4d66f5ffadf0873a7fd384e2ddf8336f Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 1 Feb 2016 23:10:40 +0100
Subject: use test timeouts

---
 .appveyor.yml | 2 +-
 .travis.yml   | 2 +-
 setup.py      | 1 +
 3 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/.appveyor.yml b/.appveyor.yml
index 55aed6aa..689ad5e5 100644
--- a/.appveyor.yml
+++ b/.appveyor.yml
@@ -11,7 +11,7 @@ install:
   - "pip install --user --src .. -r requirements.txt"
   - "python -c \"from OpenSSL import SSL; print(SSL.SSLeay_version(SSL.SSLEAY_VERSION))\""
 test_script:
-  - "py.test -s --cov libmproxy"
+  - "py.test -s --cov libmproxy --timeout 30"
 cache:
   - C:\Users\appveyor\AppData\Local\pip\cache
 after_test:
diff --git a/.travis.yml b/.travis.yml
index 2545899c..be78031d 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -61,7 +61,7 @@ before_script:
   - "openssl version -a"
 
 script:
-  - "py.test -s --cov libmproxy"
+  - "py.test -s --cov libmproxy --timeout 30"
 
 after_success:
   - coveralls
diff --git a/setup.py b/setup.py
index 522857fd..f70eda46 100644
--- a/setup.py
+++ b/setup.py
@@ -46,6 +46,7 @@ dev_deps = {
     "pytest>=2.8.0",
     "pytest-xdist>=1.13.1",
     "pytest-cov>=2.1.0",
+    "pytest-timeout>=1.0.0",
     "coveralls>=0.4.1",
     "pathod>=%s, <%s" % (version.MINORVERSION, version.NEXT_MINORVERSION),
     "sphinx>=1.3.1",
-- 
cgit v1.2.3


From af6c2571312132d4309ffc43e86053c40132c849 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 1 Feb 2016 23:27:17 +0100
Subject: fix flow == None errors

---
 libmproxy/protocol/http.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 6a669ae1..9b4d4d8f 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -188,7 +188,7 @@ class HttpLayer(Layer):
 
                 self.log("response", "debug", [repr(flow.response)])
                 flow = self.channel.ask("response", flow)
-                if flow == Kill:
+                if not flow or flow == Kill:
                     raise Kill()
                 self.send_response_to_client(flow)
 
-- 
cgit v1.2.3


From b007ff3f9b00165bd87c18292d8b23d4acc83ab9 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 1 Feb 2016 23:27:37 +0100
Subject: fix locking issues

---
 libmproxy/protocol/http2.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index 4b582f51..4b3ef0ed 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -72,6 +72,7 @@ class SafeH2Connection(H2Connection):
             while position < len(chunk):
                 self.lock.acquire()
                 if is_zombie(self, stream_id):
+                    self.lock.release()
                     return
                 max_outbound_frame_size = self.max_outbound_frame_size
                 frame_chunk = chunk[position:position + max_outbound_frame_size]
-- 
cgit v1.2.3


From 738094e1674ae78f92ff32020e608510ff4af45a Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Mon, 1 Feb 2016 23:27:50 +0100
Subject: improve test reliability

---
 test/test_protocol_http2.py | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index 831f70ab..dce8c5da 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -48,7 +48,10 @@ class _Http2ServerBase(netlib_tservers.ServerTestBase):
             self.wfile.flush()
 
             while True:
-                events = h2_conn.receive_data(b''.join(http2_read_raw_frame(self.rfile)))
+                try:
+                    events = h2_conn.receive_data(b''.join(http2_read_raw_frame(self.rfile)))
+                except:
+                    break
                 self.wfile.write(h2_conn.data_to_send())
                 self.wfile.flush()
 
@@ -362,8 +365,8 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
             ('foo', 'bar')
         ])
 
-        done = False
-        while not done:
+        streams = 0
+        while streams != 3:
             try:
                 events = h2_conn.receive_data(b''.join(http2_read_raw_frame(client.rfile)))
             except:
@@ -373,8 +376,9 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
 
             for event in events:
                 if isinstance(event, h2.events.StreamEnded) and event.stream_id == 1:
-                    done = True
+                    streams += 1
                 elif isinstance(event, h2.events.PushedStreamReceived):
+                    streams += 1
                     h2_conn.reset_stream(event.pushed_stream_id, error_code=0x8)
                     client.wfile.write(h2_conn.data_to_send())
                     client.wfile.flush()
-- 
cgit v1.2.3


From 74e62903c13d9f1f1545a31ff019cdfc5e83ddda Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Tue, 2 Feb 2016 09:57:11 +0100
Subject: fix exception classes

---
 libmproxy/protocol/http.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 9b4d4d8f..d9803a37 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -235,7 +235,8 @@ class HttpLayer(Layer):
         try:
             response = make_error_response(code, message)
             self.send_response(response)
-        except NetlibException as H2Error:
+        except (NetlibException, H2Error):
+            self.log(traceback.format_exc(), "debug")
             pass
 
     def change_upstream_proxy_server(self, address):
-- 
cgit v1.2.3


From 6bc1755750f8a7986ab26ff28ea0e90ad0ccaacd Mon Sep 17 00:00:00 2001
From: Maximilian Hils <git@maximilianhils.com>
Date: Tue, 2 Feb 2016 12:32:58 +0100
Subject: add comment that explains OrderedDict use

---
 libmproxy/models/http.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/libmproxy/models/http.py b/libmproxy/models/http.py
index a52597b4..8a0b226d 100644
--- a/libmproxy/models/http.py
+++ b/libmproxy/models/http.py
@@ -14,6 +14,10 @@ from .flow import Flow
 from collections import OrderedDict
 
 class MessageMixin(stateobject.StateObject):
+    # The restoration order is important currently, e.g. because
+    # of .content setting .headers["content-length"] automatically.
+    # Using OrderedDict is the short term fix, restoring state should
+    # be implemented without side-effects again.
     _stateobject_attributes = OrderedDict(
         http_version=bytes,
         headers=Headers,
-- 
cgit v1.2.3


From 68bcc82b8e4c219b024bff0081741c799b9cbd74 Mon Sep 17 00:00:00 2001
From: Maximilian Hils <git@maximilianhils.com>
Date: Tue, 2 Feb 2016 15:49:21 +0100
Subject: do not send RST if there is not upstream stream openend yet

---
 libmproxy/protocol/http2.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index 4b3ef0ed..fe9f8695 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -161,7 +161,8 @@ class Http2Layer(Layer):
                     other_stream_id = self.streams[eid].client_stream_id
                 else:
                     other_stream_id = self.streams[eid].server_stream_id
-                other_conn.h2.safe_reset_stream(other_stream_id, event.error_code)
+                if other_stream_id is not None:
+                    other_conn.h2.safe_reset_stream(other_stream_id, event.error_code)
         elif isinstance(event, RemoteSettingsChanged):
             new_settings = dict([(id, cs.new_value) for (id, cs) in event.changed_settings.iteritems()])
             other_conn.h2.safe_update_settings(new_settings)
-- 
cgit v1.2.3


From ca5cc34d0b70f3306f62004be7ceb3f0c2053da7 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Tue, 2 Feb 2016 17:48:09 +0100
Subject: cleanup

---
 libmproxy/protocol/http2.py | 14 +++++++-------
 test/test_protocol_http2.py |  2 +-
 2 files changed, 8 insertions(+), 8 deletions(-)

diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index fe9f8695..e617f77c 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -61,7 +61,7 @@ class SafeH2Connection(H2Connection):
 
     def safe_send_headers(self, is_zombie, stream_id, headers):
         with self.lock:
-            if is_zombie(self, stream_id):
+            if is_zombie():
                 return
             self.send_headers(stream_id, headers)
             self.conn.send(self.data_to_send())
@@ -71,7 +71,7 @@ class SafeH2Connection(H2Connection):
             position = 0
             while position < len(chunk):
                 self.lock.acquire()
-                if is_zombie(self, stream_id):
+                if is_zombie():
                     self.lock.release()
                     return
                 max_outbound_frame_size = self.max_outbound_frame_size
@@ -85,7 +85,7 @@ class SafeH2Connection(H2Connection):
                 self.lock.release()
                 position += max_outbound_frame_size
         with self.lock:
-            if is_zombie(self, stream_id):
+            if is_zombie():
                 return
             self.end_stream(stream_id)
             self.conn.send(self.data_to_send())
@@ -246,10 +246,8 @@ class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
         self.response_arrived = threading.Event()
         self.data_finished = threading.Event()
 
-    def is_zombie(self, h2_conn, stream_id):
-        if self.zombie:
-            return True
-        return False
+    def is_zombie(self):
+        return self.zombie is not None
 
     def read_request(self):
         self.data_finished.wait()
@@ -300,6 +298,8 @@ class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
         )
 
     def send_request(self, message):
+        if self.zombie:
+            return
         with self.server_conn.h2.lock:
             self.server_stream_id = self.server_conn.h2.get_next_available_stream_id()
             self.server_to_client_stream_ids[self.server_stream_id] = self.client_stream_id
diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index dce8c5da..cc62f734 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -384,6 +384,6 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
                     client.wfile.flush()
 
         bodies = [flow.response.body for flow in self.master.state.flows if flow.response]
-        assert len(bodies) == 3
+        assert len(bodies) >= 1
         assert b'regular_stream' in bodies
         # the other two bodies might not be transmitted before the reset
-- 
cgit v1.2.3


From cf8c063773b70ad37ab0a2125f5ed03c35e17336 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Tue, 2 Feb 2016 23:54:35 +0100
Subject: fix http2 race condition

---
 libmproxy/protocol/http2.py | 67 +++++++++++++++++++++++++++++++++++----------
 test/test_protocol_http2.py | 66 ++++++++++++++++++++++++++++++++++++--------
 2 files changed, 107 insertions(+), 26 deletions(-)

diff --git a/libmproxy/protocol/http2.py b/libmproxy/protocol/http2.py
index e617f77c..de068836 100644
--- a/libmproxy/protocol/http2.py
+++ b/libmproxy/protocol/http2.py
@@ -167,7 +167,8 @@ class Http2Layer(Layer):
             new_settings = dict([(id, cs.new_value) for (id, cs) in event.changed_settings.iteritems()])
             other_conn.h2.safe_update_settings(new_settings)
         elif isinstance(event, ConnectionTerminated):
-            other_conn.h2.safe_close_connection(event.error_code)
+            # Do not immediately terminate the other connection.
+            # Some streams might be still sending data to the client.
             return False
         elif isinstance(event, PushedStreamReceived):
             # pushed stream ids should be uniq and not dependent on race conditions
@@ -183,7 +184,7 @@ class Http2Layer(Layer):
             self.streams[event.pushed_stream_id].pushed = True
             self.streams[event.pushed_stream_id].parent_stream_id = parent_eid
             self.streams[event.pushed_stream_id].timestamp_end = time.time()
-            self.streams[event.pushed_stream_id].data_finished.set()
+            self.streams[event.pushed_stream_id].request_data_finished.set()
             self.streams[event.pushed_stream_id].start()
         elif isinstance(event, TrailersReceived):
             raise NotImplementedError()
@@ -240,18 +241,50 @@ class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
         self.server_stream_id = None
         self.request_headers = request_headers
         self.response_headers = None
-        self.data_queue = Queue.Queue()
-        self.queued_data_length = 0
+        self.pushed = False
+
+        self.request_data_queue = Queue.Queue()
+        self.request_queued_data_length = 0
+        self.request_data_finished = threading.Event()
 
         self.response_arrived = threading.Event()
-        self.data_finished = threading.Event()
+        self.response_data_queue = Queue.Queue()
+        self.response_queued_data_length = 0
+        self.response_data_finished = threading.Event()
+
+    @property
+    def data_queue(self):
+        if self.response_arrived.is_set():
+            return self.response_data_queue
+        else:
+            return self.request_data_queue
+
+    @property
+    def queued_data_length(self):
+        if self.response_arrived.is_set():
+            return self.response_queued_data_length
+        else:
+            return self.request_queued_data_length
+
+    @property
+    def data_finished(self):
+        if self.response_arrived.is_set():
+            return self.response_data_finished
+        else:
+            return self.request_data_finished
+
+    @queued_data_length.setter
+    def queued_data_length(self, v):
+        if self.response_arrived.is_set():
+            return self.response_queued_data_length
+        else:
+            return self.request_queued_data_length
 
     def is_zombie(self):
         return self.zombie is not None
 
     def read_request(self):
-        self.data_finished.wait()
-        self.data_finished.clear()
+        self.request_data_finished.wait()
 
         authority = self.request_headers.get(':authority', '')
         method = self.request_headers.get(':method', 'GET')
@@ -279,8 +312,8 @@ class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
         port = int(port)
 
         data = []
-        while self.data_queue.qsize() > 0:
-            data.append(self.data_queue.get())
+        while self.request_data_queue.qsize() > 0:
+            data.append(self.request_data_queue.get())
         data = b"".join(data)
 
         return HTTPRequest(
@@ -298,9 +331,15 @@ class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
         )
 
     def send_request(self, message):
-        if self.zombie:
+        if self.pushed:
+            # nothing to do here
             return
+
         with self.server_conn.h2.lock:
+            # We must not assign a stream id if we are already a zombie.
+            if self.zombie:
+                return
+
             self.server_stream_id = self.server_conn.h2.get_next_available_stream_id()
             self.server_to_client_stream_ids[self.server_stream_id] = self.client_stream_id
 
@@ -333,12 +372,12 @@ class Http2SingleStreamLayer(_HttpTransmissionLayer, threading.Thread):
     def read_response_body(self, request, response):
         while True:
             try:
-                yield self.data_queue.get(timeout=1)
+                yield self.response_data_queue.get(timeout=1)
             except Queue.Empty:
                 pass
-            if self.data_finished.is_set():
-                while self.data_queue.qsize() > 0:
-                    yield self.data_queue.get()
+            if self.response_data_finished.is_set():
+                while self.response_data_queue.qsize() > 0:
+                    yield self.response_data_queue.get()
                 return
             if self.zombie:
                 return
diff --git a/test/test_protocol_http2.py b/test/test_protocol_http2.py
index cc62f734..38cfdfc3 100644
--- a/test/test_protocol_http2.py
+++ b/test/test_protocol_http2.py
@@ -5,6 +5,7 @@ import pytest
 import traceback
 import os
 import tempfile
+import sys
 
 from libmproxy.proxy.config import ProxyConfig
 from libmproxy.proxy.server import ProxyServer
@@ -47,9 +48,11 @@ class _Http2ServerBase(netlib_tservers.ServerTestBase):
             self.wfile.write(h2_conn.data_to_send())
             self.wfile.flush()
 
-            while True:
+            done = False
+            while not done:
                 try:
-                    events = h2_conn.receive_data(b''.join(http2_read_raw_frame(self.rfile)))
+                    raw = b''.join(http2_read_raw_frame(self.rfile))
+                    events = h2_conn.receive_data(raw)
                 except:
                     break
                 self.wfile.write(h2_conn.data_to_send())
@@ -58,10 +61,12 @@ class _Http2ServerBase(netlib_tservers.ServerTestBase):
                 for event in events:
                     try:
                         if not self.server.handle_server_event(event, h2_conn, self.rfile, self.wfile):
+                            done = True
                             break
                     except Exception as e:
                         print(repr(e))
                         print(traceback.format_exc())
+                        done = True
                         break
 
     def handle_server_event(self, h2_conn, rfile, wfile):
@@ -182,7 +187,10 @@ class TestSimple(_Http2TestBase, _Http2ServerBase):
 
         done = False
         while not done:
-            events = h2_conn.receive_data(b''.join(http2_read_raw_frame(client.rfile)))
+            try:
+                events = h2_conn.receive_data(b''.join(http2_read_raw_frame(client.rfile)))
+            except:
+                break
             client.wfile.write(h2_conn.data_to_send())
             client.wfile.flush()
 
@@ -248,7 +256,10 @@ class TestWithBodies(_Http2TestBase, _Http2ServerBase):
 
         done = False
         while not done:
-            events = h2_conn.receive_data(b''.join(http2_read_raw_frame(client.rfile)))
+            try:
+                events = h2_conn.receive_data(b''.join(http2_read_raw_frame(client.rfile)))
+            except:
+                break
             client.wfile.write(h2_conn.data_to_send())
             client.wfile.flush()
 
@@ -303,14 +314,16 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
             wfile.write(h2_conn.data_to_send())
             wfile.flush()
 
-            h2_conn.send_headers(2, [(':status', '202')])
-            h2_conn.send_headers(4, [(':status', '204')])
+            h2_conn.send_headers(2, [(':status', '200')])
+            h2_conn.send_headers(4, [(':status', '200')])
             wfile.write(h2_conn.data_to_send())
             wfile.flush()
 
             h2_conn.send_data(1, b'regular_stream')
             h2_conn.send_data(2, b'pushed_stream_foo')
             h2_conn.send_data(4, b'pushed_stream_bar')
+            wfile.write(h2_conn.data_to_send())
+            wfile.flush()
             h2_conn.end_stream(1)
             h2_conn.end_stream(2)
             h2_conn.end_stream(4)
@@ -330,11 +343,14 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
             ('foo', 'bar')
         ])
 
+        done = False
         ended_streams = 0
         pushed_streams = 0
-        while ended_streams != 3:
+        responses = 0
+        while not done:
             try:
-                events = h2_conn.receive_data(b''.join(http2_read_raw_frame(client.rfile)))
+                raw = b''.join(http2_read_raw_frame(client.rfile))
+                events = h2_conn.receive_data(raw)
             except:
                 break
             client.wfile.write(h2_conn.data_to_send())
@@ -345,7 +361,19 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
                     ended_streams += 1
                 elif isinstance(event, h2.events.PushedStreamReceived):
                     pushed_streams += 1
+                elif isinstance(event, h2.events.ResponseReceived):
+                    responses += 1
+                if isinstance(event, h2.events.ConnectionTerminated):
+                    done = True
 
+            if responses == 3 and ended_streams == 3 and pushed_streams == 2:
+                done = True
+
+        h2_conn.close_connection()
+        client.wfile.write(h2_conn.data_to_send())
+        client.wfile.flush()
+
+        assert ended_streams == 3
         assert pushed_streams == 2
 
         bodies = [flow.response.body for flow in self.master.state.flows]
@@ -365,8 +393,11 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
             ('foo', 'bar')
         ])
 
-        streams = 0
-        while streams != 3:
+        done = False
+        ended_streams = 0
+        pushed_streams = 0
+        responses = 0
+        while not done:
             try:
                 events = h2_conn.receive_data(b''.join(http2_read_raw_frame(client.rfile)))
             except:
@@ -376,12 +407,23 @@ class TestPushPromise(_Http2TestBase, _Http2ServerBase):
 
             for event in events:
                 if isinstance(event, h2.events.StreamEnded) and event.stream_id == 1:
-                    streams += 1
+                    ended_streams += 1
                 elif isinstance(event, h2.events.PushedStreamReceived):
-                    streams += 1
+                    pushed_streams += 1
                     h2_conn.reset_stream(event.pushed_stream_id, error_code=0x8)
                     client.wfile.write(h2_conn.data_to_send())
                     client.wfile.flush()
+                elif isinstance(event, h2.events.ResponseReceived):
+                    responses += 1
+                if isinstance(event, h2.events.ConnectionTerminated):
+                    done = True
+
+            if responses >= 1 and ended_streams >= 1 and pushed_streams == 2:
+                done = True
+
+        h2_conn.close_connection()
+        client.wfile.write(h2_conn.data_to_send())
+        client.wfile.flush()
 
         bodies = [flow.response.body for flow in self.master.state.flows if flow.response]
         assert len(bodies) >= 1
-- 
cgit v1.2.3


From 9759207c8d7a7f29f3a0ff630325f7abd3b1d2b1 Mon Sep 17 00:00:00 2001
From: Maximilian Hils <git@maximilianhils.com>
Date: Wed, 3 Feb 2016 00:05:02 +0100
Subject: check for channel error location

---
 libmproxy/protocol/http.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index d9803a37..377e1670 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -219,6 +219,10 @@ class HttpLayer(Layer):
                 else:
                     six.reraise(ProtocolException, ProtocolException(
                         "Error in HTTP connection: %s" % repr(e)), sys.exc_info()[2])
+            except Exception:
+                import traceback
+                traceback.print_exc()
+                six.reraise(*sys.exc_info()[:3])
             finally:
                 flow.live = False
 
-- 
cgit v1.2.3


From 547dd4163e7da909b9d3da6f562d22d36dcf80a1 Mon Sep 17 00:00:00 2001
From: Maximilian Hils <git@maximilianhils.com>
Date: Wed, 3 Feb 2016 00:08:25 +0100
Subject: fix import

---
 libmproxy/protocol/http.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index 377e1670..f0f84e4a 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -220,7 +220,6 @@ class HttpLayer(Layer):
                     six.reraise(ProtocolException, ProtocolException(
                         "Error in HTTP connection: %s" % repr(e)), sys.exc_info()[2])
             except Exception:
-                import traceback
                 traceback.print_exc()
                 six.reraise(*sys.exc_info()[:3])
             finally:
-- 
cgit v1.2.3


From 07c36542f0ce12300b45b7c853bf52ff23a80fb6 Mon Sep 17 00:00:00 2001
From: Maximilian Hils <git@maximilianhils.com>
Date: Thu, 4 Feb 2016 02:58:51 +0100
Subject: fix travis caching

---
 .travis.yml | 10 ----------
 1 file changed, 10 deletions(-)

diff --git a/.travis.yml b/.travis.yml
index be78031d..2c105a11 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -90,17 +90,7 @@ notifications:
     on_success: always
     on_failure: always
 
-# exclude cryptography from cache
-# it depends on libssl-dev version
-# which needs to be compiled specifically to each version
-before_cache:
-  - pip uninstall -y cryptography
-
 cache:
   directories:
     - $HOME/.cache/pip
-    - $HOME/virtualenv/python2.7.9/lib/python2.7/site-packages
-    - $HOME/virtualenv/python2.7.9/bin
     - $HOME/.pyenv
-    - $HOME/virtualenv/pypy-2.5.0/site-packages
-    - $HOME/virtualenv/pypy-2.5.0/bin
-- 
cgit v1.2.3


From 69df00c19f3c358482243a9e2fff15b315cd8ff5 Mon Sep 17 00:00:00 2001
From: Maximilian Hils <git@maximilianhils.com>
Date: Thu, 4 Feb 2016 03:09:39 +0100
Subject: remove debug output

---
 libmproxy/protocol/http.py | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py
index f0f84e4a..f3240b85 100644
--- a/libmproxy/protocol/http.py
+++ b/libmproxy/protocol/http.py
@@ -188,7 +188,7 @@ class HttpLayer(Layer):
 
                 self.log("response", "debug", [repr(flow.response)])
                 flow = self.channel.ask("response", flow)
-                if not flow or flow == Kill:
+                if flow == Kill:
                     raise Kill()
                 self.send_response_to_client(flow)
 
@@ -219,9 +219,6 @@ class HttpLayer(Layer):
                 else:
                     six.reraise(ProtocolException, ProtocolException(
                         "Error in HTTP connection: %s" % repr(e)), sys.exc_info()[2])
-            except Exception:
-                traceback.print_exc()
-                six.reraise(*sys.exc_info()[:3])
             finally:
                 flow.live = False
 
-- 
cgit v1.2.3


From 98f54d21b6156ceaa8450072be9f53dfde137248 Mon Sep 17 00:00:00 2001
From: Maximilian Hils <git@maximilianhils.com>
Date: Thu, 4 Feb 2016 03:19:39 +0100
Subject: travis: cache wheels on osx, allow pypy failures

---
 .travis.yml | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/.travis.yml b/.travis.yml
index 2c105a11..8ea3ed32 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -32,6 +32,8 @@ matrix:
     - python: 2.7
       env: DOCS=1
       script: 'cd docs && make html'
+  allow_failures:
+    - python: pypy
 
 install:
   - |
@@ -94,3 +96,4 @@ cache:
   directories:
     - $HOME/.cache/pip
     - $HOME/.pyenv
+    - $HOME/Library/Caches/pip
\ No newline at end of file
-- 
cgit v1.2.3