vulncheck
3b482b3560
services/urs_service.py implements the Unified Risk Score per the
tester's formula:
AVS (Asset Vulnerability Score)
hybrid default: 0.7 × avg(CPR) + 0.3 × max(CPR)
'avg' and 'max' modes also exposed so admin can pick policy
ASS (Asset Security Score)
mean of weighted_score across the asset's policies
weighted_score per policy = Σ(impact × passed) / Σ(impact)
falls back to simple score% when no impacts loaded for the policy
URS = round((AVS + ASS) / 2) × criticality_factor capped at 100
criticality factor: low ×0.7 | normal ×1.0 | high ×1.3 | crit ×1.5
Severity bands per spec:
90-100 CRITICAL | 70-89 HIGH | 40-69 MEDIUM | 1-39 LOW | 0 NONE
compute_urs() optionally writes a daily snapshot (per asset, one row
per snapshot_date) so the dashboard can show a 7-day trend arrow.
compute_urs_for_all() loops every asset — called by the nightly
scheduler job in E8 and after each Wazuh/Nessus/Vulnrichment sync.
get_urs_trend() returns the URS delta vs N days ago (positive =
degrading, negative = improving) — used by the dashboard widget.
Endpoints + UI consume this in the next commits.
326 lines
9.5 KiB
Python
326 lines
9.5 KiB
Python
"""
|
||
Unified Risk Score (URS) calculation.
|
||
|
||
Inputs:
|
||
AVS — Asset Vulnerability Score (0-100, from CVE/CPR data)
|
||
ASS — Asset Security Score (0-100, from SCA weighted score)
|
||
|
||
Output:
|
||
URS = round((AVS + ASS) / 2) × criticality_multiplier (capped 100)
|
||
|
||
Severity bands (per tester spec):
|
||
90-100 CRITICAL → eskalation
|
||
70-89 HIGH → 24-48 h
|
||
40-69 MEDIUM → 1-2 Wochen
|
||
1-39 LOW → routine
|
||
0 NONE
|
||
|
||
Defaults:
|
||
AVS hybrid = 0.7 × avg(CPR) + 0.3 × max(CPR)
|
||
ASS = avg weighted score across all SCA policies of the asset
|
||
impact-weighted when ComplianceImpact rows are loaded for
|
||
the policy's cis_ids, plain pass% otherwise.
|
||
Asset criticality multiplier:
|
||
low ×0.7 | normal ×1.0 | high ×1.3 | critical ×1.5
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from datetime import datetime
|
||
from typing import Dict, Iterable, List, Optional, Tuple
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models.asset import Asset
|
||
from app.models.compliance import (
|
||
AssetRiskSnapshot, ComplianceCheck, ComplianceImpact, ComplianceResult,
|
||
)
|
||
from app.models.vulnerability import Vulnerability, VulnerabilityStatus
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# --------------------------------------------------------------
|
||
# Severity bands
|
||
# --------------------------------------------------------------
|
||
SEVERITY_BANDS = [
|
||
(90.0, 100.0, "CRITICAL"),
|
||
(70.0, 89.999, "HIGH"),
|
||
(40.0, 69.999, "MEDIUM"),
|
||
(0.001, 39.999, "LOW"),
|
||
(0.0, 0.0, "NONE"),
|
||
]
|
||
|
||
|
||
def urs_severity(urs: Optional[float]) -> str:
|
||
if urs is None:
|
||
return "NONE"
|
||
for lo, hi, label in SEVERITY_BANDS:
|
||
if lo <= urs <= hi:
|
||
return label
|
||
return "NONE"
|
||
|
||
|
||
# --------------------------------------------------------------
|
||
# Criticality multiplier
|
||
# --------------------------------------------------------------
|
||
_CRIT_FACTOR = {
|
||
"low": 0.7,
|
||
"normal": 1.0,
|
||
"high": 1.3,
|
||
"critical": 1.5,
|
||
}
|
||
|
||
|
||
def _criticality_factor(value: Optional[str]) -> float:
|
||
return _CRIT_FACTOR.get((value or "normal").lower(), 1.0)
|
||
|
||
|
||
# --------------------------------------------------------------
|
||
# AVS — Asset Vulnerability Score
|
||
# --------------------------------------------------------------
|
||
def compute_avs(
|
||
db: Session,
|
||
asset_id: int,
|
||
mode: str = "hybrid",
|
||
) -> Optional[float]:
|
||
"""
|
||
Compute AVS for one asset from its open vulnerabilities' CPR scores.
|
||
|
||
mode:
|
||
'hybrid' (default) → 0.7 × avg(CPR) + 0.3 × max(CPR)
|
||
'avg' → mean(CPR)
|
||
'max' → max(CPR)
|
||
"""
|
||
vulns = (
|
||
db.query(Vulnerability)
|
||
.filter(
|
||
Vulnerability.asset_id == asset_id,
|
||
Vulnerability.status == VulnerabilityStatus.open,
|
||
)
|
||
.all()
|
||
)
|
||
cprs = [v.calculate_cpr_score() for v in vulns]
|
||
cprs = [c for c in cprs if c is not None]
|
||
if not cprs:
|
||
return None
|
||
if mode == "avg":
|
||
return round(sum(cprs) / len(cprs), 2)
|
||
if mode == "max":
|
||
return round(max(cprs), 2)
|
||
# hybrid
|
||
avg = sum(cprs) / len(cprs)
|
||
return round(0.7 * avg + 0.3 * max(cprs), 2)
|
||
|
||
|
||
# --------------------------------------------------------------
|
||
# ASS — Asset Security Score (impact-weighted SCA)
|
||
# --------------------------------------------------------------
|
||
def _impact_lookup(
|
||
db: Session,
|
||
cis_ids: Iterable[str],
|
||
benchmarks: Iterable[str],
|
||
) -> Dict[str, int]:
|
||
"""
|
||
Bulk-fetch impacts for a set of cis_ids, falling back across
|
||
benchmarks. Returns {cis_id → impact}.
|
||
"""
|
||
cis_ids = list({c for c in cis_ids if c})
|
||
benchmarks = list({b for b in benchmarks if b})
|
||
if not cis_ids:
|
||
return {}
|
||
rows = (
|
||
db.query(ComplianceImpact)
|
||
.filter(ComplianceImpact.cis_id.in_(cis_ids))
|
||
.all()
|
||
)
|
||
result: Dict[str, int] = {}
|
||
benchmark_set = set(benchmarks)
|
||
# Prefer rows whose benchmark matches the asset's policies; else
|
||
# take the first hit (CSV cross-OS overlap).
|
||
for r in rows:
|
||
if r.cis_id not in result:
|
||
result[r.cis_id] = r.impact
|
||
if r.benchmark in benchmark_set:
|
||
result[r.cis_id] = r.impact
|
||
return result
|
||
|
||
|
||
def _weighted_score_for_policy(
|
||
db: Session,
|
||
result_row: ComplianceResult,
|
||
) -> Optional[float]:
|
||
"""
|
||
Recompute weighted_score for one ComplianceResult by inspecting its
|
||
cached ComplianceCheck rows (when present) and the loaded impacts.
|
||
|
||
Falls back to result_row.score when no checks cached or no impacts
|
||
match.
|
||
"""
|
||
checks: List[ComplianceCheck] = list(result_row.checks or [])
|
||
if not checks:
|
||
return result_row.score
|
||
impacts = _impact_lookup(
|
||
db,
|
||
(c.check_id for c in checks),
|
||
[result_row.policy_id],
|
||
)
|
||
if not impacts:
|
||
return result_row.score
|
||
total_impact = 0
|
||
earned_impact = 0
|
||
for c in checks:
|
||
imp = impacts.get(c.check_id, 50)
|
||
if (c.result or "").lower() in ("not applicable", "n/a"):
|
||
continue
|
||
total_impact += imp
|
||
if (c.result or "").lower() == "passed":
|
||
earned_impact += imp
|
||
if total_impact == 0:
|
||
return result_row.score
|
||
return round(100.0 * earned_impact / total_impact, 2)
|
||
|
||
|
||
def compute_ass(db: Session, asset_id: int) -> Tuple[Optional[float], int]:
|
||
"""
|
||
Compute ASS as the mean of (weighted_score OR score) across all
|
||
policy results of the asset. Returns (ass, policy_count).
|
||
"""
|
||
results = (
|
||
db.query(ComplianceResult)
|
||
.filter(ComplianceResult.asset_id == asset_id)
|
||
.all()
|
||
)
|
||
if not results:
|
||
return None, 0
|
||
scored: List[float] = []
|
||
for r in results:
|
||
# Refresh weighted_score if we have cached checks; else use it
|
||
# as-is or fall back to the simple score.
|
||
w = _weighted_score_for_policy(db, r)
|
||
if w != r.weighted_score:
|
||
r.weighted_score = w
|
||
chosen = w if w is not None else r.score
|
||
if chosen is not None:
|
||
scored.append(chosen)
|
||
if not scored:
|
||
return None, len(results)
|
||
return round(sum(scored) / len(scored), 2), len(results)
|
||
|
||
|
||
# --------------------------------------------------------------
|
||
# URS — Unified Risk Score
|
||
# --------------------------------------------------------------
|
||
def compute_urs(
|
||
db: Session,
|
||
asset_id: int,
|
||
avs_mode: str = "hybrid",
|
||
persist_snapshot: bool = True,
|
||
) -> Dict[str, Optional[float]]:
|
||
"""
|
||
Combine AVS + ASS into URS and optionally write a daily snapshot.
|
||
"""
|
||
asset = db.query(Asset).filter(Asset.id == asset_id).first()
|
||
if not asset:
|
||
return {"asset_id": asset_id, "avs": None, "ass": None, "urs": None,
|
||
"severity": "NONE", "criticality": None}
|
||
|
||
avs = compute_avs(db, asset_id, mode=avs_mode)
|
||
ass, policy_count = compute_ass(db, asset_id)
|
||
crit_factor = _criticality_factor(asset.criticality)
|
||
|
||
parts = [v for v in (avs, ass) if v is not None]
|
||
if not parts:
|
||
urs: Optional[float] = None
|
||
else:
|
||
base = sum(parts) / len(parts)
|
||
urs = min(round(base * crit_factor, 1), 100.0)
|
||
# Spec wants integer URS — round to nearest int but keep one
|
||
# decimal in storage so trend arrows can detect 0.5-point moves.
|
||
urs = round(urs, 1)
|
||
sev = urs_severity(urs)
|
||
|
||
if persist_snapshot:
|
||
_upsert_daily_snapshot(db, asset_id, avs, ass, urs, sev)
|
||
db.commit()
|
||
|
||
return {
|
||
"asset_id": asset_id,
|
||
"avs": avs,
|
||
"ass": ass,
|
||
"urs": urs,
|
||
"severity": sev,
|
||
"criticality": asset.criticality,
|
||
"criticality_factor": crit_factor,
|
||
"policy_count": policy_count,
|
||
}
|
||
|
||
|
||
def _upsert_daily_snapshot(
|
||
db: Session,
|
||
asset_id: int,
|
||
avs: Optional[float],
|
||
ass: Optional[float],
|
||
urs: Optional[float],
|
||
severity: str,
|
||
) -> None:
|
||
today = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
|
||
row = (
|
||
db.query(AssetRiskSnapshot)
|
||
.filter(
|
||
AssetRiskSnapshot.asset_id == asset_id,
|
||
AssetRiskSnapshot.snapshot_date == today,
|
||
)
|
||
.first()
|
||
)
|
||
if row is None:
|
||
row = AssetRiskSnapshot(asset_id=asset_id, snapshot_date=today)
|
||
db.add(row)
|
||
row.avs = avs
|
||
row.ass = ass
|
||
row.urs = urs
|
||
row.severity = severity
|
||
|
||
|
||
def compute_urs_for_all(
|
||
db: Session,
|
||
avs_mode: str = "hybrid",
|
||
) -> Dict[str, int]:
|
||
"""Recompute URS for every asset and persist daily snapshots."""
|
||
assets = db.query(Asset).all()
|
||
stats = {"assets": 0, "with_urs": 0, "errors": 0}
|
||
for a in assets:
|
||
try:
|
||
r = compute_urs(db, a.id, avs_mode=avs_mode, persist_snapshot=True)
|
||
stats["assets"] += 1
|
||
if r["urs"] is not None:
|
||
stats["with_urs"] += 1
|
||
except Exception as e:
|
||
stats["errors"] += 1
|
||
logger.warning("URS calc failed for asset=%s: %s", a.id, e)
|
||
return stats
|
||
|
||
|
||
def get_urs_trend(
|
||
db: Session,
|
||
asset_id: int,
|
||
days: int = 7,
|
||
) -> Optional[float]:
|
||
"""
|
||
Trend delta: today's URS minus URS `days` ago. Positive = degrading.
|
||
None when not enough history.
|
||
"""
|
||
snaps = (
|
||
db.query(AssetRiskSnapshot)
|
||
.filter(AssetRiskSnapshot.asset_id == asset_id)
|
||
.order_by(AssetRiskSnapshot.snapshot_date.desc())
|
||
.limit(days + 1)
|
||
.all()
|
||
)
|
||
if len(snaps) < 2 or snaps[0].urs is None:
|
||
return None
|
||
older = next((s.urs for s in snaps[1:] if s.urs is not None), None)
|
||
if older is None:
|
||
return None
|
||
return round(snaps[0].urs - older, 1)
|