From 2aa175a6ca657db0b4faa2aeb84a78b7ef3c4761 Mon Sep 17 00:00:00 2001
From: Aldo Cortesi <aldo@nullcube.com>
Date: Tue, 29 Jan 2013 10:55:19 +1300
Subject: Stub implementation of a server connection pool.

---
 libmproxy/proxy.py | 48 +++++++++++++++++++++++++++---------------------
 1 file changed, 27 insertions(+), 21 deletions(-)

(limited to 'libmproxy')

diff --git a/libmproxy/proxy.py b/libmproxy/proxy.py
index f14e4e3e..3bbb82ba 100644
--- a/libmproxy/proxy.py
+++ b/libmproxy/proxy.py
@@ -107,12 +107,30 @@ class ServerConnection(tcp.TCPClient):
         except IOError:
             pass
 
+class ServerConnectionPool:
+    def __init__(self, config):
+        self.config = config
+        self.conn = None
+
+    def get_connection(self, scheme, host, port):
+        sc = self.conn
+        if self.conn and (host, port) != (sc.host, sc.port):
+            sc.terminate()
+            self.conn = None
+        if not self.conn:
+            try:
+                self.conn = ServerConnection(self.config, host, port)
+                self.conn.connect(scheme)
+            except tcp.NetLibError, v:
+                raise ProxyError(502, v)
+        return self.conn
+
 
 class ProxyHandler(tcp.BaseHandler):
     def __init__(self, config, connection, client_address, server, mqueue, server_version):
         self.mqueue, self.server_version = mqueue, server_version
         self.config = config
-        self.server_conn = None
+        self.server_conn_pool = ServerConnectionPool(config)
         self.proxy_connect_state = None
         self.sni = None
         tcp.BaseHandler.__init__(self, connection, client_address, server)
@@ -133,18 +151,6 @@ class ProxyHandler(tcp.BaseHandler):
         )
         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)
-
     def handle_request(self, cc):
         try:
             request, err = None, None
@@ -173,21 +179,21 @@ class ProxyHandler(tcp.BaseHandler):
                         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()
+                    sc = self.server_conn_pool.get_connection(scheme, host, port)
+                    sc.send(request)
+                    sc.rfile.reset_timestamps()
                     httpversion, code, msg, headers, content = http.read_response(
-                        self.server_conn.rfile,
+                        sc.rfile,
                         request.method,
                         self.config.body_size_limit
                     )
                     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 = response._send(self.mqueue)
                     if response is None:
-                        self.server_conn.terminate()
+                        sc.terminate()
                 if response is None:
                     return
                 self.send_response(response)
@@ -310,7 +316,7 @@ class ProxyHandler(tcp.BaseHandler):
                     self.rfile.first_byte_timestamp, utils.timestamp()
                )
 
-    
+
     def read_request_proxy(self, client_conn):
         line = self.get_line(self.rfile)
         if line == "":
-- 
cgit v1.2.3


From 782bbee8c0a7d14be25e87d61c9c6db03b532eb7 Mon Sep 17 00:00:00 2001
From: Aldo Cortesi <aldo@nullcube.com>
Date: Tue, 29 Jan 2013 11:35:57 +1300
Subject: Unit tests for ServerConnectionPool

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

(limited to 'libmproxy')

diff --git a/libmproxy/proxy.py b/libmproxy/proxy.py
index 3bbb82ba..f2c6db1f 100644
--- a/libmproxy/proxy.py
+++ b/libmproxy/proxy.py
@@ -107,6 +107,7 @@ class ServerConnection(tcp.TCPClient):
         except IOError:
             pass
 
+
 class ServerConnectionPool:
     def __init__(self, config):
         self.config = config
-- 
cgit v1.2.3


From aaf892e3afc682b2dc2a166a96872420e50092cd Mon Sep 17 00:00:00 2001
From: Aldo Cortesi <aldo@nullcube.com>
Date: Sun, 17 Feb 2013 12:42:48 +1300
Subject: Significantly refactor the master/slave message passing interface.

