@dataclass
class Finding:
"""Unified finding — one model for all issue types across all sources.
Phase 1 covers CVE findings (migrated from BlastRadius).
Phase 2 will add cloud CIS, proxy, SAST, skill findings.
"""
# Core identity
finding_type: FindingType
source: FindingSource
asset: Asset
severity: str # mirrors Severity enum value; str for forward-compat
# Explicit scope (issue #3946) — carried on the finding for query/filter
# convenience and mirrored onto the asset at ingest. All optional/nullable
# so existing findings serialize unchanged.
provider: Optional[str] = None # aws | azure | gcp | snowflake | ...
account_ref: Optional[str] = None # normalized ``<provider>:<account>``
region: Optional[str] = None
environment: Optional[str] = None # prod | staging | dev | ...
# Vendor severity (from source scanner) vs normalised CVSS severity
vendor_severity: Optional[str] = None # severity as reported by vendor/scanner
cvss_severity: Optional[str] = None # normalised from CVSS base score
# Finding content
title: str = ""
description: str = ""
cve_id: Optional[str] = None # e.g. "CVE-2024-1234"
cwe_ids: list[str] = field(default_factory=list) # e.g. ["CWE-79"]
cvss_score: Optional[float] = None
cvss_vector: Optional[str] = None
attack_vector: Optional[str] = None
attack_complexity: Optional[str] = None
privileges_required: Optional[str] = None
user_interaction: Optional[str] = None
network_exploitable: bool = False
epss_score: Optional[float] = None
is_kev: bool = False # CISA Known Exploited Vulnerability
is_malicious: bool = False # Known-malicious package (MAL- IDs, typosquat, etc.)
malicious_reason: Optional[str] = None
# Remediation
fixed_version: Optional[str] = None
remediation_guidance: Optional[str] = None
# Structured advisory remediation (fix + least-privilege-to-apply + optional
# artifact). Additive + optional: findings without it serialize unchanged.
# Read-only forever — agent-bom recommends, the user applies.
remediation: Optional["Remediation"] = None
# Compliance mappings (same tags as BlastRadius for parity)
compliance_tags: list[str] = field(default_factory=list) # all framework tags combined
# Framework slugs that govern this finding (set by compliance_hub.apply_hub_classification).
# Distinct from the per-framework `*_tags` fields below, which hold control codes.
applicable_frameworks: list[str] = field(default_factory=list)
controls: list[ControlTag] = field(default_factory=list)
owasp_tags: list[str] = field(default_factory=list)
atlas_tags: list[str] = field(default_factory=list)
attack_tags: list[str] = field(default_factory=list)
nist_ai_rmf_tags: list[str] = field(default_factory=list)
owasp_mcp_tags: list[str] = field(default_factory=list)
owasp_agentic_tags: list[str] = field(default_factory=list)
eu_ai_act_tags: list[str] = field(default_factory=list)
nist_csf_tags: list[str] = field(default_factory=list)
iso_27001_tags: list[str] = field(default_factory=list)
soc2_tags: list[str] = field(default_factory=list)
cis_tags: list[str] = field(default_factory=list)
cmmc_tags: list[str] = field(default_factory=list)
nist_800_53_tags: list[str] = field(default_factory=list)
fedramp_tags: list[str] = field(default_factory=list)
pci_dss_tags: list[str] = field(default_factory=list)
# Graph / correlation
related_findings: list[str] = field(default_factory=list) # IDs of related findings
evidence: dict = field(default_factory=dict) # raw evidence payload
# First-class graph FKs (optional, additive). ``node_id`` is the estate /
# asset UnifiedNode this finding attaches to; ``finding_node_id`` is the
# vulnerability/misconfiguration node (e.g. ``vuln:CVE-…``) when materialised.
# ``entity_type`` mirrors EntityType.value for the asset node when known.
node_id: Optional[str] = None
finding_node_id: Optional[str] = None
entity_type: Optional[str] = None
# Risk
risk_score: float = 0.0 # 0-10 unified risk score
reachability: Optional[str] = None
is_actionable: Optional[bool] = None
impact_category: Optional[str] = None
# Suppression state (mirrors BlastRadius; preserved through the unified stream
# so a suppressed finding never appears unsuppressed downstream)
suppressed: bool = False
suppression_id: Optional[str] = None
suppression_state: Optional[str] = None
suppression_reason: Optional[str] = None
unsuppressed_risk_score: Optional[float] = None
# AI-native risk context (mirrors BlastRadius)
ai_risk_context: Optional[str] = None
ai_summary: Optional[str] = None
attack_vector_summary: Optional[str] = None
# Reach / blast-radius lists — kept structured rather than collapsed to counts
affected_servers: list[str] = field(default_factory=list) # MCP server names on the impacted path
affected_agents: list[str] = field(default_factory=list) # agent names reachable along the path
exposed_credentials: list[str] = field(default_factory=list) # credential env-var names at risk
exposed_tools: list[str] = field(default_factory=list) # tool names accessible through the path
# CWPP runtime/EDR workload evidence (optional, additive). Never implies the
# workload is clean — summaries carry clean_workload_assertion=False.
workload_runtime_evidence: Optional[dict] = None
# Unique ID — deterministic UUID v5 based on content (computed in __post_init__)
# Pass an explicit id= to override (e.g. when ingesting from external scanner)
id: str = field(default="")
def __post_init__(self) -> None:
"""Compute stable ID from finding content if not explicitly set."""
from agent_bom.graph.severity import normalize_severity
self.severity = normalize_severity(self.severity)
# Keep finding scope and asset scope consistent: mirror finding-level
# scope down to the asset when the asset does not already carry it (and
# lift asset scope up when only the asset was populated). Non-cloud
# findings leave every field None, so this is a no-op for them.
for _scope_field in ("provider", "account_ref", "region", "environment"):
finding_val = getattr(self, _scope_field)
asset_val = getattr(self.asset, _scope_field, None)
if finding_val is not None and asset_val is None:
setattr(self.asset, _scope_field, finding_val)
elif finding_val is None and asset_val is not None:
setattr(self, _scope_field, asset_val)
if self.vendor_severity is not None:
self.vendor_severity = normalize_severity(self.vendor_severity)
if self.cvss_severity is not None:
self.cvss_severity = normalize_severity(self.cvss_severity)
self.controls = _dedupe_control_tags(
[
*(tag if isinstance(tag, ControlTag) else ControlTag.from_dict(tag) for tag in self.controls),
*self._legacy_control_tags(),
]
)
# Derive entity_type from asset_type aliases without mutating asset_type
# (asset_type feeds stable_id / finding id — must stay byte-stable).
if not self.entity_type:
try:
from agent_bom.graph.asset_entity import entity_type_for_asset_type
mapped = entity_type_for_asset_type(self.asset.asset_type)
if mapped is not None:
self.entity_type = mapped.value
except Exception: # noqa: BLE001 — finding construction must stay resilient
pass
if not self.id:
# Deterministic ID: same CVE on same asset always same ID
cve_part = self.vulnerability_id or self.title
pkg_name = ""
pkg_version = ""
if self.asset.asset_type == "package" and self.asset.identifier:
# purl like "pkg:pypi/torch@2.3.0" — extract name/version
purl = self.asset.identifier
pkg_part = purl.split("/")[-1] if "/" in purl else purl
if "@" in pkg_part:
pkg_name, pkg_version = pkg_part.rsplit("@", 1)
elif isinstance(self.evidence, dict):
# Asset is a server/container/etc — the affected package lives in
# evidence. Fold it into the discriminator so two CVEs on distinct
# packages under one asset don't collide on the same id.
pkg_name = str(self.evidence.get("package_name") or "")
pkg_version = str(self.evidence.get("package_version") or "")
self.id = canonical_finding_id(
self.asset.stable_id,
cve_part,
pkg_name,
pkg_version,
)
@property
def canonical_id(self) -> str:
"""Canonical alias for id used by report and graph consumers."""
return self.id
@property
def vulnerability_id(self) -> Optional[str]:
"""Canonical advisory identity, regardless of CVE/GHSA/OSV namespace.
``cve_id`` remains the wire-compatible legacy field. New producers may
populate ``evidence.vulnerability_id`` and consumers should prefer this
namespace-neutral alias when joining findings to advisories.
"""
if self.cve_id:
return self.cve_id
raw = self.evidence.get("vulnerability_id") if isinstance(self.evidence, dict) else None
return str(raw).strip() or None if raw is not None else None
@property
def advisory_ids(self) -> list[str]:
"""Return deterministic, de-duplicated CVE/GHSA/OSV advisory aliases."""
raw: list[object] = [self.vulnerability_id]
if isinstance(self.evidence, dict):
raw.extend(self.evidence.get("cve_ids") or [])
raw.extend(self.evidence.get("advisory_aliases") or [])
raw.extend(self.evidence.get("advisory_ids") or [])
seen: set[str] = set()
result: list[str] = []
for value in raw:
item = str(value or "").strip()
if item and item not in seen:
seen.add(item)
result.append(item)
return result
@property
def finding_category(self) -> str:
"""Stable category for consumers while legacy finding types remain intact."""
if self.finding_type is FindingType.CVE:
return "vulnerability"
if self.finding_type in {FindingType.CIS_FAIL, FindingType.CIS_ERROR}:
return "compliance"
return self.finding_type.value.lower()
def _legacy_control_tags(self) -> list[ControlTag]:
"""Return normalized controls derived from legacy tag arrays."""
tags: list[ControlTag] = []
for field_name, framework in LEGACY_CONTROL_FIELDS:
values = getattr(self, field_name)
for value in values:
if value:
tags.append(
ControlTag(
framework=framework,
control=str(value),
version=_LEGACY_CONTROL_VERSION_BY_FRAMEWORK.get(framework, "legacy"),
confidence=0.75,
source=f"legacy:{field_name}",
via=field_name,
)
)
return tags
def normalized_controls(self) -> list[ControlTag]:
"""Return deduplicated structured controls for this finding."""
return _dedupe_control_tags([*self.controls, *self._legacy_control_tags()])
def all_compliance_tags(self) -> list[str]:
"""Return deduplicated union of all compliance tag lists."""
seen: set[str] = set()
result: list[str] = []
for tag in (
self.compliance_tags
+ self.owasp_tags
+ self.atlas_tags
+ self.attack_tags
+ self.nist_ai_rmf_tags
+ self.owasp_mcp_tags
+ self.owasp_agentic_tags
+ self.eu_ai_act_tags
+ self.nist_csf_tags
+ self.iso_27001_tags
+ self.soc2_tags
+ self.cis_tags
+ self.cmmc_tags
+ self.nist_800_53_tags
+ self.fedramp_tags
+ self.pci_dss_tags
+ [tag.control for tag in self.normalized_controls()]
):
if tag not in seen:
seen.add(tag)
result.append(tag)
return result
def effective_severity(self) -> str:
"""Return the best severity value: vendor > cvss > base severity."""
return self.vendor_severity or self.cvss_severity or self.severity
@property
def security_domain(self) -> str:
"""Derived posture lane: one of cspm/vuln/aspm/dspm/aispm.
A pure function of source + finding type (+ evidence for the cloud
data-vs-config split), so the overview and findings surfaces route each
finding to exactly one coverage lane without double counting.
"""
from agent_bom.finding_scope import security_domain_for
return security_domain_for(self.source, self.finding_type, self.evidence)
def to_dict(self) -> dict:
"""Return a JSON-serializable finding payload."""
return {
"schema_version": FINDING_SCHEMA_VERSION,
"id": self.id,
"canonical_id": self.canonical_id,
"finding_type": self.finding_type.value,
"finding_category": self.finding_category,
"source": self.source.value,
"asset": {
"name": self.asset.name,
"asset_type": self.asset.asset_type,
"identifier": self.asset.identifier,
"location": self.asset.location,
"stable_id": self.asset.stable_id,
"canonical_id": self.asset.canonical_id,
"source_ids": self.asset.source_ids,
"provider": self.asset.provider,
"account_ref": self.asset.account_ref,
"region": self.asset.region,
"environment": self.asset.environment,
},
# First-class scope + taxonomy (issue #3946)
"provider": self.provider,
"account_ref": self.account_ref,
"region": self.region,
"environment": self.environment,
"security_domain": self.security_domain,
"severity": self.severity,
"effective_severity": self.effective_severity(),
"vendor_severity": self.vendor_severity,
"cvss_severity": self.cvss_severity,
"title": self.title,
"description": self.description,
"cve_id": self.cve_id,
"vulnerability_id": self.vulnerability_id,
"advisory_ids": self.advisory_ids,
"cve_ids": self.evidence.get("cve_ids") or ([self.cve_id] if self.cve_id else []),
"match_confidence_tier": self.evidence.get("match_confidence_tier"),
"advisory_aliases": self.evidence.get("advisory_aliases") or [],
"cwe_ids": self.cwe_ids,
"cvss_score": self.cvss_score,
"cvss_vector": self.cvss_vector,
"attack_vector": self.attack_vector,
"attack_complexity": self.attack_complexity,
"privileges_required": self.privileges_required,
"user_interaction": self.user_interaction,
"network_exploitable": self.network_exploitable,
"epss_score": self.epss_score,
"is_kev": self.is_kev,
"is_malicious": self.is_malicious,
"malicious_reason": self.malicious_reason,
"fixed_version": self.fixed_version,
"remediation_guidance": self.remediation_guidance,
# Structured advisory remediation — emitted only when populated so
# findings without it keep their existing serialization shape.
**({"remediation": self.remediation.to_dict()} if self.remediation is not None else {}),
"compliance_tags": self.all_compliance_tags(),
"applicable_frameworks": list(self.applicable_frameworks),
"controls": [tag.to_dict() for tag in self.normalized_controls()],
"owasp_tags": self.owasp_tags,
"atlas_tags": self.atlas_tags,
"attack_tags": self.attack_tags,
"nist_ai_rmf_tags": self.nist_ai_rmf_tags,
"owasp_mcp_tags": self.owasp_mcp_tags,
"owasp_agentic_tags": self.owasp_agentic_tags,
"eu_ai_act_tags": self.eu_ai_act_tags,
"nist_csf_tags": self.nist_csf_tags,
"iso_27001_tags": self.iso_27001_tags,
"soc2_tags": self.soc2_tags,
"cis_tags": self.cis_tags,
"cmmc_tags": self.cmmc_tags,
"nist_800_53_tags": self.nist_800_53_tags,
"fedramp_tags": self.fedramp_tags,
"pci_dss_tags": self.pci_dss_tags,
"related_findings": self.related_findings,
"evidence": self.evidence,
"node_id": self.node_id,
"finding_node_id": self.finding_node_id,
"entity_type": self.entity_type,
"risk_score": self.risk_score,
"reachability": self.reachability,
"is_actionable": self.is_actionable,
"impact_category": self.impact_category,
# Suppression state — a suppressed finding must never surface as
# unsuppressed downstream (mirrors BlastRadius / SARIF suppressions[]).
"suppressed": self.suppressed,
"suppression_id": self.suppression_id,
"suppression_state": self.suppression_state,
"suppression_reason": self.suppression_reason,
"unsuppressed_risk_score": self.unsuppressed_risk_score,
# AI-native risk context
"ai_risk_context": self.ai_risk_context,
"ai_summary": self.ai_summary,
"attack_vector_summary": self.attack_vector_summary,
# Structured reach / blast-radius lists (not collapsed to counts)
"affected_servers": list(self.affected_servers),
"affected_agents": list(self.affected_agents),
"exposed_credentials": list(self.exposed_credentials),
"exposed_tools": list(self.exposed_tools),
# CWPP runtime/EDR — omit when unset so plain findings stay unchanged
**(
{"workload_runtime_evidence": dict(self.workload_runtime_evidence)}
if isinstance(self.workload_runtime_evidence, dict)
else {}
),
}