admin / Synapse-NetscanXi

"""
Scan manager: a single background worker draining a job queue, so manual and
scheduled scans coexist without colliding. Tracks live progress + supports
cancellation. Persists results, computes diffs vs. previous scan, enriches
vulns, and fires alerts.
"""

import queue
import threading
import time

from . import db, diff, enrich, alerts, passive, compliance, docker_scan
from .parser import parse_nmap_xml
from .scanner import run_nmap, ScanProcess, PROFILES, DEFAULT_PROFILE


class ScanManager:
    def __init__(self, db_path=db.DEFAULT_DB_PATH, use_nvd=False):
        self.db_path = db_path
        self.use_nvd = use_nvd
        self._q = queue.Queue()
        self._lock = threading.Lock()
        self._live = {
            "active": False, "scan_id": None, "target": None, "profile": None,
            "status": "idle", "phase": "", "progress": 0.0,
            "queued": 0, "triggered_by": None,
            "started_at": None,      # epoch seconds when current scan began
            "elapsed": 0.0,          # seconds elapsed (live)
            "options": None,
            "current_ip": None,      # v8: host nmap is actively scanning
            "tenant": "default",     # v8: owning tenant of the running scan
            "scan_type": "active",   # v9: active | passive
        }
        self._handle = None
        self._worker = threading.Thread(target=self._run_loop, daemon=True)
        self._worker.start()

    # -- public API ---------------------------------------------------------
    def enqueue(self, target, profile=DEFAULT_PROFILE, triggered_by="manual",
                options=None, tenant="default", scan_type="active"):
        if profile not in PROFILES:
            profile = DEFAULT_PROFILE
        self._q.put((target, profile, triggered_by, options, tenant, scan_type))
        with self._lock:
            self._live["queued"] = self._q.qsize()
        return True

    def cancel_current(self):
        with self._lock:
            handle = self._handle
            active = self._live["active"]
        if active and handle:
            handle.cancel()
            return True
        return False

    def status(self):
        with self._lock:
            s = dict(self._live)
        s["queued"] = self._q.qsize()
        # Live elapsed time while a scan is active.
        if s.get("active") and s.get("started_at"):
            s["elapsed"] = round(time.time() - s["started_at"], 1)
        return s

    # -- worker -------------------------------------------------------------
    def _set(self, **kw):
        with self._lock:
            self._live.update(kw)

    def _progress(self, pct, phase, ip=None):
        with self._lock:
            self._live["progress"] = max(0.0, min(100.0, round(pct, 1)))
            self._live["phase"] = phase
            if ip:
                self._live["current_ip"] = ip

    def _run_loop(self):
        while True:
            job = self._q.get()
            # Backward/forward compatible unpack (options/tenant may be absent).
            target, profile, triggered_by = job[0], job[1], job[2]
            options = job[3] if len(job) > 3 else None
            tenant = job[4] if len(job) > 4 else "default"
            scan_type = job[5] if len(job) > 5 else "active"
            self._execute(target, profile, triggered_by, options, tenant,
                          scan_type)
            self._q.task_done()

    def _execute(self, target, profile, triggered_by, options=None,
                 tenant="default", scan_type="active"):
        scan_type = "passive" if scan_type == "passive" else "active"
        scan_id = db.create_scan(target, profile, triggered_by, tenant=tenant,
                                 scan_type=scan_type, db_path=self.db_path)
        handle = ScanProcess()
        started_at = time.time()
        self._set(active=True, scan_id=scan_id, target=target, profile=profile,
                  status="running", phase="Starting", progress=0.0,
                  triggered_by=triggered_by, started_at=started_at, elapsed=0.0,
                  options=options, current_ip=None, tenant=tenant,
                  scan_type=scan_type)
        with self._lock:
            self._handle = handle

        try:
            if scan_type == "passive":
                # Passive: no probe traffic; harvest from ARP + announcements.
                # The user may set how long (0-10 min) to listen for self-
                # announcements; fall back to the module default when unset.
                listen_seconds = passive.LISTEN_SECONDS
                if options and options.get("passive_seconds") is not None:
                    try:
                        listen_seconds = max(0, int(options["passive_seconds"]))
                    except (TypeError, ValueError):
                        pass
                hosts = passive.passive_scan(progress_cb=self._progress,
                                             listen_seconds=listen_seconds)
            else:
                xml = run_nmap(target, profile, progress_cb=self._progress,
                               handle=handle, options=options)
                if handle.cancelled:
                    db.finish_scan(scan_id, "cancelled", [],
                                   error="cancelled by user", db_path=self.db_path)
                    self._set(active=False, status="cancelled", phase="Cancelled")
                    return
                self._progress(99.0, "Parsing & enriching")
                hosts = parse_nmap_xml(xml)
                # vuln enrichment if profile OR user options ran vuln scripts
                opt_vuln = bool(options.get("vulns")) if options else False
                wants_vuln = PROFILES[profile].get("vuln") or opt_vuln
                enrich.enrich_hosts(hosts, use_nvd=self.use_nvd and wants_vuln)
                # v10r1: deepen Docker hosts with authenticated API inventory,
                # image CVE scanning and a CIS Docker audit. Best-effort and
                # gated by the docker_deep option (on by default).
                if not options or options.get("docker_deep", True):
                    self._progress(99.5, "Scanning Docker containers")
                    try:
                        docker_scan.enrich_hosts(hosts)
                        # Container CVEs feed KEV/EPSS enrichment + risk scoring.
                        enrich.enrich_container_vulns(hosts, use_nvd=self.use_nvd and wants_vuln)
                    except Exception:
                        pass

            # v9: bind every host to its persistent 8-char asset ID (keyed on
            # MAC) and apply any sticky user type-override.
            self._assign_asset_ids(hosts, tenant)

            # Diff against previous done scan BEFORE saving this one.
            prev = db.previous_done_scan(scan_id, tenant=tenant, db_path=self.db_path)
            changes = []
            if prev:
                old_hosts = db.get_hosts(prev["id"], db_path=self.db_path)
                changes = diff.diff_scans(old_hosts, hosts)

            db.finish_scan(scan_id, "done", hosts, db_path=self.db_path)
            db.record_changes(scan_id, changes, db_path=self.db_path)

            # v9.1: capture a risk + compliance posture snapshot for trend graphs.
            try:
                self._record_snapshot(scan_id, tenant, hosts)
            except Exception:
                pass

            # v9.1: apply the data-retention policy automatically if enabled.
            try:
                self._apply_retention()
            except Exception:
                pass

            if changes:
                try:
                    alerts.dispatch(target, changes)
                except Exception:
                    pass

            self._set(active=False, status="done", phase="Complete", progress=100.0,
                      elapsed=round(time.time() - started_at, 1))
        except Exception as exc:
            db.finish_scan(scan_id, "error", [], error=str(exc), db_path=self.db_path)
            self._set(active=False, status="error", phase="Error",
                      elapsed=round(time.time() - started_at, 1))
        finally:
            with self._lock:
                self._handle = None

    def _record_snapshot(self, scan_id, tenant, hosts):
        """v9.1: persist an aggregate risk/compliance datapoint per scan so the
        UI can plot 30/90/365-day trends."""
        risks = [h.get("risk") for h in hosts if h.get("risk") is not None]
        # Network risk score = the worst-case posture, blended with the mean so
        # one critical host dominates but breadth still matters.
        if risks:
            mx = max(risks)
            avg = sum(risks) / len(risks)
            risk_score = round(0.6 * mx + 0.4 * avg, 1)
        else:
            risk_score = 0.0
        comp = compliance.summarize(hosts)
        total = comp.get("hosts_total", 0) or 0
        at_risk = comp.get("hosts_at_risk", 0) or 0
        compliance_pct = round(100.0 * (total - at_risk) / total, 1) if total else 100.0
        kev = sum(1 for h in hosts for v in h.get("vulns", []) if v.get("kev"))
        high_risk = sum(1 for h in hosts if (h.get("risk") or 0) >= 70)
        vulns = sum(h.get("vuln_count", 0) for h in hosts)
        db.add_risk_snapshot(scan_id, tenant, risk_score, compliance_pct,
                             hosts=len(hosts), vulns=vulns, kev=kev,
                             high_risk=high_risk, db_path=self.db_path)

    def _apply_retention(self):
        """v9.1: honour an auto-apply data-retention policy after each scan."""
        pol = db.get_setting("retention_policy", None, db_path=self.db_path)
        if not pol or not pol.get("auto_apply"):
            return
        age = pol.get("max_age_days") or None
        cap = pol.get("max_scans") or None
        if age or cap:
            db.purge_old_scans(max_age_days=age, max_scans=cap,
                               db_path=self.db_path)

    def _assign_asset_ids(self, hosts, tenant):
        """v9: attach a stable 8-char asset_id (bound to MAC) to each host and
        apply any user-corrected device type. New vulnerabilities are therefore
        never duplicated for the same physical device across scans."""
        for h in hosts:
            try:
                rec = db.resolve_asset(h.get("mac"), h.get("ip"), tenant=tenant,
                                       db_path=self.db_path)
            except Exception:
                continue
            h["asset_id"] = rec["asset_id"]
            h["first_seen"] = rec.get("first_seen")
            if rec.get("type_override"):
                h["device_type"] = rec["type_override"]
                h["type_override"] = rec["type_override"]