---
 libmproxy/console/__init__.py |  6 +--
 libmproxy/console/common.py   |  4 +-
 libmproxy/controller.py       | 85 +++++++++++++++++++++++++++++++------------
 libmproxy/dump.py             | 22 +++++------
 libmproxy/flow.py             | 50 ++++++++++++-------------
 libmproxy/proxy.py            | 43 +++++++++++-----------
 6 files changed, 122 insertions(+), 88 deletions(-)

(limited to 'libmproxy')

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..c36bb9df 100644
--- a/libmproxy/controller.py
+++ b/libmproxy/controller.py
@@ -17,37 +17,73 @@ 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=False):
         if not self.acked:
             self.acked = True
-            if data is None:
-                self.q.put(data)
+            if msg is None:
+                self.q.put(msg)
             else:
-                self.q.put(data or self)
+                self.q.put(msg or self.obj)
 
-    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):
+        """
+            Send a message to the master, and 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:
+                continue
+            return g
+
+    def tell(self, m):
+        """
+            Send a message to the master, and keep going. 
+        """
+        m.reply = None
+        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 +91,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 +120,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 9238cfbf..0f5fb563 100644
--- a/libmproxy/flow.py
+++ b/libmproxy/flow.py
@@ -196,7 +196,7 @@ class decoded(object):
             self.o.encode(self.ce)
 
 
-class HTTPMsg(controller.Msg):
+class HTTPMsg:
     def get_decoded_content(self):
         """
             Returns the decoded content based on the current Content-Encoding header.
@@ -252,6 +252,7 @@ class HTTPMsg(controller.Msg):
             return 0
         return len(self.content)
 
+
 class Request(HTTPMsg):
     """
         An HTTP request.
@@ -289,7 +290,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
@@ -396,7 +396,6 @@ class Request(HTTPMsg):
             Returns a copy of this object.
         """
         c = copy.copy(self)
-        c.acked = True
         c.headers = self.headers.copy()
         return c
 
@@ -603,7 +602,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):
@@ -708,7 +706,6 @@ class Response(HTTPMsg):
             Returns a copy of this object.
         """
         c = copy.copy(self)
-        c.acked = True
         c.headers = self.headers.copy()
         return c
 
@@ -773,7 +770,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 +779,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:
     """
         A single client connection. Each connection can result in multiple HTTP
         Requests.
@@ -807,7 +803,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()
@@ -838,11 +833,10 @@ class ClientConnect(controller.Msg):
             Returns a copy of this object.
         """
         c = copy.copy(self)
-        c.acked = True
         return c
 
 
-class Error(controller.Msg):
+class Error:
     """
         An Error.
 
