Merge remote-tracking branch 'upstream/master'

6 files changed, 250 insertions, 195 deletions

diff --git a/libmproxy/console/__init__.py b/libmproxy/console/__init__.py
index d6c7f5a2..a16cc4dc 100644
--- a/libmproxy/console/__init__.py
+++ b/libmproxy/console/__init__.py
@@ -580,7 +580,7 @@ class ConsoleMaster(flow.FlowMaster):
 
         self.view_flowlist()
 
-        self.server.start_slave(controller.Slave, self.masterq)
+        self.server.start_slave(controller.Slave, controller.Channel(self.masterq))
 
         if self.options.rfile:
             ret = self.load_flows(self.options.rfile)
@@ -1002,7 +1002,7 @@ class ConsoleMaster(flow.FlowMaster):
         if self.state.intercept and f.match(self.state.intercept) and not f.request.is_replay():
             f.intercept()
         else:
-            r._ack()
+            r.reply()
         self.sync_list_view()
         self.refresh_flow(f)
 
@@ -1023,7 +1023,7 @@ class ConsoleMaster(flow.FlowMaster):
     # Handlers
     def handle_log(self, l):
         self.add_event(l.msg)
-        l._ack()
+        l.reply()
 
     def handle_error(self, r):
         f = flow.FlowMaster.handle_error(self, r)
diff --git a/libmproxy/console/common.py b/libmproxy/console/common.py
index 2da7f802..1cc0b5b9 100644
--- a/libmproxy/console/common.py
+++ b/libmproxy/console/common.py
@@ -184,7 +184,7 @@ def format_flow(f, focus, extended=False, padding=2):
         req_timestamp = f.request.timestamp_start,
         req_is_replay = f.request.is_replay(),
         req_method = f.request.method,
-        req_acked = f.request.acked,
+        req_acked = f.request.reply.acked,
         req_url = f.request.get_url(),
 
         err_msg = f.error.msg if f.error else None,
@@ -200,7 +200,7 @@ def format_flow(f, focus, extended=False, padding=2):
         d.update(dict(
             resp_code = f.response.code,
             resp_is_replay = f.response.is_replay(),
-            resp_acked = f.response.acked,
+            resp_acked = f.response.reply.acked,
             resp_clen = contentdesc
         ))
         t = f.response.headers["content-type"]
diff --git a/libmproxy/controller.py b/libmproxy/controller.py
index f38d1edb..bb22597d 100644
--- a/libmproxy/controller.py
+++ b/libmproxy/controller.py
@@ -17,37 +17,75 @@ import Queue, threading
 
 should_exit = False
 
-class Msg:
+
+class DummyReply:
+    """
+        A reply object that does nothing. Useful when we need an object to seem
+        like it has a channel, and during testing.
+    """
     def __init__(self):
+        self.acked = False
+
+    def __call__(self, msg=False):
+        self.acked = True
+
+
+class Reply:
+    """
+        Messages sent through a channel are decorated with a "reply" attribute.
+        This object is used to respond to the message through the return
+        channel.
+    """
+    def __init__(self, obj):
+        self.obj = obj
         self.q = Queue.Queue()
         self.acked = False
 
-    def _ack(self, data=False):
+    def __call__(self, msg=None):
         if not self.acked:
             self.acked = True
-            if data is None:
-                self.q.put(data)
+            if msg is None:
+                self.q.put(self.obj)
             else:
-                self.q.put(data or self)
+                self.q.put(msg)
 
-    def _send(self, masterq):
-        self.acked = False
-        try:
-            masterq.put(self, timeout=3)
-            while not should_exit: # pragma: no cover
-                try:
-                    g = self.q.get(timeout=0.5)
-                except Queue.Empty:
-                    continue
-                return g
-        except (Queue.Empty, Queue.Full): # pragma: no cover
-            return None
+
+class Channel:
+    def __init__(self, q):
+        self.q = q
+
+    def ask(self, m):
+        """
+            Decorate a message with a reply attribute, and send it to the
+            master.  then wait for a response.
+        """
+        m.reply = Reply(m)
+        self.q.put(m)
+        while not should_exit:
+            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
+                continue
+            return g
+
+    def tell(self, m):
+        """
+            Decorate a message with a dummy reply attribute, send it to the
+            master, then return immediately.
+        """
+        m.reply = DummyReply()
+        self.q.put(m)
 
 
 class Slave(threading.Thread):
