Skip to content

Scanners

Vulnerability scanners that query OSV, NVD, and other databases.

agent_bom.scanners

Vulnerability scanning — re-exports for agent_bom.scanners public surface.

IncompleteScanError

Bases: RuntimeError

Raised when a scan cannot produce a trustworthy verdict.

Source code in src/agent_bom/scanners/package_scan.py 
183
184
class IncompleteScanError(RuntimeError):
    """Raised when a scan cannot produce a trustworthy verdict."""

ScanOptions dataclass

Immutable per-scan scanner controls.

The legacy module-level setters remain for CLI compatibility, but request handlers and concurrent scan callers should pass explicit options so tenant policy cannot bleed through shared module state.

Source code in src/agent_bom/scanners/package_scan.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@dataclass(frozen=True)
class ScanOptions:
    """Immutable per-scan scanner controls.

    The legacy module-level setters remain for CLI compatibility, but request
    handlers and concurrent scan callers should pass explicit options so tenant
    policy cannot bleed through shared module state.
    """

    offline: bool = False
    compliance_enabled: bool = False
    resolve_transitive: bool = False
    prefer_local_db: bool = False
    demo_advisories: bool = False
    # Target project directory (the ``-p`` path) for local install resolution.
    # When set, bare/floating versions are resolved against the target — its
    # virtualenv for pip, the dir itself for npm/go — never the scanner host.
    project_dir: Optional[str] = None

ScannerExecutionState

Bases: str, Enum

Whether a driver is executable today or declared as a roadmap slot.

Source code in src/agent_bom/scanners/base.py 
30
31
32
33
34
35
class ScannerExecutionState(str, Enum):
    """Whether a driver is executable today or declared as a roadmap slot."""

    ACTIVE = "active"
    PASSIVE = "passive"
    PLANNED = "planned"

ScannerFailureMode

Bases: str, Enum

How the orchestrator should treat driver failures.

Source code in src/agent_bom/scanners/base.py 
38
39
40
41
42
43
class ScannerFailureMode(str, Enum):
    """How the orchestrator should treat driver failures."""

    FAIL_CLOSED = "fail_closed"
    WARN_AND_CONTINUE = "warn_and_continue"
    SKIP_WHEN_UNAVAILABLE = "skip_when_unavailable"

ScannerPhase

Bases: str, Enum

Pipeline phase where a scanner driver contributes evidence.

Source code in src/agent_bom/scanners/base.py 
19
20
21
22
23
24
25
26
27
class ScannerPhase(str, Enum):
    """Pipeline phase where a scanner driver contributes evidence."""

    DISCOVERY = "discovery"
    EXTRACTION = "extraction"
    SCANNING = "scanning"
    ENRICHMENT = "enrichment"
    ANALYSIS = "analysis"
    OUTPUT = "output"

ScannerRegistration dataclass

Bases: RegistryEntry

Registry metadata for a scanner driver implementation.

Source code in src/agent_bom/scanners/base.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@dataclass(frozen=True)
class ScannerRegistration(RegistryEntry):
    """Registry metadata for a scanner driver implementation."""

    phase: ScannerPhase = ScannerPhase.SCANNING
    execution_state: ScannerExecutionState = ScannerExecutionState.ACTIVE
    failure_mode: ScannerFailureMode = ScannerFailureMode.WARN_AND_CONTINUE
    enabled_by_default: bool = True
    run_attr: str = ""
    input_types: tuple[str, ...] = ()
    output_types: tuple[str, ...] = ()
    finding_types: tuple[str, ...] = ()
    skip_when: tuple[str, ...] = ()
    telemetry_keys: tuple[str, ...] = ()
    standards: tuple[str, ...] = ()
    summary: str = ""

    def to_dict(self) -> dict[str, Any]:
        """Return a stable API/CLI-safe representation."""

        return {
            "name": self.name,
            "module": self.module,
            "source": self.source,
            "phase": self.phase.value,
            "execution_state": self.execution_state.value,
            "failure_mode": self.failure_mode.value,
            "enabled_by_default": self.enabled_by_default,
            "run_attr": self.run_attr,
            "input_types": list(self.input_types),
            "output_types": list(self.output_types),
            "finding_types": list(self.finding_types),
            "skip_when": list(self.skip_when),
            "telemetry_keys": list(self.telemetry_keys),
            "standards": list(self.standards),
            "summary": self.summary,
            "capabilities": {
                "scan_modes": list(self.capabilities.scan_modes),
                "required_scopes": list(self.capabilities.required_scopes),
                "permissions_used": list(self.capabilities.permissions_used),
                "outbound_destinations": list(self.capabilities.outbound_destinations),
                "data_boundary": self.capabilities.data_boundary,
                "writes": self.capabilities.writes,
                "network_access": self.capabilities.network_access,
                "guarantees": list(self.capabilities.guarantees),
            },
        }

to_dict 

to_dict() -> dict[str, Any]

Return a stable API/CLI-safe representation.

Source code in src/agent_bom/scanners/base.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def to_dict(self) -> dict[str, Any]:
    """Return a stable API/CLI-safe representation."""

    return {
        "name": self.name,
        "module": self.module,
        "source": self.source,
        "phase": self.phase.value,
        "execution_state": self.execution_state.value,
        "failure_mode": self.failure_mode.value,
        "enabled_by_default": self.enabled_by_default,
        "run_attr": self.run_attr,
        "input_types": list(self.input_types),
        "output_types": list(self.output_types),
        "finding_types": list(self.finding_types),
        "skip_when": list(self.skip_when),
        "telemetry_keys": list(self.telemetry_keys),
        "standards": list(self.standards),
        "summary": self.summary,
        "capabilities": {
            "scan_modes": list(self.capabilities.scan_modes),
            "required_scopes": list(self.capabilities.required_scopes),
            "permissions_used": list(self.capabilities.permissions_used),
            "outbound_destinations": list(self.capabilities.outbound_destinations),
            "data_boundary": self.capabilities.data_boundary,
            "writes": self.capabilities.writes,
            "network_access": self.capabilities.network_access,
            "guarantees": list(self.capabilities.guarantees),
        },
    }

expand_blast_radius_hops 

expand_blast_radius_hops(blast_radii: list[BlastRadius], agents: list[Agent], max_depth: int = 1) -> None

Expand blast radii with multi-hop delegation chain analysis.

Source code in src/agent_bom/scanners/blast_radius.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def expand_blast_radius_hops(
    blast_radii: list[BlastRadius],
    agents: list[Agent],
    max_depth: int = 1,
) -> None:
    """Expand blast radii with multi-hop delegation chain analysis."""
    max_depth = max(1, min(max_depth, 5))
    if max_depth <= 1:
        return

    server_to_agents: dict[str, list[Agent]] = {}
    for agent in agents:
        for server in agent.mcp_servers:
            server_to_agents.setdefault(server.name, []).append(agent)

    agent_to_servers: dict[str, list[str]] = {}
    for agent in agents:
        agent_to_servers[agent.name] = [server.name for server in agent.mcp_servers]

    for blast_radius in blast_radii:
        direct_agent_names = {agent.name for agent in blast_radius.affected_agents}
        direct_server_names = {server.name for server in blast_radius.affected_servers}

        visited_agents: set[str] = set(direct_agent_names)
        visited_servers: set[str] = set(direct_server_names)
        transitive_agents: list[dict] = []
        transitive_credentials: list[str] = []
        chains: list[str] = []

        queue: list[tuple[str, int, list[str]]] = []
        for agent in blast_radius.affected_agents:
            for server_name in agent_to_servers.get(agent.name, []):
                if server_name not in direct_server_names:
                    queue.append((agent.name, 1, [agent.name, server_name]))
                    visited_servers.add(server_name)

        max_hop_reached = 1
        while queue:
            _agent_name, hop, chain = queue.pop(0)
            if hop >= max_depth:
                continue

            current_server = chain[-1]
            for next_agent in server_to_agents.get(current_server, []):
                if next_agent.name in visited_agents:
                    continue
                visited_agents.add(next_agent.name)
                next_hop = hop + 1
                max_hop_reached = max(max_hop_reached, next_hop)

                new_chain = chain + [next_agent.name]
                chain_str = "\u2192".join(new_chain)
                chains.append(chain_str)

                agent_creds: list[str] = []
                for server in next_agent.mcp_servers:
                    agent_creds.extend(server.credential_names)
                agent_creds = list(set(agent_creds))

                transitive_agents.append(
                    {
                        "name": next_agent.name,
                        "type": next_agent.agent_type.value,
                        "hop": next_hop,
                        "chain": chain_str,
                    }
                )
                transitive_credentials.extend(agent_creds)

                if next_hop < max_depth:
                    for server_name in agent_to_servers.get(next_agent.name, []):
                        if server_name not in visited_servers:
                            visited_servers.add(server_name)
                            queue.append((next_agent.name, next_hop, new_chain + [server_name]))

        if transitive_agents:
            blast_radius.hop_depth = max_hop_reached
            blast_radius.delegation_chain = chains
            blast_radius.transitive_agents = transitive_agents
            blast_radius.transitive_credentials = list(set(transitive_credentials))
            factor = _HOP_RISK_FACTORS.get(max_hop_reached, 0.25)
            blast_radius.transitive_risk_score = round(blast_radius.risk_score * factor, 2)

build_vulnerabilities 

build_vulnerabilities(vuln_data_list: list[dict], package: Package) -> list[Vulnerability]

Convert OSV response data to Vulnerability objects.

Filters out false positives by verifying the package version falls within OSV affected ranges. Deduplicates by canonical CVE ID.

Source code in src/agent_bom/scanners/package_scan.py 
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
def build_vulnerabilities(vuln_data_list: list[dict], package: Package) -> list[Vulnerability]:
    """Convert OSV response data to Vulnerability objects.

    Filters out false positives by verifying the package version falls
    within OSV affected ranges.  Deduplicates by canonical CVE ID.
    """
    vulns = []
    seen_ids: set[str] = set()

    for vuln_data in vuln_data_list:
        vuln_id = vuln_data.get("id", "unknown")

        # Version-range filter: skip vulns that don't affect our version
        if package.version and package.version not in ("unknown", "latest"):
            if not _is_version_affected(
                vuln_data,
                package.name,
                package.version,
                package.ecosystem,
                source_package=package.source_package,
            ):
                _logger.debug(
                    "Filtered %s: version %s not in affected range for %s",
                    vuln_id,
                    package.version,
                    package.name,
                )
                continue

        from agent_bom.advisory_ids import canonical_vulnerability_id, match_confidence_tier

        aliases = vuln_data.get("aliases", [])
        canonical_id, all_aliases = canonical_vulnerability_id(vuln_id, aliases)

        # Deduplicate by canonical ID AND raw ID — prevents PYSEC/GHSA duplicates
        if canonical_id in seen_ids or vuln_id in seen_ids:
            continue
        seen_ids.add(canonical_id)
        seen_ids.add(vuln_id)
        # Also mark all aliases as seen to prevent future duplicates
        for alias in aliases:
            seen_ids.add(alias)

        severity, cvss_score, sev_source = parse_osv_severity(vuln_data)
        fixed = parse_fixed_version(
            vuln_data,
            package.name,
            package.ecosystem,
            current_version=package.version or "",
            source_package=package.source_package,
        )

        references = [ref.get("url", "") for ref in vuln_data.get("references", []) if ref.get("url")]

        summary = vuln_data.get("summary", vuln_data.get("details", "No description available"))[:200]

        # Extract CWE IDs from database_specific (GHSA entries store them here)
        cwe_ids: list[str] = []
        db_specific = vuln_data.get("database_specific", {})
        if isinstance(db_specific, dict):
            raw_cwes = db_specific.get("cwe_ids", [])
            if isinstance(raw_cwes, list):
                cwe_ids = [c for c in raw_cwes if isinstance(c, str) and c.startswith("CWE-")]

        from agent_bom.reachability_cve import (
            advisory_affected_symbols_by_path,
            advisory_affected_symbols_list,
        )

        affected_symbols = advisory_affected_symbols_list(vuln_data)
        affected_symbols_by_path = advisory_affected_symbols_by_path(vuln_data)

        vulns.append(
            Vulnerability(
                id=canonical_id,
                summary=summary,
                severity=severity,
                severity_source=sev_source,
                cvss_score=cvss_score,
                fixed_version=fixed,
                references=references,
                published_at=vuln_data.get("published"),
                modified_at=vuln_data.get("modified"),
                aliases=all_aliases,
                cwe_ids=cwe_ids,
                affected_symbols=affected_symbols,
                affected_symbols_by_path=affected_symbols_by_path,
                advisory_sources=["osv"],
                match_confidence_tier=match_confidence_tier(
                    advisory_source="osv",
                    db_ecosystem=None,
                    package_ecosystem=package.ecosystem,
                    fixed_version=fixed,
                ),
            )
        )

    _apply_distro_release_ambiguity(package, vulns)
    return vulns