@@ -860,7 +854,6 @@ 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"]
@@ -871,7 +864,6 @@ class Error(controller.Msg):
             Returns a copy of this object.
         """
         c = copy.copy(self)
-        c.acked = True
         return c
 
     def _get_state(self):
@@ -1180,10 +1172,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(None)
+        elif self.response and not self.response.reply.acked:
+            self.response.reply(None)
         master.handle_error(self.error)
         self.intercepting = False
 
@@ -1199,10 +1192,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):
@@ -1464,7 +1457,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
@@ -1491,10 +1484,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
 
@@ -1522,7 +1518,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:
@@ -1561,11 +1557,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)
@@ -1573,7 +1569,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):
@@ -1596,7 +1592,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 f2c6db1f..1fbb6d58 100644
--- a/libmproxy/proxy.py
+++ b/libmproxy/proxy.py
@@ -29,9 +29,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
 
 
@@ -51,7 +50,7 @@ class ProxyConfig:
 
 class RequestReplayThread(threading.Thread):
     def __init__(self, config, flow, masterq):
-        self.config, self.flow, self.masterq = config, flow, masterq
+        self.config, self.flow, self.channel = config, flow, controller.Channel(masterq)
         threading.Thread.__init__(self)
 
     def run(self):
@@ -66,10 +65,10 @@ class RequestReplayThread(threading.Thread):
             response = flow.Response(
                 self.flow.request, httpversion, code, msg, headers, content, server.cert
             )
-            response._send(self.masterq)
+            self.channel.ask(response)
         except (ProxyError, http.HttpError, tcp.NetLibError), v:
             err = flow.Error(self.flow.request, str(v))
-            err._send(self.masterq)
+            self.channel.ask(err)
 
 
 class ServerConnection(tcp.TCPClient):
@@ -128,8 +127,8 @@ class ServerConnectionPool:
 
 
 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_pool = ServerConnectionPool(config)
         self.proxy_connect_state = None
@@ -139,18 +138,18 @@ class ProxyHandler(tcp.BaseHandler):
     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)
+        self.channel.ask(cd)
 
     def handle_request(self, cc):
         try:
@@ -167,14 +166,14 @@ class ProxyHandler(tcp.BaseHandler):
                     self.log(cc, "Error in wsgi app.", err.split("\n"))
                     return
             else:
-                request = request._send(self.mqueue)
+                request = self.channel.ask(request)
                 if request is None:
                     return
 
                 if isinstance(request, flow.Response):
                     response = request
                     request = False
-                    response = response._send(self.mqueue)
+                    response = self.channel.ask(response)
                 else:
                     if self.config.reverse_proxy:
                         scheme, host, port = self.config.reverse_proxy
@@ -192,7 +191,7 @@ class ProxyHandler(tcp.BaseHandler):
                         request, httpversion, code, msg, headers, content, sc.cert,
                         sc.rfile.first_byte_timestamp, utils.timestamp()
                     )
-                    response = response._send(self.mqueue)
+                    response = self.channel.ask(response)
                     if response is None:
                         sc.terminate()
                 if response is None:
@@ -214,7 +213,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()]
@@ -235,7 +234,7 @@ class ProxyHandler(tcp.BaseHandler):
             msg.append("  -> "+i)
         msg = "\n".join(msg)
         l = Log(msg)
-        l._send(self.mqueue)
+        self.channel.ask(l)
 
     def find_cert(self, host, port, sni):
         if self.config.certfile:
@@ -438,18 +437,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()
@@ -487,7 +486,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):
-- 
cgit v1.2.3


From 7800b7c9103ae119a13b81048a84f626850cf04f Mon Sep 17 00:00:00 2001
From: Aldo Cortesi <aldo@nullcube.com>
Date: Sat, 23 Feb 2013 14:08:28 +1300
Subject: Refactor proxy core communications to be clearer.

---
 libmproxy/controller.py | 10 ++++++----
 libmproxy/flow.py       | 45 ++++++++++++++-------------------------------
 libmproxy/proxy.py      | 46 ++++++++++++++++++++++++++--------------------
 3 files changed, 46 insertions(+), 55 deletions(-)

(limited to 'libmproxy')

diff --git a/libmproxy/controller.py b/libmproxy/controller.py
index c36bb9df..849d998b 100644
--- a/libmproxy/controller.py
+++ b/libmproxy/controller.py
@@ -56,13 +56,14 @@ class Channel:
 
     def ask(self, m):
         """
-            Send a message to the master, and wait for a response.
+            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. 
+                # The timeout is here so we can handle a should_exit event.
                 g = m.reply.q.get(timeout=0.5)
             except Queue.Empty:
                 continue
@@ -70,9 +71,10 @@ class Channel:
 
     def tell(self, m):
         """
-            Send a message to the master, and keep going. 
+            Decorate a message with a dummy reply attribute, send it to the
+            master, then return immediately.
         """
-        m.reply = None
+        m.reply = DummyReply()
         self.q.put(m)
 
 
