From 654e70b17a767c019d00cc6a9cb579536b011375 Mon Sep 17 00:00:00 2001 From: Gabriel Lluch Date: Sat, 1 Aug 2026 16:29:23 -0700 Subject: [PATCH 1/2] feat(scan): work out the network you are on instead of demanding a CIDR --- netdiff/cli.py | 36 +++++++++++----- netdiff/scan.py | 62 ++++++++++++++++++++++++++++ tests/test_scan_and_store.py | 79 ++++++++++++++++++++++++++++++++++++ 3 files changed, 167 insertions(+), 10 deletions(-) diff --git a/netdiff/cli.py b/netdiff/cli.py index 9629e41..13a45b9 100644 --- a/netdiff/cli.py +++ b/netdiff/cli.py @@ -15,7 +15,7 @@ from . import audit as audit_rules from . import mdns, oui, report, store, upnp from .diff import diff, summarise -from .scan import discover, grab_banners +from .scan import discover, grab_banners, local_subnet DEFAULT_PORTS = (22, 80, 443, 445, 554, 1883, 3389, 5000, 8080, 8443) @@ -47,7 +47,20 @@ def device_label(hostname: str, vendor: str, services: str) -> str: return f"{main} ({services})" if services and services != main else main +def resolve_subnet(args) -> None: + """Fill in the subnet from the network we are on, and say that we did. + + Announcing it is not decoration: a scan whose target was inferred has to + show its target, or the report is about a network the reader never chose. + """ + if args.subnet: + return + args.subnet = local_subnet() + print(f"no subnet given - scanning {args.subnet}, the network this machine is on") + + def cmd_scan(args) -> int: + resolve_subnet(args) ports = () if args.no_ports else tuple(args.ports) devices = discover( args.subnet, @@ -62,7 +75,7 @@ def cmd_scan(args) -> int: recent = store.recent_scan_ids(conn, limit=1) previous = store.load_scan(conn, recent[0]) if recent else [] scan_id = store.record_scan(conn, args.subnet, devices) - changes = diff(previous, devices) + changes = diff(previous, devices, compare_ports=not args.no_ports) if args.json: print( @@ -182,12 +195,7 @@ def placeholders(text): 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 - + resolve_subnet(args) devices = discover( args.subnet, ports=tuple(args.ports), @@ -314,7 +322,11 @@ def build_parser() -> argparse.ArgumentParser: sub = parser.add_subparsers(dest="command", required=True) scan = sub.add_parser("scan", help="scan a subnet, record it, report changes") - scan.add_argument("subnet", help="CIDR to scan, e.g. 192.168.1.0/24") + scan.add_argument( + "subnet", + nargs="?", + help="CIDR to scan, e.g. 192.168.1.0/24 - defaults to the network you are on", + ) scan.add_argument("--ports", type=int, nargs="*", default=list(DEFAULT_PORTS)) scan.add_argument("--no-ports", action="store_true", help="skip the port scan") scan.add_argument("--no-resolve", action="store_true", help="skip reverse DNS") @@ -338,7 +350,11 @@ def build_parser() -> argparse.ArgumentParser: "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( + "subnet", + nargs="?", + help="CIDR to audit, e.g. 192.168.1.0/24 - defaults to the network you are on", + ) 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( diff --git a/netdiff/scan.py b/netdiff/scan.py index b156039..8fd5c7d 100644 --- a/netdiff/scan.py +++ b/netdiff/scan.py @@ -75,6 +75,68 @@ def key(self) -> str: return self.mac +# The address and mask of the interface we would actually use, in the three +# shapes the usual commands print them: `ip -o -4 addr` gives a CIDR, macOS +# ifconfig gives a hex mask, Linux ifconfig gives a dotted one. +_INET_CIDR = re.compile(r"inet\s+(\d+\.\d+\.\d+\.\d+)/(\d+)") +_MASK_HEX = re.compile(r"netmask\s+0x([0-9a-fA-F]{8})") +_MASK_DOTTED = re.compile(r"netmask\s+(\d+\.\d+\.\d+\.\d+)") + + +def local_address() -> str: + """Our own address on the network we would actually route through. + + A UDP socket that is `connect`ed sends nothing - the kernel just picks the + route and binds a source address, which is exactly the question being asked. + The destination is TEST-NET-1, which is reserved and routed nowhere, so this + stays true even if it were ever to send. + """ + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.connect(("192.0.2.1", NUDGE_PORT)) + return sock.getsockname()[0] + + +def _prefix_length(text: str, ip: str): + """Mask width for `ip`, from whichever command's output this is.""" + for line in text.splitlines(): + if ip not in line: + continue + cidr = _INET_CIDR.search(line) + if cidr and cidr.group(1) == ip: + return int(cidr.group(2)) + hexmask = _MASK_HEX.search(line) + if hexmask: + return bin(int(hexmask.group(1), 16)).count("1") + dotted = _MASK_DOTTED.search(line) + if dotted: + return sum(bin(int(o)).count("1") for o in dotted.group(1).split(".")) + return None + + +def local_subnet(runner=subprocess.run) -> str: + """The CIDR of the network this machine is on. + + So that walking into somewhere new and running `netdiff scan` works without + first having to go and read your own IP settings. Guessing /24 would be right + most of the time, which is exactly the kind of nearly-true this tool refuses + elsewhere - if the mask cannot be read, say so and ask for one. + """ + ip = local_address() + for cmd in (["ip", "-o", "-4", "addr"], ["ifconfig"]): + try: + proc = runner(cmd, capture_output=True, text=True, timeout=10) + except (OSError, subprocess.SubprocessError): + continue + if proc.returncode == 0 and proc.stdout: + prefix = _prefix_length(proc.stdout, ip) + if prefix: + return str(ipaddress.ip_network(f"{ip}/{prefix}", strict=False)) + raise ValueError( + f"could not work out the subnet for {ip} - pass one explicitly, " + f"e.g. netdiff scan {ip.rsplit('.', 1)[0]}.0/24" + ) + + def normalise_mac(mac: str) -> str: """Canonical lower-case colon form, zero-padded. diff --git a/tests/test_scan_and_store.py b/tests/test_scan_and_store.py index 2b3b89a..d404ff0 100644 --- a/tests/test_scan_and_store.py +++ b/tests/test_scan_and_store.py @@ -107,6 +107,85 @@ def test_a_subnet_that_fits_is_not_refused(monkeypatch): assert scan.discover("192.168.1.0/24", settle=0) == [] +# --- working out where we are ----------------------------------------------- + +IP_ADDR_OUTPUT = ( + "1: lo inet 127.0.0.1/8 scope host lo\\ valid_lft forever\n" + "2: eth0 inet 192.168.1.190/24 brd 192.168.1.255 scope global eth0\n" +) + +MACOS_IFCONFIG = """\ +lo0: flags=8049 mtu 16384 +\tinet 127.0.0.1 netmask 0xff000000 +en0: flags=8863 mtu 1500 +\tinet 192.168.1.190 netmask 0xffffff00 broadcast 192.168.1.255 +""" + +LINUX_IFCONFIG = """\ +eth0: flags=4163 mtu 1500 + inet 192.168.1.190 netmask 255.255.255.0 broadcast 192.168.1.255 +""" + + +@pytest.fixture +def here(monkeypatch): + """Pretend this machine is 192.168.1.190, without touching a socket.""" + monkeypatch.setattr(scan, "local_address", lambda: "192.168.1.190") + + +def runner_for(output, ok=("ip", "ifconfig")): + def run(cmd, **kwargs): + if cmd[0] not in ok: + raise FileNotFoundError(cmd[0]) + + class Result: + returncode, stdout = 0, output + + return Result() + + return run + + +@pytest.mark.parametrize( + "output", [IP_ADDR_OUTPUT, MACOS_IFCONFIG, LINUX_IFCONFIG], ids=["ip", "bsd", "gnu"] +) +def test_the_subnet_is_read_from_whichever_command_exists(here, output): + """A CIDR, a hex mask and a dotted mask all mean the same /24.""" + assert scan.local_subnet(runner=runner_for(output)) == "192.168.1.0/24" + + +def test_a_mask_that_is_not_a_24_is_not_rounded_to_one(here): + output = IP_ADDR_OUTPUT.replace("192.168.1.190/24", "192.168.1.190/22") + assert scan.local_subnet(runner=runner_for(output)) == "192.168.0.0/22" + + +def test_an_unreadable_mask_asks_rather_than_assuming_24(here): + """Guessing would be right most of the time, which is the wrong kind of right.""" + with pytest.raises(ValueError, match="pass one explicitly"): + scan.local_subnet(runner=runner_for("no interfaces here")) + + +def test_the_loopback_line_is_not_mistaken_for_ours(here): + """127.0.0.1/8 appears first and would give the whole of 127/8.""" + assert scan.local_subnet(runner=runner_for(MACOS_IFCONFIG)) == "192.168.1.0/24" + + +def test_an_omitted_subnet_is_filled_in_and_announced(monkeypatch, capsys): + monkeypatch.setattr(cli, "local_subnet", lambda: "10.1.2.0/24") + args = type("Args", (), {"subnet": None})() + cli.resolve_subnet(args) + assert args.subnet == "10.1.2.0/24" + assert "10.1.2.0/24" in capsys.readouterr().out, "a scan must show its target" + + +def test_an_explicit_subnet_is_left_alone(monkeypatch, capsys): + monkeypatch.setattr(cli, "local_subnet", lambda: pytest.fail("should not detect")) + args = type("Args", (), {"subnet": "192.168.9.0/24"})() + cli.resolve_subnet(args) + assert args.subnet == "192.168.9.0/24" + assert capsys.readouterr().out == "" + + def test_a_bad_subnet_is_a_message_not_a_traceback(capsys): assert cli.main(["scan", "not-a-subnet"]) == 2 assert capsys.readouterr().err.strip(), "the reason has to reach the user" From 90bd420901585b0e45638ec502a84a4660858fab Mon Sep 17 00:00:00 2001 From: Gabriel Lluch Date: Sat, 1 Aug 2026 16:29:23 -0700 Subject: [PATCH 2/2] fix(diff): a scan that skipped ports must not report every port as closed --- netdiff/diff.py | 10 +++++++++- tests/test_diff.py | 21 +++++++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/netdiff/diff.py b/netdiff/diff.py index e0dc941..942f5d2 100644 --- a/netdiff/diff.py +++ b/netdiff/diff.py @@ -45,11 +45,17 @@ def __str__(self) -> str: return f"[{self.kind}] {label} {self.device.ip} {self.device.mac}{suffix}" -def diff(previous, current) -> list[Change]: +def diff(previous, current, compare_ports: bool = True) -> list[Change]: """Changes between two device lists, most significant first. Devices are matched by MAC, so a DHCP lease change is reported as `ip-changed` on one device rather than as a departure plus an arrival. + + `compare_ports=False` for a scan that did not look at ports. A device with + no ports scanned and a device with no ports open are both an empty tuple + here, so without this a `--no-ports` run reports every previously-open port + as `port-closed` - a change that never happened, which is the one failure + mode this module exists to avoid. """ before = {d.key(): d for d in previous} after = {d.key(): d for d in current} @@ -78,6 +84,8 @@ def diff(previous, current) -> list[Change]: f"{was.hostname or '-'} -> {now.hostname or '-'}", ) ) + if not compare_ports: + continue opened = sorted(set(now.ports) - set(was.ports)) closed = sorted(set(was.ports) - set(now.ports)) if opened: diff --git a/tests/test_diff.py b/tests/test_diff.py index 2865673..67128de 100644 --- a/tests/test_diff.py +++ b/tests/test_diff.py @@ -93,3 +93,24 @@ def test_summarise_counts_each_kind(): changes = diff([dev()], [dev(mac="11:22:33:44:55:66", ip="192.168.1.77")]) assert "1 appeared" in summarise(changes) assert "1 vanished" in summarise(changes) + + +def test_a_scan_that_skipped_ports_reports_no_port_changes(): + """`--no-ports` must not read as "every port closed". + + A device with nothing scanned and a device with nothing open are the same + empty tuple here, so this is the one case the data cannot distinguish on its + own - the caller has to say. Reporting it wrongly is a false alert, which is + worse than no alert. + """ + was = [Device(mac="aa:bb:cc:00:00:01", ip="192.168.1.10", ports=(22, 80))] + now = [Device(mac="aa:bb:cc:00:00:01", ip="192.168.1.10", ports=())] + + assert [c.kind for c in diff(was, now)] == ["port-closed"] + assert diff(was, now, compare_ports=False) == [] + + +def test_skipping_ports_still_reports_everything_else(): + was = [Device(mac="aa:bb:cc:00:00:01", ip="192.168.1.10", ports=(22,))] + now = [Device(mac="aa:bb:cc:00:00:01", ip="192.168.1.99", ports=())] + assert [c.kind for c in diff(was, now, compare_ports=False)] == ["ip-changed"]