create_client 

create_client(timeout: float | None = None, max_redirects: int = 0, *, cert: str | tuple[str, str] | None = None, verify: bool | str = True) -> httpx.AsyncClient

Create an httpx.AsyncClient with connection-level retries.

Uses httpx's built-in transport retry for connection failures (DNS, TCP reset). Application-level retries (429, 5xx) are handled by request_with_retry.

Parameters:

Name Type Description Default
timeout float | None

Per-request timeout in seconds.

 None
max_redirects int

Maximum redirects available if a caller explicitly enables redirects on a request. Redirect following is disabled by default so SSRF validation cannot be bypassed by a Location header.

 0
Source code in src/agent_bom/http_client.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def create_client(
    timeout: float | None = None,
    max_redirects: int = 0,
    *,
    cert: str | tuple[str, str] | None = None,
    verify: bool | str = True,
) -> httpx.AsyncClient:
    """Create an httpx.AsyncClient with connection-level retries.

    Uses httpx's built-in transport retry for connection failures (DNS, TCP reset).
    Application-level retries (429, 5xx) are handled by ``request_with_retry``.

    Args:
        timeout: Per-request timeout in seconds.
        max_redirects: Maximum redirects available if a caller explicitly
            enables redirects on a request. Redirect following is disabled by
            default so SSRF validation cannot be bypassed by a Location header.
    """
    check_offline()
    from agent_bom.config import HTTP_DEFAULT_TIMEOUT

    if timeout is None:
        timeout = HTTP_DEFAULT_TIMEOUT
    transport = None if _env_proxy_configured() else httpx.AsyncHTTPTransport(retries=2)
    return httpx.AsyncClient(
        timeout=timeout,
        transport=transport,
        follow_redirects=False,
        max_redirects=max_redirects,
        verify=verify,
        cert=cert,
    )

deduplicate_packages 

deduplicate_packages(packages: list) -> list

Remove duplicate packages across discovery sources.

Deduplicates by (ecosystem, normalized_name, version) fingerprint. When duplicates exist, the first occurrence is kept (preserves source ordering).

This prevents redundant OSV API calls and duplicate vulnerability findings when the same package is discovered from multiple sources (local, K8s, cloud).

Parameters:

Name Type Description Default
packages list

List of Package objects from one or more discovery sources.

 required

Returns:

Type Description
list

Deduplicated list, preserving first-seen order.
Source code in src/agent_bom/scanners/package_scan.py 
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
def deduplicate_packages(packages: list) -> list:
    """Remove duplicate packages across discovery sources.

    Deduplicates by (ecosystem, normalized_name, version) fingerprint.
    When duplicates exist, the first occurrence is kept (preserves source ordering).

    This prevents redundant OSV API calls and duplicate vulnerability findings
    when the same package is discovered from multiple sources (local, K8s, cloud).

    Args:
        packages: List of Package objects from one or more discovery sources.

    Returns:
        Deduplicated list, preserving first-seen order.
    """
    seen: set[tuple[str, str, str]] = set()
    result = []
    for pkg in packages:
        # Use normalized name for dedup (PEP 503: torch == Torch == pytorch)
        name = getattr(pkg, "name", "") or ""
        ecosystem = getattr(pkg, "ecosystem", "") or ""
        version = getattr(pkg, "version", "") or ""
        key = canonical_package_identity(name, version, ecosystem, getattr(pkg, "purl", None))
        if key not in seen:
            seen.add(key)
            result.append(pkg)
    return result

default_scan_options 

default_scan_options(*, compliance_enabled: bool = False, resolve_transitive: bool = False, prefer_local_db: bool | None = None, offline: bool | None = None, demo_advisories: bool = False, project_dir: str | None = None) -> ScanOptions

Build per-scan options while preserving legacy offline defaults.

Source code in src/agent_bom/scanners/package_scan.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def default_scan_options(
    *,
    compliance_enabled: bool = False,
    resolve_transitive: bool = False,
    prefer_local_db: bool | None = None,
    offline: bool | None = None,
    demo_advisories: bool = False,
    project_dir: str | None = None,
) -> ScanOptions:
    """Build per-scan options while preserving legacy offline defaults."""

    return ScanOptions(
        offline=_scanners_patchable("offline_mode") if offline is None else offline,
        compliance_enabled=compliance_enabled,
        resolve_transitive=resolve_transitive,
        prefer_local_db=(prefer_local_db if prefer_local_db is not None else _scanners_patchable("prefer_local_db")),
        demo_advisories=demo_advisories,
        project_dir=project_dir,
    )

parse_fixed_version 

parse_fixed_version(vuln_data: dict, package_name: str, ecosystem: str = '', current_version: str = '', source_package: str | None = None, allow_prerelease: bool = False) -> Optional[str]

Extract fixed version from OSV affected data.

Source code in src/agent_bom/scanners/osv.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def parse_fixed_version(
    vuln_data: dict,
    package_name: str,
    ecosystem: str = "",
    current_version: str = "",
    source_package: str | None = None,
    allow_prerelease: bool = False,
) -> Optional[str]:
    """Extract fixed version from OSV affected data."""
    from agent_bom.version_utils import (
        compare_version_order,
        is_prerelease_version,
        version_in_range,
    )

    norm_inputs = candidate_package_names(package_name, ecosystem, source_package)
    prerelease_candidate: Optional[str] = None
    # ``same_branch_fix`` is the fix from the affected branch that actually
    # CONTAINS the installed version (introduced <= current < fixed); it is
    # preferred so a multi-branch advisory never advises a cross-branch jump
    # (e.g. urllib3 1.26.4 -> 1.26.18, not the 2.x fix 2.0.7). ``fallback_fix``
    # (earliest valid fix) is used only when no branch contains the version.
    same_branch_fix: Optional[str] = None
    fallback_fix: Optional[str] = None
    has_current = bool(current_version and current_version not in ("unknown", "latest", ""))

    def _consider_fallback(candidate: str) -> None:
        nonlocal fallback_fix
        if fallback_fix is None:
            fallback_fix = candidate
            return
        order = compare_version_order(candidate, fallback_fix, ecosystem)
        if order is not None and order < 0:
            fallback_fix = candidate

    for affected in vuln_data.get("affected", []):
        pkg = affected.get("package", {})
        pkg_name = pkg.get("name", "")
        if not pkg_name:
            _logger.debug("Skipping affected entry with empty package name in %s", vuln_data.get("id", "?"))
            continue
        osv_eco = pkg.get("ecosystem", ecosystem)
        if not ecosystem_matches(osv_eco, ecosystem):
            _logger.debug(
                "Skipping cross-ecosystem affected entry %s/%s (want %s) in %s",
                osv_eco,
                pkg_name,
                ecosystem,
                vuln_data.get("id", "?"),
            )
            continue
        osv_norm = normalize_package_name(pkg_name, osv_eco)
        if osv_norm not in norm_inputs:
            continue
        for rng in affected.get("ranges", []):
            introduced: Optional[str] = None
            for event in rng.get("events", []):
                if "introduced" in event:
                    introduced = event.get("introduced") or None
                    continue
                if "fixed" not in event:
                    continue
                fixed = event["fixed"]
                if not is_valid_fix_version(fixed):
                    continue
                try:
                    if has_current:
                        current_cmp = compare_version_order(current_version, fixed, ecosystem)
                        if current_cmp is not None and current_cmp > 0:
                            _logger.debug(
                                "Skipping fix %s < current %s for %s",
                                fixed,
                                current_version,
                                package_name,
                            )
                            continue
                    prerelease = is_prerelease_version(fixed, ecosystem)
                except Exception as exc:  # noqa: BLE001
                    _logger.debug("Version parse failed for %r: %s", fixed, exc)
                    if has_current:
                        current_cmp = compare_version_order(current_version, fixed, ecosystem)
                        if current_cmp is not None and current_cmp > 0:
                            continue
                    prerelease = False

                if prerelease:
                    if prerelease_candidate is None:
                        prerelease_candidate = fixed
                    continue

                if has_current and same_branch_fix is None and version_in_range(current_version, introduced, fixed, None, ecosystem):
                    same_branch_fix = fixed
                _consider_fallback(fixed)

    if same_branch_fix is not None:
        return same_branch_fix
    if fallback_fix is not None:
        return fallback_fix
    if allow_prerelease:
        return prerelease_candidate
    if prerelease_candidate:
        _logger.debug("Suppressing prerelease-only fix %s for %s", prerelease_candidate, package_name)
    return None

query_osv_batch async 

query_osv_batch(packages: list[Package]) -> dict[str, list[dict]]

Query OSV API for vulnerabilities in batch.

Source code in src/agent_bom/scanners/package_scan.py 
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
async def query_osv_batch(packages: list[Package]) -> dict[str, list[dict]]:
    """Query OSV API for vulnerabilities in batch."""
    return await query_osv_batch_impl(
        packages,
        console=console,
        get_scan_cache=_scanners_patchable("_get_scan_cache"),
        get_api_semaphore=_get_api_semaphore,
        bump_scan_perf=_bump_scan_perf,
        enrich_results_if_needed_fn=_scanners_patchable("_enrich_results_if_needed"),
        record_scan_warning=_scanners_patchable("record_scan_warning"),
        osv_ecosystems_for_package=_osv_ecosystems_for_package,
        non_osv_ecosystems=_NON_OSV_ECOSYSTEMS,
        create_client_fn=_scanners_patchable("create_client"),
        request_with_retry_fn=_scanners_patchable("request_with_retry"),
    )

request_with_retry async 

request_with_retry(client: AsyncClient, method: str, url: str, max_retries: int = MAX_RETRIES, **kwargs: Any) -> Optional[httpx.Response]

Make an HTTP request with exponential backoff on retryable errors.

Handles: - 429 Too Many Requests (respects Retry-After header) - 5xx server errors - Connection timeouts and network errors

Returns:

Type Description
Optional[Response]

