detect recursive self-connects and stop them

mitmproxy/addons/proxyserver.py

@@ -5,7 +5,7 @@ from typing import Dict, Optional, Tuple
 from mitmproxy import command, controller, ctx, flow, http, log, master, options, platform, tcp, websocket
 from mitmproxy.flow import Error, Flow
 from mitmproxy.proxy import commands, events
-from mitmproxy.proxy import server
+from mitmproxy.proxy import server, server_hooks
 from mitmproxy.proxy.layers.tcp import TcpMessageInjected
 from mitmproxy.proxy.layers.websocket import WebSocketMessageInjected
 from mitmproxy.utils import asyncio_utils, human, strutils
@@ -180,3 +180,8 @@ class Proxyserver:
             self.inject_event(event)
         except ValueError as e:
             ctx.log.warn(str(e))
+
+    def server_connected(self, ctx: server_hooks.ServerConnectionHookData):
+        # check if the outbound part of this connection appeared as a new client.
+        if ctx.server.sockname in self._connections:
+            ctx.server.error = "Stopped mitmproxy from recursively connecting to itself."

mitmproxy/proxy/server.py

@@ -177,32 +177,31 @@ class ConnectionHandler(metaclass=abc.ABCMeta):
                 else:
                     addr = human.format_address(command.connection.address)
                 self.log(f"server connect {addr}")
-                connected_hook = asyncio_utils.create_task(
-                    self.handle_hook(server_hooks.ServerConnectedHook(hook_data)),
-                    name=f"handle_hook(server_connected) {addr}",
-                    client=self.client.peername,
-                )
-                if not connected_hook:
-                    return  # this should not be needed, see asyncio_utils.create_task
+                await self.handle_hook(server_hooks.ServerConnectedHook(hook_data))
 
-                self.server_event(events.OpenConnectionCompleted(command, None))
+                if errmsg := command.connection.error:
+                    self.log(f"server connection to {addr} killed: {errmsg}")
+                    self.server_event(events.OpenConnectionCompleted(command, f"Connection killed: {errmsg}"))
+                    del self.transports[command.connection]
+                    writer.close()
+                else:
+                    self.server_event(events.OpenConnectionCompleted(command, None))
 
-                # during connection opening, this function is the designated handler that can be cancelled.
-                # once we have a connection, we do want the teardown here to happen in any case, so we
-                # reassign the handler to .handle_connection and then clean up here once that is done.
-                new_handler = asyncio_utils.create_task(
-                    self.handle_connection(command.connection),
-                    name=f"server connection handler for {addr}",
-                    client=self.client.peername,
-                )
-                if not new_handler:
-                    return  # this should not be needed, see asyncio_utils.create_task
-                self.transports[command.connection].handler = new_handler
-                await asyncio.wait([new_handler])
+                    # during connection opening, this function is the designated handler that can be cancelled.
+                    # once we have a connection, we do want the teardown here to happen in any case, so we
+                    # reassign the handler to .handle_connection and then clean up here once that is done.
+                    new_handler = asyncio_utils.create_task(
+                        self.handle_connection(command.connection),
+                        name=f"server connection handler for {addr}",
+                        client=self.client.peername,
+                    )
+                    if not new_handler:
+                        return  # this should not be needed, see asyncio_utils.create_task
+                    self.transports[command.connection].handler = new_handler
+                    await asyncio.wait([new_handler])
 
                 self.log(f"server disconnect {addr}")
                 command.connection.timestamp_end = time.time()
-                await connected_hook  # wait here for this so that closed always comes after connected.
                 await self.handle_hook(server_hooks.ServerDisconnectedHook(hook_data))
 
     async def handle_connection(self, connection: Connection) -> None:

test/mitmproxy/addons/test_proxyserver.py

@@ -160,3 +160,28 @@ async def test_warn_no_nextlayer():
         await tctx.master.await_log("Proxy server listening at", level="info")
         assert tctx.master.has_log("Warning: Running proxyserver without nextlayer addon!", level="warn")
         await ps.shutdown_server()
+
+
+@pytest.mark.asyncio
+async def test_self_connect():
+    ps = Proxyserver()
+    with taddons.context(ps) as tctx:
+        state = HelperAddon()
+        state.layers = [
+            lambda ctx: layers.modes.ReverseProxy(ctx),
+            lambda ctx: layers.HttpLayer(ctx, HTTPMode.transparent),
+            lambda ctx: layers.modes.ReverseProxy(ctx),
+        ]
+        tctx.master.addons.add(state)
+
+        tctx.configure(ps, listen_host="127.0.0.1", listen_port=0)
+        ps.running()
+        await tctx.master.await_log("Proxy server listening", level="info")
+        assert ps.server
+        proxy_addr = ps.server.sockets[0].getsockname()[:2]
+
+        tctx.options.mode = f"reverse:{':'.join(str(x) for x in proxy_addr)}"
+        reader, writer = await asyncio.open_connection(*proxy_addr)
+        writer.write(b"GET / HTTP/1.1\r\n\r\n")
+        assert b"502 Bad Gateway" in await reader.readuntil(b"\r\n\r\n")
+        assert b"Stopped mitmproxy from recursively connecting" in await reader.readuntil(b"</html>")