admin / Synapse-NetscanXi

"""
Passive scanning (v9).

Active scanning sends probes (SYN/version/OS/NSE) at targets. PASSIVE scanning
sends *no* probe traffic to discovered hosts: it only reads what is already
observable locally and listens for self-announcements on the local segment.

Sources used (all zero-probe):
  * The host ARP/neighbour cache (devices that have already talked on the LAN).
  * A short, read-only listen for multicast self-announcements: mDNS
    (224.0.0.251:5353), SSDP/UPnP (239.255.255.250:1900) and NetBIOS/LLMNR
    name chatter. We never transmit a query - we only receive what is broadcast.
  * Offline OUI vendor lookup on the MAC prefix.

Output is intentionally limited (see the User Guide / passive-scan spec):
no open ports, no service versions, no OS accuracy, no CVEs/vulnerabilities,
no TLS/AD/Docker/credential findings - those all require active probing.

The result is a list of "host" dicts shaped like the active parser's output so
the rest of the pipeline (asset IDs, dashboard, export) just works, but with the
active-only fields left empty and a passive flag set.
"""

import re
import socket
import struct
import subprocess
import time

# How long to passively listen for self-announcements (seconds).
LISTEN_SECONDS = 8

_MAC_RE = re.compile(r"([0-9a-fA-F]{2}(?:[:-][0-9a-fA-F]{2}){5})")
_IP_RE = re.compile(r"(\d{1,3}(?:\.\d{1,3}){3})")

# Minimal embedded OUI map (offline). Extend as needed; unknown -> "".
# Keys are the upper-cased first 3 MAC octets (OUI), no separators.
OUI_VENDORS = {
    "001A11": "Google", "3C5AB4": "Google", "F4F5E8": "Google",
    "F0D5BF": "Apple", "A4C361": "Apple", "ACBC32": "Apple", "DCA904": "Apple",
    "B827EB": "Raspberry Pi Foundation", "DCA632": "Raspberry Pi Trading",
    "E45F01": "Raspberry Pi Trading",
    "001132": "Synology", "0011D8": "Asustek", "F832E4": "Asustek",
    "00156D": "Ubiquiti", "245A4C": "Ubiquiti", "FCECDA": "Ubiquiti",
    "0024A5": "Buffalo", "001E2A": "Netgear", "A040A0": "Netgear",
    "EC086B": "TP-Link", "50C7BF": "TP-Link", "C46E1F": "TP-Link",
    "001583": "Hikvision", "44A642": "Axis", "00408C": "Axis",
    "001801": "Actiontec", "0017C8": "Kyocera", "0021B7": "Lexmark",
    "001B78": "HP", "3C52A1": "Hon Hai (Foxconn)", "001E8F": "Canon",
    "00000C": "Cisco", "002155": "Cisco", "F09FC2": "Ubiquiti",
    "FCFBFB": "Cisco", "B4FBE4": "Ubiquiti",
    "002608": "Apple", "9027E4": "Apple", "8863DF": "Apple",
    "001D0F": "TP-Link", "D8478F": "MitraStar", "60019B": "Sagemcom",
    "ECB5FA": "Philips Lighting", "001788": "Philips Lighting (Hue)",
    "B0C554": "D-Link", "1CBDB9": "D-Link", "000FB5": "Netgear",
    "5CCF7F": "Espressif (ESP)", "240AC4": "Espressif (ESP)", "A020A6": "Espressif",
    "D0737F": "Murata (IoT)", "AC233F": "Shenzhen Minew (BLE/IoT)",
    "001132AA": "Synology",
}


def vendor_for_mac(mac):
    if not mac:
        return ""
    oui = mac.replace(":", "").replace("-", "").upper()[:6]
    return OUI_VENDORS.get(oui, "")


# ---------------------------------------------------------------------------
# ARP / neighbour cache (no traffic sent)
# ---------------------------------------------------------------------------
def _read_arp_cache():
    """Return {ip: mac} from the OS neighbour cache. Cross-platform best effort."""
    out = {}
    cmds = []
    import os
    if os.name == "nt":
        cmds = [["arp", "-a"]]
    else:
        cmds = [["ip", "neigh"], ["arp", "-an"]]
    for cmd in cmds:
        try:
            res = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
        except (OSError, subprocess.TimeoutExpired):
            continue
        text = (res.stdout or "") + "\n" + (res.stderr or "")
        for line in text.splitlines():
            ipm = _IP_RE.search(line)
            macm = _MAC_RE.search(line)
            if ipm and macm:
                ip = ipm.group(1)
                mac = macm.group(1).upper().replace("-", ":")
                if mac in ("00:00:00:00:00:00", "FF:FF:FF:FF:FF:FF"):
                    continue
                out.setdefault(ip, mac)
        if out:
            break
    return out


