diff --git a/backend/app/scanner/__init__.py b/backend/app/scanner/__init__.py index fc2e3c4..3f67818 100644 --- a/backend/app/scanner/__init__.py +++ b/backend/app/scanner/__init__.py @@ -12,7 +12,9 @@ from dataclasses import dataclass, field from app.scanner.arp_scan import ArpHost, run_arp_scan +from app.scanner.dns_lookup import resolve_hostnames from app.scanner.nmap_scan import NmapHost, run_nmap_scan +from app.scanner.os_inference import enrich_os_guesses logger = logging.getLogger(__name__) @@ -64,6 +66,12 @@ def orchestrate_scan( if not nh.mac and nh.ip in arp_by_ip: nh.mac = arp_by_ip[nh.ip].mac + # Hostname fallback: reverse DNS for hosts nmap couldn't name + resolve_hostnames(nmap_hosts) + + # Passive OS inference from service/version banners + enrich_os_guesses(nmap_hosts) + # Hosts arp found but nmap returned nothing for (e.g. ICMP-filtered) arp_only = [h for h in arp_hosts if h.ip not in nmap_ips] diff --git a/backend/app/scanner/dns_lookup.py b/backend/app/scanner/dns_lookup.py new file mode 100644 index 0000000..624023c --- /dev/null +++ b/backend/app/scanner/dns_lookup.py @@ -0,0 +1,77 @@ +""" +dns_lookup.py — Reverse DNS lookup fallback for hostname resolution. + +resolve_hostnames() is the only public function; it operates on a list of +NmapHost objects in-place and fills in hostname where nmap returned none. +Pure socket calls — no subprocess, no elevated privileges required. +""" + +from __future__ import annotations + +import logging +import signal +import socket +from contextlib import contextmanager + +from app.scanner.nmap_scan import NmapHost + +logger = logging.getLogger(__name__) + +_LOOKUP_TIMEOUT_SECONDS = 1 + + +@contextmanager +def _timeout(seconds: int): + """SIGALRM-based timeout context for blocking socket calls.""" + + def _handler(signum, frame): # noqa: ANN001 — signal handler signature is fixed by the stdlib + raise TimeoutError + + old = signal.signal(signal.SIGALRM, _handler) + signal.alarm(seconds) + try: + yield + finally: + signal.alarm(0) + signal.signal(signal.SIGALRM, old) + + +def _rdns(ip: str) -> str: + """ + Return the PTR hostname for *ip*, or empty string on any failure. + + Uses a SIGALRM timeout so a single slow DNS server cannot stall + the entire scan cycle. + """ + try: + with _timeout(_LOOKUP_TIMEOUT_SECONDS): + hostname, _, _ = socket.gethostbyaddr(ip) + return hostname + except (OSError, TimeoutError): + return "" + + +def resolve_hostnames(hosts: list[NmapHost]) -> None: + """ + Fill in missing hostnames on *hosts* via reverse DNS (in-place). + + Only queries IPs where nmap returned no hostname. Each lookup is + individually timeout-guarded so a non-responsive resolver doesn't + block the scan. + """ + missing = [h for h in hosts if not h.hostname] + if not missing: + return + + logger.debug("Reverse DNS lookup for %d host(s) with no hostname", len(missing)) + resolved = 0 + for host in missing: + name = _rdns(host.ip) + if name: + host.hostname = name + resolved += 1 + logger.debug("rDNS %s → %s", host.ip, name) + else: + logger.debug("rDNS %s → (no PTR record)", host.ip) + + logger.info("Reverse DNS resolved %d/%d hostname(s)", resolved, len(missing)) diff --git a/backend/app/scanner/os_inference.py b/backend/app/scanner/os_inference.py new file mode 100644 index 0000000..e227b3e --- /dev/null +++ b/backend/app/scanner/os_inference.py @@ -0,0 +1,109 @@ +""" +os_inference.py — Passive OS hints from nmap service/version banners. + +infer_os() is the single public function. It inspects a host's port +banners and service names to produce a best-effort OS label without +requiring elevated privileges or -O. + +All logic is pure string matching so it is fully unit-testable. +""" + +from __future__ import annotations + +import re + +from app.scanner.nmap_scan import NmapHost + +# ── SSH banner patterns ─────────────────────────────────────────────────────── +# OpenSSH embeds OS/distro hints in the version comment field, e.g. +# "OpenSSH 8.9p1 Ubuntu-3ubuntu0.6" +# "OpenSSH 8.4p1 Debian-2+deb11u2" +# "OpenSSH 7.9 (protocol 2.0)" ← FreeBSD default +# "OpenSSH_for_Windows_8.1" + +_SSH_OS_PATTERNS: list[tuple[re.Pattern, str]] = [ + (re.compile(r"Ubuntu", re.IGNORECASE), "Linux (Ubuntu)"), + (re.compile(r"Debian", re.IGNORECASE), "Linux (Debian)"), + (re.compile(r"Raspbian", re.IGNORECASE), "Linux (Raspbian)"), + (re.compile(r"CentOS", re.IGNORECASE), "Linux (CentOS)"), + (re.compile(r"Fedora", re.IGNORECASE), "Linux (Fedora)"), + (re.compile(r"openSUSE|SUSE", re.IGNORECASE), "Linux (openSUSE)"), + (re.compile(r"FreeBSD", re.IGNORECASE), "FreeBSD"), + (re.compile(r"NetBSD", re.IGNORECASE), "NetBSD"), + (re.compile(r"OpenBSD", re.IGNORECASE), "OpenBSD"), + (re.compile(r"for_Windows|Windows", re.IGNORECASE), "Windows"), + # Generic Linux when distro is absent but OpenSSH is present + (re.compile(r"OpenSSH", re.IGNORECASE), "Linux"), +] + +# ── HTTP Server header patterns ─────────────────────────────────────────────── +_HTTP_OS_PATTERNS: list[tuple[re.Pattern, str]] = [ + (re.compile(r"DD-WRT", re.IGNORECASE), "Linux (DD-WRT)"), + (re.compile(r"OpenWrt", re.IGNORECASE), "Linux (OpenWrt)"), + (re.compile(r"Tomato", re.IGNORECASE), "Linux (Tomato)"), + (re.compile(r"Synology", re.IGNORECASE), "Linux (Synology DSM)"), + (re.compile(r"QNAP", re.IGNORECASE), "Linux (QNAP QTS)"), + (re.compile(r"Unraid", re.IGNORECASE), "Linux (Unraid)"), + (re.compile(r"FreeNAS|TrueNAS", re.IGNORECASE), "FreeBSD (TrueNAS)"), + (re.compile(r"lighttpd", re.IGNORECASE), "Linux"), + (re.compile(r"mini_httpd", re.IGNORECASE), "Embedded Linux"), + (re.compile(r"Cisco", re.IGNORECASE), "Cisco IOS"), + (re.compile(r"MikroTik", re.IGNORECASE), "MikroTik RouterOS"), + (re.compile(r"Windows", re.IGNORECASE), "Windows"), +] + +# ── Service-name fallbacks ──────────────────────────────────────────────────── +_SERVICE_OS_HINTS: dict[str, str] = { + "msrpc": "Windows", + "netbios-ssn": "Windows", + "microsoft-ds": "Windows", + "ms-wbt-server": "Windows", # RDP +} + + +def _match_patterns(text: str, patterns: list[tuple[re.Pattern, str]]) -> str: + for pattern, label in patterns: + if pattern.search(text): + return label + return "" + + +def infer_os(host: NmapHost) -> str: + """ + Return a best-effort OS label for *host* derived from its port banners. + + Returns empty string when nothing can be inferred. Does not modify + the host object — callers decide whether to apply the result. + """ + for port in host.ports: + # SSH banner is the most reliable source + if port.service_name == "ssh" and port.version_banner: + label = _match_patterns(port.version_banner, _SSH_OS_PATTERNS) + if label: + return label + + for port in host.ports: + # HTTP Server header + if port.service_name in ("http", "http-alt", "https") and port.version_banner: + label = _match_patterns(port.version_banner, _HTTP_OS_PATTERNS) + if label: + return label + + for port in host.ports: + # Service-name hints (Windows RPC/SMB/RDP) + hint = _SERVICE_OS_HINTS.get(port.service_name, "") + if hint: + return hint + + return "" + + +def enrich_os_guesses(hosts: list[NmapHost]) -> None: + """ + Fill in missing os_guess on *hosts* via passive banner inference (in-place). + + Only updates hosts where os_guess is currently empty. + """ + for host in hosts: + if not host.os_guess: + host.os_guess = infer_os(host) diff --git a/backend/tests/test_enrichment.py b/backend/tests/test_enrichment.py new file mode 100644 index 0000000..65d41e6 --- /dev/null +++ b/backend/tests/test_enrichment.py @@ -0,0 +1,213 @@ +""" +test_enrichment.py — Unit tests for dns_lookup and os_inference modules. + +Markers: unit +""" + +from __future__ import annotations + +from unittest.mock import patch + +import pytest +from app.scanner.dns_lookup import resolve_hostnames +from app.scanner.nmap_scan import NmapHost, PortInfo +from app.scanner.os_inference import enrich_os_guesses, infer_os + +# ═══════════════════════════════════════════════════════════════════════════════ +# os_inference — infer_os() +# ═══════════════════════════════════════════════════════════════════════════════ + + +def _host_with_ssh(banner: str) -> NmapHost: + return NmapHost( + ip="192.168.1.1", + ports=[ + PortInfo( + port_number=22, + protocol="tcp", + state="open", + service_name="ssh", + version_banner=banner, + ) + ], + ) + + +def _host_with_http(banner: str) -> NmapHost: + return NmapHost( + ip="192.168.1.1", + ports=[ + PortInfo( + port_number=80, + protocol="tcp", + state="open", + service_name="http", + version_banner=banner, + ) + ], + ) + + +class TestInferOs: + @pytest.mark.unit + def test_ubuntu_ssh_banner(self): + assert infer_os(_host_with_ssh("OpenSSH 8.9p1 Ubuntu-3ubuntu0.6")) == "Linux (Ubuntu)" + + @pytest.mark.unit + def test_debian_ssh_banner(self): + assert infer_os(_host_with_ssh("OpenSSH 8.4p1 Debian-2+deb11u2")) == "Linux (Debian)" + + @pytest.mark.unit + def test_raspbian_ssh_banner(self): + assert infer_os(_host_with_ssh("OpenSSH 8.4p1 Raspbian-5+rpi1")) == "Linux (Raspbian)" + + @pytest.mark.unit + def test_freebsd_ssh_banner(self): + assert infer_os(_host_with_ssh("OpenSSH 7.9 FreeBSD-20200214")) == "FreeBSD" + + @pytest.mark.unit + def test_windows_ssh_banner(self): + assert infer_os(_host_with_ssh("OpenSSH_for_Windows_8.1")) == "Windows" + + @pytest.mark.unit + def test_generic_openssh_fallback(self): + assert infer_os(_host_with_ssh("OpenSSH 9.0 (protocol 2.0)")) == "Linux" + + @pytest.mark.unit + def test_openwrt_http_banner(self): + assert infer_os(_host_with_http("OpenWrt/uhttpd")) == "Linux (OpenWrt)" + + @pytest.mark.unit + def test_synology_http_banner(self): + assert infer_os(_host_with_http("nginx/Synology")) == "Linux (Synology DSM)" + + @pytest.mark.unit + def test_windows_smb_service(self): + host = NmapHost( + ip="192.168.1.1", + ports=[ + PortInfo( + port_number=445, + protocol="tcp", + state="open", + service_name="microsoft-ds", + version_banner="", + ) + ], + ) + assert infer_os(host) == "Windows" + + @pytest.mark.unit + def test_no_ports_returns_empty(self): + assert infer_os(NmapHost(ip="192.168.1.1")) == "" + + @pytest.mark.unit + def test_unrecognised_banner_returns_empty(self): + assert infer_os(_host_with_ssh("some-unknown-server 1.0")) == "" + + @pytest.mark.unit + def test_ssh_takes_priority_over_http(self): + """SSH banner should be used before HTTP when both are present.""" + host = NmapHost( + ip="192.168.1.1", + ports=[ + PortInfo( + port_number=22, + protocol="tcp", + state="open", + service_name="ssh", + version_banner="OpenSSH 8.9p1 Ubuntu-3", + ), + PortInfo( + port_number=80, + protocol="tcp", + state="open", + service_name="http", + version_banner="OpenWrt/uhttpd", + ), + ], + ) + assert infer_os(host) == "Linux (Ubuntu)" + + +class TestEnrichOsGuesses: + @pytest.mark.unit + def test_fills_empty_os_guess(self): + host = _host_with_ssh("OpenSSH 8.9p1 Ubuntu-3") + host.os_guess = "" + enrich_os_guesses([host]) + assert host.os_guess == "Linux (Ubuntu)" + + @pytest.mark.unit + def test_does_not_overwrite_existing_os_guess(self): + host = _host_with_ssh("OpenSSH 8.9p1 Ubuntu-3") + host.os_guess = "Linux 5.4" # already set (e.g. from nmap -O in future) + enrich_os_guesses([host]) + assert host.os_guess == "Linux 5.4" + + @pytest.mark.unit + def test_mixed_list(self): + h1 = _host_with_ssh("OpenSSH 8.4p1 Debian-2") + h1.os_guess = "" + h2 = NmapHost(ip="192.168.1.2") # no ports, no inference possible + h2.os_guess = "" + enrich_os_guesses([h1, h2]) + assert h1.os_guess == "Linux (Debian)" + assert h2.os_guess == "" + + +# ═══════════════════════════════════════════════════════════════════════════════ +# dns_lookup — resolve_hostnames() +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestResolveHostnames: + @pytest.mark.unit + def test_fills_missing_hostname_via_rdns(self): + host = NmapHost(ip="192.168.1.1", hostname="") + with patch( + "app.scanner.dns_lookup.socket.gethostbyaddr", + return_value=("router.local", [], ["192.168.1.1"]), + ): + resolve_hostnames([host]) + assert host.hostname == "router.local" + + @pytest.mark.unit + def test_does_not_overwrite_existing_hostname(self): + host = NmapHost(ip="192.168.1.1", hostname="already-named.local") + with patch("app.scanner.dns_lookup.socket.gethostbyaddr") as mock_dns: + resolve_hostnames([host]) + mock_dns.assert_not_called() + assert host.hostname == "already-named.local" + + @pytest.mark.unit + def test_gracefully_handles_no_ptr_record(self): + host = NmapHost(ip="192.168.1.1", hostname="") + with patch("app.scanner.dns_lookup.socket.gethostbyaddr", side_effect=OSError("no PTR")): + resolve_hostnames([host]) + assert host.hostname == "" + + @pytest.mark.unit + def test_gracefully_handles_timeout(self): + host = NmapHost(ip="192.168.1.1", hostname="") + with patch("app.scanner.dns_lookup.socket.gethostbyaddr", side_effect=TimeoutError): + resolve_hostnames([host]) + assert host.hostname == "" + + @pytest.mark.unit + def test_only_queries_hosts_without_hostname(self): + h1 = NmapHost(ip="192.168.1.1", hostname="existing.local") + h2 = NmapHost(ip="192.168.1.2", hostname="") + with patch( + "app.scanner.dns_lookup.socket.gethostbyaddr", + return_value=("resolved.local", [], []), + ) as mock_dns: + resolve_hostnames([h1, h2]) + mock_dns.assert_called_once_with("192.168.1.2") + assert h2.hostname == "resolved.local" + + @pytest.mark.unit + def test_empty_list_is_noop(self): + with patch("app.scanner.dns_lookup.socket.gethostbyaddr") as mock_dns: + resolve_hostnames([]) + mock_dns.assert_not_called()