autoscan — Copilot Tasks v2
Extending: CVE Feed, Confidence Scoring, H1 Auto-Draft, Live Probing
Read ARCHITECTURE.md + HOW_TO_EXTEND.md + autoscan_copilot_tasks.md first.
TASK 15 — Separate CVE Data from Code
Goal: Move hardcoded CVE_TRIGGERS dict out of Python into a JSON file. The scanner reads from JSON at runtime. The JSON gets auto-updated by Task 16.
Files to create/modify
- Create:
scanners/cve_data.json - Modify:
scanners/cve_trigger_runner.py - Create:
scanners/cve_data_schema.py(dataclass + loader)
cve_data.json structure
{
"version": "1.0",
"updated": "2026-05-21T00:00:00Z",
"entries": [
{
"cve_id": "PYSEC-2026-139",
"package": "torch",
"title": "pt2 Loading Handler deserialization RCE",
"triggers": ["torch.load("],
"safe_variants": ["torch.load($X, weights_only=True"],
"severity": "ERROR",
"confidence": "confirmed",
"owasp": "A06:2021-Vulnerable_and_Outdated_Components",
"category": "ml-security",
"remediation": "Add weights_only=True and upgrade torch>=2.6",
"affected_versions": "<2.10.0",
"fix_version": "2.10.0",
"references": ["https://osv.dev/vulnerability/PYSEC-2026-139"]
}
]
}
cve_data_schema.py
from dataclasses import dataclass
from pathlib import Path
import json
CVE_DATA_PATH = Path(__file__).parent / "cve_data.json"
@dataclass
class CVEEntry:
cve_id: str
package: str
title: str
triggers: list[str]
safe_variants: list[str]
severity: str
confidence: str
owasp: str
category: str
remediation: str
affected_versions: str
fix_version: str
references: list[str]
def load_cve_data() -> list[CVEEntry]:
"""Load CVE entries from JSON. Returns empty list on parse error."""
try:
data = json.loads(CVE_DATA_PATH.read_text(encoding="utf-8"))
return [CVEEntry(**e) for e in data.get("entries", [])]
except Exception:
return []
def save_cve_data(entries: list[CVEEntry], updated: str) -> None:
"""Persist updated entries back to JSON."""
CVE_DATA_PATH.write_text(
json.dumps({
"version": "1.0",
"updated": updated,
"entries": [e.__dict__ for e in entries],
}, indent=2, ensure_ascii=False),
encoding="utf-8",
)
Modify cve_trigger_runner.py
Replace the hardcoded CVE_TRIGGERS dict with:
from scanners.cve_data_schema import load_cve_data
def cve_trigger(work: str, pip_audit_findings=None):
entries = load_cve_data() # reads from JSON
cve_map = {e.cve_id: e for e in entries}
# rest of logic unchanged — iterate cve_map instead of CVE_TRIGGERS dict
TASK 16 — CVE Feed Updater (Scheduled Job)
File: sentinel/jobs/cve_feed.py
Trigger: APScheduler weekly job — add to sentinel/jobs/scheduler.py
Goal: Fetch new CVEs from OSV.dev, GitHub Advisories, NVD.
For each new/updated CVE, update scanners/cve_data.json.
Optionally call an LLM to extract trigger patterns (see LLM section below).
Function signatures
async def refresh_cve_feed() -> dict:
"""
Main entry point called by APScheduler weekly.
Returns summary: {added: N, updated: N, skipped: N, errors: [...]}
"""
async def fetch_osv(packages: list[str]) -> list[dict]:
"""
Query OSV.dev batch API for all packages.
Endpoint: POST https://api.osv.dev/v1/querybatch
No auth required.
"""
async def fetch_github_advisories(packages: list[str], token: str) -> list[dict]:
"""
Query GitHub Advisory Database via GraphQL.
Endpoint: POST https://api.github.com/graphql
Requires GITHUB_TOKEN env var (free, read-only).
"""
async def fetch_nvd(keyword: str, api_key: str | None = None) -> list[dict]:
"""
Query NVD CVE API v2.
Endpoint: GET https://services.nvd.nist.gov/rest/json/cves/2.0
No auth required (rate limited). Optional NVD_API_KEY for higher limits.
"""
async def extract_trigger_pattern(cve: dict) -> CVEEntry | None:
"""
Given raw CVE data, extract trigger pattern.
Strategy: rule-based first, LLM fallback (see below).
"""
Packages to watch
WATCHED_PACKAGES = [
"torch", "torchvision", "torchaudio",
"transformers", "diffusers", "accelerate",
"gradio", "gradio-client",
"numpy", "scipy",
"joblib", "scikit-learn",
"keras", "tensorflow",
"safetensors", "huggingface-hub",
"basicsr", "opencv-python",
"langchain", "langchain-core", "openai", "anthropic",
"datasets", "evaluate",
]
OSV.dev batch query
async def fetch_osv(packages: list[str]) -> list[dict]:
import httpx
queries = [
{"package": {"name": pkg, "ecosystem": "PyPI"}}
for pkg in packages
]
async with httpx.AsyncClient(timeout=30) as client:
r = await client.post(
"https://api.osv.dev/v1/querybatch",
json={"queries": queries},
)
r.raise_for_status()
results = r.json().get("results", [])
vulns = []
for pkg, result in zip(packages, results):
for v in result.get("vulns", []):
v["_package"] = pkg # tag which package
vulns.append(v)
return vulns
GitHub Advisory GraphQL query
GITHUB_ADVISORY_QUERY = """
query($after: String) {
securityAdvisories(
ecosystem: PIP
first: 100
after: $after
orderBy: {field: UPDATED_AT, direction: DESC}
) {
nodes {
ghsaId
summary
description
severity
publishedAt
updatedAt
cvss { score vectorString }
vulnerabilities(first: 10) {
nodes {
package { name ecosystem }
firstPatchedVersion { identifier }
vulnerableVersionRange
}
}
references { url }
}
pageInfo { hasNextPage endCursor }
}
}
"""
async def fetch_github_advisories(packages: list[str], token: str) -> list[dict]:
import httpx
pkg_set = set(p.lower() for p in packages)
results = []
cursor = None
async with httpx.AsyncClient(timeout=30) as client:
while True:
r = await client.post(
"https://api.github.com/graphql",
json={"query": GITHUB_ADVISORY_QUERY, "variables": {"after": cursor}},
headers={"Authorization": f"Bearer {token}"},
)
data = r.json()["data"]["securityAdvisories"]
for node in data["nodes"]:
for vuln in node["vulnerabilities"]["nodes"]:
if vuln["package"]["name"].lower() in pkg_set:
node["_package"] = vuln["package"]["name"]
node["_fix_version"] = (
vuln.get("firstPatchedVersion", {}) or {}
).get("identifier", "unknown")
results.append(node)
break
if not data["pageInfo"]["hasNextPage"]:
break
cursor = data["pageInfo"]["endCursor"]
return results
Trigger pattern extraction — how it works
This is the core intelligence question. Three strategies, use in order:
Strategy 1 — Rule-based keyword matching (fast, free, ~80% coverage)
TRIGGER_KEYWORDS = {
"torch.load": ["torch.load("],
"pickle": ["pickle.load(", "pickle.loads("],
"deserialization": ["torch.load(", "pickle.load(", "joblib.load("],
"from_pretrained": ["from_pretrained("],
"trust_remote_code": ["trust_remote_code=True"],
"jit.script": ["torch.jit.script("],
"allow_pickle": ["allow_pickle=True"],
"markdown": ["markdown.markdown(", "Markdown("],
"subprocess": ["subprocess.run(", "subprocess.call(", "os.system("],
"path traversal": ["open(", "send_file(", "FileResponse("],
"ssrf": ["requests.get(", "httpx.get(", "urllib.request"],
"code injection": ["exec(", "eval(", "compile("],
"oauth": ["gr.LoginButton(", "/login/huggingface"],
}
def extract_trigger_rule_based(description: str, summary: str) -> list[str]:
text = (description + " " + summary).lower()
triggers = []
for keyword, patterns in TRIGGER_KEYWORDS.items():
if keyword in text:
triggers.extend(patterns)
return list(set(triggers))
Strategy 2 — LLM call (for CVEs rule-based misses)
Use LLM when extract_trigger_rule_based() returns empty list.
Which LLM to use:
- Local Ollama (free, private): Best for production. No API costs. Use
qwen2.5-coder:7b— it understands code patterns well. Your RTX 5090 handles it trivially. - Claude API (accurate, costs money): Best quality. Use claude-haiku-20240307 for cost (cheap per call — ~$0.001 per CVE).
- OpenRouter (flexible): Good if you want to switch models without changing code. Use
meta-llama/llama-3.1-8b-instruct:freefor free tier.
Recommendation: Local Ollama first, Claude API fallback for low-confidence results.
# sentinel/config.py — add these settings
CVE_LLM_MODE: str = "local" # "local" | "claude" | "openrouter" | "off"
CVE_LLM_MODEL: str = "qwen2.5-coder:7b"
CVE_LLM_ENDPOINT: str = "http://localhost:11434/v1/chat/completions"
OPENROUTER_API_KEY: str = ""
# sentinel/jobs/cve_feed.py
CVE_TRIGGER_PROMPT = """You are a security scanner developer.
Given this CVE description, extract the Python function/pattern that
an attacker would call or that must be present in code for this CVE
to be exploitable.
CVE ID: {cve_id}
Package: {package}
Summary: {summary}
Description: {description}
Respond ONLY with a JSON object, no markdown, no explanation:
{{
"triggers": ["pattern1(", "pattern2("],
"safe_variants": ["pattern_with_safe_arg("],
"confidence": "confirmed|likely|possible",
"reasoning": "one sentence"
}}
Rules:
- triggers: Python code patterns to grep for (include the opening paren)
- safe_variants: patterns that indicate the safe usage (skip these in scanner)
- confidence: confirmed if trigger is obvious, possible if uncertain
- If no code-level trigger exists (e.g. CVE is in build tooling): return {{"triggers": []}}
"""
async def extract_trigger_llm(cve: dict, config) -> dict:
"""Call LLM to extract trigger pattern. Returns parsed JSON or {}."""
import httpx, json
prompt = CVE_TRIGGER_PROMPT.format(
cve_id=cve.get("id", ""),
package=cve.get("_package", ""),
summary=cve.get("summary", ""),
description=cve.get("details", "")[:2000], # truncate
)
if config.CVE_LLM_MODE == "off":
return {}
if config.CVE_LLM_MODE == "local":
# Ollama OpenAI-compatible endpoint
payload = {
"model": config.CVE_LLM_MODEL,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.0,
}
url = config.CVE_LLM_ENDPOINT
elif config.CVE_LLM_MODE == "claude":
# Anthropic API
import anthropic
client = anthropic.AsyncAnthropic()
msg = await client.messages.create(
model="claude-haiku-20240307",
max_tokens=256,
messages=[{"role": "user", "content": prompt}],
)
text = msg.content[0].text
try:
return json.loads(text)
except json.JSONDecodeError:
return {}
elif config.CVE_LLM_MODE == "openrouter":
payload = {
"model": "meta-llama/llama-3.1-8b-instruct:free",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.0,
}
url = "https://openrouter.ai/api/v1/chat/completions"
# For local/openrouter: shared httpx call
headers = {}
if config.CVE_LLM_MODE == "openrouter":
headers["Authorization"] = f"Bearer {config.OPENROUTER_API_KEY}"
async with httpx.AsyncClient(timeout=60) as client:
r = await client.post(url, json=payload, headers=headers)
text = r.json()["choices"][0]["message"]["content"]
try:
# Strip markdown fences if model added them
clean = text.strip().removeprefix("```json").removesuffix("```").strip()
return json.loads(clean)
except json.JSONDecodeError:
return {}
Full extraction pipeline
async def extract_trigger_pattern(cve: dict, config) -> CVEEntry | None:
"""
1. Try rule-based first (free, fast)
2. If empty, try LLM
3. If still empty, skip (no code-level trigger found)
"""
from datetime import datetime
# Rule-based
triggers = extract_trigger_rule_based(
cve.get("details", ""),
cve.get("summary", ""),
)
confidence = "likely"
if not triggers:
# LLM fallback
llm_result = await extract_trigger_llm(cve, config)
triggers = llm_result.get("triggers", [])
confidence = llm_result.get("confidence", "possible")
if not triggers:
return None # no code-level trigger — skip
# Map CVE severity to scanner severity
ghsa_sev = cve.get("severity", [{}])
cvss_score = (ghsa_sev[0].get("score", 0) if ghsa_sev else 0)
severity = "ERROR" if cvss_score >= 7 else "WARNING" if cvss_score >= 4 else "INFO"
return CVEEntry(
cve_id=cve.get("id", "UNKNOWN"),
package=cve.get("_package", ""),
title=cve.get("summary", "")[:200],
triggers=triggers,
safe_variants=[], # LLM may populate this
severity=severity,
confidence=confidence,
owasp="A06:2021-Vulnerable_and_Outdated_Components",
category="ml-security",
remediation=f"See {cve.get('references', [{}])[0].get('url', '')}",
affected_versions="",
fix_version=cve.get("_fix_version", "unknown"),
references=[r.get("url", "") for r in cve.get("references", [])],
)
Merge logic
async def refresh_cve_feed() -> dict:
from datetime import datetime, timezone
from scanners.cve_data_schema import load_cve_data, save_cve_data
existing = {e.cve_id: e for e in load_cve_data()}
summary = {"added": 0, "updated": 0, "skipped": 0, "errors": []}
# Fetch from all sources
raw_cves = []
raw_cves += await fetch_osv(WATCHED_PACKAGES)
if settings.GITHUB_TOKEN:
raw_cves += await fetch_github_advisories(WATCHED_PACKAGES, settings.GITHUB_TOKEN)
seen_ids = set()
for cve in raw_cves:
cve_id = cve.get("id", "")
if cve_id in seen_ids:
continue
seen_ids.add(cve_id)
try:
entry = await extract_trigger_pattern(cve, settings)
if entry is None:
summary["skipped"] += 1
continue
if cve_id in existing:
existing[cve_id] = entry # update
summary["updated"] += 1
else:
existing[cve_id] = entry # new
summary["added"] += 1
except Exception as e:
summary["errors"].append(f"{cve_id}: {e}")
save_cve_data(
list(existing.values()),
updated=datetime.now(timezone.utc).isoformat(),
)
# Create Notification so UI shows the update
# (use existing Notification model pattern from sentinel)
return summary
Register in APScheduler
# sentinel/jobs/scheduler.py — add:
from sentinel.jobs.cve_feed import refresh_cve_feed
scheduler.add_job(
refresh_cve_feed,
trigger="cron",
day_of_week="mon",
hour=6,
minute=0,
id="cve-feed-refresh",
replace_existing=True,
)
Add to sentinel/config.py
GITHUB_TOKEN: str = ""
NVD_API_KEY: str = "" # optional — higher rate limits
CVE_LLM_MODE: str = "local" # "local" | "claude" | "openrouter" | "off"
CVE_LLM_MODEL: str = "qwen2.5-coder:7b"
CVE_LLM_ENDPOINT: str = "http://localhost:11434/v1/chat/completions"
OPENROUTER_API_KEY: str = ""
TASK 17 — Confidence Scoring Layer
File: core/scoring.py
Goal: Replace binary found/not-found with a 0–10 score per finding.
High score = likely H1 reportable. Low score = noise.
Scoring factors
from dataclasses import dataclass
@dataclass
class ScoreFactors:
# Finding properties
severity: str # ERROR=3, WARNING=2, INFO=1
confidence: str # confirmed=3, likely=2, possible=1
has_user_input: bool # trigger receives user-controlled data +3
has_upload_widget: bool # gr.File/gr.UploadButton present in same file +2
is_employee_space: bool # owned by known HF employee +1
has_mcp_server: bool # mcp_server=True in repo +1
fix_version_exists: bool # known fix available (reportable) +1
cve_has_cvss: bool # CVSS score assigned (more credible) +1
EMPLOYEE_ACCOUNTS = frozenset({
"clem", "lysandre", "thomwolf", "julien-c",
"osanseviero", "sgugger", "victorsanh",
"reach-vb", "pcuenq", "nielsr",
})
def score_finding(finding: dict, repo_url: str, all_findings: list[dict]) -> int:
"""
Return integer score 0-10.
0-3 = noise / informational
4-6 = worth reviewing
7-8 = likely reportable
9-10 = H1 critical
"""
score = 0
# Base from severity + confidence
sev_map = {"ERROR": 3, "WARNING": 2, "INFO": 1}
conf_map = {"confirmed": 3, "likely": 2, "possible": 1}
score += sev_map.get(finding.get("severity", "INFO"), 1)
score += conf_map.get(finding.get("confidence", "possible"), 1)
# Has a user-input path to the trigger?
# Heuristic: same file has gr.Textbox, gr.File, or gr.Number
file_findings = [f for f in all_findings if f.get("file") == finding.get("file")]
has_input = any(
"gr.File" in f.get("message", "") or
"gr.UploadButton" in f.get("message", "") or
"gr.Textbox" in f.get("message", "")
for f in file_findings
)
if has_input:
score += 3
# Upload widget makes RCE findings immediately exploitable
has_upload = any(
"gr.File" in f.get("message", "") or
"gr.UploadButton" in f.get("message", "")
for f in all_findings
)
if has_upload and finding.get("severity") == "ERROR":
score += 2
# Employee account — higher impact
owner = repo_url.split("/")[-2] if "/" in repo_url else ""
if owner.lower() in EMPLOYEE_ACCOUNTS:
score += 1
# MCP server present — broader attack surface
has_mcp = any("mcp_server=True" in f.get("message", "") for f in all_findings)
if has_mcp:
score += 1
return min(score, 10)
Wire into post-processing in core/scanner.py
# In scan_repo(), after dedup_findings() and sort_findings():
from core.scoring import score_finding
for f in findings:
f["score"] = score_finding(f, repo_url, findings)
# Re-sort by score descending (within same severity bucket)
findings.sort(key=lambda f: (
SEVERITY_ORDER.get(f["severity"], 0),
CONFIDENCE_ORDER.get(f["confidence"], 0),
f.get("score", 0),
), reverse=True)
Display in Sentinel UI
In sentinel/templates/scan.html, add a score badge next to each finding:
<!-- Score badge — color by score value -->
{% if finding.score >= 9 %}
<span class="badge bg-red-100 text-red-800">Score {{ finding.score }}/10 — H1 Critical</span>
{% elif finding.score >= 7 %}
<span class="badge bg-orange-100 text-orange-800">Score {{ finding.score }}/10 — Reportable</span>
{% elif finding.score >= 4 %}
<span class="badge bg-yellow-100 text-yellow-800">Score {{ finding.score }}/10 — Review</span>
{% else %}
<span class="badge bg-gray-100 text-gray-500">Score {{ finding.score }}/10</span>
{% endif %}
Add to Finding ORM model
# sentinel/models/db.py — add to Finding model:
score: Mapped[int] = mapped_column(Integer, default=0)
Run alembic migration after adding.
TASK 18 — H1 Report Auto-Draft
File: sentinel/services/h1_draft.py
Goal: For findings with score >= 7, auto-generate a structured H1
report draft using the AI Explainer infrastructure.
Function signature
async def generate_h1_draft(finding: Finding, scan: Scan, target: Target) -> str:
"""
Generate a Markdown H1 report draft.
Uses ai_explain.py infrastructure (Ollama/Claude/OpenRouter).
Returns markdown string.
"""
Prompt template
H1_PROMPT = """You are a security researcher writing a HackerOne vulnerability report.
Write a professional, factual H1 report for this finding.
Target: {target_url}
Tool: {tool}
Rule: {rule}
Severity: {severity}
Confidence: {confidence}
Score: {score}/10
File: {file}
Line: {line}
Finding: {message}
Remediation: {remediation}
CVE: {cve}
OWASP: {owasp}
Write ONLY the markdown report with these exact sections:
## Summary
(2-3 sentences: what is vulnerable, what can an attacker do)
## Steps to Reproduce
(numbered list — assume local clone of the repo)
## Impact
(what data/systems are at risk, who is affected)
## Proof of Concept
(minimal curl or Python code — use placeholder URLs)
## Suggested Fix
(specific code change or version upgrade)
## CVSS
(one line: AV:N/AC:L/PR:N/UI:N/S:C/C:H/I:H/A:H or appropriate vector)
Be precise. Do not invent details not supported by the finding data.
Do not include a title line — the H1 platform adds that separately.
"""
LLM routing (reuse config from Task 16)
async def generate_h1_draft(finding, scan, target) -> str:
from sentinel.config import settings
prompt = H1_PROMPT.format(
target_url=target.url,
tool=finding.tool,
rule=finding.rule,
severity=finding.severity,
confidence=finding.confidence,
score=finding.score,
file=finding.file,
line=finding.line,
message=finding.message,
remediation=finding.remediation or "See CVE references",
cve=finding.rule if finding.rule.startswith(("CVE-", "PYSEC-")) else "N/A",
owasp=finding.owasp or "N/A",
)
# Reuse existing ai_explain infrastructure
from sentinel.services.ai_explain import _explain_with_prompt
return await _explain_with_prompt(prompt, max_tokens=1024)
Trigger condition
Auto-generate draft when:
- Finding is persisted with
score >= 7, AND AI_EXPLAINER_MODE != "off"
# In sentinel/services/scanner.py _persist_findings():
for f in high_score_findings:
if f.score >= 7 and settings.AI_EXPLAINER_MODE != "off":
draft = await generate_h1_draft(f, scan, target)
f.h1_draft = draft # add h1_draft column to Finding model
Add to Finding ORM model
# sentinel/models/db.py:
h1_draft: Mapped[str | None] = mapped_column(Text, nullable=True)
Sentinel UI — H1 Draft panel
In sentinel/templates/scan.html, for findings with score >= 7:
{% if finding.h1_draft %}
<details class="mt-2">
<summary class="text-xs font-medium text-blue-600 cursor-pointer">
📋 H1 Report Draft
</summary>
<div class="mt-2 p-3 bg-gray-50 rounded text-xs font-mono whitespace-pre-wrap">
{{ finding.h1_draft }}
</div>
<button onclick="copyToClipboard('{{ finding.id }}-draft')"
class="mt-1 text-xs text-blue-500 hover:underline">
Copy to clipboard
</button>
</details>
{% endif %}
TASK 19 — Weekly Prompt Template (for manual runs)
File: docs/weekly_update_prompt.md
Purpose: Paste this into Claude Project or Claude Code weekly.
# Weekly autoscan CVE Update
Search for new CVEs published in the last 7 days affecting these packages:
torch, transformers, gradio, numpy, joblib, keras, safetensors,
huggingface-hub, datasets, langchain, basicsr, diffusers
Sources to check:
- https://osv.dev/list?ecosystem=PyPI
- https://github.com/advisories?query=ecosystem%3Apip
- https://huntr.com/repos/huggingface/transformers
- https://huntr.com/repos/gradio-app/gradio
For each new CVE found:
1. Identify the Python function/pattern that must be present in code
for the CVE to be exploitable (the "trigger")
2. Output a JSON entry to add to scanners/cve_data.json:
{
"cve_id": "CVE-2026-XXXXX",
"package": "package-name",
"title": "Short title",
"triggers": ["function_name("],
"safe_variants": ["function_name(..., safe_param=True"],
"severity": "ERROR|WARNING|INFO",
"confidence": "confirmed|likely|possible",
"owasp": "A0X:2021-Category",
"category": "ml-security|security|llm",
"remediation": "Specific fix instruction",
"affected_versions": "<X.Y.Z",
"fix_version": "X.Y.Z",
"references": ["https://..."]
}
3. If a new Semgrep rule is needed (not covered by existing triggers),
output the YAML rule to add to the appropriate rules/*.yaml file.
4. Output the remediation.py entry:
"RULE-ID": "Plain English fix instruction.",
Output format: one JSON block per CVE, then YAML rules, then remediation entries.
Do not output anything else.
TASK 20 — Deep Research Prompt (Quarterly)
File: docs/quarterly_research_prompt.md
Purpose: Use Claude Deep Research quarterly to find new attack patterns.
# Quarterly ML Security Research
Research new attack patterns and vulnerability classes for HuggingFace Spaces
and ML deployment platforms published in the last 90 days.
Focus areas:
1. New model file format vulnerabilities (safetensors, GGUF, ONNX, NeMo, Flax)
2. New Gradio/Streamlit attack surfaces
3. HuggingFace Hub supply chain attacks
4. LLM agent tool-use security (MCP, function calling, tool injection)
5. New deserialization vectors in ML frameworks
6. Side-channel attacks on shared GPU inference
7. Model theft via API timing/output analysis
For each new pattern found:
- Describe the attack precisely
- Identify what code pattern to grep/semgrep for
- Estimate H1 severity (Critical/High/Medium)
- Describe what a scanner module would look like
Output as a structured list with one section per pattern.
Include academic paper references and CVE numbers where available.
TASK 21 — Notification: New CVE Alert in Sentinel UI
File: modify sentinel/jobs/cve_feed.py + sentinel/models/db.py
Goal: When CVE feed adds entries, create a Notification visible in the UI.
# At end of refresh_cve_feed():
if summary["added"] > 0:
async with AsyncSessionLocal() as db:
notif = Notification(
title=f"CVE Feed Updated: {summary['added']} new CVEs",
body=(
f"Added {summary['added']} new CVEs, "
f"updated {summary['updated']} existing. "
f"Re-scan active targets to apply new rules."
),
level="info",
created_at=datetime.now(timezone.utc),
)
db.add(notif)
await db.commit()
Add a "Rescan with new rules" button to the notification that triggers a fresh scan of all targets that were last scanned before the feed update.
Checklist — all new tasks
[ ] TASK 15: scanners/cve_data.json created with all entries from Task 01
[ ] TASK 15: scanners/cve_data_schema.py created
[ ] TASK 15: scanners/cve_trigger_runner.py reads from JSON not hardcode
[ ] TASK 16: sentinel/jobs/cve_feed.py created
[ ] TASK 16: fetch_osv() implemented and tested
[ ] TASK 16: fetch_github_advisories() implemented
[ ] TASK 16: extract_trigger_rule_based() implemented
[ ] TASK 16: extract_trigger_llm() implemented for all 3 modes
[ ] TASK 16: refresh_cve_feed() wired into APScheduler (weekly Monday 06:00)
[ ] TASK 16: sentinel/config.py has CVE_LLM_MODE, GITHUB_TOKEN, NVD_API_KEY
[ ] TASK 17: core/scoring.py created
[ ] TASK 17: score_finding() wired into scan_repo() post-processing
[ ] TASK 17: Finding ORM model has score column + migration
[ ] TASK 17: Score badge shown in scan.html
[ ] TASK 18: sentinel/services/h1_draft.py created
[ ] TASK 18: generate_h1_draft() called for score >= 7 findings
[ ] TASK 18: Finding ORM model has h1_draft column + migration
[ ] TASK 18: H1 draft panel shown in scan.html
[ ] TASK 19: docs/weekly_update_prompt.md created
[ ] TASK 20: docs/quarterly_research_prompt.md created
[ ] TASK 21: Notification created after feed update
[ ] TASK 21: "Rescan with new rules" button in notification
[ ] tests/test_cve_feed.py: covers fetch_osv, rule-based extraction, merge logic
[ ] tests/test_scoring.py: covers all score factors
[ ] tests/test_h1_draft.py: covers prompt construction, LLM routing
Architecture diagram — how everything connects
OSV.dev API ──────────────────────────┐
GitHub Advisories ────────────────────┤
NVD API ──────────────────────────────┤
▼
sentinel/jobs/cve_feed.py
(weekly APScheduler job)
│
rule-based extraction
│ (miss)
LLM extraction ──── local Ollama (default)
(Task 16) │ Claude API (fallback)
│ OpenRouter (alternative)
▼
scanners/cve_data.json ◄── git committed weekly
│
▼
scanners/cve_trigger_runner.py
(reads JSON at scan time)
│
▼
core/scanner.py scan_repo()
(runs all scanners in parallel)
│
▼
core/scoring.py score_finding()
(scores each finding 0-10)
│
score >= 7 ─┤─ score < 7
│ │
▼ ▼
h1_draft.py normal finding
LLM → markdown (no draft)
│
▼
sentinel DB: Finding(score=9, h1_draft="## Summary...")
│
▼
sentinel/templates/scan.html
Shows: score badge + H1 draft panel + copy button