mirror of
https://github.com/yogeshojha/rengine.git
synced 2026-10-21 07:41:28 +02:00
Yogesh OjhaandClaude Opus 5
2cb8d73280
The comment rule in CLAUDE.md is one line maximum, never narration: no multi-line blocks, no module-header explainers, no sentences that explain or justify the code to the reader. The tree had drifted a long way from it, and most of the drift is mine. Removed: 232 lines of multi-line comment blocks 203 single-line comments that argued instead of stating 154 comments joined by so/because/otherwise/rather-than 143 narrating comments in the frontend 112 test docstrings (the test name already says what it asserts) 25 test module docstrings and 25 decorative section dividers Collapsed: 324 multi-line docstrings to one factual line 271 one-line docstrings that had a justifying clause bolted on Kept: functional directives, section dividers, short field labels, alembic revision headers, and the eight FastAPI route docstrings that render in /docs. Where a fact was load-bearing it is in CLAUDE.md, which is where it belongs. 284 python tests and 34 frontend tests pass, ruff and ruff-format clean across ten packages, svelte-check 0 errors, prettier and eslint clean, app.main imports. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013dKUV96WkQQETZBMoAG98C
466 lines
16 KiB
Python
466 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.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()
|
|
|
|
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
|
|
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)
|
|
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
|