mirror of
https://github.com/yogeshojha/rengine.git
synced 2026-10-21 15:51:30 +02:00
Yogesh OjhaandClaude Opus 5
e3b310b5b1
One evidence axis runs across both findings dimensions
(shared/definitions/evidence.py): inferred < observed < corroborated <
proven. Software rows store inferred or corroborated, set in match_scan from
a host and CVE join to non-info, non-suppressed findings; vulnerabilities
store observed or proven, proven being an out-of-band interaction recorded at
write. Both grammars gain an evidence: field and the old is:proven becomes
is:recorded. GET /cves and GET /cves/{cve} answer "am I affected by X" across
both dimensions, with an MCP tool and a toolbox lookup beside them. Ladder
counts are count(*) FILTER over the same predicates the searches compile, so
a rung opens exactly the rows it counted; tests/grammar/test_evidence.py
asserts it. match_scan carries discovered_at forward through a
software_previous temp table and returns the newly exposed delta for the
nightly notification; nvd_corpus.load skips the download when the mirror
stamp is unchanged.
The Attack surface group had grown to eight items with three of them sharing
a word. Vulnerabilities is now one entry parenting three sibling routes
joined by a tab rail: Findings, Software and CVEs. The routes are unchanged,
so vuln_q and sw_q state, deep links and per-route caching survive, and
NavItem.match keeps the parent lit on the siblings. The Software dimension
label drops to "Software" while its rows keep the noun "software CVE".
The CVE index gets the layout its siblings have: coverage strip, a search bar
attached to the card, a severity rail, a filter row with sort and refresh,
ListHeader rows and pagination. components/search-bar.svelte is the
plain-text twin of QueryBar for surfaces with no grammar. GET /cves gained
page, size, sort, dir, severity, kev, ransomware and evidence and returns
severity_counts; severity and the top rung are resolved in SQL from
SEVERITY_RANK and EVIDENCE_RANK rather than in Python.
Counts verified against the rows they open: overview.cves 1,165 equals the
index total, the Software tab 91,039 equals coverage.findings, and the
severity rail 157/373/583/52 sums to 1,165. _INDEX_TOTAL now excludes the
triage-suppressed findings the listing already excluded, and the software
dimension is counted exactly rather than capped at COUNT_CAP, so the tab
matches the page. The distinct-CVE count is 10s with JIT on and 91ms under
SET LOCAL jit = off, leaving the overview at 137ms cold and 17ms warm.
Also fixes software-table, which passed a 1-based page to ResultsPagination
where every other table passes a 0-based one, so page one read
"Showing 51-100".
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MirAGiQ2fSJJtikWi6YRNj
470 lines
16 KiB
Python
470 lines
16 KiB
Python
import uuid
|
|
from datetime import date, datetime
|
|
|
|
from pydantic import BaseModel
|
|
from pydantic import Field as PydanticField
|
|
from sqlalchemy import Column, Text
|
|
from sqlalchemy.types import JSON
|
|
from sqlmodel import Field, SQLModel, UniqueConstraint
|
|
|
|
from shared.definitions.asset_query import MAX_QUERY_LENGTH
|
|
from shared.definitions.evidence import Evidence
|
|
from shared.definitions.vulnerabilities import (
|
|
CoverageStatus,
|
|
Protocol,
|
|
Scanner,
|
|
Severity,
|
|
VulnState,
|
|
)
|
|
from shared.models.asset_query import MatchEvidence, QueryError
|
|
from shared.utils.datetime import utc_now
|
|
|
|
|
|
def _json_list() -> Field:
|
|
return Field(default_factory=list, sa_column=Column(JSON, nullable=False))
|
|
|
|
|
|
def _json_dict() -> Field:
|
|
return Field(default_factory=dict, sa_column=Column(JSON, nullable=False))
|
|
|
|
|
|
def _text() -> Field:
|
|
return Field(default=None, sa_column=Column(Text, nullable=True))
|
|
|
|
|
|
class Vulnerability(SQLModel, table=True):
|
|
__tablename__ = "vulnerabilities"
|
|
__table_args__ = (
|
|
UniqueConstraint("scan_id", "fingerprint", name="uq_vuln_scan_fingerprint"),
|
|
)
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
scan_id: uuid.UUID = Field(foreign_key="scans.id", index=True, ondelete="CASCADE")
|
|
target_id: uuid.UUID = Field(
|
|
foreign_key="targets.id", index=True, ondelete="CASCADE"
|
|
)
|
|
project_id: uuid.UUID = Field(foreign_key="projects.id", index=True)
|
|
|
|
fingerprint: str = Field(max_length=64, index=True)
|
|
scanner: str = Field(default=Scanner.NUCLEI.value, max_length=32, index=True)
|
|
|
|
# the check that fired
|
|
template_id: str = Field(max_length=200, index=True)
|
|
template_name: str = Field(max_length=500)
|
|
template_path: str | None = Field(default=None, max_length=500)
|
|
template_url: str | None = Field(default=None, max_length=1000)
|
|
severity: str = Field(default=Severity.UNKNOWN.value, max_length=16, index=True)
|
|
protocol: str = Field(default=Protocol.OTHER.value, max_length=16, index=True)
|
|
matcher_name: str | None = Field(default=None, max_length=200)
|
|
extractor_name: str | None = Field(default=None, max_length=200)
|
|
extracted_results: list = _json_list()
|
|
|
|
# what it says
|
|
description: str | None = _text()
|
|
impact: str | None = _text()
|
|
remediation: str | None = _text()
|
|
references: list = _json_list()
|
|
tags: list = _json_list()
|
|
authors: list = _json_list()
|
|
|
|
# how it is classified
|
|
cve_ids: list = _json_list()
|
|
cwe_ids: list = _json_list()
|
|
cvss_metrics: str | None = Field(default=None, max_length=200)
|
|
cvss_score: float | None = Field(default=None)
|
|
epss_score: float | None = Field(default=None)
|
|
epss_percentile: float | None = Field(default=None)
|
|
cpe: str | None = Field(default=None, max_length=300)
|
|
is_kev: bool = Field(default=False, index=True)
|
|
extra: dict = _json_dict()
|
|
|
|
exploit_score: int = Field(default=0, index=True)
|
|
intel_kinds: list = _json_list()
|
|
kev_ransomware: bool = Field(default=False, index=True)
|
|
kev_due_date: date | None = Field(default=None)
|
|
poc_count: int | None = Field(default=None)
|
|
template_available: bool | None = Field(default=None)
|
|
hackerone_reports: int | None = Field(default=None)
|
|
intel_at: datetime | None = Field(default=None)
|
|
|
|
# where it is
|
|
matched_at: str = Field(max_length=2000)
|
|
host: str | None = Field(default=None, max_length=500, index=True)
|
|
ip: str | None = Field(default=None, max_length=45, index=True)
|
|
port: int | None = Field(default=None, index=True)
|
|
scheme: str | None = Field(default=None, max_length=16)
|
|
url: str | None = Field(default=None, max_length=2000)
|
|
path: str | None = Field(default=None, max_length=2000)
|
|
|
|
subdomain_id: uuid.UUID | None = Field(default=None, index=True)
|
|
http_asset_id: uuid.UUID | None = Field(default=None, index=True)
|
|
port_id: uuid.UUID | None = Field(default=None, index=True)
|
|
|
|
# the proof
|
|
request: str | None = _text()
|
|
response: str | None = _text()
|
|
curl_command: str | None = _text()
|
|
interaction: dict = _json_dict()
|
|
evidence: str = Field(default=Evidence.OBSERVED.value, max_length=16, index=True)
|
|
|
|
observed_at: datetime | None = Field(default=None)
|
|
discovered_at: datetime = Field(default_factory=utc_now, index=True)
|
|
created_at: datetime = Field(default_factory=utc_now)
|
|
|
|
|
|
class VulnerabilityTriage(SQLModel, table=True):
|
|
"""Review state carried across every scan of a target, keyed by the finding's fingerprint."""
|
|
|
|
__tablename__ = "vulnerability_triage"
|
|
__table_args__ = (
|
|
UniqueConstraint("target_id", "fingerprint", name="uq_vulntriage_target_fp"),
|
|
)
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
project_id: uuid.UUID = Field(foreign_key="projects.id", index=True)
|
|
target_id: uuid.UUID = Field(
|
|
foreign_key="targets.id", index=True, ondelete="CASCADE"
|
|
)
|
|
fingerprint: str = Field(max_length=64, index=True)
|
|
template_id: str = Field(max_length=200)
|
|
matched_at: str = Field(max_length=2000)
|
|
state: str = Field(default=VulnState.OPEN.value, max_length=20, index=True)
|
|
note: str | None = Field(default=None, max_length=2000)
|
|
decided_by: uuid.UUID | None = Field(default=None)
|
|
created_at: datetime = Field(default_factory=utc_now)
|
|
updated_at: datetime = Field(default_factory=utc_now)
|
|
|
|
|
|
class VulnerabilityCoverage(SQLModel, table=True):
|
|
"""One scanner invocation's account of what it covered."""
|
|
|
|
__tablename__ = "vulnerability_coverage"
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
scan_id: uuid.UUID = Field(foreign_key="scans.id", index=True, ondelete="CASCADE")
|
|
target_id: uuid.UUID = Field(
|
|
foreign_key="targets.id", index=True, ondelete="CASCADE"
|
|
)
|
|
project_id: uuid.UUID = Field(foreign_key="projects.id", index=True)
|
|
|
|
scanner: str = Field(default=Scanner.NUCLEI.value, max_length=32)
|
|
group: str = Field(max_length=120)
|
|
status: str = Field(default=CoverageStatus.COMPLETED.value, max_length=16)
|
|
|
|
severities: list = _json_list()
|
|
template_sets: list = _json_list()
|
|
templates_selected: int | None = Field(default=None)
|
|
templates_loaded: int | None = Field(default=None)
|
|
custom_templates: int = Field(default=0)
|
|
|
|
hosts_total: int = Field(default=0)
|
|
hosts_scanned: int | None = Field(default=None)
|
|
hosts_dropped: list = _json_list()
|
|
requests_sent: int | None = Field(default=None)
|
|
requests_planned: int | None = Field(default=None)
|
|
matched: int | None = Field(default=None)
|
|
errors: int | None = Field(default=None)
|
|
|
|
rate_limit: int | None = Field(default=None)
|
|
concurrency: int | None = Field(default=None)
|
|
command: str | None = _text()
|
|
error: str | None = Field(default=None, max_length=2000)
|
|
|
|
started_at: datetime = Field(default_factory=utc_now)
|
|
ended_at: datetime | None = Field(default=None)
|
|
duration_seconds: float | None = Field(default=None)
|
|
|
|
|
|
class AssetContext(BaseModel):
|
|
"""The asset a finding sits on, as the scan already knows it."""
|
|
|
|
host: str | None = None
|
|
url: str | None = None
|
|
status_code: int | None = None
|
|
title: str | None = None
|
|
webserver: str | None = None
|
|
tech: list[str] = PydanticField(default_factory=list)
|
|
screenshot_path: str | None = None
|
|
ip: str | None = None
|
|
asn: int | None = None
|
|
asn_org: str | None = None
|
|
country: str | None = None
|
|
is_cdn: bool = False
|
|
cdn_name: str | None = None
|
|
waf: str | None = None
|
|
open_ports: int = 0
|
|
service_name: str | None = None
|
|
|
|
|
|
class Corroboration(BaseModel):
|
|
template_id: str
|
|
template_name: str
|
|
severity: str
|
|
scanner: str
|
|
basis: str
|
|
shared: list[str] = PydanticField(default_factory=list)
|
|
|
|
|
|
class VulnerabilityRead(BaseModel):
|
|
id: uuid.UUID
|
|
scan_id: uuid.UUID
|
|
target_id: uuid.UUID
|
|
target_value: str | None = None
|
|
fingerprint: str
|
|
scanner: str
|
|
template_id: str
|
|
template_name: str
|
|
template_path: str | None = None
|
|
template_url: str | None = None
|
|
severity: str
|
|
protocol: str
|
|
matcher_name: str | None = None
|
|
extractor_name: str | None = None
|
|
extracted_results: list[str] = PydanticField(default_factory=list)
|
|
description: str | None = None
|
|
impact: str | None = None
|
|
remediation: str | None = None
|
|
references: list[str] = PydanticField(default_factory=list)
|
|
tags: list[str] = PydanticField(default_factory=list)
|
|
authors: list[str] = PydanticField(default_factory=list)
|
|
cve_ids: list[str] = PydanticField(default_factory=list)
|
|
cwe_ids: list[str] = PydanticField(default_factory=list)
|
|
cvss_metrics: str | None = None
|
|
cvss_score: float | None = None
|
|
epss_score: float | None = None
|
|
epss_percentile: float | None = None
|
|
cpe: str | None = None
|
|
is_kev: bool = False
|
|
exploit_score: int = 0
|
|
intel_kinds: list[str] = PydanticField(default_factory=list)
|
|
kev_ransomware: bool = False
|
|
kev_due_date: date | None = None
|
|
poc_count: int | None = None
|
|
template_available: bool | None = None
|
|
hackerone_reports: int | None = None
|
|
matched_at: str
|
|
host: str | None = None
|
|
ip: str | None = None
|
|
port: int | None = None
|
|
scheme: str | None = None
|
|
url: str | None = None
|
|
path: str | None = None
|
|
curl_command: str | None = None
|
|
request: str | None = None
|
|
response: str | None = None
|
|
interaction: dict = PydanticField(default_factory=dict)
|
|
observed_at: datetime | None = None
|
|
discovered_at: datetime
|
|
state: str = VulnState.OPEN.value
|
|
note: str | None = None
|
|
is_new: bool = False
|
|
host_count: int = 1
|
|
sets: list[str] = PydanticField(default_factory=list)
|
|
corroborated_by: list[Corroboration] = PydanticField(default_factory=list)
|
|
colocated: int = 0
|
|
evidence: str = Evidence.OBSERVED.value
|
|
asset: AssetContext | None = None
|
|
matched_in: list[MatchEvidence] = PydanticField(default_factory=list)
|
|
|
|
|
|
class VulnerabilityFilter(BaseModel):
|
|
q: str | None = PydanticField(default=None, max_length=MAX_QUERY_LENGTH)
|
|
severities: list[str] = PydanticField(default_factory=list, max_length=8)
|
|
states: list[str] = PydanticField(default_factory=list, max_length=8)
|
|
protocols: list[str] = PydanticField(default_factory=list, max_length=12)
|
|
templates: list[str] = PydanticField(default_factory=list, max_length=200)
|
|
tags: list[str] = PydanticField(default_factory=list, max_length=100)
|
|
hosts: list[str] = PydanticField(default_factory=list, max_length=200)
|
|
scanners: list[str] = PydanticField(default_factory=list, max_length=8)
|
|
kev: bool = False
|
|
cve: bool = False
|
|
new: bool = False
|
|
corroborated: bool = False
|
|
include_info: bool = True
|
|
include_suppressed: bool = False
|
|
sort: str = PydanticField(default="risk", max_length=20)
|
|
order: str = PydanticField(default="desc", max_length=4)
|
|
limit: int = PydanticField(default=50, ge=1, le=200)
|
|
offset: int = PydanticField(default=0, ge=0, le=100_000_000)
|
|
|
|
def has_facets(self) -> bool:
|
|
return bool(
|
|
self.severities
|
|
or self.states
|
|
or self.protocols
|
|
or self.templates
|
|
or self.tags
|
|
or self.hosts
|
|
or self.scanners
|
|
or self.kev
|
|
or self.cve
|
|
or self.new
|
|
or self.corroborated
|
|
or not self.include_info
|
|
or self.include_suppressed
|
|
)
|
|
|
|
|
|
class VulnerabilityPage(BaseModel):
|
|
items: list[VulnerabilityRead] = PydanticField(default_factory=list)
|
|
total: int = 0
|
|
total_capped: bool = False
|
|
error: QueryError | None = None
|
|
|
|
|
|
class IssueRead(BaseModel):
|
|
"""One weakness: every observation of the same check, rolled up across the assets it fired on."""
|
|
|
|
template_id: str
|
|
template_name: str
|
|
template_url: str | None = None
|
|
scanner: str
|
|
severity: str
|
|
protocol: str
|
|
tags: list[str] = PydanticField(default_factory=list)
|
|
sets: list[str] = PydanticField(default_factory=list)
|
|
cve_ids: list[str] = PydanticField(default_factory=list)
|
|
cwe_ids: list[str] = PydanticField(default_factory=list)
|
|
cvss_score: float | None = None
|
|
epss_score: float | None = None
|
|
is_kev: bool = False
|
|
description: str | None = None
|
|
remediation: str | None = None
|
|
findings: int = 0
|
|
hosts: int = 0
|
|
addresses: int = 0
|
|
locations: int = 0
|
|
sample_hosts: list[str] = PydanticField(default_factory=list)
|
|
corroborated: int = 0
|
|
new_count: int = 0
|
|
states: dict[str, int] = PydanticField(default_factory=dict)
|
|
first_seen: datetime
|
|
last_seen: datetime
|
|
sample_id: uuid.UUID
|
|
|
|
|
|
class IssuePage(BaseModel):
|
|
items: list[IssueRead] = PydanticField(default_factory=list)
|
|
total: int = 0
|
|
total_capped: bool = False
|
|
error: QueryError | None = None
|
|
|
|
|
|
class SeverityCount(BaseModel):
|
|
severity: str
|
|
label: str
|
|
count: int
|
|
|
|
|
|
class VulnFacet(BaseModel):
|
|
name: str
|
|
count: int
|
|
label: str | None = None
|
|
|
|
|
|
class VulnerabilityFacets(BaseModel):
|
|
severity: list[VulnFacet] = PydanticField(default_factory=list)
|
|
state: list[VulnFacet] = PydanticField(default_factory=list)
|
|
protocol: list[VulnFacet] = PydanticField(default_factory=list)
|
|
template: list[VulnFacet] = PydanticField(default_factory=list)
|
|
tag: list[VulnFacet] = PydanticField(default_factory=list)
|
|
host: list[VulnFacet] = PydanticField(default_factory=list)
|
|
scanner: list[VulnFacet] = PydanticField(default_factory=list)
|
|
issue_severity: list[VulnFacet] = PydanticField(default_factory=list)
|
|
|
|
|
|
class CoverageRead(BaseModel):
|
|
id: uuid.UUID
|
|
scanner: str
|
|
group: str
|
|
status: str
|
|
severities: list[str] = PydanticField(default_factory=list)
|
|
template_sets: list[str] = PydanticField(default_factory=list)
|
|
templates_selected: int | None = None
|
|
templates_loaded: int | None = None
|
|
custom_templates: int = 0
|
|
hosts_total: int = 0
|
|
hosts_scanned: int | None = None
|
|
hosts_dropped: list[dict] = PydanticField(default_factory=list)
|
|
hosts_dropped_count: int = 0
|
|
requests_sent: int | None = None
|
|
matched: int | None = None
|
|
errors: int | None = None
|
|
rate_limit: int | None = None
|
|
duration_seconds: float | None = None
|
|
error: str | None = None
|
|
|
|
|
|
class RankedFinding(BaseModel):
|
|
id: uuid.UUID
|
|
template_id: str
|
|
name: str
|
|
severity: str
|
|
host: str | None = None
|
|
matched_at: str
|
|
host_count: int = 1
|
|
is_kev: bool = False
|
|
is_new: bool = False
|
|
cve_ids: list[str] = PydanticField(default_factory=list)
|
|
epss_score: float | None = None
|
|
signals: list[str] = PydanticField(default_factory=list)
|
|
|
|
|
|
class VulnHostRow(BaseModel):
|
|
host: str
|
|
total: int
|
|
worst: str
|
|
counts: list[SeverityCount] = PydanticField(default_factory=list)
|
|
query: str
|
|
|
|
|
|
class ScanVulnerabilities(BaseModel):
|
|
"""The Overview panel's model: the finding first, then the account of what was checked."""
|
|
|
|
ran: bool = False
|
|
total: int = 0
|
|
issues: int = 0
|
|
actionable: int = 0
|
|
new_count: int = 0
|
|
kev_count: int = 0
|
|
cve_count: int = 0
|
|
suppressed: int = 0
|
|
by_severity: list[SeverityCount] = PydanticField(default_factory=list)
|
|
headline: str = ""
|
|
headline_detail: str = ""
|
|
headline_query: str = ""
|
|
headline_tone: str = "muted"
|
|
affected_hosts: int = 0
|
|
scanned_hosts: int = 0
|
|
top_findings: list[RankedFinding] = PydanticField(default_factory=list)
|
|
top_hosts: list[VulnHostRow] = PydanticField(default_factory=list)
|
|
coverage: list[CoverageRead] = PydanticField(default_factory=list)
|
|
templates_run: int | None = None
|
|
requests_sent: int | None = None
|
|
duration_seconds: float | None = None
|
|
|
|
|
|
class TriageUpdate(BaseModel):
|
|
state: str
|
|
note: str | None = PydanticField(default=None, max_length=2000)
|
|
|
|
|
|
class TriageResult(BaseModel):
|
|
fingerprint: str
|
|
state: str
|
|
note: str | None = None
|
|
updated: int = 0
|
|
|
|
|
|
class BulkTriageUpdate(BaseModel):
|
|
fingerprints: list[str] = PydanticField(default_factory=list, max_length=500)
|
|
template_ids: list[str] = PydanticField(default_factory=list, max_length=100)
|
|
state: str
|
|
note: str | None = PydanticField(default=None, max_length=2000)
|
|
|
|
|
|
class BulkTriageResult(BaseModel):
|
|
state: str
|
|
fingerprints: int = 0
|
|
updated: int = 0
|