diff --git a/README.md b/README.md index 7a2dfd0..6f96baa 100644 --- a/README.md +++ b/README.md @@ -9,10 +9,10 @@ Then `netdiff audit` asks the question those tools do not: **which of these is r ```console $ netdiff scan 192.168.1.0/24 scan 7: 12 device(s) on 192.168.1.0/24 - 192.168.1.1 00:1d:c9:0a:1b:2c router.local ports 53,80,443 - 192.168.1.23 b8:27:eb:aa:bb:cc Raspberry Pi (SSH, Web interface) ports 22 + 192.168.1.1 00:1d:c9:0a:1b:2c router.local Linux, macOS or BSD? (TTL 64) ports 53,80,443 + 192.168.1.23 b8:27:eb:aa:bb:cc Raspberry Pi (SSH, Web interface) Linux, macOS or BSD? (TTL 64) ports 22 192.168.1.64 54:60:09:11:22:33 Google (Chromecast) - 192.168.1.71 d8:3a:dd:aa:bb:cc Mac15,7, AirPlay + 192.168.1.71 d8:3a:dd:aa:bb:cc Mac15,7, AirPlay Windows? (TTL 128) ... changes since last scan: 1 appeared, 1 port-opened @@ -28,13 +28,17 @@ Most people assume NAT is a firewall - nothing outside can reach in unless they ```console $ netdiff audit 192.168.1.0/24 -audit 12: 192.168.1.0/24 - 1 critical, 2 high, 1 medium, 1 info +audit 12: 192.168.1.0/24 - 1 critical, 3 high, 2 medium, 3 info critical nas.local (192.168.1.23:8080) is reachable from the internet on port 8080 [NEW] - high Telnet on port 23 sends usernames, passwords and every keystroke of the - session in cleartext - high port 8080 asks for a password over unencrypted HTTP + high 192.168.1.23 Telnet on port 23 sends usernames, passwords and every + keystroke of the session in cleartext + high 192.168.1.23 port 8080 asks for a password over unencrypted HTTP + high 192.168.1.40 SMBv1 file sharing is enabled on port 445 medium the router lets any device on the LAN open its firewall + medium 192.168.1.23 SSH on port 22 still offers 3 deprecated algorithm(s) + info 192.168.1.1 port 80 identifies itself as lighttpd 1.4.59 + info 192.168.1.1 port 443 uses a certificate that vouches for itself info 7 open port(s) observed, and not reported as problems -v adds the evidence each line rests on, why it matters, how to fix it, @@ -45,7 +49,7 @@ A report nobody finishes reading teaches nothing, so depth is something you ask ```console $ netdiff audit 192.168.1.0/24 -v -audit 12: 192.168.1.0/24 - 1 critical, 2 high, 1 medium, 1 info +audit 12: 192.168.1.0/24 - 1 critical, 3 high, 2 medium, 3 info CRITICAL nas.local (192.168.1.23:8080) is reachable from the internet on port 8080 [NEW] @@ -87,7 +91,13 @@ Findings are recorded alongside scans, so a repeat audit marks what is `[NEW]` s | `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 | +| `smb-v1` | high | A file server accepted the 1996 SMB dialect - the one EternalBlue and WannaCry travelled over | | `upnp-control-open` | medium | The router answered an unauthenticated control request - so would it for anything else on the LAN | +| `ssh-weak-algorithms` | medium | An SSH server still offers RC4, CBC ciphers, MD5 integrity or SHA-1 key exchange | +| `tls-cert-expired` | medium | A certificate is past its own notAfter date | +| `service-version` | info | The software a service names when asked, quoted rather than guessed | +| `tls-cert-untrusted` | info | Explicitly **not** a problem. A self-signed certificate - the normal case on a LAN | +| `dns-recursion-open` | info | Explicitly **not** a problem, usually. A device resolves internet names for anyone who asks | | `open-ports-noted` | info | Explicitly **not** a problem. See below. | **And a vendor is not a device type.** "Espressif" covers a smart plug, a doorbell and someone's weekend project equally, so a MAC lookup alone leaves the most useful column nearly empty. Rather than guess a device type from its open ports - which is how the tool netdiff replaced arrived at "Managed Web Server" for a printer - netdiff asks the network the question every phone on it asks continuously, and reads the answer. Chromecasts, printers, Sonos, HomeKit gear and Apple devices all announce their services over multicast DNS, unprompted, to anyone on the segment. `Mac15,7` in the output is the device's own word for itself, not an inference. A device that announces nothing is left blank, because not knowing is the normal case. @@ -96,12 +106,16 @@ Findings are recorded alongside scans, so a repeat audit marks what is `[NEW]` s **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. +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. netdiff prints the version a service names and stops there - the question worth asking is "is this still supported", which no vulnerability database answers. + +**Four questions a port number cannot answer.** Port 445 being open says nothing about which SMB dialect a server accepts, so netdiff offers it the 1996 dialect alone and reports what comes back. Port 443 being open says nothing about the certificate behind it, so the certificate is read and judged on its own dates and names. An SSH server announces the algorithms it will negotiate with before authentication happens at all, so those are read from the handshake rather than by a failed login. And a device either resolves a name it has no authority over or it does not. Each is one exchange, sends nothing a server stores, and yields evidence rather than an inference. ### 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. +The depth probes hold the same line. The TLS handshake is completed and abandoned. The SMB negotiate offers a dialect and reads the answer - no tree connect, no share enumeration, no null session. The SSH probe swaps version strings and reads the algorithm list the server sends unprompted, which is the alternative to the usual trick of grabbing it with a failed login. The DNS query asks for `example.com`, a name IANA reserves for exactly this. None of them authenticate, and none of them leave a record beyond a connection. + 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. @@ -129,9 +143,11 @@ 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 --ports top100 # nmap's 100 most common ports, not the default 10 netdiff scan 192.168.1.0/24 --no-ports # discovery only, no TCP connections netdiff scan 192.168.1.0/24 --no-mdns # skip asking devices what they are netdiff audit 192.168.1.0/24 # what this network exposes, and why it matters +netdiff audit --ports top100 3000 5432 # a set, plus whatever else you run netdiff inventory # every device ever seen, first and last sighting netdiff history # diff the two most recent scans @@ -168,7 +184,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. `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. +- **`port-opened` means a TCP handshake completed**, nothing about what is listening. `netdiff audit` adds banner reading, certificate reading and protocol handshakes on top of that, but there is still no CVE matching, on purpose. +- **The OS hint is a hint.** It is the TTL of one ping reply, which narrows a device to a family and nothing more - and only when the TTL is one of the three common starting values. Anything else is printed as the bare number, because 32 is not "nearly 64". Real fingerprinting needs crafted packets and a raw socket, which needs root. A device that drops ICMP has no hint at all, which is common. +- **`--ports top100` is 100 ports, not 65535.** It is nmap's frequency ranking, which is a good answer to "what is worth a timeout" and a bad answer to "what is definitely closed". A service on an unusual port is invisible to both the default set and this one. - **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: @@ -179,7 +197,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, completes TCP handshakes, reads banners services volunteer, asks the standard DNS-SD question over multicast and reads the replies, 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. +Only scan networks you are responsible for. netdiff is deliberately read-only. In full, what it sends: empty UDP datagrams to provoke ARP, one ICMP echo per device, TCP handshakes, `HEAD /` to HTTP ports, a TLS ClientHello, an SMB negotiate offering one dialect, an SSH version string, a DNS query for `example.com`, the standard DNS-SD question over multicast, and one UPnP request asking 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 @@ -189,12 +207,16 @@ pip install pytest && pytest -q The tests never touch the network. ARP parsing runs against captured `arp -an` and `ip neigh` output, UPnP parsing against captured router XML, mDNS parsing against hand-built packets, and the one end-to-end test stands up a throwaway HTTP server on loopback. The database is a temp file. +`test_probe.py` covers the protocol parsers. Its two SMB fixtures are real replies captured from Samba - one configured to allow SMBv1 and one to refuse it - because the refusal is the shape that matters: a parser that only handles the happy path reports every modern server as running SMBv1. The certificate is a throwaway generated by `openssl`, and the parser has to arrive at the same dates `openssl x509 -noout -dates` prints for it. + `test_mdns.py` builds its packets with its own helpers rather than with the encoder in `mdns.py`, because a decoder tested only against its own encoder agrees with itself however wrong both are. Half of that file is malformed input - a name pointing at itself, a record claiming to be longer than the packet carrying it - because anything able to send a UDP datagram can send those. `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. +The split holds across three modules: `scan.py` finds what is here, `probe.py` asks protocols the questions a port number cannot answer, and `audit.py` decides what any of it means. `probe.py` keeps its parsers separate from its sockets for the same reason - the fiddly half is bytes in, a value out. + ## License MIT diff --git a/netdiff/audit.py b/netdiff/audit.py index 671fcff..872b7b6 100644 --- a/netdiff/audit.py +++ b/netdiff/audit.py @@ -23,7 +23,9 @@ from __future__ import annotations +import re from dataclasses import dataclass +from datetime import date from .scan import HTTP_PORTS @@ -44,6 +46,36 @@ 5900: ("VNC", "the screen contents, and often the password too", True), } +# SSH algorithms that are deprecated rather than merely old, with the reason - +# the reason is the finding, since "weak" on its own is a word not an argument. +# A server offering these still works; it also still accepts them, which is what +# a client downgrade attack needs. +WEAK_SSH_ALGORITHMS = { + "diffie-hellman-group1-sha1": "1024-bit key exchange, breakable by Logjam", + "diffie-hellman-group14-sha1": "SHA-1 key exchange", + "diffie-hellman-group-exchange-sha1": "SHA-1 key exchange", + "ssh-rsa": "SHA-1 signatures, disabled by OpenSSH 8.8 in 2021", + "ssh-dss": "1024-bit DSA, removed from OpenSSH in 2015", + "arcfour": "RC4, biased keystream", + "arcfour128": "RC4, biased keystream", + "arcfour256": "RC4, biased keystream", + "3des-cbc": "56-bit effective key", + "aes128-cbc": "CBC in SSH is encrypt-and-MAC, which leaks plaintext", + "aes192-cbc": "CBC in SSH is encrypt-and-MAC, which leaks plaintext", + "aes256-cbc": "CBC in SSH is encrypt-and-MAC, which leaks plaintext", + "blowfish-cbc": "64-bit block cipher in CBC mode", + "cast128-cbc": "64-bit block cipher in CBC mode", + "hmac-md5": "MD5 integrity", + "hmac-md5-96": "MD5 integrity, truncated", + "hmac-sha1-96": "truncated SHA-1 integrity", + "none": "no encryption at all, if a client asks for it", +} + +# `lighttpd/1.4.35`, `OpenSSH_7.4`, `Boa/0.94`. Two characters of name, then a +# separator, then a dotted number - narrow on purpose, because a looser pattern +# reads version numbers out of dates, ETags and session cookies. +_VERSION = re.compile(r"\b([A-Za-z][A-Za-z0-9.+-]{1,30})[/_](\d+(?:\.\d+)+[\w.-]*)") + @dataclass(frozen=True) class Finding: @@ -63,6 +95,19 @@ class Finding: verify: str +def headline(finding) -> str: + """The title, prefixed with the device when the title does not name it. + + Both renderers show findings as a flat list with no device column, which was + readable while a quiet network produced one or two of them. It stopped being + readable the moment a report could say "port 80 identifies itself as + lighttpd 1.4.59" three times about three different devices. + """ + if finding.device in ("", "network") or finding.device in finding.title: + return finding.title + return f"{finding.device} {finding.title}" + + RULES = { "internet-exposed-service": { "severity": "critical", @@ -211,6 +256,143 @@ class Finding: "nc {device} {port}\nThe first line it prints is the version it speaks." ), }, + "smb-v1": { + "severity": "high", + "title": "SMBv1 file sharing is enabled on port {port}", + "why": ( + "This device agreed to speak the 1996 version of the Windows file " + "sharing protocol. SMBv1 cannot verify who it is talking to, so a " + "device on the same network can sit in the middle of a file transfer " + "unnoticed. It is also the protocol EternalBlue and WannaCry travelled " + "over, and worms built on it are still circulating years later because " + "the devices still answering are the ones nobody updates. Microsoft " + "stopped installing it by default in 2017." + ), + "fix": ( + "On Windows: Control Panel, Turn Windows features on or off, untick " + "'SMB 1.0/CIFS File Sharing Support'. On a NAS, look for a minimum SMB " + "protocol setting and set it to SMB2 or SMB3 - every client made in the " + "last decade speaks those. If a device only offers SMBv1 and cannot be " + "updated, it should not be on the same network as anything you care about." + ), + "verify": ( + "nmap --script smb-protocols -p445 {device}\n" + "Anything listing a dialect of 'NT LM 0.12' is SMBv1." + ), + }, + "ssh-weak-algorithms": { + "severity": "medium", + "title": "SSH on port {port} still offers {count} deprecated algorithm(s)", + "why": ( + "SSH negotiates its cryptography with each client, and this server is " + "still willing to accept: {detail}. A current client will pick something " + "stronger, so this is not a break - it is a downgrade waiting for an old " + "client or someone able to interfere with the negotiation. It is also a " + "reliable sign of firmware that has not been updated in years, which is " + "usually the more useful thing to learn from it." + ), + "fix": ( + "Update the device's firmware first - modern OpenSSH drops these by " + "default and this list disappears on its own. Where that is not " + "possible, set KexAlgorithms, Ciphers and MACs explicitly in sshd_config " + "to the current defaults." + ), + "verify": ( + "ssh -vv {device} 2>&1 | grep 'peer server KEXINIT'\n" + "Or: nmap --script ssh2-enum-algos -p {port} {device}" + ), + }, + "tls-cert-expired": { + "severity": "medium", + "title": "the certificate on port {port} expired on {not_after}", + "why": ( + "Encryption still works, so traffic is not readable - but every browser " + "and app reaching this device now shows a warning, and the only way to " + "keep using it is to click through that warning. Once clicking through " + "is routine, a genuine interception looks exactly like the thing you " + "already dismiss every day. That is the real cost, and it is a habit " + "rather than a vulnerability." + ), + "fix": ( + "Reissue the certificate in the device's admin page. Many devices " + "regenerate a self-signed one on request, or after a factory reset. If " + "the device cannot, its firmware is old enough that the certificate is " + "the smaller problem." + ), + "verify": ( + "openssl s_client -connect {device}:{port} /dev/null" + " | openssl x509 -noout -dates" + ), + }, + "tls-cert-untrusted": { + "severity": "info", + "title": "port {port} uses a certificate that vouches for itself", + "why": ( + "Explicitly not a problem on its own, and the normal case for home " + "equipment - a router or NAS has no way to obtain a certificate a " + "browser would trust for a private address. The traffic is genuinely " + "encrypted. What is missing is identity: nothing here proves the device " + "answering is the one you meant, so the padlock says the connection is " + "private without saying who it is private with. Worth knowing because it " + "is why this device shows a warning, and why that warning is not one to " + "chase." + ), + "fix": ( + "Nothing, for most devices. If you want the warning gone properly, " + "run a local certificate authority or use a tool that issues real " + "certificates for internal names. Do not disable TLS to avoid the " + "warning - encrypted-but-unverified beats plaintext every time." + ), + "verify": ( + "openssl s_client -connect {device}:{port} /dev/null" + " | openssl x509 -noout -subject -issuer\n" + "The same name on both lines means it signed its own certificate." + ), + }, + "dns-recursion-open": { + "severity": "info", + "title": "{device} resolves internet names for anything that asks it", + "why": ( + "This device answered a query for a name it does not own, which makes it " + "a recursive resolver. On a home network that is usually just the router " + "doing its job, and it is not a problem while it is only reachable " + "from inside. It becomes one if the same device is reachable from the " + "internet: an open resolver is the classic amplifier for denial of " + "service attacks, because a small forged query produces a large reply " + "sent to whoever the attacker named. Check whether port 53 appears in " + "any exposure finding above." + ), + "fix": ( + "Nothing, if this is your router and nothing forwards port 53 inward. " + "If it is not your router, ask why that device is running a resolver at " + "all. Where the option exists, restrict recursion to the local subnet." + ), + "verify": ( + "dig @{device} example.com +short\n" + "An answer means it resolved a name it is not authoritative for." + ), + }, + "service-version": { + "severity": "info", + "title": "port {port} identifies itself as {product} {version}", + "why": ( + "Not a finding, a fact: this is the software the service names when " + "asked, quoted rather than guessed. netdiff does not match it against a " + "vulnerability database - home equipment rarely announces a precise " + "enough version for that to be honest, and a list of maybe-CVEs reads " + "as alarming while meaning nothing. What this is good for is the " + "question a database cannot answer: is this version still supported? A " + "web server from 2014 on a device with no firmware updates left is worth " + "more of your attention than any severity score." + ), + "fix": ( + "Search for '{product} {version} release date' and for the vendor's " + "support page for this device. If the version predates the last " + "firmware update you can install, install it. If the vendor has stopped " + "shipping updates, that is the finding." + ), + "verify": "curl -sI http://{device}:{port}/ | grep -i server", + }, "open-ports-noted": { "severity": "info", "title": "{count} open port(s) observed, and not reported as problems", @@ -343,12 +525,122 @@ def rule_ssh_v1(ip: str, port: int, banner: str): return finding("ssh-v1", ip, banner.splitlines()[0].strip(), port=port) -BANNER_RULES = (rule_plaintext_protocol, rule_http_auth_plaintext, rule_ssh_v1) +def _version_line(banner: str) -> str: + """The line of a banner that names software, if one does. + + An HTTP `Server:` header and an SSH version string are both declarations of + identity. Anything else falls back to the first line, which is where every + protocol that greets you puts its name - except an HTTP status line, where + `HTTP/1.0` is the version of the protocol and says nothing about the + software. Reporting that as a product would be the exact failure this tool + exists to avoid: a confident sentence about nothing. + """ + for line in banner.splitlines(): + if line.lower().startswith("server:") or line.startswith("SSH-"): + return line.strip() + lines = [ + line for line in banner.strip().splitlines() if not line.startswith("HTTP/") + ] + return lines[0].strip() if lines else "" + + +def rule_service_version(ip: str, port: int, banner: str): + """The software a service names, quoted rather than interpreted.""" + line = _version_line(banner) + match = _VERSION.search(line) + if not match: + return None + return finding( + "service-version", + ip, + line, + port=port, + product=match.group(1), + version=match.group(2), + ) + + +BANNER_RULES = ( + rule_plaintext_protocol, + rule_http_auth_plaintext, + rule_ssh_v1, + rule_service_version, +) + + +def rule_smbv1(ip: str, port: int, dialect: str): + """The server accepted an offer of the 1996 dialect and nothing else.""" + if not dialect: + return None + return finding( + "smb-v1", + ip, + f"negotiated dialect {dialect!r} when offered no other option", + port=port, + ) + + +def rule_ssh_weak_algorithms(ip: str, port: int, algorithms): + """Deprecated algorithms in what the server offered to negotiate with.""" + weak = [name for name in algorithms if name in WEAK_SSH_ALGORITHMS] + if not weak: + return None + return finding( + "ssh-weak-algorithms", + ip, + "offered: " + ", ".join(weak), + port=port, + count=len(weak), + detail="; ".join(f"{name} ({WEAK_SSH_ALGORITHMS[name]})" for name in weak), + ) + + +def rule_tls_cert_expired(ip: str, port: int, cert, today: str): + """Past its notAfter, by its own dates.""" + if cert is None or not cert.not_after or cert.not_after >= today: + return None + return finding( + "tls-cert-expired", + ip, + f"certificate valid {cert.not_before} to {cert.not_after}, today is {today}", + port=port, + not_after=cert.not_after, + ) + + +def rule_tls_cert_untrusted(ip: str, port: int, cert): + """Issuer and subject are the same name, so nothing external vouches for it.""" + if cert is None or not cert.self_signed: + return None + return finding( + "tls-cert-untrusted", + ip, + f"subject and issuer are both {cert.subject!r}; " + f"valid {cert.not_before} to {cert.not_after}", + port=port, + ) -def audit(devices, gateway=None, banners=None) -> list[Finding]: - """Apply every rule. `banners` maps (ip, port) -> whatever the service said.""" +def rule_dns_recursion(ip: str, evidence: str): + """It resolved a name it has no authority over.""" + if not evidence: + return None + return finding("dns-recursion-open", ip, evidence) + + +def audit( + devices, gateway=None, banners=None, probes=None, today=None +) -> list[Finding]: + """Apply every rule. + + `banners` maps (ip, port) -> whatever the service said. `probes` is the dict + `probe.collect()` returns: certificates, SMB dialects and SSH algorithms per + (ip, port), and DNS recursion evidence per ip. Both are plain data gathered + by the caller, which is what keeps this module free of sockets. + """ banners = banners or {} + probes = probes or {} + today = today or date.today().isoformat() findings = [] control = rule_upnp_control_open(gateway) @@ -359,15 +651,26 @@ def audit(devices, gateway=None, banners=None) -> list[Finding]: if mapping.enabled: findings.append(rule_mapping(mapping, devices)) + certs = probes.get("certs", {}) + smb = probes.get("smb", {}) + ssh = probes.get("ssh", {}) + resolvers = probes.get("dns", {}) + 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) + pair = (device.ip, port) + banner = banners.get(pair, "") + hits = [rule(device.ip, port, banner) for rule in BANNER_RULES] + hits.append(rule_smbv1(device.ip, port, smb.get(pair, ""))) + hits.append(rule_ssh_weak_algorithms(device.ip, port, ssh.get(pair, ()))) + hits.append(rule_tls_cert_expired(device.ip, port, certs.get(pair), today)) + hits.append(rule_tls_cert_untrusted(device.ip, port, certs.get(pair))) + findings.extend(hit for hit in hits if hit) + resolver = rule_dns_recursion(device.ip, resolvers.get(device.ip, "")) + if resolver: + findings.append(resolver) if open_ports: findings.append( diff --git a/netdiff/cli.py b/netdiff/cli.py index 13a45b9..c5df7d8 100644 --- a/netdiff/cli.py +++ b/netdiff/cli.py @@ -13,11 +13,16 @@ from pathlib import Path from . import audit as audit_rules -from . import mdns, oui, report, store, upnp +from . import mdns, oui, probe, report, store, upnp from .diff import diff, summarise -from .scan import discover, grab_banners, local_subnet - -DEFAULT_PORTS = (22, 80, 443, 445, 554, 1883, 3389, 5000, 8080, 8443) +from .scan import ( + DEFAULT_PORTS, + PORT_SETS, + discover, + grab_banners, + local_subnet, + resolve_ports, +) def send_webhook(url: str, payload: dict, timeout: float = 10) -> str: @@ -61,7 +66,7 @@ def resolve_subnet(args) -> None: def cmd_scan(args) -> int: resolve_subnet(args) - ports = () if args.no_ports else tuple(args.ports) + ports = () if args.no_ports else resolve_ports(args.ports) devices = discover( args.subnet, ports=ports, @@ -106,7 +111,8 @@ def cmd_scan(args) -> int: if device.ports else "" ) - print(f" {device.ip:<15} {device.mac} {label}{open_ports}") + hint = f" {device.os_hint}" if device.os_hint else "" + print(f" {device.ip:<15} {device.mac} {label}{hint}{open_ports}") print(f"\nchanges since last scan: {summarise(changes)}") for change in changes: print(f" {change}") @@ -160,12 +166,13 @@ def print_headline(finding, is_new: bool = False) -> None: above a group, so any single line still says what it is once it has been copied somewhere else. """ - print_field(finding.severity, f"{finding.title}{' [NEW]' if is_new else ''}") + title = audit_rules.headline(finding) + print_field(finding.severity, f"{title}{' [NEW]' if is_new else ''}") 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(f" {audit_rules.headline(finding)}{' [NEW]' if is_new else ''}") print_field("evidence", finding.evidence, wrap=False) print_field("why", finding.why) print_field("fix", finding.fix) @@ -198,7 +205,7 @@ def placeholders(text): resolve_subnet(args) devices = discover( args.subnet, - ports=tuple(args.ports), + ports=resolve_ports(args.ports), lookup_vendor=oui.lookup, resolve_names=not args.no_resolve, services={} if args.no_mdns else mdns.discover(), @@ -206,8 +213,9 @@ def placeholders(text): banners = grab_banners( (device.ip, port) for device in devices for port in device.ports ) + probes = probe.collect(devices, banners) gateway = None if args.no_upnp else upnp.probe_gateway(args.subnet) - findings = audit_rules.audit(devices, gateway, banners) + findings = audit_rules.audit(devices, gateway, banners, probes) conn = store.connect(args.db) scan_id = store.record_scan(conn, args.subnet, devices) @@ -290,7 +298,8 @@ def cmd_inventory(args) -> int: print(f"{len(rows)} device(s) ever seen\n") for row in rows: label = device_label(row["hostname"], row["vendor"], row["services"]) - print(f"{row['ip']:<15} {row['mac']} {label}") + hint = f" {row['os_hint']}" if row["os_hint"] else "" + print(f"{row['ip']:<15} {row['mac']} {label}{hint}") print( f" first {row['first_seen']} last {row['last_seen']} seen {row['times_seen']}x" ) @@ -327,7 +336,13 @@ def build_parser() -> argparse.ArgumentParser: 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( + "--ports", + nargs="*", + default=list(DEFAULT_PORTS), + metavar="PORT|SET", + help=f"port numbers, or a named set ({', '.join(sorted(PORT_SETS))})", + ) 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") scan.add_argument( @@ -355,7 +370,13 @@ def build_parser() -> argparse.ArgumentParser: 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( + "--ports", + nargs="*", + default=list(DEFAULT_PORTS), + metavar="PORT|SET", + help=f"port numbers, or a named set ({', '.join(sorted(PORT_SETS))})", + ) aud.add_argument("--no-resolve", action="store_true", help="skip reverse DNS") aud.add_argument( "--no-mdns", action="store_true", help="skip asking devices what they are" diff --git a/netdiff/probe.py b/netdiff/probe.py new file mode 100644 index 0000000..d7b809e --- /dev/null +++ b/netdiff/probe.py @@ -0,0 +1,359 @@ +"""Ask four specific questions of a service that a port number cannot answer. + +`scan.py` finds out what is here and which ports accept a connection. +`audit.py` decides what that means. This module is the layer between: it speaks +just enough of TLS, SMB, DNS and SSH to turn "port 445 is open" into "this +server accepted the 1996 version of SMB", which is a different kind of claim. + +Every probe here obeys the same contract as the rest of the tool. It sends the +opening move of a protocol and reads the reply that any client would get. It +sends no credentials, writes nothing a server would store, and asks for nothing +beyond what the handshake volunteers before authentication happens at all. + +The parsers are separated from the sockets on purpose - `parse_certificate`, +`parse_smb_negotiate`, `parse_kexinit` and `parse_dns_reply` are pure functions +over bytes, so the fiddly part is tested against captured shapes rather than +against a live host that happens to be on this desk today. +""" + +from __future__ import annotations + +import socket +import ssl +import struct +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass + +from .mdns import encode_name +from .scan import SCAN_WORKERS + +# ---------------------------------------------------------------- TLS + +# Ports where a TLS handshake is the expected greeting. Any other open port that +# stayed silent when spoken to is also tried, because silence is what a TLS port +# sounds like to a plaintext banner grab. +TLS_PORTS = (443, 465, 563, 636, 853, 989, 990, 993, 995, 8443, 8883, 9443) + + +@dataclass(frozen=True) +class Certificate: + """The three things about a certificate that can be judged without a CA.""" + + subject: str + issuer: str + not_before: str + not_after: str + + @property + def self_signed(self) -> bool: + return bool(self.subject) and self.subject == self.issuer + + +# A DER certificate is nested TLV. Rather than carry an ASN.1 parser for the two +# fields that matter, both are found by their shape: +# +# validity - the first SEQUENCE holding two adjacent time values. Times appear +# nowhere else in the structure before it, and the sequence is short enough +# that its length is always a single byte. +# +# commonName - the OID 2.5.4.3 followed by its string. Issuer precedes subject +# in a certificate, so the first two are the issuer's and the subject's. +# +# ponytail: shape-matching, not parsing. It reads the two fields nothing else +# can produce; it cannot read a third. Wanting key size or the SAN list is the +# day to write the real walker. +_UTC_TIME, _GENERALIZED_TIME = 0x17, 0x18 +_TIME_LENGTHS = {_UTC_TIME: 13, _GENERALIZED_TIME: 15} +_CN_OID = b"\x06\x03\x55\x04\x03" +_STRING_TAGS = (0x0C, 0x13, 0x14, 0x16, 0x1E) + + +def _read_time(der: bytes, at: int): + """(iso date, offset after it) if a time value starts at `at`, else None.""" + if at + 2 > len(der): + return None + tag, length = der[at], der[at + 1] + if _TIME_LENGTHS.get(tag) != length or at + 2 + length > len(der): + return None + text = der[at + 2 : at + 2 + length].decode("ascii", "replace") + if not text[:-1].isdigit(): + return None + if tag == _UTC_TIME: + # RFC 5280: a two-digit year below 50 means the 2000s. + year = int(text[:2]) + year += 2000 if year < 50 else 1900 + rest = text[2:] + else: + year, rest = int(text[:4]), text[4:] + return f"{year:04d}-{rest[0:2]}-{rest[2:4]}", at + 2 + length + + +def _common_names(der: bytes) -> list[str]: + names = [] + at = der.find(_CN_OID) + while at != -1: + tag, length = der[at + 5 : at + 6], der[at + 6 : at + 7] + if tag and length and tag[0] in _STRING_TAGS and length[0] < 128: + start = at + 7 + names.append(der[start : start + length[0]].decode("utf-8", "replace")) + at = der.find(_CN_OID, at + 5) + return names + + +def parse_certificate(der: bytes): + """Subject, issuer and validity from a DER certificate, or None.""" + validity = None + for at in range(len(der) - 4): + if der[at] != 0x30: # SEQUENCE + continue + first = _read_time(der, at + 2) + second = _read_time(der, first[1]) if first else None + if second: + validity = (first[0], second[0]) + break + if validity is None: + return None + names = _common_names(der) + return Certificate( + subject=names[1] if len(names) > 1 else "", + issuer=names[0] if names else "", + not_before=validity[0], + not_after=validity[1], + ) + + +def tls_certificate(ip: str, port: int, timeout: float = 3.0, server_name: str = ""): + """The certificate a port presents, or None if it does not speak TLS. + + Verification is off deliberately. Nearly every certificate on a home network + is self-signed and would fail a check against the system trust store, and + refusing the handshake would mean learning nothing about the most common + case. The certificate is read and judged on its own contents instead - which + is also why the DER is parsed here rather than trusting `getpeercert()`, as + that returns an empty dict for exactly the certificates worth looking at. + + `server_name` is sent as SNI when there is one. A server hosting several + names on one port refuses a handshake that does not say which was wanted, + and a reverse-DNS name is the best guess available - it costs nothing when + wrong, since the name is not checked against the certificate either way. + """ + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + try: + with socket.create_connection((ip, port), timeout) as raw: + with context.wrap_socket(raw, server_hostname=server_name or None) as tls: + der = tls.getpeercert(binary_form=True) + except (OSError, ssl.SSLError, ValueError): + return None + return parse_certificate(der) if der else None + + +# ---------------------------------------------------------------- SMB + +# An SMB1 NEGOTIATE PROTOCOL request offering exactly one dialect: the one from +# 1996 that WannaCry travelled over. Offering it alone is the whole technique - +# a server that also speaks SMB2 would answer any wider offer with SMB2 and hide +# whether SMB1 is still enabled, so the question has to be asked on its own. +_SMB1_DIALECT = b"NT LM 0.12" +# Exactly 32 bytes, and `parse_smb_negotiate` reads the reply's word count at +# offset 32 on that basis - so SMB_HEADER_SIZE is asserted rather than trusted. +SMB_HEADER_SIZE = 32 +_SMB_HEADER = ( + b"\xffSMB" # 4 protocol id + b"\x72" # 1 SMB_COM_NEGOTIATE + b"\x00\x00\x00\x00" # 4 status + b"\x18" # 1 flags: canonical paths, case sensitive + b"\x01\x28" # 2 flags2: long names, extended security + + b"\x00" * 2 # 2 pid high + + b"\x00" * 8 # 8 security signature + + b"\x00\x00" # 2 reserved + + b"\x00\x00" # 2 tree id + + b"\xff\xfe" # 2 process id + + b"\x00\x00" # 2 user id + + b"\x00\x00" # 2 multiplex id +) + + +def smb1_negotiate_request() -> bytes: + dialects = b"\x02" + _SMB1_DIALECT + b"\x00" + body = b"\x00" + struct.pack(" str: + """The dialect the server accepted, or '' if it refused SMBv1. + + A server with SMBv1 disabled either drops the connection or answers with + dialect index 0xFFFF, which is the protocol's way of saying "none of those". + """ + body = data[4:] + if not body.startswith(b"\xffSMB") or len(body) < SMB_HEADER_SIZE + 3: + return "" + if body[4] != 0x72 or int.from_bytes(body[5:9], "little") != 0: + return "" + at = SMB_HEADER_SIZE + if body[at] == 0: # word count: an error reply carries no dialect + return "" + index = int.from_bytes(body[at + 1 : at + 3], "little") + return "" if index == 0xFFFF else _SMB1_DIALECT.decode() + + +def smb_dialect(ip: str, port: int = 445, timeout: float = 3.0) -> str: + try: + with socket.create_connection((ip, port), timeout) as sock: + sock.settimeout(timeout) + sock.sendall(smb1_negotiate_request()) + return parse_smb_negotiate(sock.recv(1024)) + except OSError: + return "" + + +# ---------------------------------------------------------------- DNS + +# A name reserved by IANA for exactly this: examples and tests. Resolving it +# tells the same story as resolving anything else and leaks nothing about who +# is asking. +DNS_PROBE_NAME = "example.com" +# Fixed rather than random: it only has to match our own reply on a socket we +# connected ourselves. Randomising it would be cargo-culted cache-poisoning +# defence for a query whose answer is thrown away. +_DNS_ID = 0x1D1F + + +def dns_query(name: str = DNS_PROBE_NAME) -> bytes: + """One A query with the recursion-desired bit set.""" + header = struct.pack("!HHHHHH", _DNS_ID, 0x0100, 1, 0, 0, 0) + return header + encode_name(name) + struct.pack("!HH", 1, 1) + + +def parse_dns_reply(data: bytes) -> str: + """Evidence that this host resolves internet names for strangers, or ''.""" + if len(data) < 12: + return "" + ident, flags, _, answers, _, _ = struct.unpack("!HHHHHH", data[:12]) + if ident != _DNS_ID or not flags & 0x8000: + return "" + if flags & 0x000F or not flags & 0x0080 or not answers: + # rcode must be 0, the recursion-available bit must be set, and there + # has to be an actual answer. A referral is not recursion. + return "" + return ( + f"answered a query for {DNS_PROBE_NAME} with {answers} record(s), " + f"recursion-available bit set" + ) + + +def dns_recursion(ip: str, timeout: float = 2.0) -> str: + try: + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.settimeout(timeout) + sock.connect((ip, 53)) + sock.send(dns_query()) + return parse_dns_reply(sock.recv(2048)) + except OSError: + return "" + + +# ---------------------------------------------------------------- SSH + +_KEXINIT = 20 +# A KEXINIT carries ten name-lists: key exchange, host keys, ciphers each way, +# MACs each way, compression each way, languages each way. Only the first six +# are read. Compression is where every server in existence offers `none`, and +# `none` is a real finding in a cipher list and the default in a compression +# one - reading both and forgetting which was which would flag every SSH server +# on earth, which is the failure mode this project is a reaction to. +_KEXINIT_LISTS = 6 + + +def parse_kexinit(data: bytes) -> tuple: + """The algorithms an SSH server offered to negotiate with. + + Deduplicated: a server offers the same cipher list in both directions, and + "offers RC4, and also offers RC4" is not two facts. + """ + if len(data) < 6: + return () + length, padding = int.from_bytes(data[:4], "big"), data[4] + payload = data[5 : 4 + length] + if padding: + payload = payload[:-padding] + if not payload or payload[0] != _KEXINIT: + return () + at = 1 + 16 # message type, then the cookie + names = [] + for _ in range(_KEXINIT_LISTS): + if at + 4 > len(payload): + break + size = int.from_bytes(payload[at : at + 4], "big") + at += 4 + chunk = payload[at : at + size].decode("ascii", "replace") + at += size + names.extend(name for name in chunk.split(",") if name) + return tuple(dict.fromkeys(names)) + + +def ssh_algorithms(ip: str, port: int = 22, timeout: float = 3.0) -> tuple: + """What an SSH server offers to negotiate with. + + A server sends its KEXINIT unprompted as soon as version strings have been + swapped, so the entire exchange is: read its banner, send ours, read the + list. Nothing is attempted against authentication, which matters - a failed + auth is the usual way of grabbing this and it lands you in the target's log + and in fail2ban. + """ + try: + with socket.create_connection((ip, port), timeout) as sock: + sock.settimeout(timeout) + banner = sock.recv(512) + if not banner.startswith(b"SSH-"): + return () + sock.sendall(b"SSH-2.0-netdiff\r\n") + return parse_kexinit(sock.recv(8192)) + except OSError: + return () + + +# ---------------------------------------------------------------- collection + + +def collect(devices, banners=None, workers: int = SCAN_WORKERS) -> dict: + """Run every applicable probe across the network at once. + + Returns the plain dict `audit()` reads, keyed the same way `banners` is: + per (ip, port) for the ones that are about a service, per ip for DNS, which + is asked over UDP of every device because a UDP port never appears in a TCP + port scan. + """ + banners = banners or {} + jobs: list = [] + + for device in devices: + jobs.append(("dns", device.ip, lambda ip=device.ip: dns_recursion(ip))) + for port in device.ports: + pair = (device.ip, port) + if port in TLS_PORTS or not banners.get(pair, "").strip(): + name = device.hostname + jobs.append( + ( + "certs", + pair, + lambda p=pair, n=name: tls_certificate(*p, server_name=n), + ) + ) + if port == 445: + jobs.append(("smb", pair, lambda p=pair: smb_dialect(*p))) + if banners.get(pair, "").startswith("SSH-"): + jobs.append(("ssh", pair, lambda p=pair: ssh_algorithms(*p))) + + results: dict = {"certs": {}, "smb": {}, "dns": {}, "ssh": {}} + if not jobs: + return results + with ThreadPoolExecutor(max_workers=min(workers, len(jobs))) as pool: + for (kind, key, _), value in zip(jobs, pool.map(lambda j: j[2](), jobs)): + if value: + results[kind][key] = value + return results diff --git a/netdiff/report.py b/netdiff/report.py index ebccd59..f00fa09 100644 --- a/netdiff/report.py +++ b/netdiff/report.py @@ -23,7 +23,7 @@ from html import escape -from .audit import SEVERITY_ORDER +from .audit import SEVERITY_ORDER, headline # Deliberately drab. A report that looks like a security product invites the # reader to skim the colours instead of the sentences, and the whole thesis here @@ -78,7 +78,7 @@ def _finding_html(finding, is_new: bool) -> str: "
" f'' f"{esc(finding.severity)}" - f"{esc(finding.title)}{new}" + f"{esc(headline(finding))}{new}" f"
{''.join(rows)}
" "
" ) diff --git a/netdiff/scan.py b/netdiff/scan.py index 8fd5c7d..39b94cb 100644 --- a/netdiff/scan.py +++ b/netdiff/scan.py @@ -37,6 +37,11 @@ # rather than by core count. SCAN_WORKERS = 32 +# The port scan is flat - every (ip, port) at once rather than a pool of devices +# each walking its own list - so `--ports top100` costs one timeout rather than +# a hundred of them per device. Same reasoning as SCAN_WORKERS, more of it. +PORT_WORKERS = 256 + # `arp -an` on macOS/BSD/Linux: "? (192.168.1.1) at ab:cd:ef:12:34:56 on en0" _ARP_LINE = re.compile( r"\((?P\d+\.\d+\.\d+\.\d+)\)\s+at\s+(?P[0-9a-fA-F:]{11,17})" @@ -53,6 +58,61 @@ # cleartext rather than TLS. HTTP_PORTS = (80, 81, 591, 5000, 8000, 8008, 8080, 8081, 8888) +# The ports worth checking when you only want to know a device is alive and +# roughly what it is. Deliberately small: this is the default because most runs +# are a change check, not an inventory, and every port added here reports itself +# as `port-opened` on everyone's next scan. +DEFAULT_PORTS = (22, 80, 443, 445, 554, 1883, 3389, 5000, 8080, 8443) + +# nmap's top 100 TCP ports, in numeric order. Not a guess - it is the published +# frequency ranking from internet-wide scanning, which is exactly the question +# "which ports are worth the timeout" already answered by someone with data. +# fmt: off +TOP_100_PORTS = ( + 7, 9, 13, 21, 22, 23, 25, 26, 37, 53, 79, 80, 81, 88, 106, 110, 111, + 113, 119, 135, 139, 143, 144, 179, 199, 389, 427, 443, 444, 445, 465, + 513, 514, 515, 543, 544, 548, 554, 587, 631, 646, 873, 990, 993, 995, + 1025, 1026, 1027, 1028, 1029, 1110, 1433, 1720, 1723, 1755, 1900, 2000, + 2001, 2049, 2121, 2717, 3000, 3128, 3306, 3389, 3986, 4899, 5000, 5009, + 5051, 5060, 5101, 5190, 5357, 5432, 5631, 5666, 5800, 5900, 6000, 6001, + 6646, 7070, 8000, 8008, 8009, 8080, 8081, 8443, 8888, 9100, 9999, 10000, + 32768, 49152, 49153, 49154, 49155, 49156, 49157 +) +# fmt: on + +PORT_SETS = {"default": DEFAULT_PORTS, "top100": TOP_100_PORTS} + + +def resolve_ports(values) -> tuple[int, ...]: + """Turn what someone typed after `--ports` into port numbers. + + Accepts a set name or a list of numbers, and mixing them, because + `--ports top100 32400` is the obvious thing to want and refusing it would + only be pedantry. + """ + ports = set() + for value in values: + text = str(value) + if text in PORT_SETS: + ports.update(PORT_SETS[text]) + continue + if not text.isdigit() or not 0 < int(text) < 65536: + known = ", ".join(sorted(PORT_SETS)) + raise ValueError(f"{text!r} is not a port number or a set ({known})") + ports.add(int(text)) + return tuple(sorted(ports)) + + +# A packet's TTL is set by the sender and decremented per hop; on one broadcast +# segment there are no hops, so what arrives is what the OS started with. These +# are the three common starting values, and only an exact match earns a name - +# a TTL of 32 is not "nearly 64", it is a device doing something else, and +# rounding it into the nearest family would be a confident sentence about +# nothing. Even an exact match is a hint: any of these can be reconfigured. +TTL_ORIGINS = {64: "Linux, macOS or BSD", 128: "Windows", 255: "network gear"} + +_TTL = re.compile(r"ttl[=\s](\d+)", re.I) + @dataclass(frozen=True) class Device: @@ -64,6 +124,7 @@ class Device: hostname: str = "" services: str = "" ports: tuple[int, ...] = field(default=()) + os_hint: str = "" def key(self) -> str: """Identity across scans. @@ -192,18 +253,68 @@ def ping_one(ip: str) -> None: list(pool.map(ping_one, hosts)) +def _port_open(ip: str, port: int, timeout: float = 0.3) -> bool: + """TCP connect. Open means the handshake completed, nothing more.""" + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.settimeout(timeout) + return sock.connect_ex((ip, port)) == 0 + except OSError: + return False + + +def scan_ports_many(pairs, timeout: float = 0.3) -> set: + """Every (ip, port) at once; returns the pairs that are open. + + Flat rather than a pool of devices each walking its own port list. With ten + ports the difference was academic; with `top100` a device that drops packets + would otherwise cost a hundred consecutive timeouts, and the scan would take + the length of the port list rather than the length of one timeout. + """ + pairs = list(pairs) + if not pairs: + return set() + with ThreadPoolExecutor(max_workers=min(PORT_WORKERS, len(pairs))) as pool: + results = pool.map(lambda pair: _port_open(*pair, timeout=timeout), pairs) + return {pair for pair, is_open in zip(pairs, results) if is_open} + + def scan_ports(ip: str, ports, timeout: float = 0.3) -> tuple[int, ...]: - """TCP connect scan. Open means the handshake completed, nothing more.""" - open_ports = [] - for port in ports: - try: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.settimeout(timeout) - if sock.connect_ex((ip, port)) == 0: - open_ports.append(port) - except OSError: - continue - return tuple(sorted(open_ports)) + """The open ports of one device.""" + found = scan_ports_many([(ip, port) for port in ports], timeout=timeout) + return tuple(sorted(port for _, port in found)) + + +def os_family(ttl: int) -> str: + """The OS family a starting TTL suggests, phrased as the guess it is. + + An unrecognised TTL is reported as the number alone. It is still an + observation worth keeping - two devices with the same odd TTL are probably + the same kind of thing - but it is not a family, so it does not get a name. + """ + if not 0 < ttl <= 255: + return "" + name = TTL_ORIGINS.get(ttl) + return f"{name}? (TTL {ttl})" if name else f"TTL {ttl}" + + +def ttl_hint(ip: str, runner=subprocess.run) -> str: + """OS family guessed from the TTL of one ping reply, or '' if it did not answer. + + This is the only honest OS detection available without root: real + fingerprinting needs crafted packets and a raw socket. A TTL narrows it to a + family and nothing more, so the output says "Windows?" and quotes the number + it inferred that from. A device that drops ICMP - plenty of IoT gear does - + simply has no hint, which is the normal case rather than an error. + """ + try: + proc = runner( + ["ping", "-c", "1", ip], capture_output=True, text=True, timeout=3 + ) + except (OSError, subprocess.SubprocessError): + return "" + match = _TTL.search(proc.stdout or "") + return os_family(int(match.group(1))) if match else "" def grab_banners(pairs, timeout: float = 2.0) -> dict: @@ -282,6 +393,10 @@ def discover( ip: mac for ip, mac in table.items() if ipaddress.ip_address(ip) in network } + open_ports: dict[str, list[int]] = {} + for ip, port in scan_ports_many([(ip, p) for ip in in_subnet for p in ports]): + open_ports.setdefault(ip, []).append(port) + def observe(item): ip, mac = item return Device( @@ -290,11 +405,12 @@ def observe(item): vendor=lookup_vendor(mac) if lookup_vendor else "", hostname=resolve_hostname(ip) if resolve_names else "", services=(services or {}).get(ip, ""), - ports=scan_ports(ip, ports) if ports else (), + ports=tuple(sorted(open_ports.get(ip, ()))), + os_hint=ttl_hint(ip), ) - # Reverse DNS and the port scan are both waits, not work, and neither depends - # on any other device. Sequentially the scan took the sum of every timeout on + # Reverse DNS and the ping are both waits, not work, and neither depends on + # any other device. Sequentially the scan took the sum of every timeout on # the network; concurrently it takes the worst single device. with ThreadPoolExecutor(max_workers=SCAN_WORKERS) as pool: devices = list(pool.map(observe, in_subnet.items())) diff --git a/netdiff/store.py b/netdiff/store.py index 149ce70..99c440b 100644 --- a/netdiff/store.py +++ b/netdiff/store.py @@ -27,6 +27,7 @@ hostname TEXT NOT NULL DEFAULT '', services TEXT NOT NULL DEFAULT '', ports TEXT NOT NULL DEFAULT '', + os_hint TEXT NOT NULL DEFAULT '', PRIMARY KEY (scan_id, mac) ); CREATE INDEX IF NOT EXISTS observations_mac ON observations(mac); @@ -60,7 +61,10 @@ def connect(path=DEFAULT_PATH) -> sqlite3.Connection: # nothing to a table that already exists, and people have history going back # months, so a new column has to be added explicitly or every read of it fails # with "no such column" on exactly the databases worth keeping. -ADDED_COLUMNS = (("observations", "services", "TEXT NOT NULL DEFAULT ''"),) +ADDED_COLUMNS = ( + ("observations", "services", "TEXT NOT NULL DEFAULT ''"), + ("observations", "os_hint", "TEXT NOT NULL DEFAULT ''"), +) def _add_missing_columns(conn: sqlite3.Connection) -> None: @@ -98,8 +102,8 @@ def record_scan(conn: sqlite3.Connection, subnet: str, devices) -> int: scan_id = cursor.lastrowid conn.executemany( "INSERT INTO observations" - " (scan_id, mac, ip, vendor, hostname, services, ports)" - " VALUES (?, ?, ?, ?, ?, ?, ?)", + " (scan_id, mac, ip, vendor, hostname, services, ports, os_hint)" + " VALUES (?, ?, ?, ?, ?, ?, ?, ?)", [ ( scan_id, @@ -109,6 +113,7 @@ def record_scan(conn: sqlite3.Connection, subnet: str, devices) -> int: d.hostname, d.services, _ports_to_text(d.ports), + d.os_hint, ) for d in devices ], @@ -118,7 +123,7 @@ def record_scan(conn: sqlite3.Connection, subnet: str, devices) -> int: def load_scan(conn: sqlite3.Connection, scan_id: int) -> list[Device]: rows = conn.execute( - "SELECT mac, ip, vendor, hostname, services, ports FROM observations" + "SELECT mac, ip, vendor, hostname, services, ports, os_hint FROM observations" " WHERE scan_id = ?", (scan_id,), ).fetchall() @@ -130,6 +135,7 @@ def load_scan(conn: sqlite3.Connection, scan_id: int) -> list[Device]: hostname=r["hostname"], services=r["services"], ports=_ports_from_text(r["ports"]), + os_hint=r["os_hint"], ) for r in rows ] @@ -223,7 +229,7 @@ def inventory(conn: sqlite3.Connection) -> list[dict]: out = [] for row in rows: latest = conn.execute( - "SELECT ip, vendor, hostname, services, ports FROM observations o" + "SELECT ip, vendor, hostname, services, ports, os_hint FROM observations o" " JOIN scans s ON s.id = o.scan_id WHERE o.mac = ?" " ORDER BY s.id DESC LIMIT 1", (row["mac"],), @@ -239,6 +245,7 @@ def inventory(conn: sqlite3.Connection) -> list[dict]: "hostname": latest["hostname"], "services": latest["services"], "ports": _ports_from_text(latest["ports"]), + "os_hint": latest["os_hint"], } ) return out diff --git a/tests/test_audit.py b/tests/test_audit.py index 9c4973a..c701799 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -9,10 +9,18 @@ important half. """ -from netdiff.audit import RULES, audit, summarise +from netdiff.audit import RULES, audit, headline, summarise +from netdiff.probe import Certificate from netdiff.scan import Device from netdiff.upnp import Gateway, Mapping +EXPIRED_SELF_SIGNED = Certificate( + subject="nas.local", + issuer="nas.local", + not_before="2015-01-01", + not_after="2016-01-01", +) + 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)) @@ -40,6 +48,13 @@ def rules_fired(findings): return [f.rule for f in findings] +def one(findings, rule): + """The single finding of a rule, failing loudly if it fired twice or not at all.""" + hits = [f for f in findings if f.rule == rule] + assert len(hits) == 1, f"{rule} fired {len(hits)} times" + return hits[0] + + # --- the anti-theater tests: presence is not vulnerability ------------------- @@ -68,6 +83,104 @@ def test_a_connection_error_is_not_a_finding(): assert [f for f in findings if f.severity != "info"] == [] +def test_port_445_open_is_not_smbv1(): + """Every Windows machine and NAS has 445 open. Only a reply is evidence.""" + findings = audit([dev(ports=[445])], probes={"smb": {}}) + assert "smb-v1" not in rules_fired(findings) + + +def test_a_modern_ssh_server_is_not_reported(): + algorithms = ("curve25519-sha256", "rsa-sha2-512", "aes256-gcm@openssh.com") + findings = audit( + [dev(ports=[22])], probes={"ssh": {("192.168.1.10", 22): algorithms}} + ) + assert "ssh-weak-algorithms" not in rules_fired(findings) + + +def test_a_certificate_that_is_current_and_signed_by_a_ca_is_not_reported(): + cert = Certificate("nas.local", "Some CA R3", "2026-01-01", "2027-01-01") + findings = audit( + [dev(ports=[443])], + probes={"certs": {("192.168.1.10", 443): cert}}, + today="2026-08-01", + ) + assert [f for f in findings if f.rule.startswith("tls-")] == [] + + +def test_a_device_that_answers_no_dns_query_is_not_a_resolver(): + findings = audit([dev(ports=[53])], probes={"dns": {"192.168.1.10": ""}}) + assert "dns-recursion-open" not in rules_fired(findings) + + +def test_an_http_version_line_is_not_a_product_version(): + """`HTTP/1.0` names the protocol. Reporting it as software would be invention.""" + banners = {("192.168.1.10", 80): "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n"} + assert "service-version" not in rules_fired( + audit([dev(ports=[80])], banners=banners) + ) + + +# --- the depth rules, when they do fire -------------------------------------- + + +def test_a_server_that_accepted_smbv1_is_high_and_quotes_the_dialect(): + hit = one( + audit( + [dev(ports=[445])], probes={"smb": {("192.168.1.10", 445): "NT LM 0.12"}} + ), + "smb-v1", + ) + assert hit.severity == "high" + assert "NT LM 0.12" in hit.evidence + + +def test_weak_ssh_algorithms_are_named_with_the_reason_each_is_weak(): + algorithms = ("curve25519-sha256", "ssh-rsa", "aes256-ctr", "hmac-md5") + hit = one( + audit([dev(ports=[22])], probes={"ssh": {("192.168.1.10", 22): algorithms}}), + "ssh-weak-algorithms", + ) + assert "2 deprecated" in hit.title + assert "ssh-rsa" in hit.evidence and "hmac-md5" in hit.evidence + assert "MD5" in hit.why # the reason, not just the name + assert "aes256-ctr" not in hit.evidence + + +def test_an_expired_certificate_reports_the_date_it_expired(): + hit = one( + audit( + [dev(ports=[443])], + probes={"certs": {("192.168.1.10", 443): EXPIRED_SELF_SIGNED}}, + today="2026-08-01", + ), + "tls-cert-expired", + ) + assert "2016-01-01" in hit.title + assert "2026-08-01" in hit.evidence + + +def test_a_self_signed_certificate_is_info_and_says_so_is_normal(): + hit = one( + audit( + [dev(ports=[443])], + probes={"certs": {("192.168.1.10", 443): EXPIRED_SELF_SIGNED}}, + today="2026-08-01", + ), + "tls-cert-untrusted", + ) + assert hit.severity == "info" + assert "not a problem" in hit.why + + +def test_a_version_banner_is_quoted_not_interpreted(): + banners = {("192.168.1.10", 80): "HTTP/1.0 200 OK\r\nServer: lighttpd/1.4.35\r\n"} + hit = one(audit([dev(ports=[80])], banners=banners), "service-version") + assert hit.severity == "info" + assert "lighttpd 1.4.35" in hit.title + assert hit.evidence == "Server: lighttpd/1.4.35" + assert "CVE" not in hit.why.upper() or "does not match" in hit.why + + def test_no_devices_produces_no_findings_at_all(): assert audit([]) == [] assert summarise([]) == "nothing to report" @@ -272,13 +385,27 @@ def test_findings_sort_worst_first(): 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]) +def everything(): + """One network arranged so that every rule in the table fires exactly once. + + Kept as a helper because two different tests need it: one asserts the rule + ids line up with the teaching table, the other that nothing renders a + placeholder. A rule added without a case here fails the first of those, + which is the point - the table and the code cannot drift apart quietly. + """ + device = dev(ip="192.168.1.23", ports=[21, 22, 23, 443, 445, 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", + ("192.168.1.23", 8080): "HTTP/1.1 401\r\nWWW-Authenticate: Basic\r\n" + "Server: lighttpd/1.4.35\r\n", + } + probes = { + "certs": {("192.168.1.23", 443): EXPIRED_SELF_SIGNED}, + "smb": {("192.168.1.23", 445): "NT LM 0.12"}, + "ssh": {("192.168.1.23", 22): ("ssh-rsa", "aes128-cbc", "hmac-md5")}, + "dns": {"192.168.1.23": "answered a query for example.com"}, } - fired = audit( + return audit( [device, dev(ip="192.168.1.30")], gw( [ @@ -288,8 +415,13 @@ def test_every_rule_id_used_by_a_rule_exists_in_the_teaching_table(): ] ), banners, + probes, + today="2026-08-01", ) - assert {f.rule for f in fired} == set(RULES) + + +def test_every_rule_id_used_by_a_rule_exists_in_the_teaching_table(): + assert {f.rule for f in everything()} == set(RULES) def test_dangling_verify_pings_a_bare_address_not_an_ip_colon_port(): @@ -304,23 +436,7 @@ def test_dangling_verify_pings_a_bare_address_not_an_ip_colon_port(): 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 f in everything(): for field in (f.title, f.why, f.fix, f.verify): assert "{" not in field and "}" not in field, f.rule @@ -343,3 +459,24 @@ def test_summarise_counts_by_severity_worst_first(): ) assert text.startswith("1 critical") assert "info" in text + + +def test_a_headline_names_the_device_it_is_about(): + """Three devices running lighttpd produce three identical lines otherwise.""" + banners = {("192.168.1.10", 80): "HTTP/1.0 200\r\nServer: lighttpd/1.4.35\r\n"} + hit = one(audit([dev(ports=[80])], banners=banners), "service-version") + assert headline(hit).startswith("192.168.1.10 ") + + +def test_a_headline_does_not_repeat_a_device_the_title_already_names(): + hit = one( + audit([dev(ports=[53])], probes={"dns": {"192.168.1.10": "answered"}}), + "dns-recursion-open", + ) + assert headline(hit) == hit.title + assert hit.title.count("192.168.1.10") == 1 + + +def test_a_finding_about_the_network_is_not_prefixed_with_the_word_network(): + hit = one(audit([dev(ports=[22])]), "open-ports-noted") + assert headline(hit) == hit.title diff --git a/tests/test_probe.py b/tests/test_probe.py new file mode 100644 index 0000000..0d537a0 --- /dev/null +++ b/tests/test_probe.py @@ -0,0 +1,229 @@ +"""The depth probes, exercised without a network. + +The sockets are three lines each; the parsers are where a mistake hides, so the +parsers are what is tested. Two of these fixtures are real captures - the SMB +replies came off actual Samba servers, one with SMBv1 enabled and one without - +because a check for a protocol nobody has run in a decade is exactly the kind +that quietly stops working and reports "nothing found" forever. + +The SSH packets are built here from RFC 4253 rather than with anything in +`probe.py`, for the reason `test_mdns.py` gives: a decoder tested against its +own encoder agrees with itself however wrong both are. +""" + +import base64 +import struct + +import pytest + +from netdiff.probe import ( + Certificate, + parse_certificate, + parse_dns_reply, + parse_kexinit, + parse_smb_negotiate, + smb1_negotiate_request, +) + +# --- TLS --------------------------------------------------------------------- + +# openssl req -x509 -newkey rsa:2048 -nodes -days 3650 \ +# -subj "/CN=nas.local/O=Home" -keyout k.pem -out c.pem +# Generated for this test. `openssl x509 -noout -dates` on it prints +# notBefore=Aug 1 2026, notAfter=Jul 29 2036 - which is what the parser has to +# arrive at from the bytes alone. +SELF_SIGNED_DER = base64.b64decode( + "MIIDJzCCAg+gAwIBAgIUHru3bCA+dYZEo/cml7W/A+THyTkwDQYJKoZIhvcNAQELBQAw" + "IzESMBAGA1UEAwwJbmFzLmxvY2FsMQ0wCwYDVQQKDARIb21lMB4XDTI2MDgwMTIzNTQ1" + "MloXDTM2MDcyOTIzNTQ1MlowIzESMBAGA1UEAwwJbmFzLmxvY2FsMQ0wCwYDVQQKDARI" + "b21lMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA5iqp4j1fNGxPTY+AxoR7" + "33wtYXDRG8+fJpThqiaEiqBx21Mgtw3ppiAhOXVUuxAL07QvoZzwgZSbdE8ln+DNLQjH" + "JGtnLfiqNQxezAr+IFOwjK/Exmeo6UMBOvGiOPEGYR5t52qqewzqzla2HRhgXm0bxqWN" + "Mu1ReNenIZ3cZC7GRoNqcyTY6WtcPCdQ6HXpiErMnsUyZXiEsKMNjtGJaBtSSRQVdq+I" + "TQWP1onxHEK7YdpRvU8/cermJqJ6bJrC28mzO9ogAfFjisjRhYcBdv1INwuZYyCkckaK" + "tJdSlU7FAGts9hDVc6K3ZCOnZW9YCDqHJqKHic6ZwG8F/v+PEwIDAQABo1MwUTAdBgNV" + "HQ4EFgQUpiDTgBoFX5tExxNw2oC9oGn8DjEwHwYDVR0jBBgwFoAUpiDTgBoFX5tExxNw" + "2oC9oGn8DjEwDwYDVR0TAQH/BAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAg0cCO4m7" + "G228oLY2U56stkH/86DnatU4/r7MxNjfLoSFhn3AS/J+sx6uNGvJ9asS0U4pU5Kbq45q" + "dnU0Jz8b6IgQ4fCx8Ewv9YK4CPZqoUGav4rQXQlqRJXFPJq7++jftWbaCNrw2vsm3g/j" + "HzXukzerDh2wqBGbwKAlauAjuq10cCayYDbgglkgo27T1Yyw0COguVmiaMyJ23TzsU8n" + "NdxH2W4B1mudJt5I+yl7tSdEJLNnOjUYcpnKnW2HCJB5WyhwdJmif339dRoAWQE0JfdH" + "ho4nszgE4XKhV5gVi9D8RicVRwn2HC/07J5j4EIKgTVNHkqugRJxGEntTZaZxw==" +) + + +def test_a_real_certificate_yields_its_names_and_dates(): + cert = parse_certificate(SELF_SIGNED_DER) + assert cert == Certificate( + subject="nas.local", + issuer="nas.local", + not_before="2026-08-01", + not_after="2036-07-29", + ) + + +def test_a_certificate_that_signed_itself_is_recognised(): + assert parse_certificate(SELF_SIGNED_DER).self_signed + + +def test_a_certificate_someone_else_signed_is_not(): + assert not Certificate( + subject="nas.local", + issuer="Some CA R3", + not_before="2026-01-01", + not_after="2027-01-01", + ).self_signed + + +def test_a_certificate_with_no_readable_name_claims_nothing(): + """Better a blank subject than a subject borrowed from the issuer.""" + assert not Certificate("", "", "2026-01-01", "2027-01-01").self_signed + + +def der_validity(not_before: bytes, not_after: bytes, tag: int = 0x17) -> bytes: + """A SEQUENCE of two times, wrapped in enough noise to have to be found.""" + times = bytes([tag, len(not_before)]) + not_before + times += bytes([tag, len(not_after)]) + not_after + return b"\x30\x82\x01\x00" + b"\x02\x01\x02" + bytes([0x30, len(times)]) + times + + +@pytest.mark.parametrize( + "year, expect", + [(b"49", "2049"), (b"50", "1950"), (b"99", "1999"), (b"26", "2026")], +) +def test_a_two_digit_year_pivots_at_fifty(year, expect): + """RFC 5280's rule. Getting it backwards dates certificates a century out.""" + cert = parse_certificate(der_validity(year + b"0101000000Z", b"300101000000Z")) + assert cert.not_before.startswith(expect) + + +def test_a_four_digit_year_is_read_as_written(): + cert = parse_certificate( + der_validity(b"20260801000000Z", b"20360729000000Z", tag=0x18) + ) + assert (cert.not_before, cert.not_after) == ("2026-08-01", "2036-07-29") + + +def test_bytes_that_are_not_a_certificate_produce_nothing(): + assert parse_certificate(b"") is None + assert parse_certificate(b"\x30\x82" + b"\xff" * 200) is None + assert parse_certificate(SELF_SIGNED_DER[:40]) is None + + +# --- SMB --------------------------------------------------------------------- + +# Both captured from Samba answering the request in `probe.py`, one configured +# `server min protocol = NT1` and one `= SMB2`. The second is the shape that +# matters most: a refusal is not an error, so a parser that only handles the +# happy path reports every modern server as running SMBv1. +SMB1_ACCEPTED = base64.b64decode( + "AAAAn/9TTUJyAAAAAIgBKAAAAAAAAAAAAAAAAAAA//4AAAAAEQAAAzIAAQAEQQAAAA" + "ABAE8AAAD984CACuml3REi3QEAAABaADlhN2JkNWNlYTUwNQAAAABgSAYGKwYBBQUC" + "oD4wPKAOMAwGCisGAQQBgjcCAgqjKjAooCYbJG5vdF9kZWZpbmVkX2luX1JGQzQxNz" + "hAcGxlYXNlX2lnbm9yZQ==" +) +SMB1_REFUSED = base64.b64decode( + "AAAAJf9TTUJyAAAAAIgDQAAAAAAAAAAAAAAAAAAA//4AAAAAAf//AAA=" +) + + +def test_the_request_is_a_well_formed_smb_message(): + """The length field and the header size are what the reply parser assumes.""" + request = smb1_negotiate_request() + assert request[0] == 0x00 + assert int.from_bytes(request[1:4], "big") == len(request) - 4 + assert request[4:8] == b"\xffSMB" + assert request[8] == 0x72 + + +def test_a_server_that_accepted_smbv1_is_reported(): + assert parse_smb_negotiate(SMB1_ACCEPTED) == "NT LM 0.12" + + +def test_a_server_that_refused_smbv1_is_not(): + assert parse_smb_negotiate(SMB1_REFUSED) == "" + + +def test_nothing_that_is_not_an_smb_reply_is_reported(): + assert parse_smb_negotiate(b"") == "" + assert parse_smb_negotiate(b"HTTP/1.1 400 Bad Request\r\n\r\n") == "" + assert parse_smb_negotiate(SMB1_ACCEPTED[:20]) == "" + + +# --- DNS --------------------------------------------------------------------- + + +def dns_reply(flags: int, answers: int = 1, ident: int = 0x1D1F) -> bytes: + return struct.pack("!HHHHHH", ident, flags, 1, answers, 0, 0) + + +def test_a_recursive_answer_is_recognised(): + # QR + recursion desired + recursion available, rcode 0. + assert parse_dns_reply(dns_reply(0x8180)) + + +def test_a_resolver_that_does_not_offer_recursion_is_not_reported(): + assert parse_dns_reply(dns_reply(0x8100)) == "" + + +def test_an_error_or_an_empty_answer_is_not_recursion(): + assert parse_dns_reply(dns_reply(0x8183)) == "" # NXDOMAIN + assert parse_dns_reply(dns_reply(0x8180, answers=0)) == "" # referral only + + +def test_a_reply_to_someone_elses_query_is_ignored(): + assert parse_dns_reply(dns_reply(0x8180, ident=0x1234)) == "" + + +def test_a_question_is_not_mistaken_for_an_answer(): + assert parse_dns_reply(dns_reply(0x0100)) == "" + assert parse_dns_reply(b"\x00" * 4) == "" + + +# --- SSH --------------------------------------------------------------------- + + +def kexinit(*name_lists: str) -> bytes: + """An SSH binary packet carrying a KEXINIT, per RFC 4253 section 6.""" + payload = bytes([20]) + b"\x00" * 16 + for names in name_lists: + raw = names.encode() + payload += struct.pack("!I", len(raw)) + raw + padding = 8 - (len(payload) + 5) % 8 + return ( + struct.pack("!IB", len(payload) + padding + 1, padding) + + payload + + (b"\x00" * padding) + ) + + +SIX = ("kex", "hostkey", "cipher-out", "cipher-in", "mac-out", "mac-in") + + +def test_every_offered_algorithm_is_read(): + packet = kexinit(*[f"{name}-a,{name}-b" for name in SIX]) + assert parse_kexinit(packet) == tuple( + f"{name}-{half}" for name in SIX for half in "ab" + ) + + +def test_the_same_algorithm_offered_both_directions_is_one_fact(): + packet = kexinit("kex", "hostkey", "aes128-cbc", "aes128-cbc", "m", "m") + assert parse_kexinit(packet) == ("kex", "hostkey", "aes128-cbc", "m") + + +def test_compression_none_is_not_read_as_an_algorithm(): + """Every SSH server offers `none` compression. Reading it would flag them all.""" + packet = kexinit(*SIX, "none", "none", "", "") + assert "none" not in parse_kexinit(packet) + + +def test_a_cipher_called_none_is_still_read(): + packet = kexinit("kex", "hostkey", "none", "none", "m", "m") + assert "none" in parse_kexinit(packet) + + +def test_anything_that_is_not_a_kexinit_offers_nothing(): + assert parse_kexinit(b"") == () + assert parse_kexinit(b"SSH-2.0-OpenSSH_9.6\r\n") == () + assert parse_kexinit(kexinit(*SIX)[:12]) == () diff --git a/tests/test_scan_and_store.py b/tests/test_scan_and_store.py index d404ff0..91f6883 100644 --- a/tests/test_scan_and_store.py +++ b/tests/test_scan_and_store.py @@ -195,7 +195,10 @@ def test_devices_are_scanned_concurrently(monkeypatch): table = {f"192.168.1.{n}": f"aa:bb:cc:00:00:{n:02x}" for n in range(1, 13)} monkeypatch.setattr(scan, "nudge", lambda *a, **k: None) monkeypatch.setattr(scan, "read_arp_table", lambda: table) - monkeypatch.setattr(scan, "scan_ports", lambda ip, ports: time.sleep(0.2) or ()) + monkeypatch.setattr(scan, "ttl_hint", lambda ip: time.sleep(0.2) or "") + monkeypatch.setattr( + scan, "_port_open", lambda ip, port, timeout=0.3: time.sleep(0.2) or False + ) start = time.monotonic() devices = scan.discover( @@ -206,7 +209,18 @@ def test_devices_are_scanned_concurrently(monkeypatch): assert [d.ip for d in devices] == sorted( table, key=lambda ip: int(ip.split(".")[3]) ) - assert elapsed < 1.0, f"12 devices x 0.2s took {elapsed:.1f}s - run sequentially?" + assert elapsed < 1.0, f"12 devices x 0.4s took {elapsed:.1f}s - run sequentially?" + + +def test_a_long_port_list_costs_one_timeout_not_a_hundred(monkeypatch): + """The point of `--ports top100`: ports overlap, they do not queue.""" + monkeypatch.setattr( + scan, "_port_open", lambda ip, port, timeout=0.3: time.sleep(0.2) or False + ) + start = time.monotonic() + assert scan.scan_ports("192.168.1.10", scan.TOP_100_PORTS) == () + elapsed = time.monotonic() - start + assert elapsed < 1.0, f"100 ports x 0.2s took {elapsed:.1f}s - run sequentially?" def test_banners_are_gathered_concurrently(monkeypatch): @@ -227,6 +241,67 @@ def test_no_open_ports_means_no_banner_work(): assert scan.grab_banners([]) == {} +# --- choosing what to scan, and guessing what answered ----------------------- + + +def test_a_named_port_set_expands_to_its_ports(): + assert scan.resolve_ports(["top100"]) == scan.TOP_100_PORTS + assert scan.resolve_ports(["default"]) == tuple(sorted(scan.DEFAULT_PORTS)) + + +def test_numbers_and_a_set_can_be_mixed_and_are_deduplicated(): + assert scan.resolve_ports(["default", "22", "32400"]) == tuple( + sorted(set(scan.DEFAULT_PORTS) | {32400}) + ) + + +@pytest.mark.parametrize("bad", ["top1000", "0", "65536", "-1", "80/tcp", ""]) +def test_something_that_is_neither_is_a_message_not_a_traceback(bad): + with pytest.raises(ValueError, match="not a port number"): + scan.resolve_ports([bad]) + + +def test_the_top_100_is_a_hundred_ports_in_order(): + assert len(set(scan.TOP_100_PORTS)) == 100 + assert list(scan.TOP_100_PORTS) == sorted(scan.TOP_100_PORTS) + + +@pytest.mark.parametrize( + "ttl, expect", [(64, "Linux"), (128, "Windows"), (255, "network gear")] +) +def test_a_ttl_suggests_an_os_family_and_says_it_is_a_guess(ttl, expect): + hint = scan.os_family(ttl) + assert expect in hint + assert "?" in hint and str(ttl) in hint, "a guess has to look like one" + + +def test_an_unrecognised_ttl_is_reported_as_a_number_not_a_family(): + """32 is not "nearly 64". Rounding it into a family would be invention.""" + assert scan.os_family(32) == "TTL 32" + + +def test_a_ttl_nothing_could_have_sent_claims_nothing(): + assert scan.os_family(0) == "" + assert scan.os_family(300) == "" + + +def test_the_ttl_is_read_from_a_real_ping_line(): + line = "64 bytes from 192.168.1.1: icmp_seq=0 ttl=64 time=2.508 ms" + assert scan.ttl_hint("192.168.1.1", runner=fake_runner(line)) == scan.os_family(64) + + +def test_a_host_that_drops_icmp_has_no_hint_rather_than_a_wrong_one(): + assert scan.ttl_hint("192.168.1.9", runner=fake_runner("")) == "" + + +def fake_runner(stdout): + class Result: + returncode = 0 + + Result.stdout = stdout + return lambda *a, **k: Result() + + @pytest.fixture def conn(tmp_path): return store.connect(tmp_path / "history.db") @@ -240,12 +315,13 @@ def test_scan_roundtrips_through_the_database(conn): vendor="Acme", services="Chromecast", ports=(22, 80), + os_hint="Linux, macOS or BSD? (TTL 64)", ), Device(mac="aa:bb:cc:00:00:02", ip="192.168.1.11", hostname="nas.local"), ] scan_id = store.record_scan(conn, "192.168.1.0/24", devices) assert store.load_scan(conn, scan_id) == devices, ( - "ports, services and blanks must survive the trip" + "ports, services, the OS hint and blanks must survive the trip" )