""" Nessus → VulnCheck sync service. Imports scan findings from Tenable Nessus and merges them onto existing Wazuh-sourced vulnerabilities using `(cve_id, asset_id)` as the dedup key. Non-CVE Nessus findings (EOL, compliance, weak ciphers, etc.) are stored under a pseudo-CVE id `NESSUS-PLUGIN-{plugin_id}` so they show up in the same Vulns list and inherit the existing workflow. Asset matching cascade: 1. assets.nessus_host_uuid (pinned after first match) 2. assets.hostname (case-insensitive) 3. assets.ip_address (string equality) 4. If no match and config.auto_create_assets is True → create new asset Backfill rule: When a Nessus scan finishes, any Vuln that previously listed `nessus` in `sources` but is missing from this scan run has `nessus` removed. If `sources` becomes empty AND status == open, mark as patched. """ from __future__ import annotations import json import logging from datetime import datetime from typing import List, Optional, Tuple from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.integrations.nessus_client import NessusClient from app.models.asset import Asset, AssetSource, AssetStatus from app.models.scan import Scan, ScanStatus, ScanType from app.models.setting import Setting from app.models.vulnerability import ( Vulnerability, VulnerabilitySeverity, VulnerabilityStatus, ) logger = logging.getLogger(__name__) SETTING_KEY = "nessus_config" SOURCE_NAME = "nessus" # --------------------------------------------------------------- # Config loader # --------------------------------------------------------------- def load_nessus_config(db: Session) -> Optional[dict]: from app.auth.setting_crypto import read_setting_value raw = read_setting_value(db, SETTING_KEY) if not raw: return None try: return json.loads(raw) except (json.JSONDecodeError, TypeError): logger.warning("nessus_config is not valid JSON") return None def _build_client(config: dict) -> NessusClient: return NessusClient( base_url=config["base_url"], access_key=config["access_key"], secret_key=config["secret_key"], verify_ssl=bool(config.get("verify_ssl", True)), ) # --------------------------------------------------------------- # Asset matching # --------------------------------------------------------------- def _find_or_create_asset( db: Session, host_info: dict, auto_create: bool, ) -> Tuple[Optional[Asset], str]: """ Returns (asset, match_method) where match_method is one of 'uuid', 'hostname', 'ip', 'created', 'skipped'. """ host_uuid = host_info.get("host_uuid") or host_info.get("uuid") hostname = (host_info.get("hostname") or host_info.get("host-fqdn") or "").strip() ip = ( host_info.get("host_ip") or host_info.get("host-ip") or host_info.get("ip") or "" ).strip() # 1) Pinned UUID if host_uuid: a = db.query(Asset).filter(Asset.nessus_host_uuid == host_uuid).first() if a: return a, "uuid" # 2) Hostname (case-insensitive) — try full + short forms so a # Nessus FQDN like 'host01.umgebung.local' matches a Wazuh # asset registered as just 'host01' (Wazuh-Agent name is # typically the short hostname). Tester reported duplicate # assets created from FQDN/short mismatches. short = hostname.split(".")[0] if hostname else "" candidates = [] if hostname: candidates.append(hostname) if short and short != hostname: candidates.append(short) for candidate in candidates: a = ( db.query(Asset) .filter(Asset.hostname.ilike(candidate)) .first() ) if a: if host_uuid and not a.nessus_host_uuid: a.nessus_host_uuid = host_uuid return a, "hostname" # 2b) Reverse — asset stored as FQDN, Nessus reports short. Find # assets whose hostname starts with 'short.' so we still # consolidate the right way around. if short: a = ( db.query(Asset) .filter(Asset.hostname.ilike(f"{short}.%")) .first() ) if a: if host_uuid and not a.nessus_host_uuid: a.nessus_host_uuid = host_uuid return a, "hostname-fqdn-prefix" # 3) IP if ip: a = db.query(Asset).filter(Asset.ip_address == ip).first() if a: if host_uuid and not a.nessus_host_uuid: a.nessus_host_uuid = host_uuid return a, "ip" # 4) Auto-create if configured. Store the SHORT hostname so a # later Wazuh sync of the same box matches via candidates[]. if auto_create and (hostname or ip): new_hostname = short or hostname or ip a = Asset( hostname=new_hostname, ip_address=ip or None, nessus_host_uuid=host_uuid, source=AssetSource.MANUAL, # not wazuh-sourced status=AssetStatus.ACTIVE, ) db.add(a) db.flush() logger.info( "Nessus sync: auto-created asset %s (from FQDN '%s', ip %s)", a.hostname, hostname, a.ip_address, ) return a, "created" return None, "skipped" # --------------------------------------------------------------- # Severity helpers # --------------------------------------------------------------- _SEV_RANK = { VulnerabilitySeverity.none: 0, VulnerabilitySeverity.low: 1, VulnerabilitySeverity.medium: 2, VulnerabilitySeverity.high: 3, VulnerabilitySeverity.critical: 4, } def _max_severity(a: VulnerabilitySeverity, b: VulnerabilitySeverity) -> VulnerabilitySeverity: return a if _SEV_RANK.get(a, 0) >= _SEV_RANK.get(b, 0) else b # --------------------------------------------------------------- # Core sync # --------------------------------------------------------------- def run_nessus_sync( db: Session, scan_ids: Optional[List[int]] = None, ) -> dict: """ Pull findings from Nessus for the requested scans (or all configured default_scan_ids) and merge them into VulnCheck. Returns a stats dict suitable for logging + API response. """ config = load_nessus_config(db) if not config: raise RuntimeError("nessus_config not set in settings") target_scan_ids: List[int] = ( scan_ids or list(config.get("default_scan_ids") or []) ) auto_create = bool(config.get("auto_create_assets", False)) stats = { "scans_processed": 0, "hosts_synced": 0, "hosts_skipped": 0, "vulns_created": 0, "vulns_merged": 0, "vulns_unchanged": 0, "vulns_marked_patched": 0, "scores_overridden": 0, "non_cve_skipped": 0, "unmatched_hosts": [], "errors": [], } newly_created_vuln_ids: List[int] = [] with _build_client(config) as client: # Auto-discover scans if no explicit IDs and no defaults if not target_scan_ids: scans = client.list_scans() target_scan_ids = [int(s["id"]) for s in scans if s.get("id")] for scan_id in target_scan_ids: try: stats["scans_processed"] += 1 scan_detail = client.get_scan(scan_id) except Exception as e: stats["errors"].append(f"scan {scan_id}: {e}") logger.warning("Nessus sync: scan %s failed: %s", scan_id, e) continue hosts = scan_detail.get("hosts") or [] logger.info("Nessus sync: scan %s has %d hosts", scan_id, len(hosts)) for host_entry in hosts: host_id = host_entry.get("host_id") if host_id is None: continue # Pull rich host detail try: host = client.get_host(scan_id, host_id) except Exception as e: stats["errors"].append(f"scan {scan_id} host {host_id}: {e}") continue host_info = host.get("info") or {} # `info` carries hostname/host-fqdn/host-ip; merge with summary entry merged_info = {**host_entry, **host_info} asset, match_method = _find_or_create_asset(db, merged_info, auto_create) if asset is None: stats["hosts_skipped"] += 1 stats["unmatched_hosts"].append({ "hostname": merged_info.get("hostname") or merged_info.get("host-fqdn"), "ip": merged_info.get("host_ip") or merged_info.get("host-ip"), "scan_id": scan_id, }) continue stats["hosts_synced"] += 1 # OS Identification (Nessus plugin 11936 + host info fields) # Only fills when the asset row has nothing — Wazuh-sourced # assets already carry structured agent OS data and we do # not want to clobber that with Nessus's longer prose form. if not asset.operating_system: os_label = ( merged_info.get("operating-system") or merged_info.get("operating_system") or merged_info.get("os") ) if isinstance(os_label, list): # Nessus sometimes returns multiple guesses — take the first os_label = next((x for x in os_label if x), None) if os_label: os_label = str(os_label).strip().splitlines()[0][:255] if os_label: asset.operating_system = os_label # Record one Scan row per host so Nessus runs appear in # the Scan-Jobs history (previously only Wazuh autoscan # populated this table). Status updated to COMPLETED / # FAILED at the end of this host's loop iteration. scan_row = Scan( asset_id=asset.id, scan_type=ScanType.NESSUS, status=ScanStatus.RUNNING, started_at=datetime.now(), ) db.add(scan_row) host_vulns_found = 0 # Track cves we observe on THIS asset in THIS sync run. # Used for backfill (drop nessus source on disappeared findings). seen_cves_for_asset: set = set() for vuln_entry in host.get("vulnerabilities") or []: plugin_id = vuln_entry.get("plugin_id") severity_int = vuln_entry.get("severity") or 0 severity_label = NessusClient.plugin_severity_to_label(severity_int) try: severity = VulnerabilitySeverity(severity_label) except ValueError: severity = VulnerabilitySeverity.none # Pull detailed plugin payload for CVE list + CVSS try: plugin_payload = client.get_plugin_output(scan_id, host_id, plugin_id) except Exception as e: logger.debug("Nessus plugin %s fetch failed: %s", plugin_id, e) plugin_payload = {} cve_list = NessusClient.extract_cves(plugin_payload) if plugin_payload else [] cvss = NessusClient.plugin_cvss(plugin_payload) if plugin_payload else None # Extra enrichment from the Nessus plugin — Tenable VPR, # exploit availability/maturity, prose description and # remediation. All optional; missing fields stay None. vpr_score = NessusClient.plugin_vpr_score(plugin_payload) if plugin_payload else None exploit_avail = NessusClient.plugin_exploit_available(plugin_payload) if plugin_payload else None exploit_mat = NessusClient.plugin_exploit_maturity(plugin_payload) if plugin_payload else None fixed_version = NessusClient.plugin_fixed_version(plugin_payload) if plugin_payload else None n_description = NessusClient.plugin_description(plugin_payload) if plugin_payload else None n_solution = NessusClient.plugin_solution(plugin_payload) if plugin_payload else None n_see_also = NessusClient.plugin_see_also(plugin_payload) if plugin_payload else [] # Compose a description with the solution appended — Nessus # provides both as separate fields, our schema has one text # column, so we glue them together for the detail view. full_description = None if n_description or n_solution: parts = [] if n_description: parts.append(n_description.strip()) if n_solution: parts.append("\n\nSolution:\n" + n_solution.strip()) full_description = "".join(parts) if not cve_list: # Skip findings without a real CVE (EOL / Compliance / Cipher / # informational plugins). Per product decision we only track # CVE-tagged vulns in VulnCheck — pseudo-CVE imports were # noisy and inflated the count. stats["non_cve_skipped"] += 1 continue plugin_name = vuln_entry.get("plugin_name") or "" # Catch-all Nessus plugins like "Patch Report" emit the # same generic title + description for every CVE they # touch. A CVE-specific plugin (e.g. "Apache Log4j # 2.0-alpha1 < 2.25.4 ... (CVE-2026-34480)") is far # better. Heuristic: the specific plugin's name contains # the CVE-ID. We use this to decide whether to overwrite # an existing title/description/package_name when merging. def _is_specific(name: Optional[str], cve_id: str) -> bool: return bool(name) and (cve_id.upper() in name.upper()) for cve_id in cve_list: seen_cves_for_asset.add(cve_id) new_is_specific = _is_specific(plugin_name, cve_id) existing = ( db.query(Vulnerability) .filter( Vulnerability.cve_id == cve_id, Vulnerability.asset_id == asset.id, ) .first() ) if existing: changed = existing.add_source(SOURCE_NAME) # Nessus wins over Wazuh on CVSS/severity, EXCEPT when # any authoritative per-CVE source has already corrected # the row via the override service. Vulnrichment / NVD / # cvelistV5 are all per-CVE feeds with the real score; # Nessus plugin-level CVSS is per-plugin and can bundle # multiple CVEs (often overstating severity). vulnrichment_locked = ( getattr(existing, "exploitation_source", None) in ("vulnrichment", "nvd", "cvelistv5") ) # Skip override only when Nessus has no CVSS at all # (would wipe a good Wazuh value) or when Vulnrichment # already pinned the authoritative score. if ( cvss is not None and existing.cvss_score != cvss and not vulnrichment_locked ): existing.cvss_score = cvss changed = True stats["scores_overridden"] += 1 # Severity: trust Nessus when it provides one (not "none"), # unless Vulnrichment has locked it via the override service. if vulnrichment_locked: pass # keep Vulnrichment severity elif severity and severity != VulnerabilitySeverity.none: if existing.severity != severity: existing.severity = severity changed = True else: new_sev = _max_severity(existing.severity, severity) if new_sev != existing.severity: existing.severity = new_sev changed = True # plugin_id: prefer the CVE-specific plugin's ID # (e.g. log4j scanner) over a generic catch-all # like "Patch Report" once we see a better one. existing_is_specific = _is_specific(existing.title, cve_id) should_replace_plugin_meta = ( new_is_specific and not existing_is_specific ) if plugin_id and ( existing.nessus_plugin_id != str(plugin_id) and (not existing.nessus_plugin_id or should_replace_plugin_meta) ): existing.nessus_plugin_id = str(plugin_id) changed = True # Title / package_name / description: write when # empty, OR when a CVE-specific plugin overrides # a generic catch-all that landed first. if plugin_name and ( not existing.package_name or should_replace_plugin_meta ): existing.package_name = plugin_name[:255] changed = True if plugin_name and ( not existing.title or should_replace_plugin_meta ): existing.title = plugin_name[:500] changed = True # Tenable VPR: always update to the latest score from # this scan — Tenable recomputes it as threat data # changes, so newer wins. if vpr_score is not None and existing.nessus_vpr_score != vpr_score: existing.nessus_vpr_score = vpr_score changed = True # exploit_available: only escalate (False → True), # don't clobber a Wazuh-confirmed True back to False. if exploit_avail is True and not existing.exploit_available: existing.exploit_available = True changed = True # exploit_maturity: fill if empty; Nessus's values # (Unproven / PoC / Functional / High) match ours. if exploit_mat and not existing.exploit_maturity: existing.exploit_maturity = exploit_mat[:50] # fixed_version: fill if empty so the # PATCH AVAILABLE badge can light up on # rows Wazuh saw first without a fix. if fixed_version and not existing.fixed_version: existing.fixed_version = fixed_version changed = True changed = True # description: write when empty, OR when a CVE- # specific plugin supplants a catch-all's generic # "missing patches" text. Do not clobber a hand- # written description (we cannot detect that, so # only replace if the existing description came # from a non-specific plugin). if full_description and ( not existing.description or should_replace_plugin_meta ): existing.description = full_description changed = True # references: append Nessus see_also URLs if the # column is empty. if n_see_also and not existing.references: existing.references = json.dumps(n_see_also) changed = True # If previously marked patched but Nessus sees it again → reopen if existing.status == VulnerabilityStatus.patched: existing.status = VulnerabilityStatus.open existing.patched_at = None changed = True if changed: existing.refresh_scores() stats["vulns_merged"] += 1 else: stats["vulns_unchanged"] += 1 continue # Create new (Nessus-only finding). # Use a savepoint so an IntegrityError (duplicate race) only # rolls back this single insert — not the entire sync transaction. try: new_vuln = Vulnerability( cve_id=cve_id, asset_id=asset.id, cvss_score=cvss, severity=severity, status=VulnerabilityStatus.open, title=plugin_name[:500] if plugin_name else None, # plugin_name is the affected-software label Nessus uses # ("Mozilla Firefox < 150.0.2", "KB5087539: ..."). Mirror # it into package_name so the Vulns table's PACKAGE # column isn't blank for Nessus-only findings. package_name=plugin_name[:255] if plugin_name else None, package_version=None, description=full_description, references=json.dumps(n_see_also) if n_see_also else None, exploit_available=bool(exploit_avail) if exploit_avail is not None else False, exploit_maturity=exploit_mat[:50] if exploit_mat else None, fixed_version=fixed_version, nessus_vpr_score=vpr_score, detected_at=datetime.now(), sources=json.dumps([SOURCE_NAME]), first_detected_by=SOURCE_NAME, nessus_plugin_id=str(plugin_id) if plugin_id else None, ) with db.begin_nested(): db.add(new_vuln) db.flush() new_vuln.refresh_scores() newly_created_vuln_ids.append(new_vuln.id) stats["vulns_created"] += 1 except IntegrityError: # Race: another request already inserted the same (cve_id, asset_id). # Savepoint was rolled back; outer transaction is intact. logger.debug( "Nessus sync: duplicate insert race for %s on asset %s", cve_id, asset.id, ) continue # Backfill: any vuln on this asset that PREVIOUSLY had `nessus` # in sources but is not in seen_cves_for_asset → drop nessus. # If sources empties out → patched. # # Safety: if this host had zero vulnerabilities AND zero raw # findings reported by Nessus this run, the response is most # likely a partial scan or an API hiccup (Nessus rarely # reports a host as truly clean). Skip the backfill to avoid # mass-patching every prior Nessus finding on this asset. raw_findings = host.get("vulnerabilities") or [] if not seen_cves_for_asset and not raw_findings: logger.warning( "Nessus reported 0 findings for host %s in scan %s — " "skipping backfill to avoid mass-patching from a " "transient empty response.", asset.hostname, scan_id, ) asset.last_scan = datetime.now() scan_row.status = ScanStatus.COMPLETED scan_row.completed_at = datetime.now() scan_row.vulnerabilities_found = 0 scan_row.error_message = "Host returned 0 findings (skipped backfill)" continue stale_nessus_vulns = ( db.query(Vulnerability) .filter( Vulnerability.asset_id == asset.id, Vulnerability.status == VulnerabilityStatus.open, Vulnerability.sources.contains('"nessus"'), ) .all() ) for v in stale_nessus_vulns: if v.cve_id in seen_cves_for_asset: continue v.remove_source(SOURCE_NAME) if not v.source_list: old_status = v.status v.status = VulnerabilityStatus.patched v.patched_at = datetime.now() stats["vulns_marked_patched"] += 1 # Audit trail — revisionssicher: who/when/how. # user_id=None marks the change as automated. try: from app.routers.vulnerabilities import log_vulnerability_change log_vulnerability_change( db, None, v.id, old_status, v.status, reason=f"Nessus rescan {scan_id} no longer reports this CVE on {asset.hostname}", cve_id=v.cve_id, source="nessus_sync", ) except Exception as e: logger.warning( "audit log for auto-patch failed (vuln_id=%s): %s", v.id, e, ) asset.last_scan = datetime.now() # Close out the per-host scan row. scan_row.status = ScanStatus.COMPLETED scan_row.completed_at = datetime.now() scan_row.vulnerabilities_found = len(seen_cves_for_asset) db.commit() # Best-effort enrichment + notification (re-use Wazuh path) if newly_created_vuln_ids: try: from app.services.email_service import dispatch_new_vuln_notifications from app.services.enrichment_service import enrich_vulnerabilities fresh = ( db.query(Vulnerability) .filter(Vulnerability.id.in_(newly_created_vuln_ids)) .all() ) try: enrich_vulnerabilities(db, fresh) except Exception as e: logger.warning("Nessus sync: enrichment failed (non-fatal): %s", e) stats["notifications"] = dispatch_new_vuln_notifications(db, fresh) except Exception as e: logger.warning("Nessus sync: post-processing failed: %s", e) logger.info( "Nessus sync done: %d hosts, %d created, %d merged, %d scores overridden, %d patched, %d unmatched", stats["hosts_synced"], stats["vulns_created"], stats["vulns_merged"], stats["scores_overridden"], stats["vulns_marked_patched"], len(stats["unmatched_hosts"]), ) return stats