diff --git a/libmproxy/flow.py b/libmproxy/flow.py
index 0f5fb563..883c7235 100644
--- a/libmproxy/flow.py
+++ b/libmproxy/flow.py
@@ -196,7 +196,15 @@ class decoded(object):
             self.o.encode(self.ce)
 
 
-class HTTPMsg:
+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.
@@ -388,13 +396,7 @@ 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.headers = self.headers.copy()
         return c
@@ -698,13 +700,7 @@ 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.headers = self.headers.copy()
         return c
@@ -782,7 +778,7 @@ class ClientDisconnect:
         self.client_conn = client_conn
 
 
-class ClientConnect:
+class ClientConnect(StateObject):
     """
         A single client connection. Each connection can result in multiple HTTP
         Requests.
@@ -804,9 +800,6 @@ class ClientConnect:
         self.requestcount = 0
         self.error = None
 
-    def __eq__(self, other):
-        return self._get_state() == other._get_state()
-
     def _load_state(self, state):
         self.close = True
         self.error = state["error"]
@@ -829,14 +822,10 @@ class ClientConnect:
             return None
 
     def copy(self):
-        """
-            Returns a copy of this object.
-        """
-        c = copy.copy(self)
-        return c
+        return copy.copy(self)
 
 
-class Error:
+class Error(StateObject):
     """
         An Error.
 
@@ -860,9 +849,6 @@ class Error:
         self.timestamp = state["timestamp"]
 
     def copy(self):
-        """
-            Returns a copy of this object.
-        """
         c = copy.copy(self)
         return c
 
@@ -880,9 +866,6 @@ class Error:
             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
@@ -1174,9 +1157,9 @@ class Flow:
         self.error = Error(self.request, "Connection killed")
         self.error.reply = controller.DummyReply()
         if self.request and not self.request.reply.acked:
-            self.request.reply(None)
+            self.request.reply(proxy.KILL)
         elif self.response and not self.response.reply.acked:
-            self.response.reply(None)
+            self.response.reply(proxy.KILL)
         master.handle_error(self.error)
         self.intercepting = False
 
diff --git a/libmproxy/proxy.py b/libmproxy/proxy.py
index 1fbb6d58..6d476c7b 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):
@@ -149,7 +151,7 @@ class ProxyHandler(tcp.BaseHandler):
             [
                 "handled %s requests"%cc.requestcount]
         )
-        self.channel.ask(cd)
+        self.channel.tell(cd)
 
     def handle_request(self, cc):
         try:
@@ -166,15 +168,15 @@ class ProxyHandler(tcp.BaseHandler):
                     self.log(cc, "Error in wsgi app.", err.split("\n"))
                     return
             else:
-                request = self.channel.ask(request)
-                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 = self.channel.ask(response)
+                    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:
@@ -191,20 +193,24 @@ class ProxyHandler(tcp.BaseHandler):
                         request, httpversion, code, msg, headers, content, sc.cert,
                         sc.rfile.first_byte_timestamp, utils.timestamp()
                     )
-                    response = self.channel.ask(response)
-                    if response is None:
+                    response_reply = self.channel.ask(response)
+                    # Not replying to the server invalidates the server connection, so we terminate.
+                    if response_reply == KILL:
                         sc.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)
@@ -234,7 +240,7 @@ class ProxyHandler(tcp.BaseHandler):
             msg.append("  -> "+i)
         msg = "\n".join(msg)
         l = Log(msg)
-        self.channel.ask(l)
+        self.channel.tell(l)
 
     def find_cert(self, host, port, sni):
         if self.config.certfile:
-- 
cgit v1.2.3


From f203881b0d7f81a7f8ecbc44b7030060242af01b Mon Sep 17 00:00:00 2001
From: Aldo Cortesi <aldo@nullcube.com>
Date: Sat, 23 Feb 2013 14:13:43 +1300
Subject: Remove redundant clause in controller.Reply

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

(limited to 'libmproxy')