# ---------------------------------------------------------------------------
# Passive multicast listen (receive-only; we never send a query)
# ---------------------------------------------------------------------------
def _listen_announcements(seconds=LISTEN_SECONDS):
    """Listen (receive-only) for mDNS + SSDP self-announcements.

    Returns {ip: {"hostname": str, "services": set(), "os_hint": str}}.
    Best-effort: any failure (no permission, no multicast) yields {}.
    """
    found = {}
    socks = []
    try:
        # mDNS 5353
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            s.bind(("", 5353))
            mreq = struct.pack("4sl", socket.inet_aton("224.0.0.251"), socket.INADDR_ANY)
            s.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
            s.setblocking(False)
            socks.append(("mdns", s))
        except OSError:
            pass
        # SSDP 1900
        try:
            s2 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
            s2.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            s2.bind(("", 1900))
            mreq2 = struct.pack("4sl", socket.inet_aton("239.255.255.250"), socket.INADDR_ANY)
            s2.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq2)
            s2.setblocking(False)
            socks.append(("ssdp", s2))
        except OSError:
            pass

        if not socks:
            return found

        import select
        deadline = time.time() + seconds
        while time.time() < deadline:
            timeout = max(0.2, deadline - time.time())
            rlist, _, _ = select.select([s for _, s in socks], [], [], timeout)
            for sk in rlist:
                kind = next((k for k, s in socks if s is sk), "")
                try:
                    data, addr = sk.recvfrom(65535)
                except OSError:
                    continue
                ip = addr[0]
                rec = found.setdefault(ip, {"hostname": "", "services": set(),
                                            "os_hint": ""})
                if kind == "mdns":
                    _parse_mdns(data, rec)
                elif kind == "ssdp":
                    _parse_ssdp(data, rec)
    finally:
        for _, s in socks:
            try:
                s.close()
            except OSError:
                pass
    # normalise sets -> sorted lists
    for ip, rec in found.items():
        rec["services"] = sorted(rec["services"])
    return found


def _parse_mdns(data, rec):
    """Very light mDNS parse: pull readable ._tcp/._udp service labels + names."""
    try:
        text = data.decode("latin-1", "ignore")
    except Exception:
        return
    for m in re.findall(r"([A-Za-z0-9_\-]+\._(?:tcp|udp))", text):
        rec["services"].add(m)
    # device/host name often appears as something.local
    nm = re.search(r"([A-Za-z0-9\-]+)\.local", text)
    if nm and not rec["hostname"]:
        rec["hostname"] = nm.group(1)
    low = text.lower()
    if "_airplay" in low or "_raop" in low or "apple" in low:
        rec["os_hint"] = rec["os_hint"] or "Apple/iOS (passive)"
    elif "_ipp" in low or "_printer" in low:
        rec["os_hint"] = rec["os_hint"] or "printer (passive)"


def _parse_ssdp(data, rec):
    try:
        text = data.decode("latin-1", "ignore")
    except Exception:
        return
    for line in text.splitlines():
        low = line.lower()
        if low.startswith("server:"):
            rec["os_hint"] = rec["os_hint"] or line.split(":", 1)[1].strip()[:60]
        if low.startswith("st:") or low.startswith("nt:"):
            val = line.split(":", 1)[1].strip()
            if val:
                rec["services"].add(val[:48])
    rec["services"].add("ssdp/upnp")


def _device_type_from_passive(vendor, services, os_hint):
    """Low-confidence device-type guess from passive signals only."""
    v = (vendor or "").lower()
    svc = " ".join(services).lower()
    oh = (os_hint or "").lower()
    if "raspberry" in v:
        return "raspberry pi"
    if any(k in v for k in ("hikvision", "axis", "dahua")):
        return "camera"
    if any(k in v for k in ("hp", "canon", "epson", "brother", "lexmark", "kyocera")) \
            or "_ipp" in svc or "_printer" in svc or "printer" in oh:
        return "printer"
    if any(k in v for k in ("ubiquiti", "tp-link", "netgear", "cisco", "asus", "d-link")):
        return "router"
    if "synology" in v or "qnap" in v:
        return "nas"
    if "_airplay" in svc or "_raop" in svc or "apple" in v or "ios" in oh:
        return "phone"
    if "espressif" in v or "shelly" in v or "tuya" in v or "philips" in v:
        return "iot"
    if "ssdp/upnp" in svc:
        return "media/iot"
    return "unknown"


def passive_scan(progress_cb=None, listen_seconds=LISTEN_SECONDS):
    """Run a passive harvest and return host dicts (active-only fields empty).

    No probe traffic is sent to any discovered host.
    """
    def _emit(pct, phase, ip=None):
        if progress_cb:
            try:
                progress_cb(pct, phase, ip)
            except TypeError:
                progress_cb(pct, phase)

    _emit(5.0, "Passive: reading ARP / neighbour cache")
    arp = _read_arp_cache()

    _emit(20.0, "Passive: listening for self-announcements")
    announce = _listen_announcements(listen_seconds)
    _emit(85.0, "Passive: correlating observations")

    # Union of IPs seen via ARP and via announcements.
    ips = set(arp) | set(announce)
    hosts = []
    for ip in sorted(ips, key=lambda x: tuple(int(o) for o in x.split(".")) if _IP_RE.match(x) else (0,)):
        mac = arp.get(ip, "")
        ann = announce.get(ip, {})
        vendor = vendor_for_mac(mac)
        services = ann.get("services", [])
        os_hint = ann.get("os_hint", "")
        hostname = ann.get("hostname", "")
        dtype = _device_type_from_passive(vendor, services, os_hint)
        h = {
            "ip": ip,
            "hostname": hostname or "-",
            "mac": mac or "-",
            "vendor": vendor or "-",
            "os": os_hint or "unknown",
            "os_accuracy": "",
            "ports": [],                 # passive: no port scan
            "port_details": [],
            "software": [],
            "software_count": 0,
            "vulns": [],                 # passive: no vuln detection
            "vuln_count": 0,
            "host_scripts": [],
            "docker": {"is_docker": False, "confidence": "none",
                       "indicators": [], "api_exposed": False,
                       "version": "", "ports": []},
            "device_type": dtype,
            "risk": None,                # n/a for purely passive assets
            "certificates": [],
            "ad": {"is_dc": False, "findings": []},
            "ad_findings": [],
            "weak_auth": [],
            "compliance": {"status": "n/a"},
            # v9 passive-specific
            "passive": True,
            "advertised_services": services,
            "passive_sources": ([] + (["arp"] if mac else [])
                                + (["mdns/ssdp"] if ann else [])),
        }
        hosts.append(h)
    _emit(99.0, "Passive: done")
    return hosts