From 99caa121aa39955bf7b70e3c23e045ef4341a3d6 Mon Sep 17 00:00:00 2001 From: Gabriel Lluch Date: Sat, 1 Aug 2026 14:08:54 -0700 Subject: [PATCH] feat: audit what the network exposes to the internet, not just what changed --- README.md | 83 +++++++- netdiff/audit.py | 391 +++++++++++++++++++++++++++++++++++ netdiff/cli.py | 156 +++++++++++++- netdiff/scan.py | 29 +++ netdiff/store.py | 56 +++++ netdiff/upnp.py | 274 ++++++++++++++++++++++++ tests/test_audit.py | 345 +++++++++++++++++++++++++++++++ tests/test_scan_and_store.py | 70 +++++++ tests/test_upnp.py | 321 ++++++++++++++++++++++++++++ 9 files changed, 1718 insertions(+), 7 deletions(-) create mode 100644 netdiff/audit.py create mode 100644 netdiff/upnp.py create mode 100644 tests/test_audit.py create mode 100644 tests/test_upnp.py diff --git a/README.md b/README.md index 355c521..fe8a40b 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,11 @@ # netdiff -Track what is on your network and tell you when it changes. **Pure standard library, no root.** +Track what is on your network, what it exposes to the internet, and when that changed. **Pure standard library, no root.** `nmap` and Fing answer "what is on my network *right now*". Neither remembers. netdiff records every scan, diffs it against the last one, and tells you what actually changed - a device that appeared at 3am, a printer that quietly opened port 8080, a laptop that moved to a new DHCP lease. +Then `netdiff audit` asks the question those tools do not: **which of these is reachable from outside your house, and why does that matter?** + ```console $ netdiff scan 192.168.1.0/24 scan 7: 12 device(s) on 192.168.1.0/24 @@ -16,6 +18,72 @@ changes since last scan: 1 appeared, 1 port-opened [port-opened] Raspberry Pi 192.168.1.23 b8:27:eb:aa:bb:cc (8080) ``` +## The audit: what your network exposes + +Most people assume NAT is a firewall - nothing outside can reach in unless they set it up. UPnP quietly breaks that. Any device on your LAN can ask the router to open a port from the internet straight to itself, with no prompt and no record anyone ever reads. The holes outlive the software that opened them. + +`netdiff audit` asks the router to list them, cross-references each forward against the devices actually present, and explains what it found. + +```console +$ netdiff audit 192.168.1.0/24 +audit 12: 192.168.1.0/24 - 1 critical, 2 high, 1 medium, 1 info + +CRITICAL + nas.local (192.168.1.23:8080) is reachable from the internet on port 8080 [NEW] + evidence *:8080/tcp -> 192.168.1.23:8080 (transmission) - and 192.168.1.23:8080 answered our scan + why Your router forwards this port from the public internet straight to this + device, so NAT is not protecting it. Anyone who scans your home IP address + reaches this service directly - and the whole internet is scanned + continuously. The service is exposed whether or not it was built to be. + fix If you did not set this up deliberately, remove the forward in your + router's admin page under Port Forwarding, then turn UPnP off so it + cannot come back. If you do need remote access, put it behind a VPN or + Tailscale instead of forwarding a port. + verify curl -s https://api.ipify.org # your public address + nc -vz THAT_ADDRESS 8080 # from a phone on cellular, NOT on your wifi + Testing from inside your own network proves nothing - most routers + answer their own public address differently from the outside world. +``` + +Every finding carries the observation that produced it, what an attacker gains, how to fix it, and **a command you run yourself to confirm it**. You should not have to take a scanner's word for anything. + +```bash +netdiff audit 192.168.1.0/24 # full report +netdiff audit 192.168.1.0/24 --json # same thing, machine-readable +netdiff audit 192.168.1.0/24 --no-upnp # skip the router check +netdiff audit --explain upnp-control-open # read a lesson without scanning +netdiff audit 192.168.1.0/24 --fail-on-finding # exit 1 on critical/high, for cron +``` + +Findings are recorded alongside scans, so a repeat audit marks what is `[NEW]` since the last one. A port forward that appeared on Tuesday is the thing worth knowing. + +### What it reports, and what it refuses to + +| Rule | Severity | Fires when | +| --- | --- | --- | +| `internet-exposed-service` | critical | A port forward points at a device, and that device answered on that port | +| `internet-exposed-port` | high | A port forward points inward, but we could not confirm what is behind it | +| `upnp-mapping-dangling` | high | A forward points at an address nothing currently holds - DHCP will hand it to something else | +| `plaintext-protocol` | high | Telnet, FTP, RTSP, MQTT or VNC - protocols with no encryption by design, confirmed by what the service said | +| `http-auth-plaintext` | high | A device sent an auth challenge over cleartext HTTP | +| `ssh-v1` | high | SSH protocol 1, deprecated since 2006 | +| `upnp-control-open` | medium | The router answered an unauthenticated control request - so would it for anything else on the LAN | +| `open-ports-noted` | info | Explicitly **not** a problem. See below. | + +**A port number is not evidence of a protocol either.** Port 23 being open does not prove telnet is behind it. So for services that greet you unprompted - FTP, Telnet, VNC - netdiff will not name the protocol until it has heard the greeting. RTSP and MQTT say nothing until spoken to, so there the evidence line states plainly that the identification is by port assignment, and the `verify` command lets you settle it. + +**An open port is not a vulnerability.** It is what a working device looks like. Tools that list every open port under a heading like "vulnerabilities found" are counting furniture and calling it a fire, and they train you to ignore the report. netdiff counts open ports and says out loud that they are not findings. A port becomes interesting when the protocol behind it is unencrypted, when it is reachable from outside the network, or when the software behind it is known-broken - and those are the rules above. + +There is no CVE matching here. Home-LAN banners rarely carry a precise enough version to map to a CVE honestly, and guessing produces a scary list that means nothing. + +### Read-only, and it means it + +The audit **never sends credentials, never writes to a scanned host, and never changes router configuration.** It reads banners that services volunteer to anyone who connects, and it calls exactly one UPnP method - `GetGenericPortMappingEntry`. There is deliberately no `AddPortMapping` code path in the source. + +This rules out checks that would otherwise be easy. Anonymous-FTP detection needs a login attempt, so it is not here. A failed SSH auth against every host on every scan - a common trick for grabbing SSH banners - lands you in the target's auth log and in fail2ban, so that is not here either. + +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. + ## 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. @@ -38,6 +106,7 @@ Python 3.9+. Nothing else - `pip show netdiff` lists no dependencies, and CI ass ```bash netdiff scan 192.168.1.0/24 # scan, record, report changes netdiff scan 192.168.1.0/24 --no-ports # discovery only, no TCP connections +netdiff audit 192.168.1.0/24 # what this network exposes, and why it matters netdiff inventory # every device ever seen, first and last sighting netdiff history # diff the two most recent scans @@ -74,7 +143,9 @@ History lives in `~/.netdiff/history.db` (override with `--db`). It is a plain S - **Randomised MACs.** Phones and laptops rotate their MAC per network by default. Those devices appear as new hardware whenever they rotate; netdiff labels them `randomised` rather than pretending to know better. If you want stable identity for a device, disable private addressing for your network on that device. - **Same broadcast segment only.** ARP does not cross routers, so this sees your subnet and nothing beyond it. That is a property of the approach, not a bug to fix. - **A device asleep during a scan is indistinguishable from one that left.** Expect `vanished`/`appeared` churn from phones. Longer intervals produce less noise. -- **`port-opened` means a TCP handshake completed**, nothing about what is listening. There is no service fingerprinting and no vulnerability scanning here on purpose - shallow banner-matching cannot compete with real scanners and only produces false confidence. +- **`port-opened` means a TCP handshake completed**, nothing about what is listening. `netdiff audit` adds banner reading, but there is still no OS fingerprinting and no CVE matching, on purpose - shallow version-guessing cannot compete with real scanners and only produces false confidence. +- **No UPnP gateway means no UPnP findings, not a clean bill of health.** A router with UPnP disabled is a good result, and it is also the common case now. Port forwards you configured by hand do not appear in the UPnP table at all - check your router's admin page for those. +- **The audit sees the LAN's exposure, not the internet's view of it.** It reads the forwarding table the router admits to. The only way to know what is actually reachable is to test from outside, which is why every exposure finding hands you that command. - **The bundled vendor table is small.** It covers common home-network hardware. For full coverage, download the IEEE registry and point `NETDIFF_OUI` at the CSV: ```bash curl -o oui.csv https://standards-oui.ieee.org/oui/oui.csv @@ -83,7 +154,7 @@ History lives in `~/.netdiff/history.db` (override with `--db`). It is a plain S ## Scope -Only scan networks you are responsible for. netdiff is deliberately read-only - it sends empty UDP datagrams and completes TCP handshakes, and never writes, authenticates, or probes a service. Even so, port scanning equipment you do not own is your problem, not the tool's. +Only scan networks you are responsible for. netdiff is deliberately read-only - it sends empty UDP datagrams, completes TCP handshakes, reads banners services volunteer, and asks the router to list its own port forwards. It never writes to a host, never authenticates, and never changes router configuration. Even so, scanning equipment you do not own is your problem, not the tool's. ## Development @@ -91,7 +162,11 @@ Only scan networks you are responsible for. netdiff is deliberately read-only - pip install pytest && pytest -q ``` -The tests never touch the network: ARP parsing runs against captured `arp -an` and `ip neigh` output, and the database is a temp file. `test_diff.py` covers the change detection, which is the part worth getting right. +The tests never touch the network. ARP parsing runs against captured `arp -an` and `ip neigh` output, UPnP parsing against captured router XML, and the one end-to-end test stands up a throwaway HTTP server on loopback. The database is a temp file. + +`test_diff.py` covers change detection. `test_audit.py` covers the rules, and roughly half of it asserts that something is *not* reported - an open port, an HTTP 200, a missing security header, a connection error. Those are the important half: the failure mode for a tool like this is not missing a finding, it is inventing one. + +Every audit rule is a pure function - evidence in, a `Finding` or `None` out - and nothing in `audit.py` opens a socket. That is what makes the security logic testable at all. `Finding.evidence` has no default value, so a finding cannot be constructed without the observation that proves it. ## License diff --git a/netdiff/audit.py b/netdiff/audit.py new file mode 100644 index 0000000..1d6c10f --- /dev/null +++ b/netdiff/audit.py @@ -0,0 +1,391 @@ +"""Turn observations into findings, and findings into lessons. + +Every rule here is a pure function: evidence in, a Finding or None out. Nothing +in this module opens a socket. That is deliberate and it is the whole design - +security logic fused to network I/O cannot be tested without a live host, so it +never gets tested, so it quietly rots into checks that raise on their first line +and report an open port as a break-in. + +Two consequences worth stating out loud: + +`Finding.evidence` has no default. You cannot construct a finding without the +observation that proves it, so "I saw a thing and it felt bad" is not +expressible. If a rule cannot quote its receipt, it is not a rule. + +Findings carry `why`, `fix` and `verify` because a scanner that only produces a +severity-coloured list teaches nothing. `verify` is a command you run yourself: +the point is that you should not have to take this tool's word for anything. + +The thesis, which the severity ladder encodes rather than asserts: +an open port is not a vulnerability, a plaintext protocol is a confidentiality +gap, and internet-reachable is attack surface. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from .scan import HTTP_PORTS + +SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "info": 3} + +# Protocols with no confidentiality by design. Naming one is a statement of +# fact about the protocol, not a guess about the device running it. +# +# The third element is whether the service greets you unprompted. Where it does, +# we refuse to name the protocol until we have heard it - a port number is a +# convention, not evidence, and "port 23 is open" does not prove telnet is +# behind it. Where it does not greet, the evidence says so in as many words. +PLAINTEXT_PROTOCOLS = { + 21: ("FTP", "usernames and passwords", True), + 23: ("Telnet", "usernames, passwords and every keystroke of the session", True), + 554: ("RTSP", "the camera stream and its credentials", False), + 1883: ("MQTT", "every message published, and any password used to connect", False), + 5900: ("VNC", "the screen contents, and often the password too", True), +} + + +@dataclass(frozen=True) +class Finding: + """One thing that is true about the network, with its receipt. + + `evidence` is positional and has no default on purpose - see the module + docstring. + """ + + rule: str + severity: str + device: str + title: str + evidence: str + why: str + fix: str + verify: str + + +RULES = { + "internet-exposed-service": { + "severity": "critical", + "title": "{label} is reachable from the internet on port {external_port}", + "why": ( + "Your router forwards this port from the public internet straight to this " + "device, so NAT is not protecting it. Anyone who scans your home IP address " + "reaches this service directly - and the whole internet is scanned " + "continuously. The service is exposed whether or not it was built to be." + ), + "fix": ( + "If you did not set this up deliberately, remove the forward in your " + "router's admin page under Port Forwarding, then turn UPnP off so it " + "cannot come back. If you do need remote access, put it behind a VPN or " + "Tailscale instead of forwarding a port." + ), + "verify": ( + "curl -s https://api.ipify.org # your public address\n" + "nc -vz THAT_ADDRESS {external_port} # from a phone on cellular, " + "NOT on your wifi\n" + "Testing from inside your own network proves nothing - most routers " + "answer their own public address differently from the outside world." + ), + }, + "internet-exposed-port": { + "severity": "high", + "title": "port {external_port} is forwarded from the internet to {internal}", + "why": ( + "Your router forwards this port in from the public internet. The target " + "did not answer our scan, so we cannot say what is behind it - but the " + "hole in the firewall is real and it is open right now." + ), + "fix": ( + "Find this entry in your router's Port Forwarding or UPnP table and " + "delete it if you do not recognise it." + ), + "verify": ( + "curl -s https://api.ipify.org # your public address\n" + "nc -vz THAT_ADDRESS {external_port} # from outside your network" + ), + }, + "upnp-mapping-dangling": { + "severity": "high", + "title": "port {external_port} is forwarded to {internal}, which is not on the network", + "why": ( + "The router is holding a door open to an address where nothing currently " + "lives. DHCP hands addresses out again, so the next device to receive this " + "one inherits an internet-facing port forward that nobody chose for it - a " + "guest's laptop, a new smart plug. This is how a forward set up for a games " + "console in 2021 ends up pointed at a camera." + ), + "fix": ( + "Delete the entry in your router's Port Forwarding table. If you need it " + "for a device that is usually online, give that device a DHCP reservation " + "so its address stops moving." + ), + "verify": ( + "ping -c1 {client}\narp -an | grep {client}\n" + "Nothing answers, and the ARP table has no entry for it." + ), + }, + "upnp-control-open": { + "severity": "medium", + "title": "the router lets any device on the LAN open its firewall", + "why": ( + "We asked the router for its port-forwarding table and it answered - no " + "password, no prompt. The same interface accepts AddPortMapping, so any " + "device here can open a path from the internet to itself and you will not " + "be told. That includes anything that gets compromised: a smart bulb, a TV, " + "a page open in a browser. This is the mechanism behind most of the other " + "findings in this report, and it is on by default on nearly every home " + "router." + ), + "fix": ( + "Turn UPnP off in your router's admin page unless something genuinely " + "needs it, and add the one or two forwards you actually want by hand. " + "Games consoles are the usual reason to leave it on; weigh that against " + "every other device on the network having the same privilege." + ), + "verify": ( + "The router answered this with no credentials. Paste it and watch:\n" + "curl -s -H 'SOAPAction: \"{service_type}#GetGenericPortMappingEntry\"' " + "-H 'Content-Type: text/xml' --data " + '\'' + '' + "0" + "' {control_url}\n" + "Nothing in that request identifies you as the owner of the network." + ), + }, + "plaintext-protocol": { + "severity": "high", + "title": "{protocol} on port {port} sends {exposed} in cleartext", + "why": ( + "{protocol} has no encryption. Anything it carries - {exposed} - travels " + "the network readable by anyone who can see the traffic: another device on " + "the same wifi, a guest, anything already compromised on the network. This " + "is a property of the protocol, not a misconfiguration of this device, " + "which is why the fix is to stop using it rather than to tune it." + ), + "fix": ( + "Prefer the encrypted equivalent - SSH instead of Telnet, SFTP or FTPS " + "instead of FTP, MQTT over TLS on 8883 instead of 1883. If the device is " + "too old to offer one, keep it on a separate VLAN or guest network and " + "never reuse its password anywhere else." + ), + "verify": ( + "nc {device} {port}\n" + "The service greets you before it authenticates you, and everything you " + "type after that crosses the network readable." + ), + }, + "http-auth-plaintext": { + "severity": "high", + "title": "port {port} asks for a password over unencrypted HTTP", + "why": ( + "This device answered with an authentication challenge on plain HTTP, so " + "the password protecting it is sent unencrypted. Anyone able to observe " + "the traffic captures it verbatim. A login prompt on HTTP protects against " + "a curious housemate and nothing else." + ), + "fix": ( + "Use the device's HTTPS interface if it has one. If it does not, treat " + "that password as public: never reuse it, and do not let this device's " + "admin page be reachable from outside the LAN." + ), + "verify": "curl -sI http://{device}:{port}/ | grep -i www-authenticate", + }, + "ssh-v1": { + "severity": "high", + "title": "SSH protocol 1 offered on port {port}", + "why": ( + "SSH version 1 has structural cryptographic flaws and has been deprecated " + "since 2006. Its integrity checking can be defeated, which means an " + "attacker positioned on the network can inject commands into a session " + "that looks encrypted and normal to both ends." + ), + "fix": ( + "Set 'Protocol 2' in the device's sshd_config, or update its firmware. A " + "device still offering SSH-1 in the present day is usually unmaintained, " + "which is its own finding." + ), + "verify": ( + "nc {device} {port}\nThe first line it prints is the version it speaks." + ), + }, + "open-ports-noted": { + "severity": "info", + "title": "{count} open port(s) observed, and not reported as problems", + "why": ( + "An open port means a service accepted a TCP handshake. That is not a " + "vulnerability - it is what a working device looks like. Tools that list " + "every open port under a heading like 'vulnerabilities found' are counting " + "furniture and calling it a fire. A port becomes interesting when the " + "protocol behind it is unencrypted, when it is reachable from outside the " + "network, or when the software behind it is known-broken. Those are the " + "things reported above." + ), + "fix": ( + "Nothing to fix. Worth skimming the device list anyway: a port you cannot " + "account for on a device you cannot identify is worth ten minutes." + ), + "verify": "netdiff inventory - every device and port this tool has ever seen here.", + }, +} + + +def finding(rule: str, device: str, evidence: str, **context) -> Finding: + """Build a Finding from the rule's teaching text. + + One source for both the report and `--explain`, so the lesson cannot drift + away from the thing that fired. + """ + spec = RULES[rule] + fields = dict(context, device=device) + return Finding( + rule=rule, + severity=spec["severity"], + device=device, + title=spec["title"].format(**fields), + evidence=evidence, + why=spec["why"].format(**fields), + fix=spec["fix"].format(**fields), + verify=spec["verify"].format(**fields), + ) + + +def rule_upnp_control_open(gateway): + """The router answered an unauthenticated control request.""" + if gateway is None: + return None + return finding( + "upnp-control-open", + "network", + f"{gateway.control_url} answered GetGenericPortMappingEntry with no " + f"credentials ({len(gateway.mappings)} mapping(s) returned)", + control_url=gateway.control_url, + service_type=gateway.service_type, + ) + + +def rule_mapping(mapping, devices): + """Classify one port forward against what is actually on the network.""" + by_ip = {d.ip: d for d in devices} + device = by_ip.get(mapping.internal_client) + internal = f"{mapping.internal_client}:{mapping.internal_port}" + + if device is None: + return finding( + "upnp-mapping-dangling", + mapping.internal_client, + str(mapping), + external_port=mapping.external_port, + internal=internal, + client=mapping.internal_client, + ) + + if mapping.internal_port in device.ports: + label = device.hostname or device.vendor or device.ip + return finding( + "internet-exposed-service", + device.ip, + f"{mapping} - and {device.ip}:{mapping.internal_port} answered our scan", + external_port=mapping.external_port, + label=f"{label} ({device.ip}:{mapping.internal_port})", + ) + + return finding( + "internet-exposed-port", + device.ip, + str(mapping), + external_port=mapping.external_port, + internal=internal, + ) + + +def rule_plaintext_protocol(ip: str, port: int, banner: str): + """A protocol with no encryption, confirmed by what the service said. + + Silence from a service that should have greeted us is not evidence of that + service, so we say nothing rather than name a protocol we did not hear. + """ + if port not in PLAINTEXT_PROTOCOLS: + return None + protocol, exposed, greets = PLAINTEXT_PROTOCOLS[port] + if greets and not banner.strip(): + return None + evidence = banner.strip() or ( + f"port {port} accepted a connection; {protocol} is the service assigned " + f"to that port and does not announce itself" + ) + return finding( + "plaintext-protocol", + ip, + evidence, + port=port, + protocol=protocol, + exposed=exposed, + ) + + +def rule_http_auth_plaintext(ip: str, port: int, banner: str): + """An HTTP auth challenge on a port that is not TLS.""" + if port not in HTTP_PORTS: + return None + for line in banner.splitlines(): + if line.lower().startswith("www-authenticate:"): + return finding("http-auth-plaintext", ip, line.strip(), port=port) + return None + + +def rule_ssh_v1(ip: str, port: int, banner: str): + """SSH-1 announces itself in the first line it sends.""" + if not banner.startswith("SSH-1."): + return None + return finding("ssh-v1", ip, banner.splitlines()[0].strip(), port=port) + + +BANNER_RULES = (rule_plaintext_protocol, rule_http_auth_plaintext, rule_ssh_v1) + + +def audit(devices, gateway=None, banners=None) -> list[Finding]: + """Apply every rule. `banners` maps (ip, port) -> whatever the service said.""" + banners = banners or {} + findings = [] + + control = rule_upnp_control_open(gateway) + if control: + findings.append(control) + if gateway is not None: + for mapping in gateway.mappings: + if mapping.enabled: + findings.append(rule_mapping(mapping, devices)) + + open_ports = 0 + for device in devices: + for port in device.ports: + open_ports += 1 + banner = banners.get((device.ip, port), "") + for rule in BANNER_RULES: + hit = rule(device.ip, port, banner) + if hit: + findings.append(hit) + + if open_ports: + findings.append( + finding( + "open-ports-noted", + "network", + f"{open_ports} open port(s) across {len(devices)} device(s)", + count=open_ports, + ) + ) + + return sorted(findings, key=lambda f: (SEVERITY_ORDER[f.severity], f.device)) + + +def summarise(findings) -> str: + """'1 critical, 2 high' - info is counted but never leads.""" + counts = {} + for item in findings: + counts[item.severity] = counts.get(item.severity, 0) + 1 + parts = [f"{counts[name]} {name}" for name in SEVERITY_ORDER if counts.get(name)] + return ", ".join(parts) if parts else "nothing to report" diff --git a/netdiff/cli.py b/netdiff/cli.py index aa08736..f20e53e 100644 --- a/netdiff/cli.py +++ b/netdiff/cli.py @@ -1,16 +1,19 @@ -"""Command line interface: scan, inventory, history, watch.""" +"""Command line interface: scan, audit, inventory, history.""" from __future__ import annotations import argparse import json +import re import sys +import textwrap import urllib.error import urllib.request -from . import oui, store +from . import audit as audit_rules +from . import oui, store, upnp from .diff import diff, summarise -from .scan import discover +from .scan import discover, grab_banner DEFAULT_PORTS = (22, 80, 443, 445, 554, 1883, 3389, 5000, 8080, 8443) @@ -99,6 +102,128 @@ def cmd_scan(args) -> int: return 0 +def print_field(label: str, text: str, wrap: bool = True) -> None: + """One labelled block, indented under its finding. + + `verify` is never wrapped: it is meant to be copied into a shell, and + reflowing a command silently corrupts it. + """ + pad = " " * 14 + if wrap: + print( + textwrap.fill( + text, 88, initial_indent=f" {label:<9} ", subsequent_indent=pad + ) + ) + return + lines = text.split("\n") + print(f" {label:<9} {lines[0]}") + for line in lines[1:]: + print(f"{pad}{line}") + + +def print_lesson(finding, is_new: bool = False) -> None: + """Render one finding as the lesson it is, not as a severity-coloured row.""" + print(f" {finding.title}{' [NEW]' if is_new else ''}") + print_field("evidence", finding.evidence, wrap=False) + print_field("why", finding.why) + print_field("fix", finding.fix) + print_field("verify", finding.verify, wrap=False) + print() + + +def cmd_audit(args) -> int: + if args.explain: + spec = audit_rules.RULES.get(args.explain) + if spec is None: + known = ", ".join(sorted(audit_rules.RULES)) + print( + f"unknown rule {args.explain!r}\nknown rules: {known}", file=sys.stderr + ) + return 2 + + # Nothing has fired, so there is no device to name. Show the + # placeholders as readable stand-ins rather than leaking "{port}". + def placeholders(text): + return re.sub(r"\{(\w+)\}", lambda m: m.group(1).upper(), text) + + print(f"{args.explain} [{spec['severity']}]") + print(f" {placeholders(spec['title'])}\n") + print_field("why", placeholders(spec["why"])) + print_field("fix", placeholders(spec["fix"])) + print_field("verify", placeholders(spec["verify"]), wrap=False) + return 0 + + if not args.subnet: + print( + "audit needs a subnet, e.g. netdiff audit 192.168.1.0/24", file=sys.stderr + ) + return 2 + + devices = discover( + args.subnet, + ports=tuple(args.ports), + lookup_vendor=oui.lookup, + resolve_names=not args.no_resolve, + ) + banners = { + (device.ip, port): grab_banner(device.ip, port) + for device in devices + for port in device.ports + } + gateway = None if args.no_upnp else upnp.probe_gateway(args.subnet) + findings = audit_rules.audit(devices, gateway, banners) + + conn = store.connect(args.db) + scan_id = store.record_scan(conn, args.subnet, devices) + # Compare against the last scan that actually audited, not merely the last + # scan - otherwise a plain `netdiff scan` in between makes everything look new. + previous_id = store.last_audited_scan_id(conn, scan_id) + seen = store.finding_keys(conn, previous_id) if previous_id else set() + store.record_findings(conn, scan_id, findings) + + # Nothing is "new" on the first audit; everything would be, which is noise. + annotated = [ + (f, bool(seen) and (f.rule, f.device, f.title) not in seen) for f in findings + ] + + if args.json: + print( + json.dumps( + { + "scan_id": scan_id, + "subnet": args.subnet, + "summary": audit_rules.summarise(findings), + "gateway": gateway.control_url if gateway else None, + "mappings": [str(m) for m in gateway.mappings] if gateway else [], + "findings": [dict(f.__dict__, is_new=new) for f, new in annotated], + }, + indent=2, + ) + ) + return 0 + + print(f"audit {scan_id}: {args.subnet} - {audit_rules.summarise(findings)}") + if gateway is None and not args.no_upnp: + print("no UPnP gateway answered - nothing here admits to forwarding ports") + print() + severity = "" + for finding, is_new in annotated: + if finding.severity != severity: + severity = finding.severity + print(severity.upper()) + print_lesson(finding, is_new) + + if not findings: + print("nothing to report - no devices answered, or none had open ports") + else: + print("Every finding above quotes the observation it rests on. Run the verify") + print("command yourself - do not take a scanner's word for anything.") + + serious = any(f.severity in ("critical", "high") for f in findings) + return 1 if args.fail_on_finding and serious else 0 + + def cmd_inventory(args) -> int: conn = store.connect(args.db) rows = store.inventory(conn) @@ -156,6 +281,31 @@ def build_parser() -> argparse.ArgumentParser: scan.add_argument("--json", action="store_true") scan.set_defaults(func=cmd_scan) + aud = sub.add_parser( + "audit", + help="what this network exposes, and why it matters", + description=( + "Read-only exposure audit. Never sends credentials, never writes to a " + "scanned host, and only ever reads the router's port-forwarding table." + ), + ) + aud.add_argument("subnet", nargs="?", help="CIDR to audit, e.g. 192.168.1.0/24") + aud.add_argument("--ports", type=int, nargs="*", default=list(DEFAULT_PORTS)) + aud.add_argument("--no-resolve", action="store_true", help="skip reverse DNS") + aud.add_argument( + "--no-upnp", action="store_true", help="skip the router port-forward check" + ) + aud.add_argument( + "--explain", metavar="RULE", help="print the lesson for a rule and exit" + ) + aud.add_argument( + "--fail-on-finding", + action="store_true", + help="exit 1 on any critical or high finding, for cron and CI", + ) + aud.add_argument("--json", action="store_true") + aud.set_defaults(func=cmd_audit) + inv = sub.add_parser("inventory", help="every device ever seen") inv.add_argument("--json", action="store_true") inv.set_defaults(func=cmd_inventory) diff --git a/netdiff/scan.py b/netdiff/scan.py index 04b3248..799a04a 100644 --- a/netdiff/scan.py +++ b/netdiff/scan.py @@ -37,6 +37,11 @@ INCOMPLETE = {"incomplete", "(incomplete)"} +# Ports we speak HTTP to rather than waiting for a greeting. Shared with the +# audit rules, which need the same list to know a challenge arrived over +# cleartext rather than TLS. +HTTP_PORTS = (80, 81, 591, 5000, 8000, 8008, 8080, 8081, 8888) + @dataclass(frozen=True) class Device: @@ -127,6 +132,30 @@ def scan_ports(ip: str, ports, timeout: float = 0.3) -> tuple[int, ...]: return tuple(sorted(open_ports)) +def grab_banner(ip: str, port: int, timeout: float = 2.0) -> str: + """Read what a service volunteers about itself. + + Most plaintext protocols greet you before they authenticate you, so + connecting and listening is the whole technique. HTTP is the exception - it + says nothing until asked - so we send HEAD, which requests headers and no + body and is the smallest thing we can ask for. + + This never sends credentials and never writes anything a server would + store. A banner is what the service tells everyone who connects. + """ + probe = b"HEAD / HTTP/1.0\r\n\r\n" if port in HTTP_PORTS else b"" + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.settimeout(timeout) + if sock.connect_ex((ip, port)) != 0: + return "" + if probe: + sock.sendall(probe) + return sock.recv(2048).decode("utf-8", "replace").strip() + except OSError: + return "" + + def resolve_hostname(ip: str) -> str: try: return socket.gethostbyaddr(ip)[0] diff --git a/netdiff/store.py b/netdiff/store.py index ff08358..4f1f665 100644 --- a/netdiff/store.py +++ b/netdiff/store.py @@ -29,6 +29,15 @@ PRIMARY KEY (scan_id, mac) ); CREATE INDEX IF NOT EXISTS observations_mac ON observations(mac); +CREATE TABLE IF NOT EXISTS findings ( + scan_id INTEGER NOT NULL REFERENCES scans(id) ON DELETE CASCADE, + rule TEXT NOT NULL, + severity TEXT NOT NULL, + device TEXT NOT NULL, + title TEXT NOT NULL, + evidence TEXT NOT NULL, + PRIMARY KEY (scan_id, rule, device, title) +); """ DEFAULT_PATH = Path.home() / ".netdiff" / "history.db" @@ -91,6 +100,53 @@ def load_scan(conn: sqlite3.Connection, scan_id: int) -> list[Device]: ] +def record_findings(conn: sqlite3.Connection, scan_id: int, findings) -> None: + """Persist the findings of one audit. + + Only what varies is stored. The teaching text lives in `audit.RULES` and is + looked up by rule id at render time, so improving an explanation improves it + everywhere including in reports already on disk. + """ + with conn: + conn.executemany( + "INSERT OR REPLACE INTO findings" + " (scan_id, rule, severity, device, title, evidence)" + " VALUES (?, ?, ?, ?, ?, ?)", + [ + (scan_id, f.rule, f.severity, f.device, f.title, f.evidence) + for f in findings + ], + ) + + +def finding_keys(conn: sqlite3.Connection, scan_id: int) -> set: + """Identity of each finding in a scan, for spotting what is new.""" + rows = conn.execute( + "SELECT rule, device, title FROM findings WHERE scan_id = ?", (scan_id,) + ).fetchall() + return {(r["rule"], r["device"], r["title"]) for r in rows} + + +def last_audited_scan_id(conn: sqlite3.Connection, before_scan_id: int): + """The most recent earlier scan that actually recorded findings. + + Plain `netdiff scan` writes no findings, so stepping back one scan would + often compare against an empty set and call everything new. + + ponytail: an audit that found literally nothing is indistinguishable from a + plain scan here, so a finding that clears and later returns is not marked + NEW. Reaching that needs a network with no open ports and no gateway, since + `open-ports-noted` fires otherwise. Add an `audits(scan_id)` table if it + ever matters. + """ + row = conn.execute( + "SELECT DISTINCT scan_id FROM findings WHERE scan_id < ?" + " ORDER BY scan_id DESC LIMIT 1", + (before_scan_id,), + ).fetchone() + return row["scan_id"] if row else None + + def recent_scan_ids(conn: sqlite3.Connection, limit: int = 2) -> list[int]: """Most recent scan ids, newest first.""" rows = conn.execute( diff --git a/netdiff/upnp.py b/netdiff/upnp.py new file mode 100644 index 0000000..47a6df1 --- /dev/null +++ b/netdiff/upnp.py @@ -0,0 +1,274 @@ +"""Ask the router what it forwards from the internet into your LAN. + +Most home users assume NAT is a firewall: nothing outside can reach inside +unless they set it up deliberately. UPnP quietly breaks that assumption. Any +device on the LAN - a console, a torrent client, a camera, a compromised smart +bulb - can ask the router to open a hole from the internet straight to itself, +with no prompt and no record anyone ever looks at. The holes outlive the +software that asked for them. + +So we ask the router to list them. That is a plain SOAP call the router already +answers for anything on the LAN, which is precisely the problem worth showing. + +Read-only by construction: this module calls `GetGenericPortMappingEntry` and +nothing else. There is deliberately no `AddPortMapping` code path here. + +Trust boundary: SSDP replies are unauthenticated UDP, so any host on the +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. +""" + +from __future__ import annotations + +import ipaddress +import socket +import urllib.error +import urllib.parse +import urllib.request +import xml.etree.ElementTree as ET +from dataclasses import dataclass + +SSDP_ADDRESS = ("239.255.255.250", 1900) +IGD_SEARCH_TARGET = "urn:schemas-upnp-org:device:InternetGatewayDevice:1" + +# Routers expose the port-mapping table under one of these two, depending on +# whether the WAN link is plain IP or PPPoE. Both answer the same SOAP action. +WAN_SERVICES = ( + "urn:schemas-upnp-org:service:WANIPConnection:1", + "urn:schemas-upnp-org:service:WANPPPConnection:1", +) + +# The router is not trusted input. Cap every read. +# ponytail: a flat byte cap, not a streaming parser - raise it if some router +# legitimately ships a description larger than this. +MAX_BODY_BYTES = 64 * 1024 + +# Enumeration ends when the router says "no such index", but a router that +# answers wrongly must not spin us forever. +MAX_MAPPINGS = 128 + +SSDP_SEARCH = ( + "M-SEARCH * HTTP/1.1\r\n" + "HOST: 239.255.255.250:1900\r\n" + 'MAN: "ssdp:discover"\r\n' + "MX: 2\r\n" + f"ST: {IGD_SEARCH_TARGET}\r\n" + "\r\n" +).encode() + +SOAP_BODY = ( + '' + '' + "" + '' + "{index}" + "" + "" +) + + +@dataclass(frozen=True) +class Mapping: + """One port forward, as the router reports it.""" + + external_port: int + protocol: str + internal_client: str + internal_port: int + description: str = "" + enabled: bool = True + remote_host: str = "" + + def __str__(self) -> str: + source = self.remote_host or "*" + label = f" ({self.description})" if self.description else "" + return ( + f"{source}:{self.external_port}/{self.protocol.lower()}" + f" -> {self.internal_client}:{self.internal_port}{label}" + ) + + +@dataclass(frozen=True) +class Gateway: + """The router's UPnP control endpoint and the forwards it admits to.""" + + control_url: str + service_type: str + mappings: tuple[Mapping, ...] = () + + +def _localname(tag: str) -> str: + """Strip the XML namespace. + + Vendors disagree about namespaces far more than they disagree about element + names, so matching on the local name is what actually survives contact with + real routers. + """ + return tag.rsplit("}", 1)[-1] + + +def _to_int(text: str) -> int: + try: + return int(text) + except (TypeError, ValueError): + return 0 + + +def parse_location(response: str) -> str: + """Pull the LOCATION header out of an SSDP reply.""" + for line in response.splitlines(): + name, _, value = line.partition(":") + if name.strip().lower() == "location": + return value.strip() + return "" + + +def is_safe_location(url: str, network) -> bool: + """Reject a LOCATION we should not fetch. + + Anyone on the LAN can forge an SSDP reply, so an unchecked LOCATION is an + attacker-chosen URL that we would fetch on their behalf. Requiring a + literal private address inside the audited subnet keeps that to a host that + is already on the network and already in the report. + """ + try: + host = urllib.parse.urlsplit(url).hostname + except ValueError: + return False + if not host: + return False + try: + address = ipaddress.ip_address(host) + except ValueError: + # A hostname could resolve anywhere, including off-network. Literal + # private IPs only. + return False + return address.is_private and address in network + + +def parse_service(description: str, base_url: str): + """Find the WAN connection service in a device description. Pure.""" + try: + root = ET.fromstring(description) + except ET.ParseError: + return None + for element in root.iter(): + if _localname(element.tag) != "service": + continue + fields = {_localname(c.tag): (c.text or "").strip() for c in element} + control = fields.get("controlURL", "") + if fields.get("serviceType") in WAN_SERVICES and control: + return urllib.parse.urljoin(base_url, control), fields["serviceType"] + return None + + +def parse_mapping(response: str): + """Turn one GetGenericPortMappingEntry response into a Mapping. Pure. + + Returns None for a SOAP fault, which is also how enumeration learns it has + reached the end of the table. + """ + try: + root = ET.fromstring(response) + except ET.ParseError: + return None + 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 + return Mapping( + external_port=_to_int(fields["NewExternalPort"]), + protocol=fields.get("NewProtocol", ""), + internal_client=fields["NewInternalClient"], + internal_port=_to_int(fields.get("NewInternalPort", "")), + description=fields.get("NewPortMappingDescription", ""), + enabled=fields.get("NewEnabled", "1") != "0", + remote_host=fields.get("NewRemoteHost", ""), + ) + + +def ssdp_search(timeout: float = 3.0) -> list[str]: + """Multicast an M-SEARCH and collect whatever replies before `timeout`.""" + replies = [] + try: + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2) + sock.settimeout(timeout) + sock.sendto(SSDP_SEARCH, SSDP_ADDRESS) + while True: + try: + data, _ = sock.recvfrom(MAX_BODY_BYTES) + except socket.timeout: + break + replies.append(data.decode("utf-8", "replace")) + except OSError: + # No multicast route, or no router that speaks UPnP. Not an error: + # a network with no IGD is a network with no UPnP forwards. + return [] + return replies + + +def _http_get(url: str, timeout: float) -> str: + try: + with urllib.request.urlopen(url, timeout=timeout) as response: + return response.read(MAX_BODY_BYTES).decode("utf-8", "replace") + except (urllib.error.URLError, OSError, ValueError): + return "" + + +def soap_post(control_url: str, service_type: str, index: int, timeout: float) -> str: + """POST one GetGenericPortMappingEntry. Returns the body, fault or not.""" + action = f"{service_type}#GetGenericPortMappingEntry" + body = SOAP_BODY.format(service=service_type, index=index).encode() + request = urllib.request.Request( + control_url, + data=body, + headers={ + "Content-Type": 'text/xml; charset="utf-8"', + "SOAPAction": f'"{action}"', + }, + ) + try: + with urllib.request.urlopen(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 + # the body is the answer rather than the failure. + return exc.read(MAX_BODY_BYTES).decode("utf-8", "replace") + except (urllib.error.URLError, OSError, ValueError): + return "" + + +def port_mappings(control_url, service_type, timeout=5.0, poster=soap_post): + """Walk the mapping table until the router runs out of entries.""" + mappings = [] + for index in range(MAX_MAPPINGS): + mapping = parse_mapping(poster(control_url, service_type, index, timeout)) + if mapping is None: + break + mappings.append(mapping) + return mappings + + +def probe_gateway(subnet: str, timeout: float = 3.0, search=ssdp_search): + """Find the IGD on `subnet` and read its port-mapping table. + + Returns None when there is no reachable UPnP gateway, which is a good + result rather than a failure - a network with no IGD has no UPnP forwards. + """ + network = ipaddress.ip_network(subnet, strict=False) + for reply in search(timeout): + location = parse_location(reply) + if not location or not is_safe_location(location, network): + continue + service = parse_service(_http_get(location, timeout), location) + if service is None: + continue + control_url, service_type = service + return Gateway( + control_url=control_url, + service_type=service_type, + mappings=tuple(port_mappings(control_url, service_type, timeout)), + ) + return None diff --git a/tests/test_audit.py b/tests/test_audit.py new file mode 100644 index 0000000..9c4973a --- /dev/null +++ b/tests/test_audit.py @@ -0,0 +1,345 @@ +"""The rules, exercised without a network. + +That this file can exist at all is the point of the design. The tool this +replaced fused its security logic to its sockets, so no check could run without +a live host, so none ever did - and nine of them raised on their first line for +months without anyone noticing. + +Half of these tests assert that something is *not* reported. Those are the +important half. +""" + +from netdiff.audit import RULES, audit, summarise +from netdiff.scan import Device +from netdiff.upnp import Gateway, Mapping + + +def dev(ip="192.168.1.10", mac="aa:bb:cc:00:00:01", ports=(), hostname="", vendor=""): + return Device(mac=mac, ip=ip, hostname=hostname, vendor=vendor, ports=tuple(ports)) + + +def gw(mappings=(), control_url="http://192.168.1.1:5000/ctl"): + return Gateway( + control_url=control_url, + service_type="urn:schemas-upnp-org:service:WANIPConnection:1", + mappings=tuple(mappings), + ) + + +def mapping(external=8080, client="192.168.1.10", internal=8080, **kw): + return Mapping( + external_port=external, + protocol="TCP", + internal_client=client, + internal_port=internal, + **kw, + ) + + +def rules_fired(findings): + return [f.rule for f in findings] + + +# --- the anti-theater tests: presence is not vulnerability ------------------- + + +def test_an_open_port_alone_is_not_a_finding(): + """The failure mode this whole rewrite exists to prevent.""" + findings = audit([dev(ports=[22, 443, 8443])]) + assert [f for f in findings if f.severity != "info"] == [] + + +def test_an_http_200_is_not_a_finding(): + banners = {("192.168.1.10", 80): "HTTP/1.0 200 OK\r\nServer: lighttpd\r\n"} + findings = audit([dev(ports=[80])], banners=banners) + assert [f for f in findings if f.severity != "info"] == [] + + +def test_a_missing_security_header_is_not_a_finding(): + """No CSP on a printer's status page is not a security problem.""" + banners = {("192.168.1.10", 80): "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n"} + assert rules_fired(audit([dev(ports=[80])], banners=banners)) == [ + "open-ports-noted" + ] + + +def test_a_connection_error_is_not_a_finding(): + findings = audit([dev(ports=[8080])], banners={("192.168.1.10", 8080): ""}) + assert [f for f in findings if f.severity != "info"] == [] + + +def test_no_devices_produces_no_findings_at_all(): + assert audit([]) == [] + assert summarise([]) == "nothing to report" + + +def test_open_ports_are_counted_as_info_and_labelled_not_a_problem(): + findings = audit([dev(ports=[22, 443])]) + noted = [f for f in findings if f.rule == "open-ports-noted"] + assert len(noted) == 1 + assert noted[0].severity == "info" + assert "2 open port" in noted[0].evidence + assert "not a vulnerability" in noted[0].why + + +# --- plaintext protocols ---------------------------------------------------- + + +def test_telnet_is_reported_as_cleartext(): + banners = {("192.168.1.10", 23): "\xff\xfbUbuntu 14.04 login:"} + findings = audit([dev(ports=[23])], banners=banners) + hit = next(f for f in findings if f.rule == "plaintext-protocol") + assert hit.severity == "high" + assert "Telnet" in hit.title + assert "login:" in hit.evidence + + +def test_ftp_and_telnet_on_one_device_are_two_findings(): + """Same rule, same device - they must not collapse into one row.""" + banners = { + ("192.168.1.10", 21): "220 ProFTPD Server ready.", + ("192.168.1.10", 23): "\xff\xfblogin:", + } + device = dev(ports=[21, 23]) + findings = [ + f for f in audit([device], banners=banners) if f.rule == "plaintext-protocol" + ] + assert len(findings) == 2 + assert {f.title for f in findings} != {findings[0].title} + + +def test_mqtt_on_1883_is_flagged_but_8883_is_not(): + """8883 is the TLS port. Same protocol, no confidentiality gap.""" + assert "plaintext-protocol" in rules_fired(audit([dev(ports=[1883])])) + assert "plaintext-protocol" not in rules_fired(audit([dev(ports=[8883])])) + + +def test_a_silent_port_23_is_not_called_telnet(): + """A port number is a convention, not evidence. Nothing greeted us, so we + cannot say what is behind it - naming it Telnet would be a guess dressed as + a finding, which is the exact habit this tool exists to avoid.""" + findings = audit([dev(ports=[23])], banners={("192.168.1.10", 23): ""}) + assert "plaintext-protocol" not in rules_fired(findings) + assert [f for f in findings if f.severity != "info"] == [] + + +def test_a_greeting_on_port_23_is_enough_to_name_it(): + banners = {("192.168.1.10", 23): "\xff\xfb\x01Ubuntu 14.04 login:"} + hit = next( + f + for f in audit([dev(ports=[23])], banners=banners) + if f.rule == "plaintext-protocol" + ) + assert "login:" in hit.evidence + + +def test_a_protocol_that_never_greets_says_so_in_its_evidence(): + """RTSP and MQTT stay silent until spoken to, so the port assignment is all + we have - the evidence must admit that rather than imply a banner.""" + hit = next(f for f in audit([dev(ports=[1883])]) if f.rule == "plaintext-protocol") + assert "does not announce itself" in hit.evidence + + +# --- HTTP auth over cleartext ----------------------------------------------- + + +def test_auth_challenge_over_http_is_reported(): + banners = { + ("192.168.1.10", 8080): ( + 'HTTP/1.1 401 Unauthorized\r\nWWW-Authenticate: Basic realm="NAS Admin"\r\n' + ) + } + hit = next( + f + for f in audit([dev(ports=[8080])], banners=banners) + if f.rule == "http-auth-plaintext" + ) + assert hit.severity == "high" + assert "NAS Admin" in hit.evidence + + +def test_auth_challenge_header_match_is_case_insensitive(): + banners = {("192.168.1.10", 80): "HTTP/1.1 401\r\nwww-authenticate: Digest\r\n"} + assert "http-auth-plaintext" in rules_fired( + audit([dev(ports=[80])], banners=banners) + ) + + +def test_the_word_authenticate_in_a_page_body_is_not_a_challenge(): + """Substring matching on response bodies is how the old tool invented findings.""" + banners = { + ("192.168.1.10", 80): "HTTP/1.1 200 OK\r\n\r\n