diff --git a/libmproxy/controller.py b/libmproxy/controller.py
index 849d998b..da097692 100644
--- a/libmproxy/controller.py
+++ b/libmproxy/controller.py
@@ -44,10 +44,7 @@ class Reply:
     def __call__(self, msg=False):
         if not self.acked:
             self.acked = True
-            if msg is None:
-                self.q.put(msg)
-            else:
-                self.q.put(msg or self.obj)
+            self.q.put(msg or self.obj)
 
 
 class Channel:
-- 
cgit v1.2.3


From 269780c57780d155c4d8bd95fcc2af2104effa5b Mon Sep 17 00:00:00 2001
From: Aldo Cortesi <aldo@nullcube.com>
Date: Sat, 23 Feb 2013 16:34:59 +1300
Subject: Unit test dummy response functions.

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

(limited to 'libmproxy')

diff --git a/libmproxy/proxy.py b/libmproxy/proxy.py
index 6d476c7b..c8fac5f4 100644
--- a/libmproxy/proxy.py
+++ b/libmproxy/proxy.py
@@ -194,7 +194,8 @@ class ProxyHandler(tcp.BaseHandler):
                         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.
+                    # Not replying to the server invalidates the server
+                    # connection, so we terminate.
                     if response_reply == KILL:
                         sc.terminate()
 
-- 
cgit v1.2.3


From 05e4d4468ec372adb73649e6980c525a185e9c07 Mon Sep 17 00:00:00 2001
From: Aldo Cortesi <aldo@nullcube.com>
Date: Sat, 23 Feb 2013 21:59:25 +1300
Subject: Test request and response kill functionality.

---
 libmproxy/controller.py | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

(limited to 'libmproxy')

diff --git a/libmproxy/controller.py b/libmproxy/controller.py
index da097692..bb22597d 100644
--- a/libmproxy/controller.py
+++ b/libmproxy/controller.py
@@ -41,10 +41,13 @@ class Reply:
         self.q = Queue.Queue()
         self.acked = False
 
-    def __call__(self, msg=False):
+    def __call__(self, msg=None):
         if not self.acked:
             self.acked = True
-            self.q.put(msg or self.obj)
+            if msg is None:
+                self.q.put(self.obj)
+            else:
+                self.q.put(msg)
 
 
 class Channel:
@@ -62,7 +65,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:
+            except Queue.Empty: # pragma: nocover
                 continue
             return g
 
-- 
cgit v1.2.3


From d0639e8925541bd6f6f386386c982d23b3828d3d Mon Sep 17 00:00:00 2001
From: Aldo Cortesi <aldo@nullcube.com>
Date: Sun, 24 Feb 2013 14:04:56 +1300
Subject: Handle server disconnects better.

Server connections can be closed for legitimate reasons, like timeouts. If
we've already pumped data over a server connection, we reconnect on error. If
not, we treat it as a legitimate error and pass it on to the client.

Fixes #85
---
 libmproxy/proxy.py | 39 +++++++++++++++++++++++++++++----------
 1 file changed, 29 insertions(+), 10 deletions(-)

(limited to 'libmproxy')

diff --git a/libmproxy/proxy.py b/libmproxy/proxy.py
index c8fac5f4..088fe94c 100644
--- a/libmproxy/proxy.py
+++ b/libmproxy/proxy.py
@@ -117,8 +117,8 @@ class ServerConnectionPool:
     def get_connection(self, scheme, host, port):
         sc = self.conn
         if self.conn and (host, port) != (sc.host, sc.port):
-            sc.terminate()
-            self.conn = None
+                sc.terminate()
+                self.conn = None
         if not self.conn:
             try:
                 self.conn = ServerConnection(self.config, host, port)
@@ -127,6 +127,9 @@ class ServerConnectionPool:
                 raise ProxyError(502, v)
         return self.conn
 
+    def del_connection(self, scheme, host, port):
+        self.conn = None
+
 
 class ProxyHandler(tcp.BaseHandler):
     def __init__(self, config, connection, client_address, server, channel, server_version):
