Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions backend/app/scanner/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)

Expand Down Expand Up @@ -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]

Expand Down
77 changes: 77 additions & 0 deletions backend/app/scanner/dns_lookup.py
Original file line number Diff line number Diff line change
@@ -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))
109 changes: 109 additions & 0 deletions backend/app/scanner/os_inference.py
Original file line number Diff line number Diff line change
@@ -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)
Loading