vulncheck
9cb9051854
Tester saw cvss_score=-1 in the DB plus cpr_score=-9.3 in the UI for
some CVEs. CPR math is purely multiplicative, so a negative CPR can
only come from a negative CVSS. Source confirmed: Wazuh-Indexer emits
-1 as a placeholder when it could not score a package, and we passed
it through to the DB unchanged. The Nessus plugin extractor had the
same gap.
Both clients now validate the extracted base_score with a
'0.0 <= f <= 10.0' guard and return None on anything outside the
CVSSv3 spec range. Downstream CPR + priority calculations already
handle None correctly (calculate_cpr_score returns None when either
cvss_score or epss_score is missing). No DB clean-up required —
new syncs overwrite the bad values.
For existing -1 rows, ops can wipe them once:
UPDATE vulnerabilities SET cvss_score = NULL
WHERE cvss_score < 0 OR cvss_score > 10;
529 lines
20 KiB
Python
529 lines
20 KiB
Python
"""
|
|
Tenable Nessus API Client
|
|
|
|
Imports vulnerability findings from Nessus Professional / Essentials /
|
|
Tenable.io scans. Nessus uses static API keys (no token refresh), so the
|
|
client is simpler than the Wazuh equivalent but still implements:
|
|
|
|
- Retry with exponential backoff on transient errors
|
|
- Strict 429 / rate-limit handling
|
|
- Optional SSL verification (default on)
|
|
- Per-host iteration helpers
|
|
|
|
API reference: https://docs.tenable.com/nessus/Content/API.htm
|
|
Generate API keys in Nessus UI: My Account → API Keys → Generate.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
import time
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import httpx
|
|
from tenacity import (
|
|
retry,
|
|
retry_if_exception_type,
|
|
stop_after_attempt,
|
|
wait_exponential,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Regex fallback used when a Nessus plugin lists no CVE in its `cve` field
|
|
# but mentions CVE IDs in the plugin_output / description text.
|
|
_CVE_REGEX = re.compile(r"CVE-\d{4}-\d{4,7}", re.IGNORECASE)
|
|
|
|
|
|
class NessusAPIError(Exception):
|
|
"""Base error for Nessus API problems."""
|
|
|
|
|
|
class NessusAuthenticationError(NessusAPIError):
|
|
"""Bad API keys or expired."""
|
|
|
|
|
|
class NessusRateLimitError(NessusAPIError):
|
|
"""Nessus returned 429."""
|
|
|
|
|
|
class NessusClient:
|
|
"""
|
|
Thin wrapper around the Nessus REST API.
|
|
|
|
Auth: `X-ApiKeys: accessKey=<a>; secretKey=<s>` header on every call.
|
|
No login/logout round-trip required.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
access_key: str,
|
|
secret_key: str,
|
|
verify_ssl: bool = True,
|
|
timeout: float = 30.0,
|
|
):
|
|
if not base_url or not access_key or not secret_key:
|
|
raise ValueError("Nessus base_url + access_key + secret_key required")
|
|
|
|
self.base_url = base_url.rstrip("/")
|
|
self.access_key = access_key
|
|
self.secret_key = secret_key
|
|
self.verify_ssl = verify_ssl
|
|
|
|
self.client = httpx.Client(
|
|
base_url=self.base_url,
|
|
verify=verify_ssl,
|
|
timeout=timeout,
|
|
headers={
|
|
"X-ApiKeys": f"accessKey={access_key}; secretKey={secret_key}",
|
|
"Accept": "application/json",
|
|
},
|
|
limits=httpx.Limits(max_connections=10, max_keepalive_connections=5),
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Context manager support so callers can `with NessusClient(...) as c:`
|
|
# and trust connections get torn down even on exception.
|
|
# ------------------------------------------------------------------
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
self.close()
|
|
|
|
def close(self):
|
|
try:
|
|
self.client.close()
|
|
except Exception:
|
|
pass
|
|
|
|
# ------------------------------------------------------------------
|
|
# Core request helper with retry
|
|
# ------------------------------------------------------------------
|
|
@retry(
|
|
retry=retry_if_exception_type(
|
|
(httpx.TimeoutException, httpx.ConnectError, NessusRateLimitError)
|
|
),
|
|
stop=stop_after_attempt(3),
|
|
wait=wait_exponential(multiplier=1, min=1, max=8),
|
|
reraise=True,
|
|
)
|
|
def _post(self, path: str, body: Optional[dict] = None) -> Any:
|
|
try:
|
|
resp = self.client.post(path, json=body or {})
|
|
except httpx.HTTPError as e:
|
|
logger.warning("Nessus POST error %s: %s", path, e)
|
|
raise
|
|
|
|
if resp.status_code == 401:
|
|
raise NessusAuthenticationError(
|
|
"Nessus API key rejected (HTTP 401). Regenerate keys in Nessus UI."
|
|
)
|
|
if resp.status_code == 429:
|
|
retry_after = resp.headers.get("Retry-After", "?")
|
|
logger.warning("Nessus 429 rate limited (Retry-After=%s)", retry_after)
|
|
raise NessusRateLimitError("Nessus 429")
|
|
if resp.status_code == 404:
|
|
raise NessusAPIError(f"Nessus 404 on POST {path}")
|
|
if not 200 <= resp.status_code < 300:
|
|
raise NessusAPIError(
|
|
f"Nessus {resp.status_code} on POST {path}: {resp.text[:200]}"
|
|
)
|
|
|
|
if resp.content:
|
|
try:
|
|
return resp.json()
|
|
except ValueError:
|
|
return {}
|
|
return {}
|
|
|
|
@retry(
|
|
retry=retry_if_exception_type(
|
|
(httpx.TimeoutException, httpx.ConnectError, NessusRateLimitError)
|
|
),
|
|
stop=stop_after_attempt(3),
|
|
wait=wait_exponential(multiplier=1, min=1, max=8),
|
|
reraise=True,
|
|
)
|
|
def _get(self, path: str, params: Optional[dict] = None) -> Any:
|
|
try:
|
|
resp = self.client.get(path, params=params or {})
|
|
except httpx.HTTPError as e:
|
|
logger.warning("Nessus request error %s: %s", path, e)
|
|
raise
|
|
|
|
if resp.status_code == 401:
|
|
raise NessusAuthenticationError(
|
|
"Nessus API key rejected (HTTP 401). Regenerate keys in Nessus UI."
|
|
)
|
|
if resp.status_code == 429:
|
|
# Honor Retry-After when present
|
|
retry_after = resp.headers.get("Retry-After", "?")
|
|
logger.warning("Nessus 429 rate limited (Retry-After=%s)", retry_after)
|
|
raise NessusRateLimitError("Nessus 429")
|
|
if resp.status_code == 404:
|
|
raise NessusAPIError(f"Nessus 404 on {path}")
|
|
if not 200 <= resp.status_code < 300:
|
|
raise NessusAPIError(
|
|
f"Nessus {resp.status_code} on {path}: {resp.text[:200]}"
|
|
)
|
|
|
|
try:
|
|
return resp.json()
|
|
except ValueError:
|
|
raise NessusAPIError(f"Nessus non-JSON response on {path}")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public API
|
|
# ------------------------------------------------------------------
|
|
def ping(self) -> dict:
|
|
"""Quick auth + connectivity probe used by the admin Test button."""
|
|
data = self._get("/server/properties")
|
|
return {
|
|
"server_version": data.get("server_version"),
|
|
"feed": data.get("feed"),
|
|
"loaded_plugin_set": data.get("loaded_plugin_set"),
|
|
}
|
|
|
|
def list_scans(self) -> List[dict]:
|
|
"""Return Nessus scans visible to the API keys."""
|
|
data = self._get("/scans")
|
|
return data.get("scans") or []
|
|
|
|
def launch_scan(
|
|
self,
|
|
scan_id: int,
|
|
alt_targets: Optional[List[str]] = None,
|
|
) -> dict:
|
|
"""
|
|
Launch a Nessus scan.
|
|
|
|
Pass ``alt_targets`` to override the scan's configured targets and
|
|
limit it to specific IPs / hostnames (e.g. for a post-patch rescan
|
|
of a single asset). Nessus returns ``{"scan_uuid": "…"}`` on success.
|
|
|
|
``POST /scans/{scan_id}/launch`` body: ``{"alt_targets": ["ip1", "ip2"]}``
|
|
|
|
Quirk handling: Nessus 10.x (especially Essentials) sometimes accepts
|
|
the launch and then drops the TCP connection without sending a
|
|
response — httpx surfaces this as ``RemoteProtocolError``. When that
|
|
happens we poll the scan's status; if it flipped to ``running`` /
|
|
``pending`` the launch did succeed and we return a synthetic OK.
|
|
"""
|
|
body: Dict[str, Any] = {}
|
|
if alt_targets:
|
|
body["alt_targets"] = alt_targets
|
|
logger.info(
|
|
"Launching Nessus scan %s (alt_targets=%s)", scan_id, alt_targets
|
|
)
|
|
|
|
# Nessus Essentials is known to drop the connection on launch POSTs
|
|
# carrying `alt_targets` — the launch is silently rejected without a
|
|
# proper HTTP error. We try once with alt_targets; if the connection
|
|
# drops AND the scan didn't actually start, we surface a helpful
|
|
# error that points to the Essentials limitation.
|
|
attempted_alt_targets = bool(alt_targets)
|
|
|
|
def _verify_started() -> Optional[dict]:
|
|
"""GET the scan and return a synthetic OK if it's running now."""
|
|
time.sleep(2)
|
|
try:
|
|
scan = self.get_scan(scan_id)
|
|
info = scan.get("info") or {}
|
|
status = str(info.get("status", "")).lower()
|
|
if status in ("running", "pending", "resuming"):
|
|
logger.info(
|
|
"Nessus scan %s is %s — launch succeeded despite dropped connection.",
|
|
scan_id, status,
|
|
)
|
|
return {
|
|
"scan_uuid": info.get("uuid"),
|
|
"status": status,
|
|
"verified_by_status_check": True,
|
|
}
|
|
logger.warning(
|
|
"Nessus scan %s status is '%s' after dropped launch — launch did not take effect.",
|
|
scan_id, status,
|
|
)
|
|
except Exception as inner:
|
|
logger.warning(
|
|
"Nessus status verify after dropped launch failed: %s", inner
|
|
)
|
|
return None
|
|
|
|
try:
|
|
return self._post(f"/scans/{scan_id}/launch", body)
|
|
except httpx.RemoteProtocolError as e:
|
|
logger.warning(
|
|
"Nessus launch %s: connection dropped without response "
|
|
"(likely already accepted): %s — verifying via status check.",
|
|
scan_id, e,
|
|
)
|
|
verified = _verify_started()
|
|
if verified is not None:
|
|
return verified
|
|
|
|
# Connection dropped AND scan didn't start. On Essentials this
|
|
# almost always means alt_targets was the culprit. Hint loudly.
|
|
if attempted_alt_targets:
|
|
raise NessusAPIError(
|
|
f"Nessus refused the launch of scan {scan_id} with "
|
|
f"alt_targets={alt_targets} (the connection was dropped "
|
|
"without an HTTP error). This is a known limitation on "
|
|
"Nessus Essentials — targeted host scans via the API "
|
|
"aren't supported by the free edition. Workarounds: "
|
|
"(1) launch the scan manually from the Nessus UI, then "
|
|
"click \"Sync Data (Nessus)\" in VulnCheck to import "
|
|
"results, or (2) upgrade to Nessus Professional / "
|
|
"Tenable.io which support API-driven launches."
|
|
) from e
|
|
|
|
raise NessusAPIError(
|
|
f"Nessus dropped the connection on launch of scan {scan_id} "
|
|
"and the scan did not enter a running state. Check the scan "
|
|
"configuration in Nessus, or try launching it from the UI."
|
|
) from e
|
|
|
|
def get_scan(self, scan_id: int, history_id: Optional[int] = None) -> dict:
|
|
"""Return scan details — includes hosts[] with host_id, hostname, vulnerability counts."""
|
|
params = {"history_id": history_id} if history_id else None
|
|
return self._get(f"/scans/{scan_id}", params=params)
|
|
|
|
def get_scan_hosts(self, scan_id: int, history_id: Optional[int] = None) -> List[dict]:
|
|
"""Convenience: return just the hosts list from a scan detail call."""
|
|
scan = self.get_scan(scan_id, history_id)
|
|
return scan.get("hosts") or []
|
|
|
|
def get_host(
|
|
self,
|
|
scan_id: int,
|
|
host_id: int,
|
|
history_id: Optional[int] = None,
|
|
) -> dict:
|
|
"""
|
|
Return host detail with `vulnerabilities[]` (plugin-id list) and `info` block.
|
|
Each vulnerability item has: plugin_id, plugin_name, severity, count, vuln_index, hostname.
|
|
"""
|
|
params = {"history_id": history_id} if history_id else None
|
|
return self._get(f"/scans/{scan_id}/hosts/{host_id}", params=params)
|
|
|
|
def get_plugin_output(
|
|
self,
|
|
scan_id: int,
|
|
host_id: int,
|
|
plugin_id: int,
|
|
history_id: Optional[int] = None,
|
|
) -> dict:
|
|
"""
|
|
Return per-plugin findings on a host. Used when we need the CVE list
|
|
and plugin metadata (description / solution / risk_factor) — not
|
|
cheap, so callers should batch carefully.
|
|
"""
|
|
params = {"history_id": history_id} if history_id else None
|
|
return self._get(
|
|
f"/scans/{scan_id}/hosts/{host_id}/plugins/{plugin_id}",
|
|
params=params,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# CVE extraction
|
|
# ------------------------------------------------------------------
|
|
@staticmethod
|
|
def extract_cves(plugin_payload: dict) -> List[str]:
|
|
"""
|
|
Pull CVE IDs from a plugin payload. Modern Nessus puts CVEs under
|
|
``pluginattributes.ref_information.ref[?].values.value`` where the
|
|
ref entry has ``name == "cve"``. Older versions had a flat
|
|
``pluginattributes.cve`` field. We check both, then fall back to a
|
|
regex over description / outputs as last resort.
|
|
"""
|
|
cves: set = set()
|
|
|
|
info = plugin_payload.get("info") or {}
|
|
attrs = (
|
|
info.get("plugindescription", {})
|
|
.get("pluginattributes", {})
|
|
)
|
|
|
|
# Path 1 (legacy): flat `cve` field directly on pluginattributes.
|
|
cve_field = attrs.get("cve")
|
|
if isinstance(cve_field, list):
|
|
for entry in cve_field:
|
|
if isinstance(entry, str):
|
|
cves.add(entry.upper())
|
|
elif isinstance(cve_field, str):
|
|
cves.add(cve_field.upper())
|
|
|
|
# Path 2 (modern): ref_information.ref[?].values.value where name=="cve".
|
|
# Layout (observed on Nessus 10.x):
|
|
# ref_information.ref = [
|
|
# {"name": "cve", "values": {"value": ["CVE-2026-8090", ...]}},
|
|
# {"name": "bid", "values": {"value": [...]}}, ...
|
|
# ]
|
|
ref_info = attrs.get("ref_information") or {}
|
|
ref_list = ref_info.get("ref") if isinstance(ref_info, dict) else None
|
|
if isinstance(ref_list, list):
|
|
for ref in ref_list:
|
|
if not isinstance(ref, dict):
|
|
continue
|
|
if str(ref.get("name", "")).lower() != "cve":
|
|
continue
|
|
values = ref.get("values") or {}
|
|
value_field = values.get("value") if isinstance(values, dict) else None
|
|
if isinstance(value_field, list):
|
|
for v in value_field:
|
|
if isinstance(v, str):
|
|
cves.add(v.upper())
|
|
elif isinstance(value_field, str):
|
|
cves.add(value_field.upper())
|
|
|
|
# Path 3 (last resort): regex over description and outputs.
|
|
text_blobs = []
|
|
desc = attrs.get("description")
|
|
if isinstance(desc, str):
|
|
text_blobs.append(desc)
|
|
|
|
for output in plugin_payload.get("outputs") or []:
|
|
ot = output.get("plugin_output")
|
|
if isinstance(ot, str):
|
|
text_blobs.append(ot)
|
|
|
|
for blob in text_blobs:
|
|
for match in _CVE_REGEX.findall(blob):
|
|
cves.add(match.upper())
|
|
|
|
return sorted(cves)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Plugin-payload helpers
|
|
# ------------------------------------------------------------------
|
|
@staticmethod
|
|
def plugin_severity_to_label(severity: int) -> str:
|
|
"""
|
|
Nessus severity int → VulnerabilitySeverity name.
|
|
Nessus scale: 0=Info, 1=Low, 2=Medium, 3=High, 4=Critical.
|
|
"""
|
|
mapping = {0: "none", 1: "low", 2: "medium", 3: "high", 4: "critical"}
|
|
return mapping.get(severity, "none")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Pluginattributes accessor (used by all extractors below)
|
|
# ------------------------------------------------------------------
|
|
@staticmethod
|
|
def _plugin_attrs(plugin_payload: dict) -> dict:
|
|
return (
|
|
(plugin_payload.get("info") or {})
|
|
.get("plugindescription", {})
|
|
.get("pluginattributes", {})
|
|
)
|
|
|
|
@staticmethod
|
|
def plugin_vpr_score(plugin_payload: dict) -> Optional[float]:
|
|
"""
|
|
Tenable VPR (Vulnerability Priority Rating, 0-10) from the plugin.
|
|
|
|
Modern Nessus puts ``vpr_score`` directly on ``pluginattributes``.
|
|
Some payloads have it as a string ("6.7"), some as a number.
|
|
"""
|
|
attrs = NessusClient._plugin_attrs(plugin_payload)
|
|
v = attrs.get("vpr_score")
|
|
if v is None or v == "":
|
|
return None
|
|
try:
|
|
return float(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
@staticmethod
|
|
def plugin_exploit_available(plugin_payload: dict) -> Optional[bool]:
|
|
"""
|
|
Returns True/False if Nessus knows of a public exploit, or None when
|
|
the field isn't present. Looks on pluginattributes directly and under
|
|
``vuln_information`` (location varies between Nessus versions).
|
|
"""
|
|
attrs = NessusClient._plugin_attrs(plugin_payload)
|
|
candidates = [attrs.get("exploit_available")]
|
|
vuln_info = attrs.get("vuln_information") or {}
|
|
if isinstance(vuln_info, dict):
|
|
candidates.append(vuln_info.get("exploit_available"))
|
|
|
|
for v in candidates:
|
|
if v is None:
|
|
continue
|
|
if isinstance(v, bool):
|
|
return v
|
|
if isinstance(v, str):
|
|
return v.strip().lower() == "true"
|
|
return None
|
|
|
|
@staticmethod
|
|
def plugin_exploit_maturity(plugin_payload: dict) -> Optional[str]:
|
|
"""``exploit_code_maturity`` from Nessus: Unproven | Proof-of-Concept |
|
|
Functional | High (matches our existing exploit_maturity column)."""
|
|
attrs = NessusClient._plugin_attrs(plugin_payload)
|
|
v = attrs.get("exploit_code_maturity")
|
|
if isinstance(v, str) and v.strip():
|
|
return v.strip()
|
|
return None
|
|
|
|
@staticmethod
|
|
def plugin_description(plugin_payload: dict) -> Optional[str]:
|
|
"""Plain-text description (`description` on pluginattributes)."""
|
|
attrs = NessusClient._plugin_attrs(plugin_payload)
|
|
v = attrs.get("description")
|
|
return v if isinstance(v, str) and v.strip() else None
|
|
|
|
@staticmethod
|
|
def plugin_solution(plugin_payload: dict) -> Optional[str]:
|
|
"""Vendor solution / remediation guidance from the Nessus plugin."""
|
|
attrs = NessusClient._plugin_attrs(plugin_payload)
|
|
v = attrs.get("solution")
|
|
return v if isinstance(v, str) and v.strip() else None
|
|
|
|
@staticmethod
|
|
def plugin_see_also(plugin_payload: dict) -> List[str]:
|
|
"""``see_also`` URL list — vendor advisories, MITRE links, etc."""
|
|
attrs = NessusClient._plugin_attrs(plugin_payload)
|
|
v = attrs.get("see_also")
|
|
if isinstance(v, list):
|
|
return [str(u) for u in v if isinstance(u, str)]
|
|
if isinstance(v, str):
|
|
# Some plugins return newline-separated URLs in a single string.
|
|
return [line.strip() for line in v.splitlines() if line.strip()]
|
|
return []
|
|
|
|
@staticmethod
|
|
def plugin_cvss(plugin_payload: dict) -> Optional[float]:
|
|
"""
|
|
Best-effort CVSS extraction — prefers v3.x base score over v2.
|
|
|
|
Modern Nessus (10.x+) groups CVSS fields under a ``risk_information``
|
|
sub-dict on ``pluginattributes``. Older versions had them flat on
|
|
``pluginattributes`` directly. Check both shapes so we don't miss
|
|
the score on either format.
|
|
"""
|
|
attrs = NessusClient._plugin_attrs(plugin_payload)
|
|
risk_info = attrs.get("risk_information") or {}
|
|
if not isinstance(risk_info, dict):
|
|
risk_info = {}
|
|
|
|
# v3 first, then v2 — prefer the modern score.
|
|
for key in ("cvss3_base_score", "cvss_base_score"):
|
|
# Try the modern (risk_information) layout first, then flat fallback.
|
|
for source in (risk_info, attrs):
|
|
v = source.get(key)
|
|
if v is None:
|
|
continue
|
|
try:
|
|
f = float(v)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
# Clamp to the CVSS spec range. Some plugins emit junk
|
|
# like -1 or 99 for un-scored vulns; treat those as
|
|
# "no score" so the downstream CPR math stays non-
|
|
# negative and within [0, 100].
|
|
if 0.0 <= f <= 10.0:
|
|
return f
|
|
return None
|