Please authenticate here

" + } + assert "http-auth-plaintext" not in rules_fired( + audit([dev(ports=[80])], banners=banners) + ) + + +# --- SSH -------------------------------------------------------------------- + + +def test_ssh_1_is_reported(): + banners = {("192.168.1.10", 22): "SSH-1.5-OpenSSH_2.9"} + hit = next( + f for f in audit([dev(ports=[22])], banners=banners) if f.rule == "ssh-v1" + ) + assert hit.evidence == "SSH-1.5-OpenSSH_2.9" + + +def test_ssh_2_is_not_reported(): + banners = {("192.168.1.10", 22): "SSH-2.0-OpenSSH_9.6"} + assert "ssh-v1" not in rules_fired(audit([dev(ports=[22])], banners=banners)) + + +def test_ssh_1_99_still_counts_because_it_accepts_protocol_1(): + """1.99 advertises "I speak 2, and I will also fall back to 1 for you".""" + banners = {("192.168.1.10", 22): "SSH-1.99-OpenSSH_3.9"} + assert "ssh-v1" in rules_fired(audit([dev(ports=[22])], banners=banners)) + + +# --- UPnP: the edge --------------------------------------------------------- + + +def test_forward_to_a_live_service_is_critical_and_names_the_device(): + device = dev(ip="192.168.1.23", hostname="nas.local", ports=[8080]) + findings = audit([device], gw([mapping(client="192.168.1.23", internal=8080)])) + hit = next(f for f in findings if f.rule == "internet-exposed-service") + assert hit.severity == "critical" + assert "nas.local" in hit.title + assert "192.168.1.23:8080" in hit.evidence + + +def test_forward_to_a_device_with_that_port_shut_is_lower_severity(): + device = dev(ip="192.168.1.23", ports=[22]) + findings = audit([device], gw([mapping(client="192.168.1.23", internal=8080)])) + hit = next(f for f in findings if f.rule == "internet-exposed-port") + assert hit.severity == "high" + + +def test_forward_to_an_absent_host_is_the_dangling_case(): + """The subtle one: DHCP will hand that address to something else.""" + findings = audit([dev(ip="192.168.1.10")], gw([mapping(client="192.168.1.47")])) + hit = next(f for f in findings if f.rule == "upnp-mapping-dangling") + assert hit.severity == "high" + assert hit.device == "192.168.1.47" + assert "DHCP" in hit.why + + +def test_a_disabled_mapping_is_not_reported(): + findings = audit([dev()], gw([mapping(enabled=False)])) + assert not [f for f in findings if "exposed" in f.rule] + + +def test_a_reachable_gateway_is_itself_a_medium_finding(): + findings = audit([], gw()) + hit = next(f for f in findings if f.rule == "upnp-control-open") + assert hit.severity == "medium" + assert "http://192.168.1.1:5000/ctl" in hit.evidence + + +def test_no_gateway_means_no_upnp_findings(): + findings = audit([dev(ports=[22])], gateway=None) + assert not [f for f in findings if "upnp" in f.rule or "exposed" in f.rule] + + +def test_mapping_evidence_carries_the_actual_forward(): + findings = audit([dev()], gw([mapping(external=32400, description="Plex")])) + hit = next(f for f in findings if f.rule.startswith("internet-exposed")) + assert "32400" in hit.evidence + assert "Plex" in hit.evidence + + +# --- structure -------------------------------------------------------------- + + +def test_every_finding_carries_evidence_and_all_three_lessons(): + device = dev(ip="192.168.1.23", ports=[23, 8080]) + banners = {("192.168.1.23", 8080): "HTTP/1.1 401\r\nWWW-Authenticate: Basic\r\n"} + findings = audit( + [device], gw([mapping(client="192.168.1.23", internal=8080)]), banners + ) + assert findings + for f in findings: + assert f.evidence.strip(), f"{f.rule} has no receipt" + assert f.why.strip() and f.fix.strip() and f.verify.strip() + + +def test_findings_sort_worst_first(): + device = dev(ip="192.168.1.23", ports=[23, 8080]) + findings = audit([device], gw([mapping(client="192.168.1.23", internal=8080)])) + severities = [f.severity for f in findings] + assert severities[0] == "critical" + assert severities[-1] == "info" + + +def test_every_rule_id_used_by_a_rule_exists_in_the_teaching_table(): + device = dev(ip="192.168.1.23", ports=[21, 23, 22, 1883, 8080]) + banners = { + ("192.168.1.23", 22): "SSH-1.5-x", + ("192.168.1.23", 8080): "HTTP/1.1 401\r\nWWW-Authenticate: Basic\r\n", + } + fired = audit( + [device, dev(ip="192.168.1.30")], + gw( + [ + mapping(client="192.168.1.23", internal=8080), # live -> critical + mapping(client="192.168.1.30", internal=9999), # shut -> high + mapping(client="10.0.0.9"), # absent -> dangling + ] + ), + banners, + ) + assert {f.rule for f in fired} == set(RULES) + + +def test_dangling_verify_pings_a_bare_address_not_an_ip_colon_port(): + """`ping 192.168.1.47:80` is not a command. The rule has two placeholders.""" + hit = next( + f + for f in audit([dev(ip="192.168.1.10")], gw([mapping(client="192.168.1.47")])) + if f.rule == "upnp-mapping-dangling" + ) + assert "ping -c1 192.168.1.47\n" in hit.verify + assert "192.168.1.47:" not in hit.verify + + +def test_no_finding_leaves_an_unsubstituted_placeholder_anywhere(): + device = dev(ip="192.168.1.23", ports=[21, 22, 23, 1883, 8080]) + banners = { + ("192.168.1.23", 22): "SSH-1.5-x", + ("192.168.1.23", 8080): "HTTP/1.1 401\r\nWWW-Authenticate: Basic\r\n", + } + fired = audit( + [device, dev(ip="192.168.1.30")], + gw( + [ + mapping(client="192.168.1.23", internal=8080), + mapping(client="192.168.1.30", internal=9999), + mapping(client="10.0.0.9"), + ] + ), + banners, + ) + for f in fired: + for field in (f.title, f.why, f.fix, f.verify): + assert "{" not in field and "}" not in field, f.rule + + +def test_verify_text_interpolates_the_real_device_and_port(): + banners = {("192.168.1.77", 23): "\xff\xfblogin:"} + hit = next( + f + for f in audit([dev(ip="192.168.1.77", ports=[23])], banners=banners) + if f.rule == "plaintext-protocol" + ) + assert "192.168.1.77 23" in hit.verify + assert "{" not in hit.verify # no unsubstituted placeholders + + +def test_summarise_counts_by_severity_worst_first(): + device = dev(ip="192.168.1.23", ports=[23, 8080]) + text = summarise( + audit([device], gw([mapping(client="192.168.1.23", internal=8080)])) + ) + assert text.startswith("1 critical") + assert "info" in text diff --git a/tests/test_scan_and_store.py b/tests/test_scan_and_store.py index 1764ad8..f98a6e1 100644 --- a/tests/test_scan_and_store.py +++ b/tests/test_scan_and_store.py @@ -4,6 +4,7 @@ import pytest from netdiff import store +from netdiff.audit import Finding from netdiff.oui import is_randomised, lookup from netdiff.scan import Device, normalise_mac, parse_arp_output, read_arp_table @@ -136,3 +137,72 @@ def test_first_seen_survives_the_device_changing_ip(conn): conn, "192.168.1.0/24", [Device(mac="aa:bb:cc:00:00:01", ip="192.168.1.55")] ) assert store.first_seen(conn, "aa:bb:cc:00:00:01") is not None + + +def find(rule="plaintext-protocol", device="192.168.1.10", title="Telnet on port 23"): + return Finding( + rule=rule, + severity="high", + device=device, + title=title, + evidence="banner", + why="w", + fix="f", + verify="v", + ) + + +def test_findings_roundtrip_by_identity(conn): + scan_id = store.record_scan(conn, "192.168.1.0/24", []) + store.record_findings(conn, scan_id, [find()]) + assert store.finding_keys(conn, scan_id) == { + ("plaintext-protocol", "192.168.1.10", "Telnet on port 23") + } + + +def test_two_findings_of_one_rule_on_one_device_do_not_collide(conn): + """FTP and Telnet on the same host are two problems, not one.""" + scan_id = store.record_scan(conn, "192.168.1.0/24", []) + store.record_findings( + conn, scan_id, [find(title="Telnet on port 23"), find(title="FTP on port 21")] + ) + assert len(store.finding_keys(conn, scan_id)) == 2 + + +def test_a_repeat_audit_finds_nothing_new(conn): + first = store.record_scan(conn, "192.168.1.0/24", []) + store.record_findings(conn, first, [find()]) + second = store.record_scan(conn, "192.168.1.0/24", []) + store.record_findings(conn, second, [find()]) + + seen = store.finding_keys(conn, store.last_audited_scan_id(conn, second)) + assert all((f.rule, f.device, f.title) in seen for f in [find()]), ( + "an unchanged network must not report NEW" + ) + + +def test_a_finding_that_appears_later_is_new(conn): + first = store.record_scan(conn, "192.168.1.0/24", []) + store.record_findings(conn, first, [find()]) + second = store.record_scan(conn, "192.168.1.0/24", []) + fresh = find(rule="internet-exposed-service", title="port 8080 exposed") + store.record_findings(conn, second, [find(), fresh]) + + seen = store.finding_keys(conn, store.last_audited_scan_id(conn, second)) + assert (fresh.rule, fresh.device, fresh.title) not in seen + assert (find().rule, find().device, find().title) in seen + + +def test_a_plain_scan_between_audits_does_not_reset_the_baseline(conn): + """`netdiff scan` writes no findings; stepping back one scan would see none.""" + audited = store.record_scan(conn, "192.168.1.0/24", []) + store.record_findings(conn, audited, [find()]) + store.record_scan(conn, "192.168.1.0/24", []) # a plain scan, no findings + latest = store.record_scan(conn, "192.168.1.0/24", []) + + assert store.last_audited_scan_id(conn, latest) == audited + + +def test_the_first_audit_ever_has_no_baseline(conn): + scan_id = store.record_scan(conn, "192.168.1.0/24", []) + assert store.last_audited_scan_id(conn, scan_id) is None diff --git a/tests/test_upnp.py b/tests/test_upnp.py new file mode 100644 index 0000000..0fa84b3 --- /dev/null +++ b/tests/test_upnp.py @@ -0,0 +1,321 @@ +"""UPnP parsing and the trust boundary in front of it. + +The XML here is shaped like what real routers actually return (MiniUPnPd, which +is what most consumer firmware ships), captured as constants so none of this +needs a network - the same approach as the ARP fixtures in +test_scan_and_store.py. +""" + +import http.server +import ipaddress +import threading + +from netdiff import upnp + +SSDP_REPLY = """HTTP/1.1 200 OK +CACHE-CONTROL: max-age=120 +ST: urn:schemas-upnp-org:device:InternetGatewayDevice:1 +USN: uuid:8bd6a1f1-7cd6-4a41-9f1a-000000000001::urn:schemas-upnp-org:device:InternetGatewayDevice:1 +EXT: +SERVER: Linux/3.4.11 UPnP/1.0 MiniUPnPd/1.9 +LOCATION: http://192.168.1.1:5000/rootDesc.xml + +""" + +DESCRIPTION = """ + + 10 + + urn:schemas-upnp-org:device:InternetGatewayDevice:1 + Home Router + + + urn:schemas-upnp-org:device:WANDevice:1 + + + urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1 + /ctl/CommonIfCfg + + + + + urn:schemas-upnp-org:device:WANConnectionDevice:1 + + + urn:schemas-upnp-org:service:WANIPConnection:1 + urn:upnp-org:serviceId:WANIPConn1 + /ctl/IPConn + /evt/IPConn + /WANIPCn.xml + + + + + + + + +""" + +MAPPING_RESPONSE = """ + + + + +32400 +TCP +32400 +192.168.1.23 +1 +Plex Media Server +0 + + +""" + +# How the router says "that index does not exist" - i.e. the end of the table. +FAULT_713 = """ + + +s:ClientUPnPError + +713 +SpecifiedArrayIndexInvalid + + +""" + +NETWORK = ipaddress.ip_network("192.168.1.0/24") + + +def test_location_is_read_from_the_ssdp_reply(): + assert upnp.parse_location(SSDP_REPLY) == "http://192.168.1.1:5000/rootDesc.xml" + + +def test_location_header_name_is_case_insensitive(): + assert ( + upnp.parse_location("Location: http://10.0.0.1/d.xml") + == "http://10.0.0.1/d.xml" + ) + + +def test_a_reply_without_a_location_yields_empty_string(): + assert upnp.parse_location("HTTP/1.1 200 OK\nST: something\n") == "" + + +# --- the trust boundary ----------------------------------------------------- +# SSDP is unauthenticated UDP: anything on the segment can forge a reply and +# choose the URL we fetch next. These tests are the whole reason that check +# exists. + + +def test_a_private_address_inside_the_subnet_is_followed(): + assert upnp.is_safe_location("http://192.168.1.1:5000/rootDesc.xml", NETWORK) + + +def test_a_public_address_is_refused(): + assert not upnp.is_safe_location("http://93.184.216.34/rootDesc.xml", NETWORK) + + +def test_a_private_address_outside_the_audited_subnet_is_refused(): + assert not upnp.is_safe_location("http://10.9.9.9/rootDesc.xml", NETWORK) + + +def test_a_hostname_is_refused_because_it_could_resolve_anywhere(): + assert not upnp.is_safe_location("http://evil.example.com/d.xml", NETWORK) + assert not upnp.is_safe_location("http://localhost/d.xml", NETWORK) + + +def test_garbage_locations_are_refused_rather_than_raising(): + for url in ("", "not a url", "http://", "file:///etc/passwd"): + assert not upnp.is_safe_location(url, NETWORK) + + +# --- description parsing ---------------------------------------------------- + + +def test_the_wan_connection_service_is_found_however_deeply_nested(): + control_url, service_type = upnp.parse_service( + DESCRIPTION, "http://192.168.1.1:5000/rootDesc.xml" + ) + assert control_url == "http://192.168.1.1:5000/ctl/IPConn" + assert service_type == "urn:schemas-upnp-org:service:WANIPConnection:1" + + +def test_the_wrong_service_is_not_mistaken_for_the_right_one(): + """WANCommonInterfaceConfig appears first and does not serve mappings.""" + control_url, _ = upnp.parse_service(DESCRIPTION, "http://192.168.1.1:5000/d.xml") + assert "CommonIfCfg" not in control_url + + +def test_an_absolute_control_url_is_left_alone(): + xml = DESCRIPTION.replace( + "/ctl/IPConn", + "http://192.168.1.1:49152/ctl", + ) + control_url, _ = upnp.parse_service(xml, "http://192.168.1.1:5000/d.xml") + assert control_url == "http://192.168.1.1:49152/ctl" + + +def test_a_description_with_no_wan_service_yields_none(): + assert upnp.parse_service("", "http://192.168.1.1/") is None + + +def test_malformed_xml_yields_none_rather_than_raising(): + assert upnp.parse_service("", "http://192.168.1.1/") is None + assert upnp.parse_mapping("}{ not xml at all") is None + + +def test_an_entity_declaration_does_not_get_expanded(): + """The router is not trusted input; a billion-laughs body must not run.""" + bomb = ( + ']>' + "" + "&a;&a;" + "" + ) + assert upnp.parse_service(bomb, "http://192.168.1.1/") is None + + +# --- mapping parsing -------------------------------------------------------- + + +def test_a_port_mapping_is_parsed_in_full(): + m = upnp.parse_mapping(MAPPING_RESPONSE) + assert m.external_port == 32400 + assert m.internal_client == "192.168.1.23" + assert m.internal_port == 32400 + assert m.protocol == "TCP" + assert m.description == "Plex Media Server" + assert m.enabled is True + + +def test_a_disabled_mapping_reports_itself_as_disabled(): + m = upnp.parse_mapping(MAPPING_RESPONSE.replace("1", "0")) + assert m.enabled is False + + +def test_a_fault_is_not_a_mapping(): + assert upnp.parse_mapping(FAULT_713) is None + + +def test_a_mapping_renders_as_the_forward_it_describes(): + assert str(upnp.parse_mapping(MAPPING_RESPONSE)) == ( + "*:32400/tcp -> 192.168.1.23:32400 (Plex Media Server)" + ) + + +def test_non_numeric_ports_degrade_to_zero_rather_than_raising(): + m = upnp.parse_mapping( + MAPPING_RESPONSE.replace("32400", "abc") + ) + assert m.internal_port == 0 + + +# --- enumeration ------------------------------------------------------------ + + +def test_enumeration_stops_at_the_first_fault(): + responses = [MAPPING_RESPONSE, MAPPING_RESPONSE, FAULT_713, MAPPING_RESPONSE] + + def poster(control_url, service_type, index, timeout): + return responses[index] + + assert len(upnp.port_mappings("http://x/ctl", "svc", poster=poster)) == 2 + + +def test_a_router_that_never_faults_is_still_bounded(): + """A misbehaving router must not spin us forever.""" + + def poster(control_url, service_type, index, timeout): + return MAPPING_RESPONSE + + mappings = upnp.port_mappings("http://x/ctl", "svc", poster=poster) + assert len(mappings) == upnp.MAX_MAPPINGS + + +def test_an_unreachable_router_yields_no_mappings(): + def poster(control_url, service_type, index, timeout): + return "" + + assert upnp.port_mappings("http://x/ctl", "svc", poster=poster) == [] + + +def test_the_soap_action_asks_only_for_the_mapping_table(): + """Read-only contract: there is no AddPortMapping path in this module.""" + source = (upnp.SOAP_BODY + upnp.soap_post.__doc__).lower() + assert "getgenericportmappingentry" in source + assert "addportmapping" not in source + + +# --- end to end against a real socket --------------------------------------- +# The parsers above are tested in isolation; this covers the glue between them, +# which is where the wiring bugs actually live. A throwaway HTTP server on +# loopback stands in for the router - no LAN, no multicast, still real sockets. + + +ENTRY = """ + + +{ext} +TCP{internal} +{client}1 +{label} +""" + + +class _FakeIGD(http.server.BaseHTTPRequestHandler): + entries = [ + {"ext": 32400, "internal": 32400, "client": "127.0.0.1", "label": "Plex"}, + {"ext": 8080, "internal": 80, "client": "127.0.0.9", "label": "webcam"}, + ] + + def log_message(self, *args): + pass + + def _send(self, body, code=200): + raw = body.encode() + self.send_response(code) + self.send_header("Content-Type", "text/xml") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def do_GET(self): + self._send(DESCRIPTION) + + def do_POST(self): + body = self.rfile.read(int(self.headers["Content-Length"])).decode() + index = int(body.split("")[1].split("<")[0]) + if index >= len(self.entries): + self._send(FAULT_713, 500) # how a real router ends the table + return + 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) + 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]) + finally: + server.shutdown() + + assert gateway.control_url == f"http://127.0.0.1:{port}/ctl/IPConn" + assert [m.external_port for m in gateway.mappings] == [32400, 8080] + assert gateway.mappings[1].internal_client == "127.0.0.9" + + +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" + assert upnp.probe_gateway("192.168.1.0/24", search=lambda timeout: [reply]) is None + + +def test_no_ssdp_replies_means_no_gateway_not_an_error(): + assert upnp.probe_gateway("192.168.1.0/24", search=lambda timeout: []) is None