vulncheck/app/services/risk_dimensions_service.py

"""
Asset "Risk Dimensions" — high-value-target (crown-jewel) role detection.

Network exposure alone (open ports) doesn't capture WHY a host matters.
A Domain Controller, a Certificate Authority, a SQL/Exchange/backup server
— once compromised — enable lateral movement, domain takeover and ransomware
spread. This service detects such roles from the data Wazuh syscollector
already provides (listening ports + process names + installed packages) and
produces:
  - a high_value_score (0-100)
  - a list of detected role dimensions ({role, label, weight})

The score feeds the URS via risk_factor() (see urs_service): roles raise an
asset's risk weighting even when the operator left criticality at "normal".

Detection sources: ports (port number + process name) and installed package
names. Package-based roles (Exchange/WSUS/MSSQL/backup/SW-distribution) are
reliable; port/process roles (DC/DNS/DHCP/WinRM) are good. Deep NTLM/Kerberos
usage analysis is NOT available from syscollector → out of scope here.
"""
from __future__ import annotations

import re
from typing import Dict, List, Optional

# role key → (label, weight, port-set, process-substrings, package-substrings)
# A role fires if ANY of its port/process/package signals match.
_ROLES: List[dict] = [
    {"role": "domain_controller", "label": "Domain Controller (AD DS)", "weight": 100,
     "ports": {88, 464}, "ports_all": {389, 88},  # kerberos+ldap together = strong DC
     "proc": ("ntds",), "pkg": ()},
    {"role": "adcs", "label": "AD Certificate Services (CA)", "weight": 100,
     "ports": set(), "proc": ("certsrv",),
     "pkg": ("active directory certificate services", "certification authority")},
    {"role": "backup", "label": "Backup server", "weight": 90,
     "ports": set(), "proc": ("veeam", "acronis", "rubrik"),
     "pkg": ("veeam", "acronis", "rubrik", "arcserve", "commvault", "netbackup",
             "veritas backup", "altaro", "nakivo")},
    {"role": "sw_distribution", "label": "Software distribution", "weight": 85,
     "ports": set(), "proc": ("ccmexec",),
     "pkg": ("configuration manager", "system center configuration", "endpoint configuration manager",
             "configmgr", "pdq deploy", "bigfix", "ivanti")},
    {"role": "exchange", "label": "Exchange (on-prem)", "weight": 85,
     "ports": set(), "proc": ("msexchange",),
     "pkg": ("microsoft exchange server",)},
    {"role": "wsus", "label": "WSUS", "weight": 80,
     "ports": {8530, 8531}, "proc": (),
     "pkg": ("windows server update services", "wsus")},
    {"role": "mssql", "label": "MS SQL Server", "weight": 70,
     "ports": {1433}, "proc": ("sqlservr",),
     "pkg": ("microsoft sql server 20", "sql server database engine")},
    {"role": "dns", "label": "DNS server", "weight": 60,
     "ports": {53}, "proc": ("dns.exe", "named", "dnsmasq"), "pkg": ()},
    {"role": "dhcp", "label": "DHCP server", "weight": 55,
     "ports": {67}, "proc": ("dhcpserver", "dhcpd"), "pkg": ("dhcp server",)},
    {"role": "winrm", "label": "WinRM / PS-Remoting", "weight": 30,
     "ports": {5985, 5986}, "proc": (), "pkg": ()},
]


def _norm(s: str) -> str:
    return re.sub(r"\s+", " ", (s or "").lower()).strip()


def _listening_ports(ports: List[dict]) -> tuple:
    """Return (set_of_listening_ports, set_of_process_substrings_lower)."""
    open_ports: set = set()
    procs: set = set()
    for p in ports or []:
        if not isinstance(p, dict):
            continue
        state = str(p.get("state") or "").lower()
        proto = str(p.get("protocol") or p.get("proto") or "").lower()
        if proto == "tcp" and state and state != "listening":
            continue
        try:
            port = int(p.get("local_port") or (p.get("local") or {}).get("port") or 0)
        except (ValueError, TypeError):
            continue
        if port > 0:
            open_ports.add(port)
        proc = _norm(p.get("process") or "")
        if proc:
            procs.add(proc)
    return open_ports, procs


def detect_risk_dimensions(ports: List[dict], packages: List[dict]) -> dict:
    """Detect crown-jewel roles → {score, dimensions:[{role,label,weight}]}.

    Pure function (no DB / network) → unit-testable.
    """
    open_ports, procs = _listening_ports(ports)
    pkg_names = [_norm(p.get("name") or "") for p in (packages or []) if isinstance(p, dict)]
    pkg_blob = " | ".join(pkg_names)

    detected: List[dict] = []
    for r in _ROLES:
        hit = False
        # ports: any of `ports`, OR all of `ports_all`
        if r.get("ports") and (open_ports & r["ports"]):
            hit = True
        if not hit and r.get("ports_all") and r["ports_all"].issubset(open_ports):
            hit = True
        # process substrings
        if not hit and r.get("proc"):
            if any(any(sub in pr for pr in procs) for sub in r["proc"]):
                hit = True
        # package substrings
        if not hit and r.get("pkg"):
            if any(sub in pkg_blob for sub in r["pkg"]):
                hit = True
        if hit:
            detected.append({"role": r["role"], "label": r["label"], "weight": r["weight"]})

    if not detected:
        return {"score": 0.0, "dimensions": []}

    weights = sorted((d["weight"] for d in detected), reverse=True)
    score = float(weights[0]) + 0.3 * sum(weights[1:])
    score = round(min(score, 100.0), 1)
    # surface highest-weight roles first
    detected.sort(key=lambda d: d["weight"], reverse=True)
    return {"score": score, "dimensions": detected}


def risk_factor(high_value_score: Optional[float]) -> float:
    """Map the high-value score to a URS multiplier band.
    >=90 → 1.5 (critical), >=70 → 1.3 (high), >=40 → 1.15, else 1.0."""
    s = high_value_score or 0.0
    if s >= 90:
        return 1.5
    if s >= 70:
        return 1.3
    if s >= 40:
        return 1.15
    return 1.0