feat(triage): patch plan ranked by exploited CVEs, with an advisory Jev config hint
The findings list says what is wrong, not what to patch first: one Chrome update closes a hundred rows, and one KEV on a single host matters more than all of them. The new /patch-plan page (GET /api/v1/vulnerabilities/patch-plan) groups open findings (open, patch_failed) on ACTIVE assets by the product to update, using the vulnerability_packages rows, else package_name, else the CVE id. Each group shows hosts, findings, CVEs, the fixed versions as reported (never compared) and summed/max priority_score. Groups with CVEs exploited in the wild (KEV / EUVD) rank first, then by summed priority. Every number comes from data the tool already scores. TypeSafe Jev (via OpenRouter Decisions, pinned typesafe/jev-1.13) is asked one question per open CVE, nightly at 06:00 after the last ingest: is it only exploitable with an optional feature or a non-default configuration? It is the one triage question no feed answers. KEV, EUVD, EPSS, SSVC and the CVSS vector already cover exploitation and attack vector, and exposure is a property of the host, not of the CVE text. The probability is stored in vulnerabilities.jev_config_dependent (migration 061); CVEs at 0.7 or above show under "Check config" on the plan. It never closes a finding (cross-source contract) and never enters priority_score. The job is off by default (Settings, AI card, or JEV_TRIAGE_ENABLED) and reuses the OpenRouter key. It sends only public CVE text. An account or network failure stops the run unstamped; a failure specific to one CVE skips only that CVE, so the job cannot stall on the same CVE every night.
This commit is contained in:
@@ -2564,3 +2564,39 @@ and retraction are the ones above.
|
||||
- Untagged images (bare `sha256:` ids) and rows without an id are skipped.
|
||||
|
||||
Test: `tests/test_dockhand_sync.py`.
|
||||
|
||||
---
|
||||
|
||||
# Patch plan + Jev config hint
|
||||
|
||||
`/patch-plan` (API `GET /api/v1/vulnerabilities/patch-plan`) answers "where do
|
||||
I start patching": open findings (`open`, `patch_failed`) on ACTIVE assets,
|
||||
grouped by the product to update (`vulnerability_packages` rows, else
|
||||
`package_name`, else the CVE id). Rank: CVEs exploited in the wild (KEV /
|
||||
EUVD) first, then summed `priority_score`. A CVE hitting two products counts
|
||||
toward both. Fixed versions are listed as reported, never compared.
|
||||
|
||||
**Jev config hint** (`jev_triage_service`, nightly 06:00, off by default —
|
||||
`jev_triage_enabled` setting or `JEV_TRIAGE_ENABLED`, plus the OpenRouter
|
||||
key). One question per CVE text to TypeSafe Jev via OpenRouter Decisions
|
||||
(`typesafe/jev-1.13`, `JEV_MODEL` overrides): *only exploitable with an optional
|
||||
feature / non-default configuration?* The probability lands in
|
||||
`vulnerabilities.jev_config_dependent` (migration 061); ≥ 0.7 lists the CVE
|
||||
under "Check config" on the patch plan.
|
||||
|
||||
What Jev deliberately does not do:
|
||||
|
||||
- **Close findings.** A finding closes when its scanners retract it (Cross-source
|
||||
contract). Auto-closing on a model's false-positive guess would hide real
|
||||
findings — accuracy over coverage.
|
||||
- **Enter the score.** Exploitation, exploitability and attack vector already
|
||||
come from KEV / EUVD / EPSS / SSVC and the CVSS vector; guessing them from
|
||||
prose would be a worse copy.
|
||||
- **Judge exposure.** Internet-facing is a property of the host, not the CVE
|
||||
text — `exposure_service` owns it.
|
||||
|
||||
Answered once per CVE; a failing API (auth, credit, rate limit, network, 5xx)
|
||||
stops the run without stamping, so the next night retries. A CVE-specific
|
||||
failure (other 4xx, unusable answer) skips only that CVE. Only public CVE text is sent.
|
||||
|
||||
Tests: `tests/test_patch_plan.py`, `tests/test_jev_triage.py`.
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
"""Add vulnerabilities.jev_config_dependent + jev_checked_at
|
||||
|
||||
Revision ID: 061
|
||||
Revises: 060
|
||||
Create Date: 2026-09-19 10:00:00.000000
|
||||
|
||||
Jev's answer to "only exploitable with a non-default configuration?", per CVE
|
||||
text. An operator hint on the patch plan; it never closes a finding and never
|
||||
enters priority_score. No backfill: the nightly Jev job fills it when enabled.
|
||||
Idempotent (IF NOT EXISTS), same shape as 053–058.
|
||||
"""
|
||||
from alembic import op
|
||||
|
||||
|
||||
revision = "061"
|
||||
down_revision = "060"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.execute("ALTER TABLE vulnerabilities ADD COLUMN IF NOT EXISTS jev_config_dependent DOUBLE PRECISION;")
|
||||
op.execute("ALTER TABLE vulnerabilities ADD COLUMN IF NOT EXISTS jev_checked_at TIMESTAMP;")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.execute("ALTER TABLE vulnerabilities DROP COLUMN IF EXISTS jev_checked_at;")
|
||||
op.execute("ALTER TABLE vulnerabilities DROP COLUMN IF EXISTS jev_config_dependent;")
|
||||
@@ -216,6 +216,13 @@ class Vulnerability(Base, TimestampMixin):
|
||||
metasploit_module_count = Column(Integer, nullable=False, default=0, server_default="0")
|
||||
exploit_intel_updated_at = Column(DateTime, nullable=True)
|
||||
|
||||
# Jev (TypeSafe, via OpenRouter Decisions) — probability that the CVE is
|
||||
# only exploitable with an optional feature / non-default configuration.
|
||||
# A HINT for the operator to check, per CVE text: it never closes a
|
||||
# finding and never enters priority_score. See jev_triage_service.
|
||||
jev_config_dependent = Column(Float, nullable=True)
|
||||
jev_checked_at = Column(DateTime, nullable=True)
|
||||
|
||||
# Relationships
|
||||
asset = relationship("Asset", back_populates="vulnerabilities")
|
||||
assigned_user = relationship("User", foreign_keys=[assigned_user_id])
|
||||
|
||||
@@ -1011,6 +1011,17 @@ def _attach_last_change(db: Session, vulns, results: list[dict]) -> None:
|
||||
logger.warning("last_change preload failed: %s", e)
|
||||
|
||||
|
||||
@router.get("/patch-plan")
|
||||
def get_patch_plan(
|
||||
limit: int = Query(100, le=500),
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Open findings grouped by the product to update, exploited first."""
|
||||
from app.services.patch_plan_service import build_patch_plan
|
||||
return build_patch_plan(db, limit=limit)
|
||||
|
||||
|
||||
@router.get("/ai-prioritization")
|
||||
def get_ai_prioritization(
|
||||
limit: int = Query(20, le=50),
|
||||
|
||||
@@ -512,6 +512,25 @@ def exploit_intel_nightly():
|
||||
db.close()
|
||||
|
||||
|
||||
def jev_triage_nightly():
|
||||
"""Jev config hint for CVEs that have none yet (jev_triage_service).
|
||||
|
||||
06:00 — after every ingest of the night (last one 05:40), so a CVE found
|
||||
tonight carries its hint when the patch plan is opened in the morning.
|
||||
No-op unless enabled; a hint only, it changes no finding and no score.
|
||||
"""
|
||||
from app.services.jev_triage_service import run_jev_triage
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
run_jev_triage(db)
|
||||
except Exception as e:
|
||||
logger.error("Jev triage nightly failed: %s", e)
|
||||
db.rollback()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def eol_check_nightly():
|
||||
"""Run endoflife.date EOL detection for every Wazuh-linked asset.
|
||||
|
||||
@@ -1427,6 +1446,17 @@ def start_scheduler():
|
||||
replace_existing=True,
|
||||
)
|
||||
|
||||
# Nightly Jev config hint at 06:00 — after the last ingest (05:40), so
|
||||
# tonight's new CVEs carry their hint on the patch plan. No-op unless
|
||||
# enabled; a hint only, changes no finding and no score.
|
||||
scheduler.add_job(
|
||||
jev_triage_nightly,
|
||||
trigger=CronTrigger(hour=6, minute=0),
|
||||
id="jev_triage_nightly",
|
||||
name="Nightly Jev Config Hint (off unless enabled)",
|
||||
replace_existing=True,
|
||||
)
|
||||
|
||||
# Nightly exploit-intel refresh (Plan M) at 03:45 — pulls
|
||||
# Exploit-DB CSV + PoC-in-GitHub + Metasploit module index, writes
|
||||
# per-vuln counts + ref lists.
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
"""
|
||||
Jev config hint — one structured question per CVE text, via OpenRouter Decisions.
|
||||
|
||||
Jev (TypeSafe) is a decision model: typed answer + probability, no prose. It
|
||||
is asked only what no feed answers — KEV / EUVD / EPSS / SSVC and the CVSS
|
||||
vector already cover "exploited?", "how exploitable?", "remote?":
|
||||
|
||||
Is this CVE only exploitable with an optional feature, module or
|
||||
non-default configuration enabled?
|
||||
|
||||
The answer (noul probability) is stored per CVE as a HINT the patch plan lists
|
||||
("check whether this applies to your setup"). It never closes a finding and
|
||||
never enters priority_score: a finding closes only when its scanners retract
|
||||
it (README.DEV.md, "Cross-source contract").
|
||||
|
||||
Off by default: JEV_TRIAGE_ENABLED env or setting `jev_triage_enabled`, plus
|
||||
the OpenRouter key the AI remediation already uses. Only public CVE text is
|
||||
sent — no hostnames, no asset data.
|
||||
"""
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Callable, Optional
|
||||
|
||||
import httpx
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.vulnerability import Vulnerability
|
||||
from app.services.ai_service import _cfg
|
||||
from app.services.patch_plan_service import open_work
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DECISIONS_URL = "https://openrouter.ai/api/alpha/decisions"
|
||||
# Pinned, not the ~latest alias: a silent model swap would shift every hint.
|
||||
DEFAULT_MODEL = "typesafe/jev-1.13"
|
||||
HTTP_TIMEOUT = 20.0
|
||||
MAX_CVES_PER_RUN = 500
|
||||
|
||||
# "noul" is Jev's yes/no answer type: the answer is the probability of yes.
|
||||
QUESTION_KEY = "non_default_config"
|
||||
QUESTION = {
|
||||
"type": "noul",
|
||||
"instructions": (
|
||||
"Based on the description, is this vulnerability only exploitable when an "
|
||||
"optional feature, module, plugin or non-default configuration is enabled, "
|
||||
"rather than in a default installation of the product?"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def parse_noul(data: dict) -> float:
|
||||
try:
|
||||
p = data["answers"][QUESTION_KEY]["noul"]
|
||||
except (KeyError, TypeError):
|
||||
raise ValueError(f"no {QUESTION_KEY} answer in Jev response")
|
||||
if isinstance(p, bool) or not isinstance(p, (int, float)) or not 0.0 <= p <= 1.0:
|
||||
raise ValueError(f"Jev returned {p!r}, not a probability")
|
||||
return float(p)
|
||||
|
||||
|
||||
def _cve_specific(e: Exception) -> bool:
|
||||
"""An unusable answer, or a 4xx that is about the request rather than the
|
||||
account (401/402/403 auth or credit, 429 rate limit)."""
|
||||
if isinstance(e, ValueError):
|
||||
return True
|
||||
if isinstance(e, httpx.HTTPStatusError):
|
||||
code = e.response.status_code
|
||||
return 400 <= code < 500 and code not in (401, 402, 403, 429)
|
||||
return False
|
||||
|
||||
|
||||
def ask_jev(api_key: str, model: str, state: dict) -> float:
|
||||
r = httpx.post(
|
||||
DECISIONS_URL,
|
||||
headers={"Authorization": f"Bearer {api_key}"},
|
||||
json={"model": model, "state": state, "questions": {QUESTION_KEY: QUESTION}},
|
||||
timeout=HTTP_TIMEOUT,
|
||||
)
|
||||
r.raise_for_status()
|
||||
return parse_noul(r.json())
|
||||
|
||||
|
||||
def run_jev_triage(db: Session, ask: Optional[Callable[[str, str, dict], float]] = None) -> dict:
|
||||
ask = ask or ask_jev
|
||||
enabled = _cfg(db, "JEV_TRIAGE_ENABLED", "jev_triage_enabled").lower() in ("1", "true", "yes")
|
||||
api_key = _cfg(db, "OPENROUTER_API_KEY", "openrouter_api_key")
|
||||
if not (enabled and api_key):
|
||||
return {"skipped": "disabled or no OpenRouter key", "checked": 0}
|
||||
model = _cfg(db, "JEV_MODEL", "jev_model", DEFAULT_MODEL)
|
||||
|
||||
# One row per CVE is enough — the question is about the CVE text.
|
||||
# ponytail: answered once, never re-asked; clear jev_checked_at to redo.
|
||||
pending = (
|
||||
open_work(db.query(Vulnerability.cve_id, Vulnerability.title,
|
||||
Vulnerability.description, Vulnerability.package_name))
|
||||
.filter(Vulnerability.jev_checked_at.is_(None))
|
||||
.filter(Vulnerability.cve_id.like("CVE-%"))
|
||||
.filter(Vulnerability.description.isnot(None))
|
||||
.order_by(Vulnerability.priority_score.desc())
|
||||
.all()
|
||||
)
|
||||
seen, stats = set(), {"checked": 0, "skipped_cves": [], "error": None}
|
||||
for cve, title, description, product in pending:
|
||||
if cve in seen:
|
||||
continue
|
||||
seen.add(cve)
|
||||
if len(seen) > MAX_CVES_PER_RUN:
|
||||
break
|
||||
try:
|
||||
p = ask(api_key, model, {"cve": cve, "product": product, "title": title,
|
||||
"description": description})
|
||||
except Exception as e:
|
||||
if _cve_specific(e):
|
||||
# This CVE's text or answer, not the API — skip it, go on;
|
||||
# otherwise it would stop the run at the same CVE every night.
|
||||
logger.warning("Jev triage skipped %s: %s", cve, e)
|
||||
stats["skipped_cves"].append(cve)
|
||||
continue
|
||||
# Auth, quota, network, server — systemic, so stop; everything
|
||||
# unanswered stays unstamped and the next run retries it.
|
||||
logger.error("Jev triage stopped at %s: %s", cve, e)
|
||||
stats["error"] = str(e)
|
||||
break
|
||||
db.query(Vulnerability).filter(Vulnerability.cve_id == cve).update(
|
||||
{"jev_config_dependent": p, "jev_checked_at": datetime.utcnow()},
|
||||
synchronize_session=False,
|
||||
)
|
||||
stats["checked"] += 1
|
||||
if stats["checked"] % 50 == 0:
|
||||
db.commit()
|
||||
db.commit()
|
||||
logger.info("Jev triage: %s", stats)
|
||||
return stats
|
||||
@@ -0,0 +1,103 @@
|
||||
"""
|
||||
Patch plan — open findings grouped by the product that has to be updated.
|
||||
|
||||
The findings list answers "what is wrong"; this answers "what do I patch
|
||||
first". Each group is one fix action (update product X) with what it removes:
|
||||
hosts, findings, CVEs, summed priority. Ranking is deterministic and uses only
|
||||
data the tool already trusts:
|
||||
|
||||
1. CVEs exploited in the wild (CISA KEV / ENISA EUVD) — any group with one
|
||||
comes first, most exploited CVEs first
|
||||
2. summed priority_score — the risk the update takes off the table
|
||||
|
||||
A finding that names several products (CVE-2026-16417: Chrome AND Edge) counts
|
||||
toward each: either update is work the operator has to do.
|
||||
|
||||
Fixed versions are listed as the scanners reported them, never compared —
|
||||
version order is family-specific and a wrong "latest" would be worse than none.
|
||||
|
||||
The Jev hint (jev_config_dependent) is only listed per group so the operator
|
||||
knows which CVEs to check for a non-default setup; it never changes the rank.
|
||||
"""
|
||||
from collections import defaultdict
|
||||
from typing import List
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.asset import Asset, AssetStatus
|
||||
from app.models.vulnerability import Vulnerability, VulnerabilityStatus
|
||||
from app.models.vulnerability_package import VulnerabilityPackage
|
||||
|
||||
# Open work only: pending_verification is already patched, awaiting a rescan.
|
||||
_WORK_STATES = (VulnerabilityStatus.open, VulnerabilityStatus.patch_failed)
|
||||
|
||||
|
||||
def open_work(query):
|
||||
"""Restrict a Vulnerability query to open work on ACTIVE assets.
|
||||
Shared with jev_triage_service so the hint covers exactly the plan's rows."""
|
||||
return (query.join(Asset, Asset.id == Vulnerability.asset_id)
|
||||
.filter(Asset.status == AssetStatus.ACTIVE)
|
||||
.filter(Vulnerability.status.in_(_WORK_STATES)))
|
||||
|
||||
|
||||
# Jev "noul" (its yes/no answer type: probability of yes) above which the
|
||||
# CVE is listed as "check your config".
|
||||
CONFIG_HINT_THRESHOLD = 0.7
|
||||
|
||||
|
||||
def build_patch_plan(db: Session, limit: int = 100) -> List[dict]:
|
||||
rows = open_work(db.query(
|
||||
Vulnerability.id, Vulnerability.asset_id, Vulnerability.cve_id,
|
||||
Vulnerability.package_name, Vulnerability.fixed_version,
|
||||
Vulnerability.priority_score, Vulnerability.kev_listed,
|
||||
Vulnerability.euvd_listed, Vulnerability.jev_config_dependent,
|
||||
)).all()
|
||||
if not rows:
|
||||
return []
|
||||
|
||||
# Per-product detail, one statement for the whole fleet.
|
||||
packages = defaultdict(list)
|
||||
for vid, name, fixed in open_work(
|
||||
db.query(VulnerabilityPackage.vulnerability_id, VulnerabilityPackage.package_name,
|
||||
VulnerabilityPackage.fixed_version)
|
||||
.join(Vulnerability, Vulnerability.id == VulnerabilityPackage.vulnerability_id)
|
||||
):
|
||||
packages[vid].append((name, fixed))
|
||||
|
||||
groups = {}
|
||||
for r in rows:
|
||||
# Findings without a product (pseudo-CVEs, OS KBs without a package)
|
||||
# become a one-CVE action of their own.
|
||||
for name, fixed in packages.get(r.id) or [(r.package_name or r.cve_id, r.fixed_version)]:
|
||||
key = name.strip().casefold()
|
||||
g = groups.setdefault(key, {
|
||||
"product": name.strip(), "hosts": set(), "findings": 0, "cves": set(),
|
||||
"fixed": set(), "exploited": set(), "config": set(),
|
||||
"priority_sum": 0.0, "priority_max": 0.0,
|
||||
})
|
||||
prio = r.priority_score or 0.0
|
||||
g["hosts"].add(r.asset_id)
|
||||
g["findings"] += 1
|
||||
g["cves"].add(r.cve_id)
|
||||
if fixed:
|
||||
g["fixed"].add(fixed)
|
||||
if r.kev_listed or r.euvd_listed:
|
||||
g["exploited"].add(r.cve_id)
|
||||
if (r.jev_config_dependent or 0) >= CONFIG_HINT_THRESHOLD:
|
||||
g["config"].add(r.cve_id)
|
||||
g["priority_sum"] += prio
|
||||
g["priority_max"] = max(g["priority_max"], prio)
|
||||
|
||||
plan = [{
|
||||
"product": g["product"],
|
||||
"hosts": len(g["hosts"]),
|
||||
"findings": g["findings"],
|
||||
"cve_count": len(g["cves"]),
|
||||
"fixed_versions": sorted(g["fixed"]),
|
||||
"exploited_cves": sorted(g["exploited"]),
|
||||
"config_dependent_cves": sorted(g["config"]),
|
||||
"priority_sum": round(g["priority_sum"], 1),
|
||||
"priority_max": round(g["priority_max"], 1),
|
||||
} for g in groups.values()]
|
||||
plan.sort(key=lambda g: (len(g["exploited_cves"]), g["priority_sum"]), reverse=True)
|
||||
return plan[:limit]
|
||||
@@ -0,0 +1,105 @@
|
||||
"use client";
|
||||
import { PageLoader } from "@/components/ui/Loading";
|
||||
|
||||
// Patch Plan — the findings list grouped into fix actions: "update X" with
|
||||
// what it removes. Ranked by the backend (patch_plan_service): exploited in
|
||||
// the wild (KEV / EUVD) first, then summed priority score. The Jev column is
|
||||
// a hint to check the setup, never part of the rank.
|
||||
import { useEffect, useState } from 'react';
|
||||
import api from '../../lib/api';
|
||||
|
||||
type Group = {
|
||||
product: string;
|
||||
hosts: number;
|
||||
findings: number;
|
||||
cve_count: number;
|
||||
fixed_versions: string[];
|
||||
exploited_cves: string[];
|
||||
config_dependent_cves: string[];
|
||||
priority_sum: number;
|
||||
priority_max: number;
|
||||
};
|
||||
|
||||
export default function PatchPlanPage() {
|
||||
const [plan, setPlan] = useState<Group[]>([]);
|
||||
const [loading, setLoading] = useState(true);
|
||||
const [error, setError] = useState('');
|
||||
|
||||
useEffect(() => {
|
||||
api.get('/api/v1/vulnerabilities/patch-plan')
|
||||
.then((r) => setPlan(r.data || []))
|
||||
.catch((e) => setError(e?.response?.data?.detail || 'Could not load the patch plan'))
|
||||
.finally(() => setLoading(false));
|
||||
}, []);
|
||||
|
||||
if (loading) return <PageLoader label="Loading patch plan…" />;
|
||||
|
||||
return (
|
||||
<div className="p-8">
|
||||
<h2 className="text-3xl font-bold text-gray-900 font-mono">Patch Plan</h2>
|
||||
<p className="mt-1 mb-6 text-sm text-gray-500">
|
||||
Open findings on active assets, grouped by the product to update. Exploited in the wild
|
||||
(CISA KEV / ENISA EUVD) first, then by the summed priority score the update removes.
|
||||
</p>
|
||||
{error && <p className="text-sm text-red-600 font-mono mb-4">{error}</p>}
|
||||
|
||||
<div className="bg-white border border-gray-200 shadow-sm rounded-sm overflow-x-auto">
|
||||
<table className="min-w-full divide-y divide-gray-200 text-sm">
|
||||
<thead className="bg-gray-50 text-xs font-mono uppercase text-gray-500">
|
||||
<tr>
|
||||
<th className="px-3 py-2 text-left">#</th>
|
||||
<th className="px-3 py-2 text-left">Update</th>
|
||||
<th className="px-3 py-2 text-left">Fixed in</th>
|
||||
<th className="px-3 py-2 text-right">Hosts</th>
|
||||
<th className="px-3 py-2 text-right">Findings</th>
|
||||
<th className="px-3 py-2 text-right">CVEs</th>
|
||||
<th className="px-3 py-2 text-left">Exploited</th>
|
||||
<th className="px-3 py-2 text-right" title="Summed / highest priority score">Priority Σ / max</th>
|
||||
<th className="px-3 py-2 text-left" title="Jev: CVE text suggests it needs an optional feature or non-default configuration. Check your setup; the rank is unaffected.">Check config</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody className="divide-y divide-gray-100 font-mono">
|
||||
{plan.length === 0 && !error && (
|
||||
<tr><td colSpan={9} className="px-3 py-4 text-gray-400">No open findings on active assets.</td></tr>
|
||||
)}
|
||||
{plan.map((g, i) => (
|
||||
<tr key={g.product} className={g.exploited_cves.length ? 'bg-red-50/40' : ''}>
|
||||
<td className="px-3 py-2 text-gray-400">{i + 1}</td>
|
||||
<td className="px-3 py-2">
|
||||
<a href={`/vulnerabilities?search=${encodeURIComponent(`"${g.product}"`)}`}
|
||||
className="font-semibold text-truevuln-blue hover:underline">{g.product}</a>
|
||||
</td>
|
||||
<td className="px-3 py-2 text-xs text-gray-600" title={g.fixed_versions.join(', ')}>
|
||||
{g.fixed_versions.slice(0, 3).join(', ') || '—'}
|
||||
{g.fixed_versions.length > 3 && ` +${g.fixed_versions.length - 3}`}
|
||||
</td>
|
||||
<td className="px-3 py-2 text-right">{g.hosts}</td>
|
||||
<td className="px-3 py-2 text-right">{g.findings}</td>
|
||||
<td className="px-3 py-2 text-right">{g.cve_count}</td>
|
||||
<td className="px-3 py-2 text-xs">
|
||||
{g.exploited_cves.length > 0 && (
|
||||
<span className="rounded px-1.5 py-0.5 bg-red-100 text-red-700 font-bold"
|
||||
title={g.exploited_cves.join(', ')}>
|
||||
{g.exploited_cves.length} KEV/EUVD
|
||||
</span>
|
||||
)}
|
||||
</td>
|
||||
<td className="px-3 py-2 text-right">
|
||||
{g.priority_sum} <span className="text-gray-400">/ {g.priority_max}</span>
|
||||
</td>
|
||||
<td className="px-3 py-2 text-xs">
|
||||
{g.config_dependent_cves.length > 0 && (
|
||||
<span className="rounded px-1.5 py-0.5 bg-amber-100 text-amber-800"
|
||||
title={g.config_dependent_cves.join(', ')}>
|
||||
{g.config_dependent_cves.length} CVE
|
||||
</span>
|
||||
)}
|
||||
</td>
|
||||
</tr>
|
||||
))}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
@@ -96,6 +96,7 @@ function OpenRouterCard() {
|
||||
const [keySet, setKeySet] = useState(false);
|
||||
const [model, setModel] = useState('openrouter/free');
|
||||
const [fallbacks, setFallbacks] = useState('');
|
||||
const [jev, setJev] = useState(false);
|
||||
const [status, setStatus] = useState<{ message: string; type: string }>({ message: '', type: '' });
|
||||
|
||||
useEffect(() => {
|
||||
@@ -108,6 +109,9 @@ function OpenRouterCard() {
|
||||
api.get('/api/v1/settings/openrouter_fallbacks')
|
||||
.then(r => { if (r.data?.value) setFallbacks(r.data.value); })
|
||||
.catch(() => {});
|
||||
api.get('/api/v1/settings/jev_triage_enabled')
|
||||
.then(r => setJev(r.data?.value === 'true'))
|
||||
.catch(() => {});
|
||||
}, []);
|
||||
|
||||
const save = async () => {
|
||||
@@ -121,6 +125,7 @@ function OpenRouterCard() {
|
||||
}
|
||||
await api.put('/api/v1/settings/openrouter_model', { value: model.trim() || 'openrouter/free' });
|
||||
await api.put('/api/v1/settings/openrouter_fallbacks', { value: fallbacks.trim() });
|
||||
await api.put('/api/v1/settings/jev_triage_enabled', { value: jev ? 'true' : 'false' });
|
||||
setStatus({ message: 'OpenRouter settings saved.', type: 'success' });
|
||||
} catch (e: any) {
|
||||
setStatus({ message: e?.response?.data?.detail || 'Save failed', type: 'error' });
|
||||
@@ -183,6 +188,15 @@ function OpenRouterCard() {
|
||||
className="block w-full rounded-md border-gray-300 shadow-sm focus:border-truevuln-blue focus:ring-truevuln-blue sm:text-sm h-11 px-3 font-mono"
|
||||
/>
|
||||
</div>
|
||||
<label className="flex items-start gap-2 text-xs text-gray-700 font-mono">
|
||||
<input type="checkbox" checked={jev} onChange={(e) => setJev(e.target.checked)}
|
||||
className="mt-0.5 h-3.5 w-3.5 rounded border-gray-300 text-truevuln-blue" />
|
||||
<span>
|
||||
Jev config hint (nightly, 06:00) — asks TypeSafe Jev per open CVE whether it only applies
|
||||
with a non-default configuration; shown on the Patch Plan. Sends public CVE text only.
|
||||
Never closes a finding or changes a score.
|
||||
</span>
|
||||
</label>
|
||||
</div>
|
||||
{status.message && (
|
||||
<p className={`mt-3 text-sm font-mono ${status.type === 'error' ? 'text-red-600' : status.type === 'success' ? 'text-green-600' : status.type === 'warning' ? 'text-amber-600' : 'text-blue-600 animate-pulse'}`}>{status.message}</p>
|
||||
|
||||
@@ -14,6 +14,7 @@ import {
|
||||
KeyIcon,
|
||||
CheckBadgeIcon,
|
||||
BellAlertIcon,
|
||||
WrenchScrewdriverIcon,
|
||||
} from '@heroicons/react/24/outline';
|
||||
import Link from 'next/link';
|
||||
import { usePathname } from 'next/navigation';
|
||||
@@ -34,6 +35,7 @@ type NavItem = {
|
||||
const navigation: NavItem[] = [
|
||||
{ name: 'Dashboard', href: '/', icon: Square2StackIcon },
|
||||
{ name: 'Vulnerabilities', href: '/vulnerabilities', icon: ShieldCheckIcon },
|
||||
{ name: 'Patch Plan', href: '/patch-plan', icon: WrenchScrewdriverIcon },
|
||||
{ name: 'Threat Intel', href: '/advisories', icon: BellAlertIcon },
|
||||
{ name: 'Assets', href: '/assets', icon: ServerIcon },
|
||||
{ name: 'Compliance', href: '/compliance', icon: CheckBadgeIcon },
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
"""Jev config hint — run: python tests/test_jev_triage.py
|
||||
|
||||
Jev (TypeSafe, via OpenRouter Decisions) answers one question per CVE text:
|
||||
"only exploitable with an optional feature / non-default configuration?".
|
||||
Nothing a feed answers — KEV, EPSS, SSVC and the CVSS vector already cover
|
||||
exploitation and attack vector — and nothing it decides: the probability is
|
||||
stored as a hint for the patch plan, never closes a finding, never enters
|
||||
priority_score.
|
||||
|
||||
Pinned here: off unless enabled, one call per CVE (not per finding), only open
|
||||
real CVEs on active hosts, a failing API stops the run without stamping
|
||||
anything, so the next night retries, and one CVE Jev cannot answer is skipped
|
||||
instead of stalling every night at the same place.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
os.environ.setdefault("JWT_SECRET_KEY", "test-key-test-key-test-key-test-key")
|
||||
|
||||
from datetime import datetime # noqa: E402
|
||||
|
||||
from sqlalchemy import create_engine # noqa: E402
|
||||
from sqlalchemy.orm import sessionmaker # noqa: E402
|
||||
|
||||
import app.models # noqa: F401,E402 (registers every mapper)
|
||||
from app.models.base import Base # noqa: E402
|
||||
from app.models.asset import Asset, AssetSource, AssetStatus # noqa: E402
|
||||
from app.models.setting import Setting # noqa: E402
|
||||
from app.models.vulnerability import ( # noqa: E402
|
||||
Vulnerability, VulnerabilitySeverity, VulnerabilityStatus,
|
||||
)
|
||||
from app.services import jev_triage_service # noqa: E402
|
||||
from app.services.jev_triage_service import parse_noul, run_jev_triage # noqa: E402
|
||||
|
||||
for var in ("JEV_TRIAGE_ENABLED", "OPENROUTER_API_KEY"):
|
||||
os.environ.pop(var, None)
|
||||
|
||||
|
||||
def _db(enabled=True):
|
||||
engine = create_engine("sqlite://")
|
||||
Base.metadata.create_all(engine)
|
||||
db = sessionmaker(bind=engine)()
|
||||
db.add(Setting(key="openrouter_api_key", value="sk-or-test"))
|
||||
db.add(Setting(key="jev_triage_enabled", value="true" if enabled else "false"))
|
||||
live = Asset(hostname="a", ip_address="10.0.0.1", source=AssetSource.MANUAL)
|
||||
gone = Asset(hostname="b", ip_address="10.0.0.2", source=AssetSource.MANUAL,
|
||||
status=AssetStatus.DECOMMISSIONED)
|
||||
db.add_all([live, gone])
|
||||
db.flush()
|
||||
|
||||
def v(asset, cve, status=VulnerabilityStatus.open, description="Apache mod_proxy ..."):
|
||||
db.add(Vulnerability(asset_id=asset.id, cve_id=cve, status=status, description=description,
|
||||
package_name="Apache HTTP Server", severity=VulnerabilitySeverity.high,
|
||||
detected_at=datetime(2026, 9, 1)))
|
||||
|
||||
v(live, "CVE-2026-1")
|
||||
v(gone, "CVE-2026-1") # same CVE, second row — no second call
|
||||
v(live, "CVE-2026-2")
|
||||
v(live, "CVE-2026-3", VulnerabilityStatus.false_positive)
|
||||
v(gone, "CVE-2026-4") # only on a decommissioned host
|
||||
v(live, "EOL-windows-10") # pseudo-CVE, not a CVE text
|
||||
v(live, "CVE-2026-5", description=None) # nothing to read
|
||||
db.commit()
|
||||
return db
|
||||
|
||||
|
||||
def _rows(db, cve):
|
||||
return db.query(Vulnerability).filter(Vulnerability.cve_id == cve).all()
|
||||
|
||||
|
||||
def test_off_unless_enabled():
|
||||
asked = []
|
||||
stats = run_jev_triage(_db(enabled=False), ask=lambda *a: asked.append(a) or 0.5)
|
||||
assert asked == [] and stats["skipped"], stats
|
||||
print("✅ disabled: Jev is never called")
|
||||
|
||||
|
||||
def test_one_call_per_open_cve():
|
||||
db = _db()
|
||||
asked = []
|
||||
|
||||
def ask(api_key, model, state):
|
||||
asked.append(state["cve"])
|
||||
return 0.9 if state["cve"] == "CVE-2026-1" else 0.1
|
||||
|
||||
stats = run_jev_triage(db, ask=ask)
|
||||
assert sorted(asked) == ["CVE-2026-1", "CVE-2026-2"], asked
|
||||
assert stats["checked"] == 2, stats
|
||||
# Every row of the CVE carries the answer, also the one on the gone host.
|
||||
assert [r.jev_config_dependent for r in _rows(db, "CVE-2026-1")] == [0.9, 0.9]
|
||||
assert all(r.jev_checked_at for r in _rows(db, "CVE-2026-1"))
|
||||
assert _rows(db, "CVE-2026-3")[0].jev_checked_at is None
|
||||
# Checked CVEs are not asked again.
|
||||
asked.clear()
|
||||
run_jev_triage(db, ask=ask)
|
||||
assert asked == [], asked
|
||||
print("✅ one call per open CVE on an active host; answered CVEs are not re-asked")
|
||||
|
||||
|
||||
def test_api_failure_stamps_nothing():
|
||||
db = _db()
|
||||
|
||||
def ask(*a):
|
||||
raise RuntimeError("HTTP 401")
|
||||
|
||||
stats = run_jev_triage(db, ask=ask)
|
||||
assert stats["error"] and stats["checked"] == 0, stats
|
||||
assert all(r.jev_checked_at is None for r in db.query(Vulnerability))
|
||||
print("✅ an API failure stops the run and leaves every CVE for the next night")
|
||||
|
||||
|
||||
def test_one_bad_cve_does_not_stall_the_run():
|
||||
db = _db()
|
||||
asked = []
|
||||
|
||||
def ask(api_key, model, state):
|
||||
asked.append(state["cve"])
|
||||
if state["cve"] == "CVE-2026-1":
|
||||
raise ValueError("Jev returned 'yes', not a probability")
|
||||
return 0.2
|
||||
|
||||
stats = run_jev_triage(db, ask=ask)
|
||||
assert sorted(asked) == ["CVE-2026-1", "CVE-2026-2"], asked
|
||||
assert stats["checked"] == 1 and stats["skipped_cves"] == ["CVE-2026-1"], stats
|
||||
assert _rows(db, "CVE-2026-2")[0].jev_checked_at is not None
|
||||
assert all(r.jev_checked_at is None for r in _rows(db, "CVE-2026-1"))
|
||||
print("✅ a CVE Jev cannot answer is skipped, the rest of the run goes on")
|
||||
|
||||
|
||||
def test_parse_noul():
|
||||
ok = {"answers": {jev_triage_service.QUESTION_KEY: {"type": "noul", "noul": 0.83}}}
|
||||
assert parse_noul(ok) == 0.83
|
||||
for bad in ({}, {"answers": {}}, {"answers": {jev_triage_service.QUESTION_KEY: {"noul": 1.7}}},
|
||||
{"answers": {jev_triage_service.QUESTION_KEY: {"noul": "yes"}}}):
|
||||
try:
|
||||
parse_noul(bad)
|
||||
except ValueError:
|
||||
continue
|
||||
raise AssertionError(f"accepted {bad}")
|
||||
print("✅ only a probability in [0, 1] is accepted from the response")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_off_unless_enabled()
|
||||
test_one_call_per_open_cve()
|
||||
test_api_failure_stamps_nothing()
|
||||
test_one_bad_cve_does_not_stall_the_run()
|
||||
test_parse_noul()
|
||||
print("\nAll Jev triage tests passed.")
|
||||
@@ -0,0 +1,116 @@
|
||||
"""Patch plan: where to start patching — run: python tests/test_patch_plan.py
|
||||
|
||||
The findings list answers "what is wrong"; it does not answer "what do I do
|
||||
first". One Chrome update closes a hundred rows, one KEV on a single host
|
||||
outranks all of them. The plan groups open findings by the product that has to
|
||||
be updated and ranks the groups: exploited-in-the-wild first, then by the
|
||||
priority the fix removes.
|
||||
|
||||
Pinned here: grouping by per-package rows (one CVE hitting two products counts
|
||||
toward both), only work that is actually open on ACTIVE hosts, exploited
|
||||
before bulk, and the Jev hint only ever annotates — it never reorders.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
os.environ.setdefault("JWT_SECRET_KEY", "test-key-test-key-test-key-test-key")
|
||||
|
||||
from datetime import datetime # noqa: E402
|
||||
|
||||
from sqlalchemy import create_engine # noqa: E402
|
||||
from sqlalchemy.orm import sessionmaker # noqa: E402
|
||||
|
||||
import app.models # noqa: F401,E402 (registers every mapper)
|
||||
from app.models.base import Base # noqa: E402
|
||||
from app.models.asset import Asset, AssetSource, AssetStatus # noqa: E402
|
||||
from app.models.vulnerability import ( # noqa: E402
|
||||
Vulnerability, VulnerabilitySeverity, VulnerabilityStatus,
|
||||
)
|
||||
from app.models.vulnerability_package import VulnerabilityPackage # noqa: E402
|
||||
from app.services.patch_plan_service import build_patch_plan # noqa: E402
|
||||
|
||||
|
||||
def _db():
|
||||
engine = create_engine("sqlite://")
|
||||
Base.metadata.create_all(engine)
|
||||
return sessionmaker(bind=engine)()
|
||||
|
||||
|
||||
def _host(db, name, status=AssetStatus.ACTIVE):
|
||||
a = Asset(hostname=name, ip_address="10.0.0.1", source=AssetSource.MANUAL, status=status)
|
||||
db.add(a)
|
||||
db.flush()
|
||||
return a
|
||||
|
||||
|
||||
def _vuln(db, asset, cve, pkg, prio, fixed=None, kev=False, status=VulnerabilityStatus.open, **kw):
|
||||
v = Vulnerability(asset_id=asset.id, cve_id=cve, package_name=pkg, fixed_version=fixed,
|
||||
priority_score=prio, kev_listed=kev, status=status,
|
||||
severity=VulnerabilitySeverity.high, detected_at=datetime(2026, 9, 1), **kw)
|
||||
db.add(v)
|
||||
db.flush()
|
||||
return v
|
||||
|
||||
|
||||
def _plan():
|
||||
db = _db()
|
||||
hosts = [_host(db, f"host-{i}") for i in range(5)]
|
||||
gone = _host(db, "gone", AssetStatus.DECOMMISSIONED)
|
||||
|
||||
# Chrome on every host: lots of priority, nothing exploited.
|
||||
for h in hosts:
|
||||
_vuln(db, h, "CVE-2026-1001", "Google Chrome", 20, fixed="140.0.1")
|
||||
_vuln(db, h, "CVE-2026-1002", "Google Chrome", 15, fixed="140.0.2")
|
||||
# One KEV on one host — little summed priority, but exploited.
|
||||
_vuln(db, hosts[0], "CVE-2026-2001", "OpenSSH", 30, fixed="9.9", kev=True)
|
||||
# Findings that are not open work must not count.
|
||||
_vuln(db, hosts[1], "CVE-2026-3001", "7-Zip", 90, status=VulnerabilityStatus.false_positive)
|
||||
_vuln(db, hosts[1], "CVE-2026-3002", "7-Zip", 90, status=VulnerabilityStatus.accepted_risk)
|
||||
_vuln(db, gone, "CVE-2026-3003", "7-Zip", 90)
|
||||
# One CVE, two products on one host (Chrome + Edge) — counts toward both.
|
||||
both = _vuln(db, hosts[2], "CVE-2026-4001", "Google Chrome", 10, jev_config_dependent=0.9)
|
||||
db.add(VulnerabilityPackage(vulnerability_id=both.id, package_name="Google Chrome", fixed_version="140.0.3"))
|
||||
db.add(VulnerabilityPackage(vulnerability_id=both.id, package_name="Microsoft Edge", fixed_version="140.0.9"))
|
||||
db.commit()
|
||||
return build_patch_plan(db)
|
||||
|
||||
|
||||
def test_exploited_ranks_before_bulk():
|
||||
plan = _plan()
|
||||
assert [g["product"] for g in plan] == ["OpenSSH", "Google Chrome", "Microsoft Edge"], plan
|
||||
assert plan[0]["exploited_cves"] == ["CVE-2026-2001"]
|
||||
print("✅ an exploited fix on one host outranks a bulk fix with more summed priority")
|
||||
|
||||
|
||||
def test_group_counts():
|
||||
chrome = next(g for g in _plan() if g["product"] == "Google Chrome")
|
||||
assert chrome["hosts"] == 5
|
||||
assert chrome["findings"] == 11
|
||||
assert chrome["cve_count"] == 3
|
||||
assert chrome["priority_sum"] == 5 * 35 + 10
|
||||
assert chrome["fixed_versions"] == ["140.0.1", "140.0.2", "140.0.3"]
|
||||
print("✅ hosts, findings, CVEs, priority and fixed versions roll up per product")
|
||||
|
||||
|
||||
def test_closed_and_inactive_are_not_work():
|
||||
assert "7-Zip" not in [g["product"] for g in _plan()]
|
||||
print("✅ false positives, accepted risk and decommissioned hosts are not in the plan")
|
||||
|
||||
|
||||
def test_jev_hint_annotates_only():
|
||||
plan = _plan()
|
||||
edge = next(g for g in plan if g["product"] == "Microsoft Edge")
|
||||
assert edge["config_dependent_cves"] == ["CVE-2026-4001"]
|
||||
chrome = next(g for g in plan if g["product"] == "Google Chrome")
|
||||
assert chrome["config_dependent_cves"] == ["CVE-2026-4001"]
|
||||
assert chrome["priority_sum"] == 185 # the hinted finding still counts in full
|
||||
print("✅ the Jev config hint is listed, the score is untouched")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_exploited_ranks_before_bulk()
|
||||
test_group_counts()
|
||||
test_closed_and_inactive_are_not_work()
|
||||
test_jev_hint_annotates_only()
|
||||
print("\nAll patch-plan tests passed.")
|
||||
Reference in New Issue
Block a user