allow_hosts/ignore_hosts option now matches against the full `host:po…

…rt` string (#6594)

Co-authored-by: Maximilian Hils <git@maximilianhils.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

CHANGELOG.md

@@ -13,6 +13,8 @@
   ([#6609](https://github.com/mitmproxy/mitmproxy/pull/6609), @mhils)
 * Fix bug where insecure HTTP requests are saved incorrectly when exporting to HAR files.
   ([#6578](https://github.com/mitmproxy/mitmproxy/pull/6578), @DaniElectra)
+* `allow_hosts`/`ignore_hosts` option now matches against the full `host:port` string.
+  ([#6594](https://github.com/mitmproxy/mitmproxy/pull/6594), @LouisAsanaka)
 
 
 ## 06 January 2024: mitmproxy 10.2.1

mitmproxy/addons/next_layer.py

@@ -218,18 +218,24 @@ def _ignore_connection(
         ) and context.server.address == ("10.0.0.53", 53):
             return False
         hostnames: list[str] = []
-        if context.server.peername and (peername := context.server.peername[0]):
-            hostnames.append(peername)
-        if context.server.address and (server_address := context.server.address[0]):
-            hostnames.append(server_address)
-            # If we already have a destination address, we can also check for HTTP Host headers.
-            # But we do need the destination, otherwise we don't know where this connection is going to.
+        if context.server.peername:
+            host, port = context.server.peername
+            hostnames.append(f"{host}:{port}")
+        if context.server.address:
+            host, port = context.server.address
+            hostnames.append(f"{host}:{port}")
+
+            # We also want to check for TLS SNI and HTTP host headers, but in order to ignore connections based on that
+            # they must have a destination address. If they don't, we don't know how to establish an upstream connection
+            # if we ignore.
             if host_header := self._get_host_header(context, data_client, data_server):
+                if not re.search(r":\d+$", host_header):
+                    host_header = f"{host_header}:{port}"
                 hostnames.append(host_header)
-        if (
-            client_hello := self._get_client_hello(context, data_client)
-        ) and client_hello.sni:
-            hostnames.append(client_hello.sni)
+            if (
+                client_hello := self._get_client_hello(context, data_client)
+            ) and client_hello.sni:
+                hostnames.append(f"{client_hello.sni}:{port}")
 
         if not hostnames:
             return False
@@ -271,7 +277,9 @@ def _get_host_header(
             rb"[A-Z]{3,}.+HTTP/", data_client, re.IGNORECASE
         )
         if host_header_expected:
-            if m := re.search(rb"\r\n(?:Host: (.+))?\r\n", data_client, re.IGNORECASE):
+            if m := re.search(
+                rb"\r\n(?:Host:\s+(.+?)\s*)?\r\n", data_client, re.IGNORECASE
+            ):
                 if host := m.group(1):
                     return host.decode("utf-8", "surrogateescape")
                 else:

test/mitmproxy/addons/test_next_layer.py

@@ -117,7 +117,25 @@ def test_configure(self):
                 ["example.com"], [], "tcp", "example.com", b"", True, id="address"
             ),
             pytest.param(
-                ["1.2.3.4"], [], "tcp", "example.com", b"", True, id="ip address"
+                ["192.0.2.1"], [], "tcp", "example.com", b"", True, id="ip address"
+            ),
+            pytest.param(
+                ["example.com:443"],
+                [],
+                "tcp",
+                "example.com",
+                b"",
+                True,
+                id="port matches",
+            ),
+            pytest.param(
+                ["example.com:123"],
+                [],
+                "tcp",
+                "example.com",
+                b"",
+                False,
+                id="port does not match",
             ),
             pytest.param(
                 ["example.com"],
@@ -177,7 +195,7 @@ def test_configure(self):
                 ["example.com"],
                 [],
                 "tcp",
-                None,
+                "192.0.2.1",
                 client_hello_with_extensions,
                 True,
                 id="sni",
@@ -186,7 +204,7 @@ def test_configure(self):
                 ["example.com"],
                 [],
                 "tcp",
-                None,
+                "192.0.2.1",
                 client_hello_with_extensions[:-5],
                 NeedsMoreData,
                 id="incomplete client hello",
@@ -195,7 +213,7 @@ def test_configure(self):
                 ["example.com"],
                 [],
                 "tcp",
-                None,
+                "192.0.2.1",
                 client_hello_no_extensions[:9] + b"\x00" * 200,
                 False,
                 id="invalid client hello",
@@ -213,7 +231,7 @@ def test_configure(self):
                 ["example.com"],
                 [],
                 "udp",
-                None,
+                "192.0.2.1",
                 dtls_client_hello_with_extensions,
                 True,
                 id="dtls sni",
@@ -222,7 +240,7 @@ def test_configure(self):
                 ["example.com"],
                 [],
                 "udp",
-                None,
+                "192.0.2.1",
                 dtls_client_hello_with_extensions[:-5],
                 NeedsMoreData,
                 id="incomplete dtls client hello",
@@ -231,7 +249,7 @@ def test_configure(self):
                 ["example.com"],
                 [],
                 "udp",
-                None,
+                "192.0.2.1",
                 dtls_client_hello_with_extensions[:9] + b"\x00" * 200,
                 False,
                 id="invalid dtls client hello",
@@ -240,7 +258,7 @@ def test_configure(self):
                 ["example.com"],
                 [],
                 "udp",
-                None,
+                "192.0.2.1",
                 quic_client_hello,
                 True,
                 id="quic sni",
@@ -297,7 +315,7 @@ def test_ignore_connection(
             ctx.client.transport_protocol = transport_protocol
             if server_address:
                 ctx.server.address = (server_address, 443)
-                ctx.server.peername = ("1.2.3.4", 443)
+                ctx.server.peername = ("192.0.2.1", 443)
             if result is NeedsMoreData:
                 with pytest.raises(NeedsMoreData):
                     nl._ignore_connection(ctx, data_client, b"")