httpx.Response on success, None on exhausted retries.
Source code in src/agent_bom/http_client.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
async def request_with_retry(
    client: httpx.AsyncClient,
    method: str,
    url: str,
    max_retries: int = MAX_RETRIES,
    **kwargs: Any,
) -> Optional[httpx.Response]:
    """Make an HTTP request with exponential backoff on retryable errors.

    Handles:
    - 429 Too Many Requests (respects Retry-After header)
    - 5xx server errors
    - Connection timeouts and network errors

    Returns:
        httpx.Response on success, None on exhausted retries.
    """
    check_offline()
    # Defense-in-depth: validate and re-derive the URL at the transport layer.
    # validate_url() raises SecurityError on SSRF attempts (private IPs,
    # localhost, metadata endpoints, non-HTTPS, DNS rebinding).
    # Re-constructing the URL from parsed components ensures CodeQL sees
    # the taint is broken.
    from urllib.parse import urlparse, urlunparse

    from agent_bom.security import validate_url as _validate_url  # noqa: E402

    _validate_url(url)  # raises SecurityError on SSRF attempts
    # Re-derive URL from parsed components to break CodeQL taint chain
    _parsed = urlparse(url)
    safe_url = urlunparse(_parsed)

    log_url = _safe_url(safe_url)
    host = _host_of(safe_url)
    backoff = INITIAL_BACKOFF

    # Breaker already open for this host: skip the network entirely so the
    # caller falls through to cached/bundled data without backoff or warnings.
    if host and registry_breaker_tripped(host):
        logger.debug("Rate-limit breaker open for %s — skipping live request to %s", host, log_url)
        return None

    for attempt in range(max_retries + 1):
        try:
            response = await client.request(method, safe_url, **kwargs)

            if not _should_retry_status(response.status_code, safe_url):
                _record_non_rate_limited(host)
                return response

            # Sustained 429s trip the per-host breaker: stop retrying this host
            # immediately and return the 429 so the caller can fall back fast.
            if response.status_code == 429 and _record_rate_limit(host):
                logger.debug("Rate-limit breaker tripped for %s on HTTP 429 — short-circuiting %s", host, log_url)
                return response

            # Retryable status — check Retry-After header
            retry_after = response.headers.get("Retry-After")
            if retry_after:
                try:
                    wait = min(float(retry_after), MAX_BACKOFF)
                except ValueError:
                    wait = backoff
            else:
                wait = backoff
            wait = _jittered_wait(wait)

            if attempt < max_retries:
                logger.info(
                    "HTTP %d from %s — retry %d/%d in %.1fs",
                    response.status_code,
                    log_url,
                    attempt + 1,
                    max_retries,
                    wait,
                )
                await asyncio.sleep(wait)
                backoff = min(backoff * 2, MAX_BACKOFF)
            else:
                logger.warning(
                    "HTTP %d from %s — exhausted %d retries",
                    response.status_code,
                    log_url,
                    max_retries,
                )
                return response

        except httpx.TimeoutException:
            if attempt < max_retries:
                wait = _jittered_wait(backoff)
                logger.info(
                    "Timeout on %s — retry %d/%d in %.1fs",
                    log_url,
                    attempt + 1,
                    max_retries,
                    wait,
                )
                await asyncio.sleep(wait)
                backoff = min(backoff * 2, MAX_BACKOFF)
            else:
                logger.warning("Timeout on %s — exhausted %d retries", log_url, max_retries)
                return None

        except httpx.HTTPError as e:
            safe_err = _sanitize_for_log(e)
            if attempt < max_retries:
                wait = _jittered_wait(backoff)
                logger.info(
                    "HTTP error on %s: %s — retry %d/%d in %.1fs",
                    log_url,
                    safe_err,
                    attempt + 1,
                    max_retries,
                    wait,
                )
                await asyncio.sleep(wait)
                backoff = min(backoff * 2, MAX_BACKOFF)
            else:
                logger.warning("HTTP error on %s: %s — exhausted %d retries", log_url, safe_err, max_retries)
                return None

    return None

scan_agents async 

scan_agents(agents: list[Agent], *, compliance_enabled: bool = False, resolve_transitive: bool = False, show_scan_banner: bool = True, options: ScanOptions | None = None) -> list[BlastRadius]

Scan all agents' MCP server packages for vulnerabilities.

Source code in src/agent_bom/scanners/package_scan.py 
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
async def scan_agents(
    agents: list[Agent],
    *,
    compliance_enabled: bool = False,
    resolve_transitive: bool = False,
    show_scan_banner: bool = True,
    options: ScanOptions | None = None,
) -> list[BlastRadius]:
    """Scan all agents' MCP server packages for vulnerabilities."""
    scan_options = options or default_scan_options(
        compliance_enabled=compliance_enabled,
        resolve_transitive=resolve_transitive,
    )
    if show_scan_banner:
        from agent_bom.output.brand_tokens import PRODUCT_NAME

        console.print(f"\n[bold cyan]{PRODUCT_NAME}[/bold cyan]  [bold]Scanning for vulnerabilities…[/bold]\n")

    def _pkg_key(pkg: Package) -> str:
        return canonical_package_key(pkg.name, pkg.version, pkg.ecosystem, pkg.purl)

    # Collect all unique packages
    all_packages = []
    pkg_to_servers: dict[str, list[MCPServer]] = {}
    pkg_to_agents: dict[str, list[Agent]] = {}

    for agent in agents:
        for server in agent.mcp_servers:
            for pkg in server.packages:
                key = _pkg_key(pkg)
                all_packages.append(pkg)

                if key not in pkg_to_servers:
                    pkg_to_servers[key] = []
                pkg_to_servers[key].append(server)

                if key not in pkg_to_agents:
                    pkg_to_agents[key] = []
                if agent not in pkg_to_agents[key]:
                    pkg_to_agents[key].append(agent)

    # Deduplicate packages for scanning — uses canonical deduplicate_packages()
    # which normalizes by (ecosystem, normalized_name, version) fingerprint.
    unique_packages = deduplicate_packages(all_packages)

    if show_scan_banner:
        console.print(f"  Scanning {len(unique_packages)} unique packages across {len(agents)} agent(s)...")

    total_vulns = await _scanners_patchable("scan_packages")(unique_packages, options=scan_options)

    # Propagate vulnerabilities back to all instances
    vuln_map = {}
    for pkg in unique_packages:
        if pkg.vulnerabilities:
            vuln_map[_pkg_key(pkg)] = pkg.vulnerabilities

    for agent in agents:
        for server in agent.mcp_servers:
            for pkg in server.packages:
                if _pkg_key(pkg) in vuln_map:
                    pkg.vulnerabilities = vuln_map[_pkg_key(pkg)]

    # Build blast radius analysis. Registry enrichment is keyed by MCP server,
    # not by vulnerable package. Keep one cache for the whole scan: a server can
    # expose many packages, and matching the same registry catalog once per
    # package turns a linear build into an avoidable package×catalog walk.
    from agent_bom.parsers import get_registry_entry

    _registry_cache: dict[tuple[str, str, tuple[str, ...], str], dict | None] = {}

    def _registry_key(server: MCPServer) -> tuple[str, str, tuple[str, ...], str]:
        return (
            server.name,
            server.command,
            tuple(server.args),
            server.url or "",
        )

    blast_radii = []
    for pkg in unique_packages:
        if not pkg.vulnerabilities:
            continue

        key = _pkg_key(pkg)
        affected_servers = pkg_to_servers.get(key, [])
        affected_agents = pkg_to_agents.get(key, [])

        # Collect exposed credentials and tools — enrich from registry when server
        # config doesn't have explicit tool/credential data.
        # Cache registry lookups per server to avoid duplicate tool creation.
        #
        # IMPORTANT: Registry-sourced tools are "phantom" — they reflect what
        # the registry CLAIMS the server has, not what was introspected.
        # We include them for visibility but mark them so blast radius
        # consumers can distinguish confirmed vs phantom tools.
        exposed_creds: list[str] = []
        exposed_tools: list = []
        phantom_tools: list = []
        for server in affected_servers:
            server_creds = server.credential_names
            server_tools = list(server.tools)  # copy — don't mutate server

            # Registry enrichment: if no tools/creds known from config, use registry
            if not server_tools or not server_creds:
                registry_key = _registry_key(server)
                if registry_key not in _registry_cache:
                    _registry_cache[registry_key] = get_registry_entry(server)
                reg = _registry_cache[registry_key]
                if reg:
                    if not server_tools and reg.get("tools"):
                        from agent_bom.models import MCPTool

                        server_tools = [
                            MCPTool(
                                name=t,
                                description="(registry — unverified)",
                                discovery_source="registry",
                                discovery_confidence="unverified",
                            )
                            for t in reg["tools"]
                        ]
                    if not server_creds and reg.get("credential_env_vars"):
                        server_creds = reg["credential_env_vars"]

            exposed_creds.extend(server_creds)
            for tool in server_tools:
                if getattr(tool, "discovery_source", None) == "registry" and getattr(tool, "discovery_confidence", None) == "unverified":
                    phantom_tools.append(tool)
                else:
                    exposed_tools.append(tool)

        # Deduplicate credentials and tools to prevent inflation
        exposed_creds_deduped = list(set(exposed_creds))
        seen_tool_names: set[str] = set()
        deduped_tools = []
        for t in exposed_tools:
            if t.name not in seen_tool_names:
                seen_tool_names.add(t.name)
                deduped_tools.append(t)
        exposed_tools = deduped_tools
        seen_phantom: set[str] = set()
        deduped_phantom = []
        for t in phantom_tools:
            if t.name not in seen_phantom:
                seen_phantom.add(t.name)
                deduped_phantom.append(t)
        phantom_tools = deduped_phantom

        # AI-native risk context: elevated when an AI framework has creds + tools
        is_ai_framework = (
            pkg.name.lower().replace("-", "_") in {n.replace("-", "_") for n in _AI_FRAMEWORK_PACKAGES}
            or pkg.name.lower() in _AI_FRAMEWORK_PACKAGES
        )
        has_creds = bool(exposed_creds_deduped)
        has_tools = bool(exposed_tools)
        has_phantom_tools = bool(phantom_tools)
        if is_ai_framework and has_creds and has_tools:
            phantom_note = f" (+{len(phantom_tools)} registry-only tool(s) excluded from score)" if has_phantom_tools else ""
            ai_risk_context = (
                f"AI framework '{pkg.name}' runs inside an agent with {len(exposed_creds_deduped)} "
                f"exposed credential(s) and {len(exposed_tools)} confirmed reachable tool(s){phantom_note}. "
                f"A compromise here gives an attacker both identity and capability."
            )
        elif is_ai_framework and has_creds:
            ai_risk_context = (
                f"AI framework '{pkg.name}' has access to {len(exposed_creds_deduped)} "
                f"credential(s). Exploitation could exfiltrate secrets via LLM output."
            )
        elif is_ai_framework:
            ai_risk_context = "AI framework package — vulnerability affects LLM inference/orchestration pipeline."
        else:
            ai_risk_context = None

        for vuln in pkg.vulnerabilities:
            if not vuln.compliance_tags:
                vuln.compliance_tags = _tag_vuln(vuln, pkg)

            # CWE-aware filtering: only expose credentials/tools the vuln
            # type can realistically reach. A DoS (CWE-400) doesn't steal
            # DATABASE_URL. An RCE (CWE-94) does.
            from agent_bom.cwe_impact import (
                build_attack_vector_summary,
                classify_cwe_impact,
                filter_credentials_by_impact,
                filter_tools_by_impact,
            )

            impact_cat = classify_cwe_impact(vuln.cwe_ids)
            filtered_creds = filter_credentials_by_impact(
                impact_cat,
                exposed_creds_deduped,
            )
            filtered_tools = filter_tools_by_impact(
                impact_cat,
                exposed_tools,
            )
            filtered_phantom = filter_tools_by_impact(
                impact_cat,
                phantom_tools,
            )
            attack_summary = build_attack_vector_summary(
                cwe_ids=vuln.cwe_ids,
                category=impact_cat,
                filtered_creds=filtered_creds,
                filtered_tools=filtered_tools,
                severity=vuln.severity.value if vuln.severity else None,
                is_kev=vuln.is_kev,
            )
            if attack_summary:
                if ai_risk_context:
                    ai_risk_context = f"{ai_risk_context} {attack_summary}"
                else:
                    ai_risk_context = attack_summary

            br = BlastRadius(
                vulnerability=vuln,
                package=pkg,
                affected_servers=affected_servers,
                affected_agents=affected_agents,
                exposed_credentials=filtered_creds,
                exposed_tools=filtered_tools,
                phantom_tools=filtered_phantom,
                ai_risk_context=ai_risk_context,
                impact_category=impact_cat,
                all_server_credentials=list(exposed_creds_deduped),
                all_server_tools=list(exposed_tools) + list(phantom_tools),
                attack_vector_summary=attack_summary,
            )
            br.calculate_risk_score()
            # Context-aware tagging remains opt-in for explicit compliance
            # views, but effective framework tags are always materialized.
            if scan_options.compliance_enabled:
                br.owasp_tags = tag_blast_radius(br)
                br.atlas_tags = tag_atlas_techniques(br)
                br.attack_tags = tag_attack_techniques(br)
                br.nist_ai_rmf_tags = tag_nist_ai_rmf(br)
                br.owasp_mcp_tags = tag_owasp_mcp(br)
                br.owasp_agentic_tags = tag_owasp_agentic(br)
                br.eu_ai_act_tags = tag_eu_ai_act(br)
                br.nist_csf_tags = tag_nist_csf(br)
                br.iso_27001_tags = tag_iso_27001(br)
                br.soc2_tags = tag_soc2(br)
                br.cis_tags = tag_cis_controls(br)
                br.cmmc_tags = tag_cmmc(br)
                br.nist_800_53_tags = tag_nist_800_53(br)
                br.fedramp_tags = tag_fedramp(br)
            apply_effective_blast_radius_tags(br)
            blast_radii.append(br)

    # Sort by risk score descending
    blast_radii.sort(key=lambda br: br.risk_score, reverse=True)

    if total_vulns:
        console.print(f"  [red]âš  Found {total_vulns} vulnerabilities across {len(blast_radii)} findings[/red]")
    else:
        console.print("  [green]✓ No known vulnerabilities found[/green]")

    _logger.info(
        "Scan summary: %d packages scanned, %d vulnerabilities, %d blast radius findings across %d agent(s)",
        len(unique_packages),
        total_vulns,
        len(blast_radii),
        len(agents),
    )

    return blast_radii

