Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions apps/backend/services/malicious/malicious_catalog.py
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,26 @@ def base_purl(purl_with_version: str) -> str:
return head[:at] if at > 0 else head


def version_from_purl(purl_with_version: str) -> str | None:
"""The version encoded in a PURL, or ``None`` when it carries none.

The catalog row's identity is ``purl_with_version``, so the verdict has to
be decided from that same string. Taking the version from a separate field
lets the two disagree: an uploaded SBOM can name
``pkg:cargo/[email protected]`` in ``purl`` while leaving ``version`` absent,
and a version-pinned advisory would then read "not one of the named
versions" and clear a row that is genuinely malicious — for every project
sharing that catalog row, not just the uploader's.
"""
head = purl_with_version.split("?", 1)[0].split("#", 1)[0]
slash = head.rfind("/")
at = head.find("@", slash + 1) if slash >= 0 else head.find("@")
if at <= 0:
return None
version = head[at + 1 :]
return version or None


@lru_cache(maxsize=1)
def load_index() -> MaliciousIndex | None:
"""Load the vendored snapshot once per process. ``None`` on any problem.
Expand Down
24 changes: 23 additions & 1 deletion apps/backend/tasks/scan_source.py
Original file line number Diff line number Diff line change
Expand Up @@ -2390,11 +2390,33 @@ def persist_sbom_components(
# case. The flagged tally feeds the scan summary log.
if malicious_evaluator is not None:
try:
verdict = malicious_evaluator.verdict_for(purl, version)
# Decide from the PURL, never from the sibling `version`
# field. The catalog row's identity is `purl_with_version`,
# and these two can disagree: an SBOM may carry
# `pkg:cargo/[email protected]` while omitting `version` (which then
# defaults to "0.0.0"). A version-pinned advisory would read
# that as "not one of the named versions" and clear a row that
# is genuinely malicious — for EVERY project sharing the
# catalog row, since this table is org-wide.
pinned_version = (
malicious_catalog.version_from_purl(purl) or version
)
verdict = malicious_evaluator.verdict_for(purl, pinned_version)
was_flagged = component_version.malicious_state == "flagged"
if malicious_catalog.stamp_component_version(
component_version, verdict, eol_now
):
malicious_stamped_count += 1
if was_flagged and verdict.state == "clear":
# Legitimate when a snapshot refresh drops a withdrawn
# advisory; suspicious when a single scan does it, so
# it is never silent.
log.warning(
"malicious_flag_cleared",
scan_id=str(scan_uuid),
purl=purl,
source=verdict.source,
)
if verdict.state == "flagged":
malicious_flagged_count += 1
except Exception: # noqa: BLE001 — enrichment is best-effort
Expand Down
37 changes: 37 additions & 0 deletions apps/backend/tests/unit/tasks/test_scan_source_malicious_stamp.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,43 @@ def test_disabled_leaves_every_column_null(
assert all(cv.malicious_evaluated_at is None for cv in cv_registry.values())


def test_a_missing_version_field_cannot_clear_a_flagged_row(
cv_registry: dict[str, _FakeComponentVersion],
) -> None:
"""An uploaded SBOM must not be able to un-flag a shared catalog row.

``component_versions`` is org-wide, so a wrong verdict here reaches every
project using the package. The hook used to decide from the SBOM's
``version`` field, which defaults to "0.0.0" when absent — against a
version-pinned advisory that reads as "not one of the named versions" and
clears the row. 12.4% of the snapshot is version-pinned, so this was not a
corner case. The verdict now comes from the PURL, which is the same string
the row is keyed on.
"""
from services.malicious import malicious_catalog

index = malicious_catalog.load_index()
assert index is not None
pinned_purl, versions = next(
(k, v) for k, v in index.versions.items() if k.startswith("pkg:")
)
named = versions[0]

sbom = {
"components": [
# `purl` names the malicious release; `version` is absent, exactly
# what an attacker-supplied document would look like.
{"name": "pinned", "purl": f"{pinned_purl}@{named}", "type": "library"}
]
}
from tasks.scan_source import persist_sbom_components

persist_sbom_components(_FakeSession(), scan_uuid=uuid.uuid4(), sbom=sbom)

row = cv_registry[f"{pinned_purl}@{named}"]
assert row.malicious_state == "flagged"


def test_flagging_never_creates_a_finding(
cv_registry: dict[str, _FakeComponentVersion],
) -> None:
Expand Down
Loading