@@ -181,14 +184,30 @@ class ProxyHandler(tcp.BaseHandler):
                         scheme, host, port = self.config.reverse_proxy
                     else:
                         scheme, host, port = request.scheme, request.host, request.port
-                    sc = self.server_conn_pool.get_connection(scheme, host, port)
-                    sc.send(request)
-                    sc.rfile.reset_timestamps()
-                    httpversion, code, msg, headers, content = http.read_response(
-                        sc.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:
+                        try:
+                            sc = self.server_conn_pool.get_connection(scheme, host, port)
+                            sc.send(request)
+                            sc.rfile.reset_timestamps()
+                            httpversion, code, msg, headers, content = http.read_response(
+                                sc.rfile,
+                                request.method,
+                                self.config.body_size_limit
+                            )
+                        except http.HttpErrorConnClosed, v:
+                            if sc.requestcount > 1:
+                                self.server_conn_pool.del_connection(scheme, host, port)
+                                continue
+                            else:
+                                raise
+                        else:
+                            break
+
                     response = flow.Response(
                         request, httpversion, code, msg, headers, content, sc.cert,
                         sc.rfile.first_byte_timestamp, utils.timestamp()
-- 
cgit v1.2.3


From 705559d65e5dc5883395efb85bacbf1459eb243c Mon Sep 17 00:00:00 2001
From: Aldo Cortesi <aldo@nullcube.com>
Date: Sun, 24 Feb 2013 17:35:24 +1300
Subject: Refactor to prepare for SNI fixes.

---
 libmproxy/proxy.py | 99 +++++++++++++++++++++++++++---------------------------
 1 file changed, 49 insertions(+), 50 deletions(-)

(limited to 'libmproxy')

diff --git a/libmproxy/proxy.py b/libmproxy/proxy.py
index 088fe94c..d92e2da9 100644
--- a/libmproxy/proxy.py
+++ b/libmproxy/proxy.py
@@ -50,36 +50,13 @@ class ProxyConfig:
         self.certstore = certutils.CertStore(certdir)
 
 
-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.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
-            )
-            self.channel.ask(response)
-        except (ProxyError, http.HttpError, tcp.NetLibError), v:
-            err = flow.Error(self.flow.request, str(v))
-            self.channel.ask(err)
-
-
 class ServerConnection(tcp.TCPClient):
     def __init__(self, config, host, port):
         tcp.TCPClient.__init__(self, host, port)
         self.config = config
         self.requestcount = 0
 
-    def connect(self, scheme):
+    def connect(self, scheme, sni):
         tcp.TCPClient.connect(self)
         if scheme == "https":
             clientcert = None
@@ -88,7 +65,7 @@ class ServerConnection(tcp.TCPClient):
                 if os.path.exists(path):
                     clientcert = path
             try:
-                self.convert_to_ssl(clientcert=clientcert, sni=self.host)
+                self.convert_to_ssl(cert=clientcert, sni=sni)
             except tcp.NetLibError, v:
                 raise ProxyError(400, str(v))
 
@@ -109,12 +86,35 @@ 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.host, r.port)
+            server.connect(r.scheme, r.host)
+            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 ServerConnectionPool:
     def __init__(self, config):
         self.config = config
         self.conn = None
 
-    def get_connection(self, scheme, host, port):
+    def get_connection(self, scheme, host, port, sni):
         sc = self.conn
         if self.conn and (host, port) != (sc.host, sc.port):
                 sc.terminate()
@@ -122,7 +122,7 @@ class ServerConnectionPool:
         if not self.conn:
             try:
                 self.conn = ServerConnection(self.config, host, port)
-                self.conn.connect(scheme)
+                self.conn.connect(scheme, sni)
             except tcp.NetLibError, v:
                 raise ProxyError(502, v)
         return self.conn