scan_agents_sync 

scan_agents_sync(agents: list[Agent], enable_enrichment: bool = False, nvd_api_key: Optional[str] = None, blast_radius_depth: int = 1, compliance_enabled: bool = False, resolve_transitive: bool = False, show_scan_banner: bool = True, offline: bool | None = None, prefer_local_db: bool | None = None, demo_advisories: bool = False, project_dir: str | None = None, options: ScanOptions | None = None) -> list[BlastRadius]

Synchronous wrapper for scan_agents.

Source code in src/agent_bom/scanners/package_scan.py 
2044
2045
2046
2047
2048
2049
2050
2051
2052
2053
2054
2055
2056
2057
2058
2059
2060
2061
2062
2063
2064
2065
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
2081
2082
2083
2084
2085
2086
2087
2088
2089
2090
def scan_agents_sync(
    agents: list[Agent],
    enable_enrichment: bool = False,
    nvd_api_key: Optional[str] = None,
    blast_radius_depth: int = 1,
    compliance_enabled: bool = False,
    resolve_transitive: bool = False,
    show_scan_banner: bool = True,
    offline: bool | None = None,
    prefer_local_db: bool | None = None,
    demo_advisories: bool = False,
    project_dir: str | None = None,
    options: ScanOptions | None = None,
) -> list[BlastRadius]:
    """Synchronous wrapper for scan_agents."""
    scan_options = options or default_scan_options(
        compliance_enabled=compliance_enabled,
        resolve_transitive=resolve_transitive,
        prefer_local_db=prefer_local_db,
        offline=offline,
        demo_advisories=demo_advisories,
        project_dir=project_dir,
    )
    if enable_enrichment:
        blast_radii = asyncio.run(
            scan_agents_with_enrichment(
                agents,
                nvd_api_key,
                enable_enrichment,
                compliance_enabled=scan_options.compliance_enabled,
                show_scan_banner=show_scan_banner,
                options=scan_options,
            )
        )
    else:
        blast_radii = asyncio.run(
            scan_agents(
                agents,
                compliance_enabled=scan_options.compliance_enabled,
                resolve_transitive=scan_options.resolve_transitive,
                show_scan_banner=show_scan_banner,
                options=scan_options,
            )
        )
    if blast_radius_depth > 1:
        expand_blast_radius_hops(blast_radii, agents, max_depth=blast_radius_depth)
    return blast_radii

scan_agents_with_enrichment async 

scan_agents_with_enrichment(agents: list[Agent], nvd_api_key: Optional[str] = None, enable_enrichment: bool = True, compliance_enabled: bool = False, show_scan_banner: bool = True, options: ScanOptions | None = None) -> list[BlastRadius]

Scan agents and enrich vulnerabilities with NVD/EPSS/KEV data.

Source code in src/agent_bom/scanners/package_scan.py 
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032
2033
2034
2035
2036
2037
2038
2039
2040
2041
async def scan_agents_with_enrichment(
    agents: list[Agent],
    nvd_api_key: Optional[str] = None,
    enable_enrichment: bool = True,
    compliance_enabled: bool = False,
    show_scan_banner: bool = True,
    options: ScanOptions | None = None,
) -> list[BlastRadius]:
    """Scan agents and enrich vulnerabilities with NVD/EPSS/KEV data."""
    scan_options = options or default_scan_options(compliance_enabled=compliance_enabled)
    # First, do normal OSV scan
    blast_radii = await scan_agents(
        agents,
        compliance_enabled=scan_options.compliance_enabled,
        show_scan_banner=show_scan_banner,
        options=scan_options,
    )

    # Then enrich with external data
    if enable_enrichment and blast_radii:
        from agent_bom.enrichment import enrich_vulnerabilities
        from agent_bom.resolver import enrich_supply_chain_metadata

        # Collect all vulnerabilities
        all_vulns = []
        all_pkgs: list[Package] = []
        for agent in agents:
            for server in agent.mcp_servers:
                for pkg in server.packages:
                    all_pkgs.append(pkg)
                    all_vulns.extend(pkg.vulnerabilities)

        if all_vulns:
            await enrich_vulnerabilities(
                all_vulns,
                nvd_api_key=nvd_api_key,
                enable_nvd=True,
                enable_epss=True,
                enable_kev=True,
                offline=offline_mode,
            )

            # Refresh CVE-level compliance tags now that CWE/KEV/EPSS data is populated
            for agent in agents:
                for server in agent.mcp_servers:
                    for pkg in server.packages:
                        for v in pkg.vulnerabilities:
                            v.compliance_tags = _tag_vuln(v, pkg)

        # Supply-chain metadata enrichment — feeds Scorecard repo resolution
        try:
            async with create_client(timeout=10.0) as client:
                await enrich_supply_chain_metadata(all_pkgs, client)
        except Exception as exc:  # noqa: BLE001
            _logger.warning("Supply chain metadata enrichment failed (scorecard coverage may be incomplete): %s", exc)
            _emit_scan_warning("supply-chain metadata enrichment failed")

        # Scorecard enrichment — adds supply-chain quality signal
        try:
            from agent_bom.scorecard import enrich_packages_with_scorecard

            if all_pkgs:
                await enrich_packages_with_scorecard(all_pkgs)
        except Exception as exc:  # noqa: BLE001
            _logger.warning("Scorecard auto-enrichment failed (risk scores may be understated): %s", exc)
            _emit_scan_warning("OpenSSF Scorecard enrichment failed")

        # Recalculate blast radius with all enriched data
        for br in blast_radii:
            br.calculate_risk_score()
            apply_effective_blast_radius_tags(br)

        # Re-sort by updated risk scores
        blast_radii.sort(key=lambda br: br.risk_score, reverse=True)

    return blast_radii

scan_packages async 

scan_packages(packages: list[Package], *, resolve_transitive: bool = False, options: ScanOptions | None = None) -> int

Scan a list of packages for vulnerabilities. Returns count of vulns found.