-    def __init__(self, masterq, server):
-        self.masterq, self.server = masterq, server
-        self.server.set_mqueue(masterq)
+    """
+        Slaves get a channel end-point through which they can send messages to
+        the master.
+    """
+    def __init__(self, channel, server):
+        self.channel, self.server = channel, server
+        self.server.set_channel(channel)
         threading.Thread.__init__(self)
 
     def run(self):
@@ -55,6 +93,9 @@ class Slave(threading.Thread):
 
 
 class Master:
+    """
+        Masters get and respond to messages from slaves.
+    """
     def __init__(self, server):
         """
             server may be None if no server is needed.
@@ -81,18 +122,18 @@ class Master:
     def run(self):
         global should_exit
         should_exit = False
-        self.server.start_slave(Slave, self.masterq)
+        self.server.start_slave(Slave, Channel(self.masterq))
         while not should_exit:
             self.tick(self.masterq)
         self.shutdown()
 
-    def handle(self, msg): # pragma: no cover
+    def handle(self, msg):
         c = "handle_" + msg.__class__.__name__.lower()
         m = getattr(self, c, None)
         if m:
             m(msg)
         else:
-            msg._ack()
+            msg.reply()
 
     def shutdown(self):
         global should_exit
diff --git a/libmproxy/dump.py b/libmproxy/dump.py
index 170c701d..3c7eee71 100644
--- a/libmproxy/dump.py
+++ b/libmproxy/dump.py
@@ -150,16 +150,6 @@ class DumpMaster(flow.FlowMaster):
             print >> self.outfile, e
             self.outfile.flush()
 
-    def handle_log(self, l):
-        self.add_event(l.msg)
-        l._ack()
-
-    def handle_request(self, r):
-        f = flow.FlowMaster.handle_request(self, r)
-        if f:
-            r._ack()
-        return f
-
     def indent(self, n, t):
         l = str(t).strip().split("\n")
         return "\n".join(" "*n + i for i in l)
@@ -210,10 +200,20 @@ class DumpMaster(flow.FlowMaster):
             self.outfile.flush()
         self.state.delete_flow(f)
 
+    def handle_log(self, l):
+        self.add_event(l.msg)
+        l.reply()
+
+    def handle_request(self, r):
+        f = flow.FlowMaster.handle_request(self, r)
+        if f:
+            r.reply()
+        return f
+
     def handle_response(self, msg):
         f = flow.FlowMaster.handle_response(self, msg)
         if f:
-            msg._ack()
+            msg.reply()
             self._process_flow(f)
         return f
 
diff --git a/libmproxy/flow.py b/libmproxy/flow.py
index af97698c..1f5d01ee 100644
--- a/libmproxy/flow.py
+++ b/libmproxy/flow.py
@@ -196,7 +196,15 @@ class decoded(object):
             self.o.encode(self.ce)
 
 
-class HTTPMsg(controller.Msg):
+class StateObject:
+    def __eq__(self, other):
+        try:
+            return self._get_state() == other._get_state()
+        except AttributeError:
+            return False
+
+
+class HTTPMsg(StateObject):
     def get_decoded_content(self):
         """
             Returns the decoded content based on the current Content-Encoding header.
@@ -252,6 +260,7 @@ class HTTPMsg(controller.Msg):
             return 0
         return len(self.content)
 