@@ -190,18 +190,18 @@ class ProxyHandler(tcp.BaseHandler):
                     # the case, we want to reconnect without sending an error
                     # to the client.
                     while 1:
+                        sc = self.server_conn_pool.get_connection(scheme, host, port, host)
+                        sc.send(request)
+                        sc.rfile.reset_timestamps()
                         try:
-                            sc = self.server_conn_pool.get_connection(scheme, host, port)
-                            sc.send(request)
-                            sc.rfile.reset_timestamps()
                             httpversion, code, msg, headers, content = http.read_response(
                                 sc.rfile,
                                 request.method,
                                 self.config.body_size_limit
                             )
                         except http.HttpErrorConnClosed, v:
+                            self.server_conn_pool.del_connection(scheme, host, port)
                             if sc.requestcount > 1:
-                                self.server_conn_pool.del_connection(scheme, host, port)
                                 continue
                             else:
                                 raise
@@ -324,25 +324,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 == "":
@@ -398,6 +379,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:
-- 
cgit v1.2.3


From 02578151410fff4b3c018303290e2f843e244a89 Mon Sep 17 00:00:00 2001
From: Aldo Cortesi <aldo@nullcube.com>
Date: Sun, 24 Feb 2013 22:24:21 +1300
Subject: Significantly simplify server connection handling, and test.

---
 libmproxy/proxy.py | 66 +++++++++++++++++++++++++++++-------------------------
 1 file changed, 35 insertions(+), 31 deletions(-)

(limited to 'libmproxy')

diff --git a/libmproxy/proxy.py b/libmproxy/proxy.py
index d92e2da9..7c229064 100644
--- a/libmproxy/proxy.py
+++ b/libmproxy/proxy.py
@@ -51,21 +51,22 @@ class ProxyConfig:
 
 
 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, sni):
+    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(cert=clientcert, sni=sni)
+                self.convert_to_ssl(cert=clientcert, sni=self.sni)
             except tcp.NetLibError, v:
                 raise ProxyError(400, str(v))
 
@@ -94,8 +95,8 @@ class RequestReplayThread(threading.Thread):
     def run(self):
         try:
             r = self.flow.request
-            server = ServerConnection(self.config, r.host, r.port)
-            server.connect(r.scheme, r.host)
+            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
@@ -109,37 +110,40 @@ class RequestReplayThread(threading.Thread):
             self.channel.ask(err)
 
 
-class ServerConnectionPool:
-    def __init__(self, config):
-        self.config = config
-        self.conn = None
-
-    def get_connection(self, scheme, host, port, sni):
-        sc = self.conn
-        if self.conn and (host, port) != (sc.host, sc.port):
-                sc.terminate()
-                self.conn = None
-        if not self.conn:
-            try:
-                self.conn = ServerConnection(self.config, host, port)
-                self.conn.connect(scheme, sni)
-            except tcp.NetLibError, v:
-                raise ProxyError(502, v)
-        return self.conn
-
-    def del_connection(self, scheme, host, port):
-        self.conn = None
-
-
 class ProxyHandler(tcp.BaseHandler):
     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_pool = ServerConnectionPool(config)
         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")
@@ -190,7 +194,7 @@ class ProxyHandler(tcp.BaseHandler):
                     # the case, we want to reconnect without sending an error
                     # to the client.
                     while 1:
-                        sc = self.server_conn_pool.get_connection(scheme, host, port, host)
+                        sc = self.get_server_connection(cc, scheme, host, port, host)
                         sc.send(request)
                         sc.rfile.reset_timestamps()
                         try:
@@ -200,7 +204,7 @@ class ProxyHandler(tcp.BaseHandler):
                                 self.config.body_size_limit
                             )
                         except http.HttpErrorConnClosed, v:
-                            self.server_conn_pool.del_connection(scheme, host, port)
+                            self.del_server_connection()
                             if sc.requestcount > 1:
                                 continue
                             else:
-- 
cgit v1.2.3