Source code in src/agent_bom/scanners/package_scan.py 
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
async def scan_packages(
    packages: list[Package],
    *,
    resolve_transitive: bool = False,
    options: ScanOptions | None = None,
) -> int:
    """Scan a list of packages for vulnerabilities. Returns count of vulns found."""
    scan_options = options or default_scan_options(resolve_transitive=resolve_transitive)
    scan_offline = scan_options.offline
    scan_resolve_transitive = scan_options.resolve_transitive
    scan_prefer_local_db = scan_options.prefer_local_db
    # Deduplicate packages across discovery sources before scanning.
    # Prevents redundant OSV API calls when the same package is discovered
    # from multiple sources (local, K8s, cloud).
    original_count = len(packages)
    packages = deduplicate_packages(packages)
    deduped = original_count - len(packages)
    reset_scan_performance()
    _bump_scan_perf("packages_seen", original_count)
    if deduped > 0:
        _bump_scan_perf("packages_deduplicated", deduped)
    if deduped > 0:
        _logger.info("Deduplicated %d duplicate packages (kept %d unique)", deduped, len(packages))

    # Normalize package names for consistent matching (PEP 503 for PyPI)
    # and strip pip extras notation (OSV doesn't understand extras)
    for pkg in packages:
        if pkg.ecosystem.lower() == "pypi":
            if "[" in pkg.name:
                pkg.name = _strip_extras(pkg.name)
            pkg.name = normalize_package_name(pkg.name, pkg.ecosystem)

    reset_scan_warnings_only()
    try:
        from agent_bom.resolver import reset_performance_stats as _reset_resolver_performance

        _reset_resolver_performance()
    except Exception:  # noqa: BLE001
        pass

    def _db_key(p: Package) -> str:
        return f"{p.ecosystem.lower()}:{normalize_package_name(p.name, p.ecosystem)}@{p.version}"

    # ── Local version resolution (installed packages) ──────────────────────
    # Try resolving versions from locally installed packages FIRST.
    # This is more accurate than registry fallback because it reflects
    # what's actually on disk (e.g. npm list, pip list).
    unresolved = [p for p in packages if p.version in ("latest", "unknown", "") and p.ecosystem.lower() in ("npm", "pypi", "go")]
    if unresolved:
        try:
            from agent_bom.resolvers.runtime_resolver import (
                resolve_go_versions,
                resolve_npm_versions,
                resolve_pip_versions,
            )

            local_resolved = 0

            # Local install resolution reflects the TARGET project (the ``-p``
            # path), never the scanner host. Without a target dir we resolve
            # nothing here and let the honest registry/floating paths take over.
            target_dir = scan_options.project_dir

            # Resolve pip packages from the target project's virtualenv only.
            pip_unresolved = [p for p in unresolved if p.ecosystem.lower() == "pypi"]
            venv_python = _target_venv_python(target_dir)
            if pip_unresolved and venv_python:
                pip_versions = resolve_pip_versions(venv_python)
                for pkg in pip_unresolved:
                    installed_ver = (
                        pip_versions.get(pkg.name.lower())
                        or pip_versions.get(pkg.name.lower().replace("-", "_"))
                        or pip_versions.get(pkg.name.lower().replace("_", "-"))
                    )
                    if installed_ver:
                        pkg.version = installed_ver
                        pkg.purl = f"pkg:{pkg.ecosystem}/{pkg.name}@{installed_ver}"
                        pkg.version_source = "installed"
                        local_resolved += 1

            # Resolve npm packages from the target project directory.
            npm_unresolved = [p for p in unresolved if p.ecosystem.lower() == "npm"]
            if npm_unresolved and target_dir:
                npm_versions = resolve_npm_versions(Path(target_dir))
                for pkg in npm_unresolved:
                    installed_ver = npm_versions.get(pkg.name)
                    if installed_ver:
                        pkg.version = installed_ver
                        pkg.purl = f"pkg:{pkg.ecosystem}/{pkg.name}@{installed_ver}"
                        pkg.version_source = "installed"
                        local_resolved += 1

            # Resolve Go packages from the target project directory.
            go_unresolved = [p for p in unresolved if p.ecosystem.lower() == "go"]
            if go_unresolved and target_dir:
                go_versions = resolve_go_versions(Path(target_dir))
                for pkg in go_unresolved:
                    installed_ver = go_versions.get(pkg.name)
                    if installed_ver:
                        pkg.version = installed_ver
                        pkg.purl = f"pkg:{pkg.ecosystem}/{pkg.name}@{installed_ver}"
                        pkg.version_source = "installed"
                        local_resolved += 1

            if local_resolved:
                console.print(f"  [green]✓[/green] Resolved {local_resolved} package version(s) from local install")
        except Exception as exc:
            _logger.debug("Local version resolution failed: %s", exc)

    # ── Registry fallback for still-unresolved versions ──────────────────
    # Only hit npm/PyPI registries for packages we couldn't resolve locally.
    # In offline mode, skip all registry calls entirely.
    still_unresolved = [p for p in packages if p.version in ("latest", "unknown", "") and p.ecosystem.lower() in ("npm", "pypi", "conda")]
    if still_unresolved and not scan_offline:
        try:
            from agent_bom.resolver import resolve_all_versions

            resolved_count = await resolve_all_versions(still_unresolved)
            if resolved_count:
                # Mark these as registry-resolved so output shows confidence
                for pkg in still_unresolved:
                    if pkg.version not in ("latest", "unknown", ""):
                        pkg.version_source = "registry_fallback"
                console.print(f"  [green]✓[/green] Auto-resolved {resolved_count} package version(s) from registry")
        except Exception as exc:
            _logger.warning("Version resolution failed for %d package(s): %s", len(still_unresolved), exc)
            console.print(f"  [yellow]âš [/yellow] Version resolution skipped: {exc}")
            _emit_scan_warning("version resolution failed for one or more packages")
    elif still_unresolved and scan_offline:
        _logger.info("Offline mode: skipping registry version resolution for %d package(s)", len(still_unresolved))

    # Capture the version-resolved direct demo inventory before optional online
    # transitive expansion. Only this curated set is closed evidence; packages
    # discovered outside it still require a real advisory lookup.
    demo_inventory_keys = {_db_key(package) for package in packages} if scan_options.demo_advisories else set()

    # ── Transitive dependency resolution (npm / PyPI / Go) ───────────────────
    if scan_resolve_transitive and not scan_offline:
        transitive_ecosystems = {"npm", "pypi", "go"}
        eligible = [p for p in packages if p.ecosystem.lower() in transitive_ecosystems]
        if eligible:
            try:
                from agent_bom.transitive import resolve_transitive_dependencies

                _logger.info("Resolving transitive dependencies for %d package(s)...", len(eligible))
                transitive_pkgs = await resolve_transitive_dependencies(eligible)
                if transitive_pkgs:
                    existing_keys = {f"{p.ecosystem.lower()}:{normalize_package_name(p.name, p.ecosystem)}@{p.version}" for p in packages}
                    new_pkgs = [
                        p
                        for p in transitive_pkgs
                        if f"{p.ecosystem.lower()}:{normalize_package_name(p.name, p.ecosystem)}@{p.version}" not in existing_keys
                    ]
                    if new_pkgs:
                        packages = packages + new_pkgs
                        packages = deduplicate_packages(packages)
                        console.print(f"  [cyan]→[/cyan] Transitive resolution: {len(new_pkgs)} additional package(s) queued")
            except Exception as exc:  # noqa: BLE001
                _logger.warning("Transitive resolution failed, scanning direct dependencies only: %s", exc)
                _emit_scan_warning("transitive dependency resolution failed")

    # SAST packages already carry vulns from Semgrep — skip OSV query for them
    scannable = [p for p in packages if p.version not in ("unknown", "latest", "") and p.ecosystem.lower() != "sast"]

    # Warn about packages that could not be resolved — no silent failures
    still_unresolved = [p for p in packages if p.version in ("unknown", "latest", "") and p.ecosystem.lower() != "sast"]
    if still_unresolved:
        names = ", ".join(f"{p.name}@{p.version}" for p in still_unresolved[:10])
        suffix = f" (+{len(still_unresolved) - 10} more)" if len(still_unresolved) > 10 else ""
        console.print(f"  [yellow]âš [/yellow] {len(still_unresolved)} package(s) skipped (unresolved version): {names}{suffix}")
        _logger.warning(
            "Skipped %d package(s) with unresolved versions: %s",
            len(still_unresolved),
            names + suffix,
        )
        _emit_scan_warning(f"{len(still_unresolved)} package(s) skipped due to unresolved versions")

    if not scannable:
        return 0

    # ── Local DB lookup (fast, offline-capable) ───────────────────────────────
    # Demo mode uses bundled advisory rows first so published first-run evidence
    # is deterministic and does not depend on a user's ambient ~/.agent-bom DB.
    local_count = 0
    db_covered: set[str] = set()
    if scan_options.demo_advisories:
        demo_count, demo_covered = _scan_packages_demo_advisories(scannable)
        local_count += demo_count
        db_covered.update(demo_covered)
        # The curated demo inventory is a closed, versioned evidence set. Some
        # entries are intentionally clean or malicious without a CVE, so a
        # missing advisory row must not fall through to the user's ambient DB
        # (or OSV) and turn deterministic demo coverage into a partial scan.
        db_covered.update(demo_inventory_keys)

    # Query the local SQLite DB for packages not already covered by the
    # deterministic demo manifest. Packages covered by the DB skip OSV calls.
    local_db_targets = [p for p in scannable if _db_key(p) not in db_covered]
    local_db_count, local_db_covered = _scanners_patchable("_scan_packages_local_db")(local_db_targets) if local_db_targets else (0, set())
    local_count += local_db_count
    db_covered.update(local_db_covered)
    if local_count:
        local_label = "vulnerability found" if local_count == 1 else "vulnerabilities found"
        source_label = "Demo advisory DB" if scan_options.demo_advisories else "Local DB"
        if scan_offline:
            mode_note = " (offline mode)"
        elif scan_prefer_local_db:
            mode_note = " (local advisory cache)"
        else:
            mode_note = ""
        console.print(f"  [green]✓[/green] {source_label}: {local_count} {local_label}{mode_note}")

    # ── Coverage-gap detection (warning only) ─────────────────────────────────
    # Flag OS releases whose advisory data the local DB does not carry (typically
    # end-of-life releases dropped by the data source). A low or zero count for
    # such a release is NOT a clean bill of health — surface it loudly so it is
    # never mistaken for a secure result. Detection never alters matching or the
    # vulnerabilities already attached to packages.
    coverage_gaps: list[dict] = []
    try:
        from agent_bom.coverage import detect_release_coverage_gaps

        coverage_gaps = detect_release_coverage_gaps(scannable)
        for gap in coverage_gaps:
            record_coverage_warning(gap)
            _emit_scan_warning(f"incomplete vulnerability coverage for {gap['release']} (likely end-of-life; results may under-report)")
            console.print(
                f"  [yellow]⚠[/yellow] [bold]Incomplete coverage:[/bold] {gap['release']} — "
                f"{gap['package_count']} package(s) present but the data source carries only "
                f"{gap['advisory_rows']} advisory row(s) for this release. Likely end-of-life; "
                "results may UNDER-report. A low or zero count is not a clean bill of health."
            )
    except Exception as exc:  # noqa: BLE001
        _logger.debug("coverage-gap detection skipped: %s", exc)

    # Only call OSV for packages not already covered by the local DB, except when
    # the release has sparse advisory coverage (EOL) — then force OSV online.
    from agent_bom.coverage import osv_fallback_db_keys

    force_osv_keys = osv_fallback_db_keys(scannable, gaps=coverage_gaps) if not scan_offline else set()
    osv_targets = [p for p in scannable if _db_key(p) not in db_covered or _db_key(p) in force_osv_keys]
    if force_osv_keys:
        _logger.info("Forcing OSV lookup for %d package(s) on sparse distro release(s)", len(force_osv_keys))
        _emit_scan_warning(f"sparse release OSV fallback for {len(force_osv_keys)} package(s)")

    if scan_offline or (scan_prefer_local_db and not osv_targets):
        if scan_offline:
            covered_ecos = _scanners_patchable("_db_covered_ecosystems")()
            if not covered_ecos and osv_targets and not scan_options.demo_advisories:
                # Genuinely empty/missing DB — nothing can be scanned offline.
                raise IncompleteScanError(
                    "Offline mode requires a populated local vulnerability DB. Run `agent-bom db update` before using `--offline`."
                )
            # A package with no DB rows whose ECOSYSTEM the DB covers is clean,
            # not a gap — the advisory DB simply has no advisory for it. Only
            # packages in an ecosystem the DB holds zero advisories for are a
            # real coverage gap. Warn loudly about those, but never discard the
            # vulnerabilities already found for covered packages (the previous
            # behaviour raised and dropped the whole report when a single
            # package — even a clean one — was "uncovered").
            uncovered = [p for p in osv_targets if not any(eco in covered_ecos for eco in _db_ecosystems_for_package(p))]
            if uncovered:
                gap_ecos = sorted({eco for p in uncovered for eco in _db_ecosystems_for_package(p)})
                skipped_names = ", ".join(f"{pkg.name}@{pkg.version}" for pkg in uncovered[:5])
                suffix = f" (+{len(uncovered) - 5} more)" if len(uncovered) > 5 else ""
                # A package can map to no DB ecosystem at all; never render an
                # empty "()" parenthetical in that case.
                eco_note = f" ({', '.join(gap_ecos)})" if gap_ecos else ""
                eco_clause = f" {', '.join(gap_ecos)}" if gap_ecos else ""
                _logger.warning(
                    "Offline mode: %d package(s) in ecosystem(s) with no local DB advisories%s skipped",
                    len(uncovered),
                    eco_note,
                )
                console.print(
                    f"  [yellow]âš [/yellow] Offline coverage gap: {len(uncovered)} package(s) in "
                    f"ecosystem(s) the local DB has no advisories for{eco_note}: "
                    f"{skipped_names}{suffix}. Run `agent-bom db update` for full coverage."
                )
                _emit_scan_warning(
                    f"offline coverage gap: {len(uncovered)} package(s) in ecosystem(s)"
                    f"{eco_clause} have no advisories in the local vulnerability DB"
                )
                # Structured signal so consumers (e.g. `check`) can fail closed
                # deterministically instead of string-matching the warning above.
                # A zero-vuln result for an ecosystem the local DB carries no
                # advisories for is NOT a clean bill of health.
                record_coverage_warning(
                    {
                        "kind": "offline_ecosystem_gap",
                        "release": f"offline:{','.join(gap_ecos)}",
                        "ecosystems": gap_ecos,
                        "package_count": len(uncovered),
                    }
                )
        results = {}
    elif scan_prefer_local_db and osv_targets:
        # DB is fresh — only query OSV for packages genuinely missing from DB
        _logger.debug("Local DB preferred: querying OSV for %d uncovered package(s) only", len(osv_targets))
        results = await _scanners_patchable("query_osv_batch")(osv_targets)
    elif osv_targets:
        results = await _scanners_patchable("query_osv_batch")(osv_targets)
    else:
        results = {}

    if not scan_offline and osv_targets:
        _flag_remote_lookup_gap(osv_targets)

    total_vulns = local_count
    for pkg in osv_targets:
        norm = normalize_package_name(pkg.name, pkg.ecosystem)
        key = f"{pkg.ecosystem.lower()}:{norm}@{pkg.version}"
        vuln_data = results.get(key, [])
        if vuln_data:
            new_vulns = _scanners_patchable("build_vulnerabilities")(vuln_data, pkg)
            # Merge: don't duplicate what the local DB already found
            existing_ids = {v.id for v in pkg.vulnerabilities}
            merged = [v for v in new_vulns if v.id not in existing_ids]
            pkg.vulnerabilities.extend(merged)
            total_vulns += len(merged)
            # Tag each CVE with compliance framework codes (pre-enrichment)
            for v in merged:
                v.compliance_tags = _tag_vuln(v, pkg)
            # Flag packages with MAL- prefixed vulnerability IDs as malicious
            flag_malicious_from_vulns(pkg)

    # Back-fill: also run OSV tagging for packages that came from local DB only
    for pkg in scannable:
        if pkg in osv_targets:
            continue  # already processed above
        for v in pkg.vulnerabilities:
            if not v.compliance_tags:
                v.compliance_tags = _tag_vuln(v, pkg)

    # Supplemental: check NVIDIA advisories for all AI framework packages.
    # nvidia_advisory.py maps NVIDIA CSAF products to bundling frameworks (torch,
    # jax, vllm, etc.) so we pass ALL AI packages — not just nvidia-prefixed ones.
    nvidia_packages = [
        p
        for p in scannable
        if p.name.lower().replace("-", "_") in _AI_FRAMEWORK_PACKAGES or p.name.lower().replace("-", "") in _AI_FRAMEWORK_PACKAGES
    ]
    if nvidia_packages and not scan_offline:
        try:
            from agent_bom.scanners.nvidia_advisory import check_nvidia_advisories

            nvidia_new = await check_nvidia_advisories(nvidia_packages)
            if nvidia_new:
                total_vulns += nvidia_new
                console.print(f"  [green]✓[/green] NVIDIA advisories: {nvidia_new} additional CVE(s)")
        except Exception as exc:
            _logger.warning("NVIDIA advisory check failed for %d package(s): %s", len(nvidia_packages), exc)
            console.print(f"  [yellow]âš [/yellow] NVIDIA advisory check skipped: {exc}")
            _emit_scan_warning("NVIDIA advisory enrichment skipped")

    # Supplemental: check AMD PSIRT advisories for ROCm / HIP packages.
    amd_packages = [
        p
        for p in scannable
        if any(
            p.name.lower().replace("-", "_").startswith(prefix)
            for prefix in (
                "rocm",
                "hip_",
                "hipcc",
                "hip_base",
                "miopen",
                "rocblas",
                "rocsolver",
                "rccl",
                "rocprim",
                "rocthrust",
                "rocrand",
                "rocfft",
                "hipsparse",
                "hipblas",
                "composablekernel",
                "tensorflow_rocm",
                "jax_rocm",
            )
        )
    ]
    if amd_packages:
        try:
            from agent_bom.scanners.amd_advisory import check_amd_advisories

            amd_new = check_amd_advisories(amd_packages)
            if amd_new:
                total_vulns += amd_new
                console.print(f"  [green]✓[/green] AMD advisories: {amd_new} additional CVE(s)")
        except Exception as exc:
            _logger.warning("AMD advisory check failed for %d package(s): %s", len(amd_packages), exc)
            console.print(f"  [yellow]âš [/yellow] AMD advisory check skipped: {exc}")
            _emit_scan_warning("AMD advisory enrichment skipped")

    # Supplemental: check GitHub Security Advisories for all packages
    if scannable and not scan_offline:
        try:
            from agent_bom.scanners.ghsa_advisory import check_github_advisories

            ghsa_new = await check_github_advisories(scannable)
            if ghsa_new:
                total_vulns += ghsa_new
                console.print(f"  [green]✓[/green] GHSA advisories: {ghsa_new} additional CVE(s)")
        except Exception as exc:
            _logger.warning("GHSA advisory check failed for %d package(s): %s", len(scannable), exc)
            console.print(f"  [yellow]âš [/yellow] GHSA advisory check skipped: {exc}")
            _emit_scan_warning("GHSA advisory enrichment skipped")

    # Typosquat detection for all scanned packages
    for pkg in scannable:
        if not pkg.is_malicious:
            target = check_typosquat(pkg.name, pkg.ecosystem)
            if target:
                pkg.is_malicious = True
                pkg.malicious_reason = f"Possible typosquat of '{target}'"
            # Dependency confusion check
            confusion_warning = check_dependency_confusion(pkg)
            if confusion_warning and not pkg.is_malicious:
                pkg.is_malicious = True
                pkg.malicious_reason = confusion_warning

    # Align OS-package reporting with mainstream scanner conventions: distro
    # advisories with no fix for the scanned release (no-dsa / won't-fix /
    # end-of-life open) are suppressed by default. Surface them with
    # set_include_unfixed(True) or AGENT_BOM_INCLUDE_UNFIXED=1.
    unfixed_suppressed = _suppress_unfixed_os_advisories(scannable)
    if unfixed_suppressed:
        total_vulns -= unfixed_suppressed
        _logger.info(
            "Suppressed %d unfixed OS-package advisory finding(s) (no-dsa/won't-fix); set AGENT_BOM_INCLUDE_UNFIXED=1 to include them",
            unfixed_suppressed,
        )
        console.print(
            f"  [dim]Suppressed {unfixed_suppressed} unfixed OS-package finding(s) "
            f"(no-dsa/won't-fix) — set AGENT_BOM_INCLUDE_UNFIXED=1 to include[/dim]"
        )

    # Apply .agent-bom-ignore suppression rules
    try:
        from agent_bom.ignore import apply_ignore_rules, load_ignore_file

        rules = load_ignore_file()
        if not rules.is_empty:
            suppressed = apply_ignore_rules(scannable, rules)
            if suppressed:
                total_vulns -= suppressed
                console.print(f"  [yellow]âš [/yellow] Suppressed {suppressed} finding(s) via .agent-bom-ignore")
    except Exception as exc:
        _logger.warning("Ignore file processing skipped: %s", exc)
        _emit_scan_warning("ignore-file processing failed")

    return total_vulns