+
 class Request(HTTPMsg):
     """
         An HTTP request.
@@ -289,7 +298,6 @@ class Request(HTTPMsg):
         self.timestamp_start = timestamp_start or utils.timestamp()
         self.timestamp_end = max(timestamp_end or utils.timestamp(), timestamp_start)
         self.close = False
-        controller.Msg.__init__(self)
 
         # Have this request's cookies been modified by sticky cookies or auth?
         self.stickycookie = False
@@ -388,15 +396,8 @@ class Request(HTTPMsg):
     def __hash__(self):
         return id(self)
 
-    def __eq__(self, other):
-        return self._get_state() == other._get_state()
-
     def copy(self):
-        """
-            Returns a copy of this object.
-        """
         c = copy.copy(self)
-        c.acked = True
         c.headers = self.headers.copy()
         return c
 
@@ -603,7 +604,6 @@ class Response(HTTPMsg):
         self.cert = cert
         self.timestamp_start = timestamp_start or utils.timestamp()
         self.timestamp_end = max(timestamp_end or utils.timestamp(), timestamp_start)
-        controller.Msg.__init__(self)
         self.replay = False
 
     def _refresh_cookie(self, c, delta):
@@ -700,15 +700,8 @@ class Response(HTTPMsg):
             state["timestamp_end"],
         )
 
-    def __eq__(self, other):
-        return self._get_state() == other._get_state()
-
     def copy(self):
-        """
-            Returns a copy of this object.
-        """
         c = copy.copy(self)
-        c.acked = True
         c.headers = self.headers.copy()
         return c
 
@@ -773,7 +766,7 @@ class Response(HTTPMsg):
             cookies.append((cookie_name, (cookie_value, cookie_parameters)))
         return dict(cookies)
 
-class ClientDisconnect(controller.Msg):
+class ClientDisconnect:
     """
         A client disconnection event.
 
@@ -782,11 +775,10 @@ class ClientDisconnect(controller.Msg):
             client_conn: ClientConnect object.
     """
     def __init__(self, client_conn):
-        controller.Msg.__init__(self)
         self.client_conn = client_conn
 
 
-class ClientConnect(controller.Msg):
+class ClientConnect(StateObject):
     """
         A single client connection. Each connection can result in multiple HTTP
         Requests.
@@ -807,10 +799,6 @@ class ClientConnect(controller.Msg):
         self.close = False
         self.requestcount = 0
         self.error = None
-        controller.Msg.__init__(self)
-
-    def __eq__(self, other):
-        return self._get_state() == other._get_state()
 
     def __str__(self):
         if self.address:
@@ -839,15 +827,10 @@ class ClientConnect(controller.Msg):
             return None
 
     def copy(self):
-        """
-            Returns a copy of this object.
-        """
-        c = copy.copy(self)
-        c.acked = True
-        return c
+        return copy.copy(self)
 
 
-class Error(controller.Msg):
+class Error(StateObject):
     """
         An Error.
 
@@ -865,18 +848,13 @@ class Error(controller.Msg):
     def __init__(self, request, msg, timestamp=None):
         self.request, self.msg = request, msg
         self.timestamp = timestamp or utils.timestamp()
-        controller.Msg.__init__(self)
 
     def _load_state(self, state):
         self.msg = state["msg"]
         self.timestamp = state["timestamp"]
 
     def copy(self):
-        """
-            Returns a copy of this object.
-        """
         c = copy.copy(self)
-        c.acked = True
         return c
 
     def _get_state(self):
@@ -893,9 +871,6 @@ class Error(controller.Msg):
             state["timestamp"],
         )
 
-    def __eq__(self, other):
-        return self._get_state() == other._get_state()
-
     def replace(self, pattern, repl, *args, **kwargs):
         """
             Replaces a regular expression pattern with repl in both the headers
@@ -1185,10 +1160,11 @@ class Flow:
             Kill this request.
         """
         self.error = Error(self.request, "Connection killed")
-        if self.request and not self.request.acked:
-            self.request._ack(None)
-        elif self.response and not self.response.acked:
-            self.response._ack(None)
+        self.error.reply = controller.DummyReply()
+        if self.request and not self.request.reply.acked:
+            self.request.reply(proxy.KILL)
+        elif self.response and not self.response.reply.acked:
+            self.response.reply(proxy.KILL)
         master.handle_error(self.error)
         self.intercepting = False
 
@@ -1204,10 +1180,10 @@ class Flow:
             Continue with the flow - called after an intercept().
         """
         if self.request:
-            if not self.request.acked:
-                self.request._ack()
-            elif self.response and not self.response.acked:
-                self.response._ack()
+            if not self.request.reply.acked:
+                self.request.reply()
+            elif self.response and not self.response.reply.acked:
+                self.response.reply()
             self.intercepting = False
 
     def replace(self, pattern, repl, *args, **kwargs):
@@ -1469,7 +1445,7 @@ class FlowMaster(controller.Master):
             flow.response = response
             if self.refresh_server_playback:
                 response.refresh()
-            flow.request._ack(response)
+            flow.request.reply(response)
             if self.server_playback.count() == 0:
                 self.stop_server_playback()
             return True
@@ -1496,10 +1472,13 @@ class FlowMaster(controller.Master):
             Loads a flow, and returns a new flow object.
         """
         if f.request:
+            f.request.reply = controller.DummyReply()
             fr = self.handle_request(f.request)
         if f.response:
+            f.response.reply = controller.DummyReply()
             self.handle_response(f.response)
         if f.error:
+            f.error.reply = controller.DummyReply()
             self.handle_error(f.error)
         return fr
 
@@ -1527,7 +1506,7 @@ class FlowMaster(controller.Master):
                 if self.kill_nonreplay:
                     f.kill(self)
                 else:
-                    f.request._ack()
+                    f.request.reply()
 
     def process_new_response(self, f):
         if self.stickycookie_state:
@@ -1566,11 +1545,11 @@ class FlowMaster(controller.Master):
 
     def handle_clientconnect(self, cc):
         self.run_script_hook("clientconnect", cc)
-        cc._ack()
+        cc.reply()
 
     def handle_clientdisconnect(self, r):
         self.run_script_hook("clientdisconnect", r)
-        r._ack()
+        r.reply()
 
     def handle_error(self, r):
         f = self.state.add_error(r)
@@ -1578,7 +1557,7 @@ class FlowMaster(controller.Master):
             self.run_script_hook("error", f)
         if self.client_playback:
             self.client_playback.clear(f)
-        r._ack()
+        r.reply()
         return f
 
     def handle_request(self, r):
@@ -1601,7 +1580,7 @@ class FlowMaster(controller.Master):
             if self.stream:
                 self.stream.add(f)
         else:
-            r._ack()
+            r.reply()
         return f
 
     def shutdown(self):
diff --git a/libmproxy/proxy.py b/libmproxy/proxy.py
index f14e4e3e..7c229064 100644
--- a/libmproxy/proxy.py
+++ b/libmproxy/proxy.py
@@ -20,6 +20,8 @@ from netlib import odict, tcp, http, wsgi, certutils, http_status
 import utils, flow, version, platform, controller
 import authentication
 
+KILL = 0
+
 
 class ProxyError(Exception):
     def __init__(self, code, msg, headers=None):
@@ -29,9 +31,8 @@ class ProxyError(Exception):
         return "ProxyError(%s, %s)"%(self.code, self.msg)
 
 
-class Log(controller.Msg):
+class Log:
     def __init__(self, msg):
-        controller.Msg.__init__(self)
         self.msg = msg
 
 
@@ -49,45 +50,23 @@ class ProxyConfig:
         self.certstore = certutils.CertStore(certdir)
 
 
-class RequestReplayThread(threading.Thread):
-    def __init__(self, config, flow, masterq):
-        self.config, self.flow, self.masterq = config, flow, masterq
-        threading.Thread.__init__(self)
-
-    def run(self):
-        try:
-            r = self.flow.request
-            server = ServerConnection(self.config, r.host, r.port)
-            server.connect(r.scheme)
-            server.send(r)
-            httpversion, code, msg, headers, content = http.read_response(
-                server.rfile, r.method, self.config.body_size_limit
-            )
-            response = flow.Response(
-                self.flow.request, httpversion, code, msg, headers, content, server.cert
-            )
-            response._send(self.masterq)
-        except (ProxyError, http.HttpError, tcp.NetLibError), v:
-            err = flow.Error(self.flow.request, str(v))
-            err._send(self.masterq)
-
-
 class ServerConnection(tcp.TCPClient):
-    def __init__(self, config, host, port):
+    def __init__(self, config, scheme, host, port, sni):
         tcp.TCPClient.__init__(self, host, port)
         self.config = config
+        self.scheme, self.sni = scheme, sni
         self.requestcount = 0
 
-    def connect(self, scheme):
+    def connect(self):
         tcp.TCPClient.connect(self)
-        if scheme == "https":
+        if self.scheme == "https":
             clientcert = None
             if self.config.clientcerts:
                 path = os.path.join(self.config.clientcerts, self.host.encode("idna")) + ".pem"
                 if os.path.exists(path):
                     clientcert = path
             try:
-                self.convert_to_ssl(clientcert=clientcert, sni=self.host)
+                self.convert_to_ssl(cert=clientcert, sni=self.sni)
             except tcp.NetLibError, v:
                 raise ProxyError(400, str(v))
 
@@ -108,42 +87,78 @@ class ServerConnection(tcp.TCPClient):
             pass
 
 
+class RequestReplayThread(threading.Thread):
+    def __init__(self, config, flow, masterq):
+        self.config, self.flow, self.channel = config, flow, controller.Channel(masterq)
+        threading.Thread.__init__(self)
+
+    def run(self):
+        try:
+            r = self.flow.request
+            server = ServerConnection(self.config, r.scheme, r.host, r.port, r.host)
+            server.connect()
+            server.send(r)
+            httpversion, code, msg, headers, content = http.read_response(
+                server.rfile, r.method, self.config.body_size_limit
+            )
+            response = flow.Response(
+                self.flow.request, httpversion, code, msg, headers, content, server.cert
+            )
+            self.channel.ask(response)
+        except (ProxyError, http.HttpError, tcp.NetLibError), v:
+            err = flow.Error(self.flow.request, str(v))
+            self.channel.ask(err)
+
+
 class ProxyHandler(tcp.BaseHandler):
-    def __init__(self, config, connection, client_address, server, mqueue, server_version):
-        self.mqueue, self.server_version = mqueue, server_version
+    def __init__(self, config, connection, client_address, server, channel, server_version):
+        self.channel, self.server_version = channel, server_version
         self.config = config
-        self.server_conn = None
         self.proxy_connect_state = None
         self.sni = None
+        self.server_conn = None
         tcp.BaseHandler.__init__(self, connection, client_address, server)
 
+    def get_server_connection(self, cc, scheme, host, port, sni):
+        sc = self.server_conn
+        if sc and (scheme, host, port, sni) != (sc.scheme, sc.host, sc.port, sc.sni):
+            sc.terminate()
+            self.server_conn = None
+            self.log(
+                cc,
+                "switching connection", [
+                    "%s://%s:%s (sni=%s) -> %s://%s:%s (sni=%s)"%(
+                        scheme, host, port, sni,
+                        sc.scheme, sc.host, sc.port, sc.sni
+                    )
+                ]
+            )
+        if not self.server_conn:
+            try:
+                self.server_conn = ServerConnection(self.config, scheme, host, port, sni)
+                self.server_conn.connect()
+            except tcp.NetLibError, v:
+                raise ProxyError(502, v)
+        return self.server_conn
+
+    def del_server_connection(self):
+        self.server_conn = None
+
     def handle(self):
         cc = flow.ClientConnect(self.client_address)
         self.log(cc, "connect")
-        cc._send(self.mqueue)
+        self.channel.ask(cc)
         while self.handle_request(cc) and not cc.close:
             pass
         cc.close = True
-        cd = flow.ClientDisconnect(cc)
 
+        cd = flow.ClientDisconnect(cc)
         self.log(
             cc, "disconnect",
             [
                 "handled %s requests"%cc.requestcount]
         )
-        cd._send(self.mqueue)
-
-    def server_connect(self, scheme, host, port):
-        sc = self.server_conn
-        if sc and (host, port) != (sc.host, sc.port):
-            sc.terminate()
-            self.server_conn = None
-        if not self.server_conn:
-            try:
-                self.server_conn = ServerConnection(self.config, host, port)
-                self.server_conn.connect(scheme)
-            except tcp.NetLibError, v:
-                raise ProxyError(502, v)
+        self.channel.tell(cd)
 
     def handle_request(self, cc):
         try:
@@ -160,45 +175,66 @@ class ProxyHandler(tcp.BaseHandler):
                     self.log(cc, "Error in wsgi app.", err.split("\n"))
                     return
             else:
-                request = request._send(self.mqueue)
-                if request is None:
+                request_reply = self.channel.ask(request)
+                if request_reply == KILL:
                     return
-
-                if isinstance(request, flow.Response):
-                    response = request
+                elif isinstance(request_reply, flow.Response):
                     request = False
-                    response = response._send(self.mqueue)
+                    response = request_reply
+                    response_reply = self.channel.ask(response)
                 else:
+                    request = request_reply
                     if self.config.reverse_proxy:
                         scheme, host, port = self.config.reverse_proxy
                     else:
                         scheme, host, port = request.scheme, request.host, request.port
-                    self.server_connect(scheme, host, port)
-                    self.server_conn.send(request)
-                    self.server_conn.rfile.reset_timestamps()
-                    httpversion, code, msg, headers, content = http.read_response(
-                        self.server_conn.rfile,
-                        request.method,
-                        self.config.body_size_limit
-                    )
+
+                    # If we've already pumped a request over this connection,
+                    # it's possible that the server has timed out. If this is
+                    # the case, we want to reconnect without sending an error
+                    # to the client.
+                    while 1:
+                        sc = self.get_server_connection(cc, scheme, host, port, host)
+                        sc.send(request)
+                        sc.rfile.reset_timestamps()
+                        try:
+                            httpversion, code, msg, headers, content = http.read_response(
+                                sc.rfile,
+                                request.method,
+                                self.config.body_size_limit
+                            )
+                        except http.HttpErrorConnClosed, v:
+                            self.del_server_connection()
+                            if sc.requestcount > 1:
+                                continue
+                            else:
+                                raise
+                        else:
+                            break
+
                     response = flow.Response(
-                        request, httpversion, code, msg, headers, content, self.server_conn.cert, self.server_conn.rfile.first_byte_timestamp, utils.timestamp()
+                        request, httpversion, code, msg, headers, content, sc.cert,
+                        sc.rfile.first_byte_timestamp, utils.timestamp()
                     )
+                    response_reply = self.channel.ask(response)
+                    # Not replying to the server invalidates the server
+                    # connection, so we terminate.
+                    if response_reply == KILL:
+                        sc.terminate()
 
-                    response = response._send(self.mqueue)
-                    if response is None:
-                        self.server_conn.terminate()
-                if response is None:
-                    return
-                self.send_response(response)
-                if request and http.request_connection_close(request.httpversion, request.headers):
-                    return
-                # We could keep the client connection when the server
-                # connection needs to go away.  However, we want to mimic
-                # behaviour as closely as possible to the client, so we
-                # disconnect.
-                if http.response_connection_close(response.httpversion, response.headers):
+                if response_reply == KILL:
                     return
+                else:
+                    response = response_reply
+                    self.send_response(response)
+                    if request and http.request_connection_close(request.httpversion, request.headers):
+                        return
+                    # We could keep the client connection when the server
+                    # connection needs to go away.  However, we want to mimic
+                    # behaviour as closely as possible to the client, so we
+                    # disconnect.
+                    if http.response_connection_close(response.httpversion, response.headers):
+                        return
         except (IOError, ProxyError, http.HttpError, tcp.NetLibDisconnect), e:
             if hasattr(e, "code"):
                 cc.error = "%s: %s"%(e.code, e.msg)
@@ -207,7 +243,7 @@ class ProxyHandler(tcp.BaseHandler):
 
             if request:
                 err = flow.Error(request, cc.error)
-                err._send(self.mqueue)
+                self.channel.ask(err)
                 self.log(
                     cc, cc.error,
                     ["url: %s"%request.get_url()]
@@ -228,7 +264,7 @@ class ProxyHandler(tcp.BaseHandler):
             msg.append("  -> "+i)
         msg = "\n".join(msg)
         l = Log(msg)
-        l._send(self.mqueue)
+        self.channel.tell(l)
 
     def find_cert(self, host, port, sni):
         if self.config.certfile:
@@ -292,25 +328,6 @@ class ProxyHandler(tcp.BaseHandler):
                     self.rfile.first_byte_timestamp, utils.timestamp()
                )
 
-    def read_request_reverse(self, client_conn):
-        line = self.get_line(self.rfile)
-        if line == "":
-            return None
-        scheme, host, port = self.config.reverse_proxy
-        r = http.parse_init_http(line)
-        if not r:
-            raise ProxyError(400, "Bad HTTP request line: %s"%repr(line))
-        method, path, httpversion = r
-        headers = self.read_headers(authenticate=False)
-        content = http.read_http_body_request(
-                    self.rfile, self.wfile, headers, httpversion, self.config.body_size_limit
-                )
-        return flow.Request(
-                    client_conn, httpversion, host, port, "http", method, path, headers, content,
-                    self.rfile.first_byte_timestamp, utils.timestamp()
-               )
-
-    
     def read_request_proxy(self, client_conn):
         line = self.get_line(self.rfile)
         if line == "":
@@ -366,6 +383,24 @@ class ProxyHandler(tcp.BaseHandler):
                         self.rfile.first_byte_timestamp, utils.timestamp()
                     )
 
+    def read_request_reverse(self, client_conn):
+        line = self.get_line(self.rfile)
+        if line == "":
+            return None
+        scheme, host, port = self.config.reverse_proxy
+        r = http.parse_init_http(line)
+        if not r:
+            raise ProxyError(400, "Bad HTTP request line: %s"%repr(line))
+        method, path, httpversion = r
+        headers = self.read_headers(authenticate=False)
+        content = http.read_http_body_request(
+                    self.rfile, self.wfile, headers, httpversion, self.config.body_size_limit
+                )
+        return flow.Request(
+                    client_conn, httpversion, host, port, "http", method, path, headers, content,
+                    self.rfile.first_byte_timestamp, utils.timestamp()
+               )
+
     def read_request(self, client_conn):
         self.rfile.reset_timestamps()
         if self.config.transparent_proxy:
@@ -431,18 +466,18 @@ class ProxyServer(tcp.TCPServer):
             tcp.TCPServer.__init__(self, (address, port))
         except socket.error, v:
             raise ProxyServerError('Error starting proxy server: ' + v.strerror)
-        self.masterq = None
+        self.channel = None
         self.apps = AppRegistry()
 
-    def start_slave(self, klass, masterq):
-        slave = klass(masterq, self)
+    def start_slave(self, klass, channel):
+        slave = klass(channel, self)
         slave.start()
 
-    def set_mqueue(self, q):
-        self.masterq = q
+    def set_channel(self, channel):
+        self.channel = channel
 
     def handle_connection(self, request, client_address):
-        h = ProxyHandler(self.config, request, client_address, self, self.masterq, self.server_version)
+        h = ProxyHandler(self.config, request, client_address, self, self.channel, self.server_version)
         h.handle()
         try:
             h.finish()
@@ -480,7 +515,7 @@ class DummyServer:
     def __init__(self, config):
         self.config = config
 
-    def start_slave(self, klass, masterq):
+    def start_slave(self, klass, channel):
         pass
 
     def shutdown(self):