feat(scan): work out the network you are on instead of demanding a CIDR

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(

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:

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.
 

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"]

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<UP,LOOPBACK,RUNNING,MULTICAST> mtu 16384
+\tinet 127.0.0.1 netmask 0xff000000
+en0: flags=8863<UP,BROADCAST,SMART,RUNNING,SIMPLEX,MULTICAST> mtu 1500
+\tinet 192.168.1.190 netmask 0xffffff00 broadcast 192.168.1.255
+"""
+
+LINUX_IFCONFIG = """\
+eth0: flags=4163<UP,BROADCAST,RUNNING,MULTICAST>  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"