set_include_unfixed 

set_include_unfixed(value: bool) -> None

Toggle surfacing of unfixed (no-dsa/won't-fix) OS-package advisories.

Source code in src/agent_bom/scanners/package_scan.py 
210
211
212
213
def set_include_unfixed(value: bool) -> None:
    """Toggle surfacing of unfixed (no-dsa/won't-fix) OS-package advisories."""
    global include_unfixed  # noqa: PLW0603
    include_unfixed = value

set_offline_mode 

set_offline_mode(value: bool) -> None

Set offline mode in both scanner and http_client transport layer.

Source code in src/agent_bom/scanners/package_scan.py 
187
188
189
190
191
192
193
def set_offline_mode(value: bool) -> None:
    """Set offline mode in both scanner and http_client transport layer."""
    global offline_mode  # noqa: PLW0603
    offline_mode = value
    from agent_bom.http_client import set_offline

    set_offline(value)

builtin_scanner_registrations 

builtin_scanner_registrations() -> list[ScannerRegistration]

Return built-in scanner registrations with capability metadata.

Source code in src/agent_bom/scanners/registry.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
def builtin_scanner_registrations() -> list[ScannerRegistration]:
    """Return built-in scanner registrations with capability metadata."""

    local_read = _local_read_capabilities()
    return [
        _registration(
            "sca-vulnerability",
            "agent_bom.scanners",
            phase=ScannerPhase.SCANNING,
            run_attr="scan_agents_sync",
            input_types=("agents", "packages"),
            output_types=("blast_radii", "vulnerabilities", "findings"),
            finding_types=("cve", "ghsa", "osv", "malicious-package", "dependency-confusion", "typosquat"),
            summary="SCA vulnerability matching with OSV/GHSA/local DB, enrichment, KEV/EPSS, compliance, and blast radius.",
            capabilities=_network_read_capabilities(destinations=("osv.dev", "github_advisory_database", "local_vulnerability_db")),
            failure_mode=ScannerFailureMode.FAIL_CLOSED,
            skip_when=("no_scan_requested", "no_packages_found"),
            telemetry_keys=("packages_scanned", "api_batches", "cache_hits", "warnings", "duration_ms"),
            standards=("OWASP", "NIST", "CIS", "SOC2", "EU_AI_ACT"),
        ),
        _registration(
            "secret-patterns",
            "agent_bom.secret_scanner",
            phase=ScannerPhase.SCANNING,
            run_attr="scan_secrets",
            input_types=("filesystem_path", "source_file", "config_file"),
            output_types=("secret_findings",),
            finding_types=("credential", "pii", "hardcoded-secret"),
            summary="Local file secret and PII pattern scanning using the shared runtime detector pattern library.",
            capabilities=local_read,
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("no_filesystem_scope", "file_too_large", "excluded_directory"),
            telemetry_keys=("files_scanned", "findings_emitted", "warnings", "duration_ms"),
            standards=("OWASP_LLM01", "CIS_16_4", "SOC2_CC6_1"),
        ),
        _registration(
            "code-native",
            "agent_bom.ast_analyzer",
            phase=ScannerPhase.ANALYSIS,
            run_attr="analyze_project",
            input_types=("code_path",),
            output_types=("ast_analysis",),
            finding_types=("prompt-risk", "guardrail", "tool-signature"),
            summary="Native Python AST analysis composed into the agent-bom code command; no Semgrep execution.",
            capabilities=local_read,
            failure_mode=ScannerFailureMode.FAIL_CLOSED,
            skip_when=("no_code_scope",),
            telemetry_keys=("files_analyzed", "components_found", "duration_ms"),
            standards=("OWASP_LLM", "MITRE_ATLAS"),
        ),
        _registration(
            "ai-component-source",
            "agent_bom.ai_components",
            phase=ScannerPhase.ANALYSIS,
            run_attr="scan_source",
            input_types=("code_path",),
            output_types=("ai_components",),
            finding_types=("ai-component", "shadow-ai", "deprecated-model", "credential-reference"),
            summary="Native multi-language AI SDK, model, and component analysis composed into the agent-bom code command.",
            capabilities=local_read,
            failure_mode=ScannerFailureMode.FAIL_CLOSED,
            skip_when=("no_code_scope",),
            telemetry_keys=("files_scanned", "components_found", "duration_ms"),
            standards=("OWASP_LLM", "MITRE_ATLAS"),
        ),
        _registration(
            "sast-semgrep",
            "agent_bom.sast",
            phase=ScannerPhase.SCANNING,
            run_attr="scan_code",
            input_types=("code_path", "sarif"),
            output_types=("sast_data", "packages", "vulnerabilities"),
            finding_types=("sast", "cwe", "owasp"),
            summary=(
                "Semgrep SAST execution and normalization into the package/vulnerability model; "
                "the legacy .sarif path input imports through the canonical parser without executing Semgrep."
            ),
            capabilities=ExtensionCapabilities(
                scan_modes=("local", "online", "offline_local_rules"),
                required_scopes=("local_project_read", "conditional_network_egress"),
                outbound_destinations=("semgrep_registry",),
                data_boundary="local_source_read_with_operator_selected_rule_source",
                network_access=True,
                guarantees=("read_only", "offline_rejects_remote_rules"),
            ),
            failure_mode=ScannerFailureMode.SKIP_WHEN_UNAVAILABLE,
            skip_when=("semgrep_missing", "no_code_scope", "offline_without_local_rules"),
            telemetry_keys=("files_scanned", "rules_loaded", "findings_emitted", "duration_ms"),
            standards=("CWE", "OWASP", "SARIF"),
        ),
        _registration(
            "external-scan-ingest",
            "agent_bom.parsers.external_scanners",
            phase=ScannerPhase.DISCOVERY,
            run_attr="detect_and_parse",
            input_types=("sarif", "cyclonedx", "spdx", "trivy_json", "grype_json", "syft_json"),
            output_types=("packages", "vulnerabilities", "findings"),
            finding_types=("external-finding", "sast", "sca", "sbom-package"),
            summary="Tool-agnostic evidence import through canonical SARIF, SBOM, and scanner-native parsers.",
            capabilities=local_read,
            failure_mode=ScannerFailureMode.FAIL_CLOSED,
            skip_when=("no_external_scan_input",),
            telemetry_keys=("format", "packages_emitted", "findings_emitted", "duration_ms"),
            standards=("SARIF", "CycloneDX", "SPDX"),
        ),
        _registration(
            "container-image",
            "agent_bom.image",
            phase=ScannerPhase.DISCOVERY,
            run_attr="scan_image",
            input_types=("image_ref", "image_tar"),
            output_types=("packages", "image_scan_strategy"),
            finding_types=("container-package", "image-vulnerability-input"),
            summary="Container image package extraction via local daemon, registry, or OCI tarball paths.",
            capabilities=ExtensionCapabilities(
                scan_modes=("image", "airgap"),
                required_scopes=("image_read",),
                permissions_used=("docker_socket_read", "registry_pull", "local_tar_read"),
                outbound_destinations=("container_registry",),
                data_boundary="image_metadata_and_layers_read_only",
                network_access=True,
                guarantees=("read_only",),
            ),
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("no_image_scope", "image_runtime_unavailable"),
            telemetry_keys=("images_scanned", "packages_emitted", "strategy", "warnings", "duration_ms"),
        ),
        _registration(
            "container-sbom-posture",
            "agent_bom.cloud.container_sbom",
            phase=ScannerPhase.ANALYSIS,
            run_attr="scan_container_image",
            input_types=("image_ref",),
            output_types=("container_sbom_posture",),
            finding_types=("unpinned-tag", "missing-sbom", "missing-provenance", "stale-image"),
            summary="Read-only OCI metadata posture check for SBOM/provenance/staleness signals.",
            capabilities=_network_read_capabilities(destinations=("docker_hub_registry_api",), scan_modes=("cloud", "container")),
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("non_docker_hub_registry_metadata_only", "registry_unavailable"),
            telemetry_keys=("images_checked", "registry_requests", "findings_emitted", "duration_ms"),
            standards=("SLSA", "CycloneDX", "SPDX"),
        ),
        _registration(
            "sbom-ingest",
            "agent_bom.sbom",
            phase=ScannerPhase.DISCOVERY,
            run_attr="load_sbom",
            input_types=("cyclonedx", "spdx", "sbom_json"),
            output_types=("packages", "sbom_metadata"),
            finding_types=("sbom-package",),
            summary="SBOM ingestion for CycloneDX/SPDX documents into the shared package model.",
            capabilities=local_read,
            failure_mode=ScannerFailureMode.FAIL_CLOSED,
            skip_when=("no_sbom_input",),
            telemetry_keys=("packages_emitted", "format", "warnings", "duration_ms"),
            standards=("CycloneDX", "SPDX"),
        ),
        _registration(
            "iac-terraform",
            "agent_bom.terraform",
            phase=ScannerPhase.DISCOVERY,
            run_attr="scan_terraform_dir",
            input_types=("terraform_dir",),
            output_types=("agents", "iac_findings"),
            finding_types=("iac", "misconfiguration", "ai-infra"),
            summary="Terraform AI infrastructure discovery and misconfiguration evidence.",
            capabilities=local_read,
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("no_terraform_scope",),
            telemetry_keys=("files_scanned", "findings_emitted", "warnings", "duration_ms"),
            standards=("MITRE_ATLAS", "CIS", "NIST"),
        ),
        _registration(
            "iac-dbt",
            "agent_bom.iac.dbt_security",
            phase=ScannerPhase.DISCOVERY,
            run_attr="scan_dbt_file",
            input_types=("dbt_project", "dbt_profiles", "dbt_packages", "dbt_models", "dbt_macros", "dbt_ci"),
            output_types=("iac_findings",),
            finding_types=("dbt-security", "credential-exposure", "supply-chain", "sql-injection", "ci-hygiene"),
            summary="dbt project, profile, package, model, macro, seed, and CI/CD security checks.",
            capabilities=local_read,
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("no_dbt_project_scope",),
            telemetry_keys=("files_scanned", "findings_emitted", "warnings", "duration_ms"),
            standards=("SLSA", "NIST", "SOC2"),
        ),
        _registration(
            "k8s-live-cluster",
            "agent_bom.k8s",
            phase=ScannerPhase.DISCOVERY,
            run_attr="scan_live_cluster_posture",
            input_types=("kubeconfig_context", "kubectl"),
            output_types=("iac_findings",),
            finding_types=(
                "kubernetes-live",
                "pod-security",
                "rbac-wildcard",
                "cluster-admin-binding",
                "kubelet-cis",
                "network-policy-gap",
            ),
            summary=(
                "Read-only live Kubernetes cluster audit via kubectl: PodSecurity on running "
                "workloads, RBAC/namespace over-broad grants, and node/kubelet CIS configuration."
            ),
            capabilities=ExtensionCapabilities(
                scan_modes=("cluster", "online"),
                required_scopes=("kubernetes_cluster_read",),
                permissions_used=("kubectl_get", "kubelet_configz_read"),
                outbound_destinations=("kubernetes_api_server",),
                data_boundary="cluster_state_read_only",
                network_access=True,
                guarantees=("read_only", "no_cluster_mutation"),
            ),
            failure_mode=ScannerFailureMode.SKIP_WHEN_UNAVAILABLE,
            skip_when=("kubectl_missing", "cluster_unreachable", "no_cluster_scope"),
            telemetry_keys=("pods_scanned", "nodes_scanned", "findings_emitted", "duration_ms"),
            standards=("CIS", "NIST"),
        ),
        _registration(
            "cicd-github-actions",
            "agent_bom.github_actions",
            phase=ScannerPhase.DISCOVERY,
            run_attr="scan_github_actions",
            input_types=("github_actions_path",),
            output_types=("agents", "workflow_findings"),
            finding_types=("workflow-permissions", "unpinned-action", "secret-exposure", "fork-pr-risk"),
            summary="GitHub Actions workflow inventory and CI/CD posture checks.",
            capabilities=local_read,
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("no_github_actions_scope",),
            telemetry_keys=("workflows_scanned", "findings_emitted", "warnings", "duration_ms"),
            standards=("SLSA", "CIS", "NIST"),
        ),
        _registration(
            "dataset-card",
            "agent_bom.parsers.dataset_cards",
            phase=ScannerPhase.DISCOVERY,
            run_attr="scan_dataset_directory",
            input_types=("dataset_directory", "dataset_card"),
            output_types=("dataset_cards",),
            finding_types=("unlicensed-dataset", "missing-card", "unversioned-data", "remote-source"),
            summary="Dataset-card provenance, license, and lineage scanner.",
            capabilities=local_read,
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("no_dataset_scope",),
            telemetry_keys=("datasets_scanned", "flagged_count", "warnings", "duration_ms"),
            standards=("EU_AI_ACT", "NIST_AI_RMF"),
        ),
        _registration(
            "dataset-pii",
            "agent_bom.parsers.dataset_pii_scanner",
            phase=ScannerPhase.SCANNING,
            run_attr="scan_directory_for_pii",
            input_types=("csv", "json", "jsonl", "dataset_directory"),
            output_types=("dataset_pii_findings",),
            finding_types=("pii", "phi", "drivers_license", "email", "ssn"),
            summary="Dataset row sampling scanner for PII/PHI exposure.",
            capabilities=local_read,
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("no_dataset_scope", "unsupported_file_type", "row_limit_reached"),
            telemetry_keys=("files_scanned", "rows_scanned", "findings_emitted", "duration_ms"),
            standards=("HIPAA", "EU_AI_ACT", "NIST_PRIVACY"),
        ),
        _registration(
            "prompt-injection",
            "agent_bom.parsers.prompt_scanner",
            phase=ScannerPhase.SCANNING,
            run_attr="scan_prompt_file",
            input_types=("prompt_file", "instruction_file"),
            output_types=("prompt_findings",),
            finding_types=("prompt-injection", "hidden-instruction", "tool-exfiltration"),
            summary="Prompt/instruction file scanning for injection and agentic abuse patterns.",
            capabilities=local_read,
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("no_prompt_scope", "unsupported_file_type"),
            telemetry_keys=("files_scanned", "findings_emitted", "duration_ms"),
            standards=("OWASP_LLM", "OWASP_AGENTIC"),
        ),
        _registration(
            "skill-audit",
            "agent_bom.parsers.skill_audit",
            phase=ScannerPhase.SCANNING,
            run_attr="audit_skill_result",
            input_types=("skill_file", "instruction_file"),
            output_types=("skill_findings", "skill_audit"),
            finding_types=("skill-risk", "mcp-blocklist", "typosquat", "shell-access", "unverified-server"),
            summary=(
                "Skill/instruction file audit: registry typosquat, MCP blocklist on extracted "
                "servers, shell access, and behavioral regex/AST risks."
            ),
            capabilities=local_read,
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("no_skill_scope", "unsupported_file_type"),
            telemetry_keys=("files_scanned", "packages_checked", "servers_checked", "findings_emitted", "duration_ms"),
            standards=("OWASP_LLM", "OWASP_AGENTIC", "OWASP_MCP"),
        ),
        _registration(
            "license-policy",
            "agent_bom.license_policy",
            phase=ScannerPhase.ANALYSIS,
            run_attr="evaluate_license_policy",
            input_types=("agents", "packages", "license_policy"),
            output_types=("license_report",),
            finding_types=("license-block", "license-warning", "unknown-license"),
            summary="SPDX license policy evaluation across discovered packages.",
            capabilities=_local_read_capabilities(scan_modes=("analysis",)),
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("no_packages_found", "license_policy_disabled"),
            telemetry_keys=("packages_evaluated", "findings_emitted", "duration_ms"),
            standards=("SPDX", "OpenChain"),
        ),
        _registration(
            "firmware-advisory",
            "agent_bom.scanners.firmware_advisory",
            phase=ScannerPhase.SCANNING,
            run_attr="scan_firmware_advisories",
            input_types=("gpu_inventory", "firmware_inventory"),
            output_types=("firmware_findings",),
            finding_types=("firmware-cve", "bmc-cve", "driver-cve"),
            summary="GPU firmware/BMC/driver advisory scanner for AI infrastructure.",
            capabilities=_network_read_capabilities(destinations=("bundled_firmware_advisory_feed",), scan_modes=("infra", "gpu")),
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("no_gpu_inventory", "advisory_feed_unavailable"),
            telemetry_keys=("devices_scanned", "advisories_matched", "duration_ms"),
            standards=("NVIDIA_CSAF", "CVE"),
        ),
        _registration(
            "model-advisory",
            "agent_bom.model_advisories",
            phase=ScannerPhase.SCANNING,
            run_attr="match_model_advisories",
            input_types=("model_card", "model_inventory"),
            output_types=("model_advisories",),
            finding_types=("unsafe-model-format", "model-card-risk", "model-advisory"),
            summary="Model-specific advisory and model-card risk matching.",
            capabilities=_local_read_capabilities(scan_modes=("model",)),
            failure_mode=ScannerFailureMode.WARN_AND_CONTINUE,
            skip_when=("no_model_inventory", "feed_missing"),
            telemetry_keys=("models_scanned", "advisories_matched", "duration_ms"),
            standards=("NIST_AI_RMF", "EU_AI_ACT"),
        ),
        _registration(
            "runtime-detectors",
            "agent_bom.runtime.detectors",
            phase=ScannerPhase.SCANNING,
            run_attr="detector_pipeline",
            input_types=("tool_call", "tool_response", "runtime_event"),
            output_types=("runtime_findings", "policy_decisions"),
            finding_types=("prompt-injection", "credential-leak", "rate-limit", "tool-drift", "vector-db-injection"),
            summary="Runtime proxy detector family for agent/tool call enforcement.",
            capabilities=ExtensionCapabilities(
                scan_modes=("runtime", "proxy"),
                required_scopes=("runtime_event_read",),
                permissions_used=("tool_call_observe",),
                outbound_destinations=(),
                data_boundary="runtime_event_inline_redacted",
                network_access=False,
                guarantees=("redacted_evidence", "policy_enforced"),
            ),
            failure_mode=ScannerFailureMode.FAIL_CLOSED,
            skip_when=("runtime_proxy_disabled",),
            telemetry_keys=("events_scanned", "findings_emitted", "policy_blocks", "duration_ms"),
            standards=("OWASP_LLM", "OWASP_MCP", "SOC2"),
        ),
        _registration(
            "yara-signature",
            "agent_bom.scanners.yara",
            phase=ScannerPhase.SCANNING,
            run_attr="scan_with_yara",
            input_types=("file", "directory", "model_file", "artifact"),
            output_types=("signature_findings",),
            finding_types=("malware-signature", "unsafe-artifact", "model-file-indicator"),
            summary="Reserved scanner-driver slot for local YARA-style artifact signatures; not executable in this release.",
            capabilities=_local_read_capabilities(scan_modes=("signature",)),
            failure_mode=ScannerFailureMode.SKIP_WHEN_UNAVAILABLE,
            execution_state=ScannerExecutionState.PLANNED,
            enabled_by_default=False,
            skip_when=("driver_not_implemented", "ruleset_missing"),
            telemetry_keys=("files_scanned", "rules_loaded", "matches", "duration_ms"),
            standards=("YARA",),
        ),
        _registration(
            "zero-day-heuristics",
            "agent_bom.scanners.zero_day",
            phase=ScannerPhase.ANALYSIS,
            run_attr="score_zero_day_residual_risk",
            input_types=("packages", "runtime_context", "exploit_signals", "graph_context"),
            output_types=("residual_risk_scores", "prioritized_findings"),
            finding_types=("residual-risk", "zero-day-susceptibility", "high-blast-radius-unfixed"),
            summary="Reserved scanner-driver slot for residual/zero-day risk scoring across SCA, runtime, and graph context.",
            capabilities=_local_read_capabilities(scan_modes=("analysis",)),
            failure_mode=ScannerFailureMode.SKIP_WHEN_UNAVAILABLE,
            execution_state=ScannerExecutionState.PLANNED,
            enabled_by_default=False,
            skip_when=("driver_not_implemented", "insufficient_context"),
            telemetry_keys=("packages_scored", "contexts_used", "scores_emitted", "duration_ms"),
            standards=("EPSS", "KEV", "NIST_AI_RMF"),
        ),
    ]

get_scanner_registration 

get_scanner_registration(name: str) -> ScannerRegistration

Return one scanner registration or raise KeyError.

Source code in src/agent_bom/scanners/registry.py 
574
575
576
577
578
def get_scanner_registration(name: str) -> ScannerRegistration:
    """Return one scanner registration or raise ``KeyError``."""

    _ensure_scanner_registry_loaded()
    return _SCANNER_REGISTRY[name]

list_registered_scanners 

list_registered_scanners(*, include_planned: bool = True) -> list[ScannerRegistration]

Return registered scanner drivers sorted by name.

Source code in src/agent_bom/scanners/registry.py 
564
565
566
567
568
569
570
571
def list_registered_scanners(*, include_planned: bool = True) -> list[ScannerRegistration]:
    """Return registered scanner drivers sorted by name."""

    _ensure_scanner_registry_loaded()
    registrations = [_SCANNER_REGISTRY[name] for name in sorted(_SCANNER_REGISTRY)]
    if include_planned:
        return registrations
    return [registration for registration in registrations if registration.execution_state != ScannerExecutionState.PLANNED]

register_scanner 

register_scanner(registration: ScannerRegistration) -> None

Register a scanner driver with capability and execution metadata.

Source code in src/agent_bom/scanners/registry.py 
539
540
541
542
543
544
545
546
def register_scanner(registration: ScannerRegistration) -> None:
    """Register a scanner driver with capability and execution metadata."""

    if not registration.name:
        raise ValueError("scanner registration must declare a name")
    if registration.name in _SCANNER_REGISTRY:
        raise ValueError(f"duplicate scanner registration: {registration.name}")
    _SCANNER_REGISTRY[registration.name] = registration

scanner_registry_summary 

scanner_registry_summary() -> dict[str, object]

Return a compact scanner registry summary for API/UI surfaces.

Source code in src/agent_bom/scanners/registry.py 
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
def scanner_registry_summary() -> dict[str, object]:
    """Return a compact scanner registry summary for API/UI surfaces."""

    registrations = list_registered_scanners(include_planned=True)
    by_phase: dict[str, int] = {}
    by_state: dict[str, int] = {}
    for registration in registrations:
        by_phase[registration.phase.value] = by_phase.get(registration.phase.value, 0) + 1
        by_state[registration.execution_state.value] = by_state.get(registration.execution_state.value, 0) + 1
    return {
        "total": len(registrations),
        "active": by_state.get(ScannerExecutionState.ACTIVE.value, 0),
        "passive": by_state.get(ScannerExecutionState.PASSIVE.value, 0),
        "planned": by_state.get(ScannerExecutionState.PLANNED.value, 0),
        "by_phase": by_phase,
        "by_state": by_state,
    }

scanner_registry_warnings 

scanner_registry_warnings() -> list[str]

Return sanitized non-fatal registry loading warnings.

Source code in src/agent_bom/scanners/registry.py 
581
582
583
584
585
def scanner_registry_warnings() -> list[str]:
    """Return sanitized non-fatal registry loading warnings."""

    _ensure_scanner_registry_loaded()
    return list(_SCANNER_REGISTRY_WARNINGS)

advisory_id_severity_fallback 

advisory_id_severity_fallback(advisory_id: str) -> tuple[Severity, Optional[str]]

Return conservative triage severity for advisory-only IDs.

Some advisory ecosystems publish IDs before CVSS/vendor severity arrives. These should not stay invisible as unknown findings in operator views, but only known advisory namespaces get this fallback. Arbitrary missing severity still remains UNKNOWN.

Source code in src/agent_bom/scanners/risk.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def advisory_id_severity_fallback(advisory_id: str) -> tuple[Severity, Optional[str]]:
    """Return conservative triage severity for advisory-only IDs.

    Some advisory ecosystems publish IDs before CVSS/vendor severity arrives.
    These should not stay invisible as ``unknown`` findings in operator views,
    but only known advisory namespaces get this fallback. Arbitrary missing
    severity still remains ``UNKNOWN``.
    """
    normalized = advisory_id.upper()
    if normalized.startswith("GHSA-"):
        return Severity.MEDIUM, "ghsa_heuristic"
    if normalized.startswith(_OSV_MEDIUM_FALLBACK_PREFIXES):
        return Severity.MEDIUM, "osv_heuristic"
    if normalized.startswith(_DISTRO_MEDIUM_FALLBACK_PREFIXES):
        return Severity.MEDIUM, "distro_advisory_heuristic"
    return Severity.UNKNOWN, None

parse_cvss_vector 

parse_cvss_vector(vector: str) -> Optional[float]

Compute CVSS base score from a vector string (v3.x and v4.0).

Source code in src/agent_bom/scanners/risk.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def parse_cvss_vector(vector: str) -> Optional[float]:
    """Compute CVSS base score from a vector string (v3.x and v4.0)."""
    try:
        if vector.startswith("CVSS:4"):
            return _parse_cvss4_vector(vector)
        if not vector.startswith("CVSS:3"):
            return None

        parts = vector.split("/")[1:]
        metrics = dict(p.split(":") for p in parts)

        av = _CVSS3_AV.get(metrics.get("AV", ""), None)
        ac = _CVSS3_AC.get(metrics.get("AC", ""), None)
        scope = metrics.get("S", "U")
        pr_map = _CVSS3_PR_C if scope == "C" else _CVSS3_PR_U
        pr = pr_map.get(metrics.get("PR", ""), None)
        ui = _CVSS3_UI.get(metrics.get("UI", ""), None)
        c = _CVSS3_CIA.get(metrics.get("C", ""), None)
        i = _CVSS3_CIA.get(metrics.get("I", ""), None)
        a = _CVSS3_CIA.get(metrics.get("A", ""), None)

        if any(value is None for value in (av, ac, pr, ui, c, i, a)):
            return None

        av, ac, pr, ui = float(av), float(ac), float(pr), float(ui)  # type: ignore[arg-type]
        c, i, a = float(c), float(i), float(a)  # type: ignore[arg-type]

        isc_base = 1.0 - (1.0 - c) * (1.0 - i) * (1.0 - a)
        if scope == "C":
            isc = 7.52 * (isc_base - 0.029) - 3.25 * ((isc_base - 0.02) ** 15)
        else:
            isc = 6.42 * isc_base

        if isc <= 0:
            return 0.0

        exploitability = 8.22 * av * ac * pr * ui
        raw = min(1.08 * (isc + exploitability), 10.0) if scope == "C" else min(isc + exploitability, 10.0)
        return math.ceil(raw * 10) / 10.0
    except Exception as exc:  # noqa: BLE001
        _logger.debug("CVSS vector parse failed for %r: %s", vector, exc)
        return None

parse_osv_severity 

parse_osv_severity(vuln_data: dict) -> tuple[Severity, Optional[float], Optional[str]]

Extract severity, CVSS score, and severity source from OSV data.

Source code in src/agent_bom/scanners/risk.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def parse_osv_severity(vuln_data: dict) -> tuple[Severity, Optional[float], Optional[str]]:
    """Extract severity, CVSS score, and severity source from OSV data."""
    cvss_score = None
    severity = Severity.UNKNOWN
    severity_source: Optional[str] = None

    for sev in vuln_data.get("severity", []):
        if sev.get("type") in ("CVSS_V3", "CVSS_V3_1", "CVSS_V4"):
            score = _normalize_cvss_score(sev.get("score"))
            if score is not None:
                cvss_score = score

    db_specific = vuln_data.get("database_specific", {})
    severity, severity_source = _first_vendor_severity(("osv_database", db_specific))

    if cvss_score is None and isinstance(db_specific, dict):
        for key in ("cvss", "cvss_score", "cvss_v3", "severity_vectors"):
            score = _normalize_cvss_score(db_specific.get(key))
            if score is not None:
                cvss_score = score
                break

    if cvss_score is None:
        score = _normalize_cvss_score(vuln_data.get("severity_vectors"))
        if score is not None:
            cvss_score = score

    if cvss_score is None:
        for affected in vuln_data.get("affected", []):
            if not isinstance(affected, dict):
                continue
            for block_name in ("database_specific", "ecosystem_specific"):
                block = affected.get(block_name)
                if not isinstance(block, dict):
                    continue
                for key in ("cvss", "cvss_score", "cvss_v3", "severity_vectors"):
                    score = _normalize_cvss_score(block.get(key))
                    if score is not None:
                        cvss_score = score
                        break
                if cvss_score is not None:
                    break
            if cvss_score is not None:
                break

    if cvss_score is not None:
        severity = cvss_to_severity(cvss_score)
        severity_source = "cvss"

    if severity == Severity.UNKNOWN:
        eco_specific = vuln_data.get("ecosystem_specific", {})
        severity, severity_source = _first_vendor_severity(("osv_ecosystem", eco_specific))

    if severity == Severity.UNKNOWN:
        for affected in vuln_data.get("affected", []):
            if not isinstance(affected, dict):
                continue
            severity, severity_source = _first_vendor_severity(
                ("osv_affected_database", affected.get("database_specific")),
                ("osv_affected_ecosystem", affected.get("ecosystem_specific")),
            )
            if severity != Severity.UNKNOWN:
                break

    if severity == Severity.UNKNOWN:
        advisory_id = str(vuln_data.get("id", "")).upper()
        fallback, source = advisory_id_severity_fallback(advisory_id)
        if fallback != Severity.UNKNOWN:
            severity = fallback
            severity_source = source

    return severity, cvss_score, severity_source

severity_from_label 

severity_from_label(raw: Any) -> Severity

Normalize scanner/vendor severity labels without inflating unknown data.

Source code in src/agent_bom/scanners/risk.py 
33
34
35
36
37
38
def severity_from_label(raw: Any) -> Severity:
    """Normalize scanner/vendor severity labels without inflating unknown data."""
    if raw is None:
        return Severity.UNKNOWN
    normalized = str(raw).strip().replace("-", "_").replace(" ", "_").upper()
    return _SEVERITY_LABELS.get(normalized, Severity.UNKNOWN)

peek_coverage_warnings 

peek_coverage_warnings() -> list[dict]

Read the structured coverage warnings without draining them.

The scanner needs to see what a sub-step recorded (e.g. an OSV lookup failure) while leaving the channel intact for the command boundary that actually consumes it.

Source code in src/agent_bom/scanners/state.py 
 93
 94
 95
 96
 97
 98
 99
100
def peek_coverage_warnings() -> list[dict]:
    """Read the structured coverage warnings without draining them.

    The scanner needs to see what a sub-step recorded (e.g. an OSV lookup
    failure) while leaving the channel intact for the command boundary that
    actually consumes it.
    """
    return list(_coverage_warnings_state())

record_coverage_warning 

record_coverage_warning(warning: dict) -> None

Record a structured per-release coverage-gap warning (deduped by release).

Source code in src/agent_bom/scanners/state.py 
84
85
86
87
88
89
90
def record_coverage_warning(warning: dict) -> None:
    """Record a structured per-release coverage-gap warning (deduped by release)."""
    warnings = _coverage_warnings_state()
    release = warning.get("release")
    if any(existing.get("release") == release for existing in warnings):
        return
    warnings.append(warning)

reset_scan_warnings 

reset_scan_warnings() -> None

Reset all scan warning channels at a command/request boundary.

Source code in src/agent_bom/scanners/state.py 
55
56
57
58
def reset_scan_warnings() -> None:
    """Reset all scan warning channels at a command/request boundary."""
    _scan_state_local.warnings = []
    _scan_state_local.coverage_warnings = []

reset_scan_warnings_only 

reset_scan_warnings_only() -> None

Reset transient scanner warnings without discarding parser coverage gaps.

Project manifest parsers can run before vulnerability scanning and record a structured coverage warning. The package scanner must clear warnings from a prior scan, but must preserve those parser warnings for the final report.

Source code in src/agent_bom/scanners/state.py 
61
62
63
64
65
66
67
68
def reset_scan_warnings_only() -> None:
    """Reset transient scanner warnings without discarding parser coverage gaps.

    Project manifest parsers can run before vulnerability scanning and record a
    structured coverage warning. The package scanner must clear warnings from a
    prior scan, but must preserve those parser warnings for the final report.
    """
    _scan_state_local.warnings = []