diff --git a/actions/scan/action.yml b/actions/scan/action.yml
index 4b6f81a..9fef049 100644
--- a/actions/scan/action.yml
+++ b/actions/scan/action.yml
@@ -64,6 +64,12 @@ outputs:
forbidden-license-count:
description: 'Distinct components carrying a forbidden-classification license.'
value: ${{ steps.gate.outputs.forbidden_license_count }}
+ malicious-component-count:
+ description: >-
+ Distinct components the malicious-package snapshot flags. These block
+ regardless of severity — remove the package and rotate the credentials
+ this build could reach; upgrading does not help.
+ value: ${{ steps.gate.outputs.malicious_component_count }}
# -----------------------------------------------------------------------------
# Steps
@@ -266,6 +272,8 @@ runs:
reason=$(echo "$json_body" | jq -r '.reason // ""')
critical=$(echo "$json_body" | jq -r '.critical_cve_count // 0')
forbidden=$(echo "$json_body" | jq -r '.forbidden_license_count // 0')
+ malicious=$(echo "$json_body" | jq -r '.malicious_component_count // 0')
+ malicious_assessed=$(echo "$json_body" | jq -r '.malicious_scan_assessed // false')
# GITHUB_OUTPUT supports multi-line via heredoc, but reason is single
# line by contract — we still defensively strip newlines.
@@ -276,6 +284,7 @@ runs:
echo "reason=${reason_oneline}"
echo "critical_cve_count=${critical}"
echo "forbidden_license_count=${forbidden}"
+ echo "malicious_component_count=${malicious}"
} >> "$GITHUB_OUTPUT"
# Job summary block — visible on the workflow run page.
@@ -287,12 +296,21 @@ runs:
echo "| Verdict | **${gate}** |"
echo "| Critical CVEs | ${critical} |"
echo "| Forbidden licenses | ${forbidden} |"
+ # Only shown when non-zero: a "Malicious packages | 0" row on every
+ # passing build would read as an all-clear, and the count is 0 when
+ # the axis is switched off too.
+ if [ "${malicious}" != "0" ]; then
+ echo "| Known-malicious packages | ${malicious} — remove, rotate credentials |"
+ elif [ "${malicious_assessed}" != "true" ]; then
+ # A zero nobody computed must not read like a clean result.
+ echo "| Known-malicious packages | not assessed on this scan |"
+ fi
if [ -n "${reason_oneline}" ]; then
echo "| Reason | ${reason_oneline} |"
fi
} >> "$GITHUB_STEP_SUMMARY"
- echo "TrustedOSS gate=${gate} critical=${critical} forbidden=${forbidden}"
+ echo "TrustedOSS gate=${gate} critical=${critical} forbidden=${forbidden} malicious=${malicious}"
# -------------------------------------------------------------------------
# 4) Post PR comment — only on pull_request events, only when caller
diff --git a/apps/backend/alembic/versions/0048_malicious_policy_exceptions.py b/apps/backend/alembic/versions/0048_malicious_policy_exceptions.py
new file mode 100644
index 0000000..9931a87
--- /dev/null
+++ b/apps/backend/alembic/versions/0048_malicious_policy_exceptions.py
@@ -0,0 +1,53 @@
+"""license_policies.malicious_exceptions (#26 MAL-2).
+
+Adds a JSONB array of temporary waivers for packages the malicious snapshot
+flags: ``[{"component_purl", "reason", "expires_at"}, ...]``.
+
+Why a separate array rather than a shape on ``license_exceptions``: that one
+keys on ``spdx_id`` and this one on a package identifier, so a single array
+would need a discriminator and both readers would have to filter. The expiry
+rule differs too — ``expires_at`` is REQUIRED here.
+
+Why the expiry is mandatory: a licence waiver can reasonably be permanent
+(counsel cleared this dependency and the answer will not change), but a
+malicious flag always resolves. Either the advisory is wrong, in which case
+challenging it upstream drops the package from the next snapshot, or it is
+right, in which case the package has to go. An open-ended waiver would only
+park an unfinished decision out of sight.
+
+The waiver removes a component from the gate count. It does not touch the
+badge, the filter or the drawer — deferring a block is not the same as hiding
+the signal.
+
+Forward-only per CLAUDE.md §6.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision: str = "0048"
+down_revision: str | None = "0047"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+ op.add_column(
+ "license_policies",
+ sa.Column(
+ "malicious_exceptions",
+ postgresql.JSONB(astext_type=sa.Text()),
+ nullable=False,
+ server_default=sa.text("'[]'::jsonb"),
+ ),
+ )
+
+
+def downgrade() -> None:
+ raise NotImplementedError("Migrations are forward-only (CLAUDE.md §6).")
diff --git a/apps/backend/api/v1/policy_gate.py b/apps/backend/api/v1/policy_gate.py
index fefaaaf..b3c50f4 100644
--- a/apps/backend/api/v1/policy_gate.py
+++ b/apps/backend/api/v1/policy_gate.py
@@ -221,6 +221,9 @@ def _build_response_body(result: GateResult) -> GateResultResponse:
epss_threshold=result.epss_threshold,
reachable_critical_cve_count=result.reachable_critical_cve_count,
reachable_gate_enforced=result.reachable_gate_enforced,
+ malicious_component_count=result.malicious_component_count,
+ malicious_gate_enforced=result.malicious_gate_enforced,
+ malicious_scan_assessed=result.malicious_scan_assessed,
project_id=result.project_id,
scan_id=result.scan_id,
evaluated_at=result.evaluated_at,
diff --git a/apps/backend/models/license_policy.py b/apps/backend/models/license_policy.py
index a0f0295..a734017 100644
--- a/apps/backend/models/license_policy.py
+++ b/apps/backend/models/license_policy.py
@@ -207,6 +207,21 @@ class LicensePolicy(Base):
JSONB, nullable=False, server_default=EMPTY_JSONB_ARR
)
+ # Array of {"component_purl","reason","expires_at"} — temporary waivers for
+ # packages the malicious snapshot flags (#26). Kept separate from
+ # ``license_exceptions`` because that array keys on ``spdx_id`` and this one
+ # on a package identifier, and because ``expires_at`` is REQUIRED here: a
+ # malicious flag is either an upstream mistake that gets retracted or a
+ # package that has to go, so an open-ended waiver only hides an unfinished
+ # decision. Shape validation lives in the Pydantic layer.
+ #
+ # The waiver removes the component from the GATE COUNT only. The badge, the
+ # filter and the drawer keep showing it — this is a deferral of blocking,
+ # not a way to make the signal disappear.
+ malicious_exceptions: Mapped[list[Any]] = mapped_column(
+ JSONB, nullable=False, server_default=EMPTY_JSONB_ARR
+ )
+
# Gate posture for licenses absent from both the override map and the static
# catalog. One of allowed|conditional|forbidden. Default ``conditional`` —
# an uncatalogued license should be reviewed, not silently allowed/blocked.
diff --git a/apps/backend/schemas/action_queue.py b/apps/backend/schemas/action_queue.py
index 4a4fb10..c76731b 100644
--- a/apps/backend/schemas/action_queue.py
+++ b/apps/backend/schemas/action_queue.py
@@ -41,6 +41,15 @@ class GateBlockedProject(BaseModel):
"when GATE_EPSS_THRESHOLD is unset, which disables that condition."
),
)
+ malicious_component_count: int = Field(
+ default=0,
+ ge=0,
+ description=(
+ "Components the malicious snapshot flags on this scan. An upper "
+ "bound: team policy waivers are resolved per project by the gate, "
+ "not here."
+ ),
+ )
class KevSlaBucket(BaseModel):
diff --git a/apps/backend/schemas/license_policy.py b/apps/backend/schemas/license_policy.py
index eb1d71a..890780a 100644
--- a/apps/backend/schemas/license_policy.py
+++ b/apps/backend/schemas/license_policy.py
@@ -145,6 +145,44 @@ def _check_purl(cls, v: str | None) -> str | None:
return v
+class MaliciousException(BaseModel):
+ """One temporary waiver for a package the malicious snapshot flags (#26).
+
+ Separate from :class:`LicenseException` rather than an extra shape on it:
+ that one keys on ``spdx_id`` and this one on a package identifier, and the
+ expiry rules differ.
+
+ ``expires_at`` is REQUIRED here, where the licence waiver treats it as
+ optional. A licence exception can reasonably be permanent — a lawyer
+ cleared this dependency and the answer does not change. A malicious flag is
+ either a mistake upstream (fixed by challenging the advisory, which lands
+ in the next snapshot) or correct (in which case the package must go). Both
+ resolve, so an open-ended waiver would only hide an unfinished decision.
+ """
+
+ model_config = ConfigDict(extra="forbid")
+
+ component_purl: str = Field(..., min_length=1, max_length=_MAX_PURL_LEN)
+ reason: str = Field(..., min_length=1, max_length=_MAX_REASON_LEN)
+ expires_at: datetime
+
+ @field_validator("component_purl")
+ @classmethod
+ def _check_purl(cls, v: str) -> str:
+ if not v.strip():
+ raise ValueError("component_purl must be non-empty")
+ if _has_control_chars(v):
+ raise ValueError("component_purl must not contain control characters")
+ return v
+
+ @field_validator("reason")
+ @classmethod
+ def _check_reason(cls, v: str) -> str:
+ if _has_control_chars(v.replace("\n", "").replace("\r", "").replace("\t", "")):
+ raise ValueError("reason must not contain control characters")
+ return v
+
+
# ---------------------------------------------------------------------------
# Upsert input
# ---------------------------------------------------------------------------
@@ -163,6 +201,7 @@ class LicensePolicyUpsertIn(BaseModel):
name: str | None = Field(default=None, max_length=_MAX_NAME_LEN)
category_overrides: dict[str, PolicyCategory] = Field(default_factory=dict)
license_exceptions: list[LicenseException] = Field(default_factory=list)
+ malicious_exceptions: list[MaliciousException] = Field(default_factory=list)
unknown_license_category: PolicyCategory = "conditional"
compound_operator_strategy: dict[CompoundOperator, CompoundStrategy] = Field(
default_factory=lambda: _default_compound_strategy()
@@ -194,6 +233,15 @@ def _check_exceptions(cls, v: list[LicenseException]) -> list[LicenseException]:
raise ValueError(f"license_exceptions exceeds {_MAX_EXCEPTIONS} entries")
return v
+ @field_validator("malicious_exceptions")
+ @classmethod
+ def _check_malicious_exceptions(
+ cls, v: list[MaliciousException]
+ ) -> list[MaliciousException]:
+ if len(v) > _MAX_EXCEPTIONS:
+ raise ValueError(f"malicious_exceptions exceeds {_MAX_EXCEPTIONS} entries")
+ return v
+
@model_validator(mode="after")
def _check_compound_keys(self) -> LicensePolicyUpsertIn:
# Pydantic already constrains keys to CompoundOperator + values to
@@ -222,6 +270,14 @@ class LicensePolicyOut(BaseModel):
name: str | None
category_overrides: dict[str, str]
license_exceptions: list[dict[str, object]]
+ malicious_exceptions: list[dict[str, object]] = Field(
+ default_factory=list,
+ description=(
+ "Time-boxed waivers for packages the malicious snapshot flags. "
+ "Readable so an operator can see which waivers are live and when "
+ "they lapse — a waiver nobody can enumerate is one nobody reviews."
+ ),
+ )
unknown_license_category: str
compound_operator_strategy: dict[str, str]
enabled: bool
diff --git a/apps/backend/schemas/policy_gate.py b/apps/backend/schemas/policy_gate.py
index 072d98a..2fed17f 100644
--- a/apps/backend/schemas/policy_gate.py
+++ b/apps/backend/schemas/policy_gate.py
@@ -81,6 +81,34 @@ class GateResultResponse(BaseModel):
"when the EPSS gate is disabled (unset/unparseable env), in which case the "
"gate behaves exactly as the critical-CVE + forbidden-license gate.",
)
+ malicious_component_count: int = Field(
+ default=0,
+ ge=0,
+ description="Distinct components on the evaluated scan that the vendored "
+ "OSV MAL- snapshot lists as known-malicious. These block regardless of "
+ "severity: a malicious package was published to attack whoever installs "
+ "it, so the response is removal plus rotating the credentials the build "
+ "could reach, not an upgrade. Not a vulnerability count — these never "
+ "appear in ``critical_cve_count`` or any severity total.",
+ )
+ malicious_scan_assessed: bool = Field(
+ default=False,
+ description=(
+ "Whether this scan's components actually carry malicious verdicts. "
+ "``false`` means the scan predates the feature, ran with flagging "
+ "off, or hit a snapshot problem — so ``malicious_component_count`` "
+ "of 0 says nothing was checked rather than nothing was found. "
+ "``true`` on a scan with no components at all (there was nothing "
+ "to check)."
+ ),
+ )
+ malicious_gate_enforced: bool = Field(
+ default=True,
+ description="Whether the known-malicious axis was active for this "
+ "evaluation (``GATE_MALICIOUS_ENABLED``, on by default). When ``false`` "
+ "the count is 0 because nothing was checked, NOT because nothing was "
+ "found — consumers must not render that as a clean result.",
+ )
reachable_critical_cve_count: int = Field(
default=0,
ge=0,
diff --git a/apps/backend/schemas/project_governance.py b/apps/backend/schemas/project_governance.py
index ab9baea..9747d08 100644
--- a/apps/backend/schemas/project_governance.py
+++ b/apps/backend/schemas/project_governance.py
@@ -41,6 +41,15 @@ class GovernanceGate(BaseModel):
ge=0,
description="Always zero when GATE_EPSS_THRESHOLD is unset, which disables that condition.",
)
+ malicious_component_count: int = Field(
+ default=0,
+ ge=0,
+ description=(
+ "Components the malicious snapshot flags on the anchored scan. "
+ "Carried so the band can say why a build is blocked when this is "
+ "the only axis failing — otherwise it renders zeros and no reason."
+ ),
+ )
scan_id: uuid.UUID | None = Field(
default=None, description="The snapshot the verdict was computed from."
)
diff --git a/apps/backend/services/action_queue_service.py b/apps/backend/services/action_queue_service.py
index 986015c..6582c1b 100644
--- a/apps/backend/services/action_queue_service.py
+++ b/apps/backend/services/action_queue_service.py
@@ -60,10 +60,12 @@
from core.security import CurrentUser
from models import (
ComponentApproval,
+ ComponentVersion,
License,
LicenseFinding,
Project,
Scan,
+ ScanComponent,
Vulnerability,
VulnerabilityFinding,
)
@@ -79,9 +81,12 @@
_latest_succeeded_scan_ids,
)
from services.license_policy_service import get_effective_policy
+from services.malicious import malicious_catalog
from services.policy_gate import (
_CLOSED_FINDING_STATUSES,
+ _active_malicious_waivers,
_count_forbidden_license_components_dynamic,
+ _flagged_purls_for_scan,
_resolve_epss_threshold,
)
@@ -158,6 +163,43 @@ async def _gate_input_counts(
return critical, forbidden
+async def _malicious_counts(
+ session: AsyncSession,
+ *,
+ scan_ids: list[uuid.UUID],
+) -> dict[uuid.UUID, int]:
+ """Components each scan carries that the malicious snapshot flags.
+
+ A project can be blocked by this term alone — the gate fails on any
+ non-empty reason clause, and this is one of five. Omitting it here would
+ drop exactly those builds from the panel whose whole purpose is listing
+ blocked builds, and the most urgent ones at that: a malicious package is
+ an attack in the build, not a defect to schedule.
+
+ Raw tally: waivers are subtracted afterwards by
+ ``_malicious_counts_under_policy``, which reuses the gate's own resolver
+ rather than restating the rule here.
+ """
+ if not scan_ids:
+ return {}
+
+ stmt = (
+ select(
+ ScanComponent.scan_id,
+ func.count(func.distinct(ScanComponent.component_version_id)),
+ )
+ .select_from(ScanComponent)
+ .join(
+ ComponentVersion,
+ ComponentVersion.id == ScanComponent.component_version_id,
+ )
+ .where(ScanComponent.scan_id.in_(scan_ids))
+ .where(ComponentVersion.malicious_state == "flagged")
+ .group_by(ScanComponent.scan_id)
+ )
+ return {sid: int(count) for sid, count in (await session.execute(stmt)).all()}
+
+
async def _epss_counts(
session: AsyncSession,
*,
@@ -244,6 +286,58 @@ async def _forbidden_counts_under_policy(
return counts
+async def _malicious_counts_under_policy(
+ session: AsyncSession,
+ *,
+ scan_by_project: dict[uuid.UUID, uuid.UUID],
+ raw_counts: dict[uuid.UUID, int],
+) -> dict[uuid.UUID, int]:
+ """Subtract active waivers so the panel agrees with the gate.
+
+ A waived component does not block, so listing its project under "blocked
+ builds" is the failure this module's docstring already describes for the
+ licence axis: the panel keeps naming a project until people stop reading
+ the panel.
+
+ No waiver rule is duplicated here — the resolution is
+ ``policy_gate._active_malicious_waivers``, reused. Only teams that run a
+ policy pay for the extra lookup, and a team with no malicious waivers
+ exits after one dict miss.
+ """
+ if not raw_counts or not scan_by_project:
+ return raw_counts
+
+ now = datetime.now(tz=UTC)
+ team_rows = (
+ await session.execute(
+ select(Project.id, Project.team_id).where(
+ Project.id.in_(list(scan_by_project))
+ )
+ )
+ ).all()
+
+ policies: dict[uuid.UUID, LicensePolicy | None] = {}
+ counts = dict(raw_counts)
+
+ for project_id, team_id in team_rows:
+ scan_id = scan_by_project.get(project_id)
+ if scan_id is None or not counts.get(scan_id):
+ continue
+ if team_id is None:
+ continue
+ if team_id not in policies:
+ policies[team_id] = await get_effective_policy(session, team_id=team_id)
+ waived = _active_malicious_waivers(policies[team_id], now)
+ if not waived:
+ continue
+ purls = await _flagged_purls_for_scan(session, scan_id)
+ counts[scan_id] = sum(
+ 1 for purl in purls if malicious_catalog.base_purl(purl) not in waived
+ )
+
+ return counts
+
+
async def _gate_blocked(
session: AsyncSession,
*,
@@ -269,11 +363,20 @@ async def _gate_blocked(
)
epss = await _epss_counts(session, scan_ids=scan_ids)
+ malicious = await _malicious_counts(session, scan_ids=scan_ids)
+ malicious = await _malicious_counts_under_policy(
+ session, scan_by_project=scan_by_project, raw_counts=malicious
+ )
blocked_scan_ids = [
sid
for sid in scan_ids
- if critical.get(sid, 0) or forbidden.get(sid, 0) or epss.get(sid, 0)
+ if (
+ critical.get(sid, 0)
+ or forbidden.get(sid, 0)
+ or epss.get(sid, 0)
+ or malicious.get(sid, 0)
+ )
]
if not blocked_scan_ids:
return []
@@ -299,6 +402,7 @@ async def _gate_blocked(
critical_cve_count=critical.get(scan_id, 0),
forbidden_license_count=forbidden.get(scan_id, 0),
epss_gate_count=epss.get(scan_id, 0),
+ malicious_component_count=malicious.get(scan_id, 0),
)
for scan_id, project_id, name in rows
]
@@ -307,7 +411,12 @@ async def _gate_blocked(
# the one dropped at the BUCKET_LIMIT boundary changes at random.
blocked.sort(
key=lambda b: (
- -(b.critical_cve_count + b.forbidden_license_count + b.epss_gate_count),
+ -(
+ b.critical_cve_count
+ + b.forbidden_license_count
+ + b.epss_gate_count
+ + b.malicious_component_count
+ ),
b.project_name,
)
)
diff --git a/apps/backend/services/license_policy_service.py b/apps/backend/services/license_policy_service.py
index 42cd93d..4bad6b9 100644
--- a/apps/backend/services/license_policy_service.py
+++ b/apps/backend/services/license_policy_service.py
@@ -115,6 +115,9 @@ def _now() -> datetime:
# precedence steps 2–4, skipping the step-1 exception match.
_DEFAULT_MAX_WAIVE_DAYS = 90
+# Shorter than the licence cap on purpose — a malicious waiver is a stopgap
+# while an advisory is challenged upstream, not a settled decision.
+_DEFAULT_MALICIOUS_WAIVE_DAYS = 30
# Default unknown-license posture when no policy row exists yet (mirrors
# ``schemas.license_policy.LicensePolicyUpsertIn.unknown_license_category``).
_DEFAULT_UNKNOWN_POSTURE = "conditional"
@@ -172,6 +175,88 @@ def _base_category(
return unknown_posture
+def _max_malicious_waive_days() -> int:
+ """Cap on a malicious-package waiver's lifetime, in days.
+
+ Deliberately shorter than the licence cap. A licence waiver can be a
+ settled decision — counsel cleared the dependency. A malicious waiver only
+ buys time to challenge the advisory upstream and wait for the next
+ snapshot, so a long one is a decision nobody finished making.
+
+ Read at call time from ``MALICIOUS_WAIVE_MAX_DAYS`` (rule #11).
+ """
+ raw = os.getenv("MALICIOUS_WAIVE_MAX_DAYS")
+ if raw is None or raw.strip() == "":
+ return _DEFAULT_MALICIOUS_WAIVE_DAYS
+ try:
+ value = int(raw.strip())
+ except (TypeError, ValueError):
+ log.warning("license_policy.bad_malicious_waive_days", raw=raw)
+ return _DEFAULT_MALICIOUS_WAIVE_DAYS
+ if value <= 0:
+ log.warning("license_policy.bad_malicious_waive_days", raw=raw)
+ return _DEFAULT_MALICIOUS_WAIVE_DAYS
+ return value
+
+
+def _prune_expired_waivers(exceptions: list[Any]) -> list[Any]:
+ """Drop waivers whose expiry has passed.
+
+ The gate already ignores them, so no verdict changes — this keeps the
+ persisted array equal to the live set, which is what makes the policy
+ response worth reading during an incident.
+ """
+ now = _now()
+ live: list[Any] = []
+ for exc in exceptions:
+ raw = exc.get("expires_at") if isinstance(exc, dict) else None
+ if not isinstance(raw, str):
+ continue
+ try:
+ parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+ except ValueError:
+ continue
+ if parsed.tzinfo is None:
+ parsed = parsed.replace(tzinfo=UTC)
+ if parsed > now:
+ live.append(exc)
+ return live
+
+
+def _enforce_malicious_ttl(exceptions: list[Any]) -> None:
+ """Reject a malicious waiver that outlives the cap.
+
+ The schema makes ``expires_at`` required; this adds the bound a required
+ field cannot express — that it is not so far out as to be permanent in
+ practice. Without it ``9999-12-31`` is a valid answer, which is exactly
+ the open-ended waiver the required field was meant to prevent.
+
+ Already-expired entries are NOT rejected. Rejecting them would lock the
+ whole policy: a waiver written 30 days ago lapses, and from then on any
+ read-edit-write round trip fails validation, including edits that have
+ nothing to do with this axis. They are pruned on write instead (see
+ ``_prune_expired_waivers``), which also keeps the stored array equal to
+ the set of live waivers.
+ """
+ max_days = _max_malicious_waive_days()
+ now = _now()
+ cap = now + timedelta(days=max_days)
+ for exc in exceptions:
+ expires_at = exc.expires_at
+ if expires_at.tzinfo is None:
+ # Naive input is read as UTC everywhere downstream; say so rather
+ # than letting an operator's local midnight drift by their offset.
+ raise LicensePolicyValidationError(
+ f"the waiver for '{exc.component_purl}' must give expires_at "
+ "with a timezone (e.g. 2026-08-20T00:00:00Z)"
+ )
+ if expires_at > cap:
+ raise LicensePolicyValidationError(
+ f"the waiver for '{exc.component_purl}' exceeds the maximum of "
+ f"{max_days} days"
+ )
+
+
def _enforce_forbidden_ttl(
exceptions: list[LicenseException],
category_overrides: dict[str, Any] | None,
@@ -233,6 +318,7 @@ def _apply_upsert(row: LicensePolicy, payload: LicensePolicyUpsertIn) -> None:
row.name = payload.name
row.category_overrides = dumped["category_overrides"]
row.license_exceptions = dumped["license_exceptions"]
+ row.malicious_exceptions = _prune_expired_waivers(dumped["malicious_exceptions"])
row.unknown_license_category = payload.unknown_license_category
row.compound_operator_strategy = dumped["compound_operator_strategy"]
row.enabled = payload.enabled
@@ -282,6 +368,7 @@ async def upsert_team_policy(
payload.category_overrides,
payload.unknown_license_category,
)
+ _enforce_malicious_ttl(payload.malicious_exceptions)
bind_audit_team(team_id)
@@ -520,6 +607,7 @@ async def upsert_org_policy(
payload.category_overrides,
payload.unknown_license_category,
)
+ _enforce_malicious_ttl(payload.malicious_exceptions)
existing = (
await session.execute(
diff --git a/apps/backend/services/policy_gate.py b/apps/backend/services/policy_gate.py
index 021c643..8eae401 100644
--- a/apps/backend/services/policy_gate.py
+++ b/apps/backend/services/policy_gate.py
@@ -86,6 +86,7 @@
LicenseFinding,
LicensePolicy,
Project,
+ ScanComponent,
VulnerabilityFinding,
)
from models import (
@@ -96,6 +97,7 @@
)
from services.license_expression import evaluate_expression
from services.license_policy_service import effective_category, get_effective_policy
+from services.malicious import malicious_catalog
from services.scan_resolution import latest_succeeded_scan_id
log = structlog.get_logger("policy_gate.service")
@@ -149,6 +151,16 @@ class GateResult:
# ``reachable_relaxation_applied`` is False — the gate ran at full strength.
# Consumers (SCA comment) use this to render an accurate advisory.
reachable_relaxation_applied: bool = False
+ # Known-malicious components on this scan (#26). Blocks regardless of
+ # severity: a malicious package has no honest version to upgrade to, so the
+ # response is removal plus credential rotation. ``malicious_gate_enforced``
+ # records whether the axis was active, so a consumer can tell "none found"
+ # from "not checked" — the same NULL-vs-clear distinction the column carries.
+ malicious_component_count: int = 0
+ malicious_gate_enforced: bool = True
+ #: False when this scan's components were never evaluated — a count of 0
+ #: then means "not checked", not "nothing found".
+ malicious_scan_assessed: bool = False
# The latest-succeeded-scan resolver was PROMOTED to ``services.scan_resolution``
@@ -293,6 +305,23 @@ def _resolve_reachable_critical_only() -> bool:
return raw.strip().lower() in {"1", "true", "yes", "on"}
+def _resolve_gate_malicious_enabled() -> bool:
+ """Read ``GATE_MALICIOUS_ENABLED`` at evaluation time (core rule #11).
+
+ Defaults to ON, unlike the EPSS and reachability knobs. Those tune how
+ strictly an existing signal is read; this one decides whether an active
+ attack blocks the build. Shipping it off by default would mean a package
+ published to steal the build's credentials passes CI until someone opts in.
+
+ Only the exact falsy tokens disable, matching ``MALICIOUS_ENABLED``.
+ """
+ return os.getenv("GATE_MALICIOUS_ENABLED", "true").strip().lower() not in {
+ "false",
+ "0",
+ "no",
+ }
+
+
def _resolve_epss_threshold() -> float | None:
"""Read ``GATE_EPSS_THRESHOLD`` at evaluation time, or None if disabled.
@@ -374,6 +403,189 @@ async def _count_forbidden_license_components(
return int(result.scalar_one())
+@dataclass(frozen=True)
+class _MaliciousScanCounts:
+ """What the scan's catalog rows say about malicious packages.
+
+ ``assessed`` exists so the gate can tell "nothing malicious" from "nobody
+ looked". Both produce a count of zero, and only one of them is reassuring.
+ Rows predating the feature, scans persisted with ``MALICIOUS_ENABLED=false``
+ and scans whose snapshot failed to load all land in the second case.
+ """
+
+ flagged_purls: list[str]
+ assessed: int
+ total: int
+
+
+async def _malicious_scan_counts(
+ session: AsyncSession,
+ scan_id: uuid.UUID,
+) -> _MaliciousScanCounts:
+ """Versioned purls of the components on ``scan_id`` the snapshot flags (#26).
+
+ Reads the catalog column rather than a findings table, because this signal
+ deliberately creates no finding: a malicious package is removed and the
+ build's credentials rotated, not patched, so putting it on the severity
+ axis would prescribe the wrong action. That makes the shape here match the
+ project overview's rollup rather than the licence axis beside it.
+
+ Only ``flagged`` counts. A row that was never assessed is NULL and a row
+ the snapshot did not list is ``clear``; neither blocks, but they mean
+ different things and the gate must not read the first as the second.
+
+ Two statements rather than one. Reading every component to derive the
+ tallies would drop the ``malicious_state = 'flagged'`` predicate, and that
+ predicate is the only reason the partial index exists — a projects page,
+ a release snapshot and a PR comment each evaluate the gate, so hauling
+ every component of a large scan into Python has a cost the axis does not
+ need to pay. The tallies come from one conditional aggregate row; the
+ purls come from an indexed lookup over the flagged minority.
+
+ The purls come back rather than a count because the caller subtracts
+ policy waivers before counting.
+ """
+ tallies = (
+ await session.execute(
+ select(
+ func.count(func.distinct(ScanComponent.component_version_id)),
+ func.count(
+ func.distinct(
+ case(
+ (
+ ComponentVersion.malicious_state.is_not(None),
+ ScanComponent.component_version_id,
+ ),
+ )
+ )
+ ),
+ )
+ .select_from(ScanComponent)
+ .join(
+ ComponentVersion,
+ ComponentVersion.id == ScanComponent.component_version_id,
+ )
+ .where(ScanComponent.scan_id == scan_id)
+ )
+ ).one()
+ total, assessed = int(tallies[0]), int(tallies[1])
+
+ return _MaliciousScanCounts(
+ flagged_purls=await _flagged_purls_for_scan(session, scan_id),
+ assessed=assessed,
+ total=total,
+ )
+
+
+async def _flagged_purls_for_scan(
+ session: AsyncSession,
+ scan_id: uuid.UUID,
+) -> list[str]:
+ """Versioned purls the snapshot flags on ``scan_id``.
+
+ Split out because the action queue needs the same list to subtract its
+ waivers, and a second implementation there is exactly what the gate-parity
+ test exists to prevent. Rides the partial index.
+ """
+ rows = (
+ await session.execute(
+ select(ComponentVersion.purl_with_version)
+ .select_from(ScanComponent)
+ .join(
+ ComponentVersion,
+ ComponentVersion.id == ScanComponent.component_version_id,
+ )
+ .where(ScanComponent.scan_id == scan_id)
+ .where(ComponentVersion.malicious_state == "flagged")
+ .distinct()
+ )
+ ).scalars()
+ return list(rows)
+
+
+def _active_malicious_waivers(
+ policy: LicensePolicy | None, now: datetime
+) -> frozenset[str]:
+ """Base purls waived by *policy* and not yet expired.
+
+ An entry without a parseable ``expires_at`` is ignored rather than treated
+ as permanent: the schema requires the field, so a row missing it reached
+ the column some other way and the conservative reading of a malformed
+ waiver is "no waiver".
+ """
+ if policy is None:
+ return frozenset()
+ entries = policy.malicious_exceptions
+ if not isinstance(entries, list):
+ # JSONB holds whatever was written to it. A shape we cannot read is
+ # not a reason to fail every gate read for this team, and it is not a
+ # reason to honour waivers we cannot parse either — so: no waivers.
+ log.warning(
+ "policy_gate.malicious_exceptions_malformed", policy_id=str(policy.id)
+ )
+ return frozenset()
+ waived: set[str] = set()
+ for entry in entries:
+ if not isinstance(entry, dict):
+ continue
+ purl = entry.get("component_purl")
+ raw_expiry = entry.get("expires_at")
+ if not isinstance(purl, str) or not purl:
+ continue
+ if not isinstance(raw_expiry, str):
+ continue
+ try:
+ expires = datetime.fromisoformat(raw_expiry.replace("Z", "+00:00"))
+ except ValueError:
+ continue
+ if expires.tzinfo is None:
+ expires = expires.replace(tzinfo=UTC)
+ if expires > now:
+ waived.add(malicious_catalog.base_purl(purl))
+ return frozenset(waived)
+
+
+async def _resolve_malicious_count(
+ session: AsyncSession,
+ *,
+ project_id: uuid.UUID,
+ scan_id: uuid.UUID,
+ now: datetime,
+) -> tuple[int, _MaliciousScanCounts]:
+ """Blocking malicious count plus the raw scan tallies behind it.
+
+ A waiver exists because the alternative is worse: an upstream false
+ positive would otherwise stop every build until the advisory is retracted,
+ which takes days. It buys time to challenge the advisory — nothing more —
+ so it is bounded by a mandatory expiry and it only affects this count. The
+ component keeps its badge everywhere else.
+
+ Waivers are matched on the BASE purl (version stripped), because an
+ advisory that names a package names it across the versions it applies to;
+ asking an operator to waive each version separately during an incident
+ would be the wrong ergonomics.
+ """
+ counts = await _malicious_scan_counts(session, scan_id)
+ if not counts.flagged_purls:
+ return 0, counts
+
+ team_id = await _team_id_for_project(session, project_id)
+ policy = (
+ await get_effective_policy(session, team_id=team_id)
+ if team_id is not None
+ else None
+ )
+ waived = _active_malicious_waivers(policy, now)
+ if not waived:
+ return len(counts.flagged_purls), counts
+ blocking = sum(
+ 1
+ for purl in counts.flagged_purls
+ if malicious_catalog.base_purl(purl) not in waived
+ )
+ return blocking, counts
+
+
def _static_default_for(spdx_id: str) -> str:
"""Return the static-catalog category for a SINGLE simple SPDX id.
@@ -587,6 +799,7 @@ def _build_reason(
forbidden_license_count: int,
epss_gate_count: int = 0,
epss_threshold: float | None = None,
+ malicious_component_count: int = 0,
*,
reachable_critical_only: bool = False,
) -> str | None:
@@ -619,6 +832,14 @@ def _build_reason(
f"{epss_gate_count} open "
f"{'CVE' if epss_gate_count == 1 else 'CVEs'} with EPSS >= {epss_threshold:g}",
)
+ if malicious_component_count > 0:
+ # Worded as an instruction, not a count: an upgrade is the wrong move
+ # here and the reason line is often all a CI reader sees.
+ parts.append(
+ f"{malicious_component_count} known-malicious "
+ f"{'package' if malicious_component_count == 1 else 'packages'} "
+ f"detected — remove and rotate exposed credentials",
+ )
if not parts:
return None
return "; ".join(parts)
@@ -662,6 +883,8 @@ async def evaluate_gate(
epss_threshold = _resolve_epss_threshold()
# Opt-in reachable-only critical mode (default OFF → legacy behaviour).
reachable_critical_only = _resolve_reachable_critical_only()
+ # #26 — on by default; see the resolver for why this one differs.
+ malicious_gate_enabled = _resolve_gate_malicious_enabled()
if scan_id is None:
# No signal: we explicitly pass. See module docstring.
@@ -679,6 +902,12 @@ async def evaluate_gate(
reachable_gate_enforced=reachable_critical_only,
# No scan → nothing analysed → the relaxation can never have applied.
reachable_relaxation_applied=False,
+ malicious_component_count=0,
+ # Nothing was evaluated because nothing was scanned. Claiming the
+ # axis was enforced here would be the same lie as reporting a
+ # clean result for an unexamined scan.
+ malicious_gate_enforced=False,
+ malicious_scan_assessed=False,
)
log.info(
"policy_gate.evaluated",
@@ -692,6 +921,9 @@ async def evaluate_gate(
reachable_critical_cve_count=0,
reachable_gate_enforced=reachable_critical_only,
reachable_relaxation_applied=False,
+ malicious_component_count=0,
+ malicious_gate_enforced=False,
+ malicious_scan_assessed=False,
reason=None,
)
return result
@@ -709,6 +941,22 @@ async def evaluate_gate(
if epss_threshold is not None
else 0
)
+ if malicious_gate_enabled:
+ malicious_component_count, malicious_counts = await _resolve_malicious_count(
+ session, project_id=project_id, scan_id=scan_id, now=evaluated_at
+ )
+ # Every component, not merely one. The persist hook disables its
+ # evaluator on the first exception and leaves the rest of the scan
+ # unstamped, so a partially evaluated scan is the realistic failure —
+ # and `assessed > 0` would call that "assessed" and report a clean
+ # zero for the 4,998 rows nobody looked at. An empty scan satisfies
+ # this trivially, which is correct: there was nothing to examine.
+ malicious_scan_assessed = (
+ malicious_counts.assessed == malicious_counts.total
+ )
+ else:
+ malicious_component_count = 0
+ malicious_scan_assessed = False
# The critical count that DRIVES the verdict.
#
@@ -759,6 +1007,7 @@ async def evaluate_gate(
forbidden_license_count,
epss_gate_count,
epss_threshold,
+ malicious_component_count,
reachable_critical_only=relaxation_applies,
)
gate: GateOutcome = "fail" if reason is not None else "pass"
@@ -780,6 +1029,9 @@ async def evaluate_gate(
reachable_critical_cve_count=reachable_critical_cve_count,
reachable_gate_enforced=reachable_critical_only,
reachable_relaxation_applied=relaxation_applies,
+ malicious_component_count=malicious_component_count,
+ malicious_gate_enforced=malicious_gate_enabled,
+ malicious_scan_assessed=malicious_scan_assessed,
)
log.info(
"policy_gate.evaluated",
@@ -796,6 +1048,9 @@ async def evaluate_gate(
forbidden_license_count=forbidden_license_count,
epss_gate_count=epss_gate_count,
epss_threshold=epss_threshold,
+ malicious_component_count=malicious_component_count,
+ malicious_gate_enforced=malicious_gate_enabled,
+ malicious_scan_assessed=malicious_scan_assessed,
reason=reason,
)
return result
diff --git a/apps/backend/services/project_governance_service.py b/apps/backend/services/project_governance_service.py
index 2316c20..7800ecd 100644
--- a/apps/backend/services/project_governance_service.py
+++ b/apps/backend/services/project_governance_service.py
@@ -194,6 +194,7 @@ async def get_project_governance(
critical_cve_count=gate_result.critical_cve_count,
forbidden_license_count=gate_result.forbidden_license_count,
epss_gate_count=gate_result.epss_gate_count,
+ malicious_component_count=gate_result.malicious_component_count,
scan_id=current_scan_id,
),
kev_sla=GovernanceKevSla(overdue=kev.overdue, due_soon=kev.due_soon),
diff --git a/apps/backend/tests/integration/test_action_queue_gate_parity.py b/apps/backend/tests/integration/test_action_queue_gate_parity.py
index 8b2b35c..4cba069 100644
--- a/apps/backend/tests/integration/test_action_queue_gate_parity.py
+++ b/apps/backend/tests/integration/test_action_queue_gate_parity.py
@@ -602,3 +602,170 @@ async def test_parity_holds_when_only_the_epss_gate_blocks(
assert verdict.forbidden_license_count == 0
assert blocked, "the queue omitted a project blocked only by the EPSS gate"
assert blocked[0].epss_gate_count == verdict.epss_gate_count
+
+
+async def test_parity_holds_when_only_a_malicious_package_blocks(
+ db_session: AsyncSession,
+) -> None:
+ """The panel must list a build blocked solely by a malicious package.
+
+ This axis is the one an operator most needs to see in a list of blocked
+ builds — the response is removal plus credential rotation, not a scheduled
+ upgrade — and it is the one an aggregate written before the axis existed
+ silently omits.
+ """
+ from services.action_queue_service import _blocked_for_projects
+ from services.policy_gate import evaluate_gate
+
+ org = await make_organization(db_session)
+ team = await make_team(db_session, organization=org)
+ project_id, scan_id = await _project_with_scan(db_session, team=team)
+
+ from models import ComponentVersion, ScanComponent
+
+ cv_id = await _component_version(db_session)
+ cv = await db_session.get(ComponentVersion, cv_id)
+ assert cv is not None
+ cv.malicious_state = "flagged"
+ cv.malicious_id = "MAL-0000-PARITY"
+ cv.malicious_source = "osv.dev@seed"
+ db_session.add(cv)
+ db_session.add(
+ ScanComponent(
+ scan_id=scan_id, component_version_id=cv_id, direct=True, raw_data={}
+ )
+ )
+ await db_session.commit()
+
+ verdict = await evaluate_gate(db_session, project_id)
+ blocked = await _blocked_for_projects(db_session, project_ids=[project_id])
+
+ assert verdict.gate == "fail"
+ assert verdict.critical_cve_count == 0
+ assert verdict.forbidden_license_count == 0
+ assert verdict.malicious_component_count == 1
+ assert blocked, "the queue omitted a project blocked only by a malicious package"
+ assert blocked[0].malicious_component_count == verdict.malicious_component_count
+
+
+async def test_parity_holds_when_a_waiver_lifts_a_malicious_block(
+ db_session: AsyncSession,
+) -> None:
+ """A waived package must leave the panel, not linger until it expires.
+
+ The panel and the gate answer the same question. If a waiver clears the
+ build but the panel keeps naming the project, the list stops meaning
+ "blocked" and people stop reading it — the failure this module's docstring
+ describes for the licence axis.
+ """
+ from models import ComponentVersion, LicensePolicy, ScanComponent
+ from services.action_queue_service import _blocked_for_projects
+ from services.policy_gate import evaluate_gate
+
+ org = await make_organization(db_session)
+ team = await make_team(db_session, organization=org)
+ project_id, scan_id = await _project_with_scan(db_session, team=team)
+
+ cv_id = await _component_version(db_session)
+ cv = await db_session.get(ComponentVersion, cv_id)
+ assert cv is not None
+ cv.malicious_state = "flagged"
+ cv.malicious_id = "MAL-0000-WAIVED"
+ cv.malicious_source = "osv.dev@seed"
+ db_session.add(cv)
+ db_session.add(
+ ScanComponent(
+ scan_id=scan_id, component_version_id=cv_id, direct=True, raw_data={}
+ )
+ )
+ await db_session.commit()
+
+ # Without a waiver both agree it blocks.
+ assert (await evaluate_gate(db_session, project_id)).gate == "fail"
+ assert await _blocked_for_projects(db_session, project_ids=[project_id])
+
+ base_purl = cv.purl_with_version.rsplit("@", 1)[0]
+ db_session.add(
+ LicensePolicy(
+ organization_id=org.id,
+ team_id=team.id,
+ name="waiver",
+ category_overrides={},
+ license_exceptions=[],
+ malicious_exceptions=[
+ {
+ "component_purl": base_purl,
+ "reason": "challenged upstream",
+ "expires_at": (
+ datetime.now(tz=UTC) + timedelta(days=7)
+ ).isoformat(),
+ }
+ ],
+ unknown_license_category="conditional",
+ enabled=True,
+ )
+ )
+ await db_session.commit()
+
+ verdict = await evaluate_gate(db_session, project_id)
+ blocked = await _blocked_for_projects(db_session, project_ids=[project_id])
+
+ assert verdict.gate == "pass"
+ assert verdict.malicious_component_count == 0
+ assert not blocked, "the panel still lists a project the waiver unblocked"
+
+
+async def test_parity_holds_when_a_malicious_waiver_has_expired(
+ db_session: AsyncSession,
+) -> None:
+ """An expired waiver puts the project back on both surfaces."""
+ from models import ComponentVersion, LicensePolicy, ScanComponent
+ from services.action_queue_service import _blocked_for_projects
+ from services.policy_gate import evaluate_gate
+
+ org = await make_organization(db_session)
+ team = await make_team(db_session, organization=org)
+ project_id, scan_id = await _project_with_scan(db_session, team=team)
+
+ cv_id = await _component_version(db_session)
+ cv = await db_session.get(ComponentVersion, cv_id)
+ assert cv is not None
+ cv.malicious_state = "flagged"
+ cv.malicious_source = "osv.dev@seed"
+ db_session.add(cv)
+ db_session.add(
+ ScanComponent(
+ scan_id=scan_id, component_version_id=cv_id, direct=True, raw_data={}
+ )
+ )
+ base_purl = cv.purl_with_version.rsplit("@", 1)[0]
+ db_session.add(
+ LicensePolicy(
+ organization_id=org.id,
+ team_id=team.id,
+ name="lapsed",
+ category_overrides={},
+ license_exceptions=[],
+ malicious_exceptions=[
+ {
+ "component_purl": base_purl,
+ "reason": "lapsed",
+ "expires_at": (
+ datetime.now(tz=UTC) - timedelta(minutes=1)
+ ).isoformat(),
+ }
+ ],
+ unknown_license_category="conditional",
+ enabled=True,
+ )
+ )
+ await db_session.commit()
+
+ verdict = await evaluate_gate(db_session, project_id)
+ blocked = await _blocked_for_projects(db_session, project_ids=[project_id])
+
+ assert verdict.gate == "fail"
+ assert verdict.malicious_component_count == 1
+ assert blocked
+ assert blocked[0].malicious_component_count == 1
+
diff --git a/apps/backend/tests/unit/services/test_malicious_waiver_lifecycle.py b/apps/backend/tests/unit/services/test_malicious_waiver_lifecycle.py
new file mode 100644
index 0000000..96017ab
--- /dev/null
+++ b/apps/backend/tests/unit/services/test_malicious_waiver_lifecycle.py
@@ -0,0 +1,274 @@
+"""
+Malicious waivers through the API, and the bounds on their lifetime (#26).
+
+These exist because the first version of this feature shipped a waiver that
+never reached the database: the schema had the field, the model had the
+column, and the upsert service quietly dropped it. Every test passed, because
+they all built policies through the ORM and never went through a request.
+
+So the round-trip here is the point — PUT, read it back, watch the gate change
+its mind. A test that constructs the row directly cannot fail the way that bug
+failed.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import uuid
+from collections.abc import AsyncIterator
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from tests._helpers import (
+ make_membership,
+ make_organization,
+ make_project,
+ make_scan,
+ make_team,
+ make_user,
+ unique_suffix,
+)
+
+BACKEND_ROOT = Path(__file__).resolve().parent.parent.parent.parent
+
+pytestmark = pytest.mark.integration
+
+
+def _require_database_url() -> str:
+ url = os.getenv("DATABASE_URL")
+ if not url:
+ pytest.skip("DATABASE_URL not set — skip malicious waiver tests")
+ return url
+
+
+@pytest.fixture(scope="module", autouse=True)
+def _migrate_once() -> None:
+ _require_database_url()
+ result = subprocess.run(
+ ["alembic", "upgrade", "head"],
+ cwd=BACKEND_ROOT,
+ capture_output=True,
+ text=True,
+ timeout=120,
+ )
+ if result.returncode != 0:
+ pytest.skip(f"alembic upgrade head failed:\n{result.stderr}")
+
+
+@pytest.fixture
+async def db_session() -> AsyncIterator[AsyncSession]:
+ from core.audit import install_audit_listeners
+ from core.config import database_url
+
+ engine = create_async_engine(database_url(), pool_pre_ping=True, future=True)
+ factory = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
+ install_audit_listeners(factory)
+ async with factory() as session:
+ yield session
+ await engine.dispose()
+
+
+def _future(days: int = 7) -> str:
+ return (datetime.now(tz=UTC) + timedelta(days=days)).isoformat()
+
+
+async def _seed(session: AsyncSession):
+ org = await make_organization(session)
+ team = await make_team(session, organization=org)
+ user = await make_user(session)
+ await make_membership(session, user=user, team=team, role="team_admin")
+ project = await make_project(session, team=team)
+ scan = await make_scan(session, project=project, status="succeeded")
+ return org, team, user, project, scan
+
+
+async def _flagged_component(session: AsyncSession, scan_id: uuid.UUID) -> str:
+ from models import Component, ComponentVersion, ScanComponent
+
+ suffix = unique_suffix()
+ purl = f"pkg:npm/waiver-{suffix}"
+ component = Component(purl=purl, package_type="npm", name=f"waiver-{suffix}")
+ session.add(component)
+ await session.commit()
+ await session.refresh(component)
+
+ cv = ComponentVersion(
+ component_id=component.id,
+ version="1.0.0",
+ purl_with_version=f"{purl}@1.0.0",
+ malicious_state="flagged",
+ malicious_id="MAL-0000-RT",
+ malicious_source="osv.dev@seed",
+ malicious_evaluated_at=datetime.now(tz=UTC),
+ )
+ session.add(cv)
+ await session.commit()
+ await session.refresh(cv)
+ session.add(ScanComponent(scan_id=scan_id, component_version_id=cv.id, direct=True))
+ await session.commit()
+ return purl
+
+
+# ---------------------------------------------------------------------------
+# The round trip
+# ---------------------------------------------------------------------------
+
+
+async def test_a_waiver_saved_through_the_service_reaches_the_gate(
+ db_session: AsyncSession,
+) -> None:
+ """PUT → read back → the gate changes its verdict.
+
+ The regression this pins: the upsert dropped `malicious_exceptions` on the
+ floor and answered 200, so an operator saw success while the build stayed
+ blocked. Nothing raised, and every ORM-built test still passed.
+ """
+ from schemas.license_policy import LicensePolicyUpsertIn, MaliciousException
+ from services.license_policy_service import get_team_policy_row, upsert_team_policy
+ from services.policy_gate import evaluate_gate
+
+ _, team, user, project, scan = await _seed(db_session)
+ purl = await _flagged_component(db_session, scan.id)
+
+ from core.security import CurrentUser
+
+ actor = CurrentUser(
+ id=user.id,
+ email=user.email,
+ role="team_admin",
+ team_ids=[team.id],
+ team_roles={team.id: "team_admin"},
+ is_superuser=False,
+ )
+
+ assert (await evaluate_gate(db_session, project.id, scan_id=scan.id)).gate == "fail"
+
+ await upsert_team_policy(
+ db_session,
+ actor,
+ team_id=team.id,
+ payload=LicensePolicyUpsertIn(
+ malicious_exceptions=[
+ MaliciousException(
+ component_purl=purl,
+ reason="challenged upstream",
+ expires_at=datetime.fromisoformat(_future()),
+ )
+ ]
+ ),
+ )
+
+ # Readable afterwards — a waiver nobody can enumerate is one nobody reviews.
+ stored = await get_team_policy_row(db_session, team_id=team.id)
+ assert stored is not None
+ assert len(stored.malicious_exceptions) == 1
+ assert stored.malicious_exceptions[0]["component_purl"] == purl
+
+ assert (await evaluate_gate(db_session, project.id, scan_id=scan.id)).gate == "pass"
+
+
+# ---------------------------------------------------------------------------
+# Lifetime bounds
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+ ("expires_at", "expected_fragment"),
+ [
+ # Beyond the cap — the open-ended waiver the required field was meant
+ # to prevent, written as a date instead of an omission.
+ ("9999-12-31T00:00:00Z", "maximum"),
+ # No timezone: read as UTC downstream, so a KST operator's "midnight"
+ # would outlive their intent by nine hours.
+ ("2099-01-01T00:00:00", "timezone"),
+ ],
+)
+async def test_waiver_expiry_bounds_are_enforced(
+ db_session: AsyncSession, expires_at: str, expected_fragment: str
+) -> None:
+ from schemas.license_policy import LicensePolicyUpsertIn, MaliciousException
+ from services.license_policy_service import (
+ LicensePolicyValidationError,
+ upsert_team_policy,
+ )
+
+ _, team, user, _, _ = await _seed(db_session)
+ from core.security import CurrentUser
+
+ actor = CurrentUser(
+ id=user.id,
+ email=user.email,
+ role="team_admin",
+ team_ids=[team.id],
+ team_roles={team.id: "team_admin"},
+ is_superuser=False,
+ )
+
+ with pytest.raises(LicensePolicyValidationError) as excinfo:
+ await upsert_team_policy(
+ db_session,
+ actor,
+ team_id=team.id,
+ payload=LicensePolicyUpsertIn(
+ malicious_exceptions=[
+ MaliciousException(
+ component_purl="pkg:npm/whatever",
+ reason="r",
+ expires_at=datetime.fromisoformat(expires_at),
+ )
+ ]
+ ),
+ )
+ assert expected_fragment in str(excinfo.value)
+
+
+async def test_an_expired_waiver_is_pruned_rather_than_blocking_the_edit(
+ db_session: AsyncSession,
+) -> None:
+ """A lapsed waiver must not lock the policy.
+
+ Rejecting already-expired entries would mean that once a waiver lapses,
+ every later edit — including ones with nothing to do with this axis —
+ fails validation until someone hand-strips the payload. They are dropped
+ on write instead, which also keeps the stored array equal to the live set.
+ """
+ from schemas.license_policy import LicensePolicyUpsertIn, MaliciousException
+ from services.license_policy_service import get_team_policy_row, upsert_team_policy
+
+ _, team, user, _, _ = await _seed(db_session)
+ from core.security import CurrentUser
+
+ actor = CurrentUser(
+ id=user.id,
+ email=user.email,
+ role="team_admin",
+ team_ids=[team.id],
+ team_roles={team.id: "team_admin"},
+ is_superuser=False,
+ )
+
+ await upsert_team_policy(
+ db_session,
+ actor,
+ team_id=team.id,
+ payload=LicensePolicyUpsertIn(
+ category_overrides={"MIT": "forbidden"},
+ malicious_exceptions=[
+ MaliciousException(
+ component_purl="pkg:npm/lapsed",
+ reason="r",
+ expires_at=datetime.now(tz=UTC) - timedelta(days=1),
+ )
+ ],
+ ),
+ )
+
+ stored = await get_team_policy_row(db_session, team_id=team.id)
+ assert stored is not None
+ assert stored.malicious_exceptions == []
+ # The unrelated part of the edit still landed.
+ assert stored.category_overrides == {"MIT": "forbidden"}
diff --git a/apps/backend/tests/unit/services/test_policy_gate_malicious.py b/apps/backend/tests/unit/services/test_policy_gate_malicious.py
new file mode 100644
index 0000000..a55e016
--- /dev/null
+++ b/apps/backend/tests/unit/services/test_policy_gate_malicious.py
@@ -0,0 +1,429 @@
+"""
+DB-backed unit tests for the malicious axis of the build gate (#26 MAL-2a).
+
+The axis blocks regardless of severity, which is the whole point: a malicious
+package has no honest version to upgrade to, so it must not be weighed against
+a CVE threshold. What needs pinning is therefore not "does it fail" alone but
+the lifecycle around it — a waiver that expires puts the block back, and
+switching the axis off must not read as a clean build.
+
+Cases:
+ - a flagged component fails the gate, with the count and reason surfaced.
+ - `clear` and never-assessed components do not fail it.
+ - GATE_MALICIOUS_ENABLED=false skips the axis (count 0, enforced False).
+ - lifecycle sequence: flagged → fail → waiver → pass → waiver expires → fail.
+ - a waiver for a different package does not lift the block.
+ - the waiver matches on base purl, so it covers the versioned row.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import uuid
+from collections.abc import AsyncIterator
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from tests._helpers import (
+ make_membership,
+ make_organization,
+ make_project,
+ make_scan,
+ make_team,
+ make_user,
+ unique_suffix,
+)
+
+BACKEND_ROOT = Path(__file__).resolve().parent.parent.parent.parent
+
+pytestmark = pytest.mark.integration
+
+
+def _require_database_url() -> str:
+ url = os.getenv("DATABASE_URL")
+ if not url:
+ pytest.skip("DATABASE_URL not set — skip malicious gate tests")
+ return url
+
+
+@pytest.fixture(scope="module", autouse=True)
+def _migrate_once() -> None:
+ _require_database_url()
+ result = subprocess.run(
+ ["alembic", "upgrade", "head"],
+ cwd=BACKEND_ROOT,
+ capture_output=True,
+ text=True,
+ timeout=120,
+ )
+ if result.returncode != 0:
+ pytest.skip(
+ f"alembic upgrade head failed; malicious gate tests cannot run\n"
+ f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}"
+ )
+
+
+@pytest.fixture
+async def db_session() -> AsyncIterator[AsyncSession]:
+ from core.audit import install_audit_listeners
+ from core.config import database_url
+
+ engine = create_async_engine(database_url(), pool_pre_ping=True, future=True)
+ factory = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
+ install_audit_listeners(factory)
+ async with factory() as session:
+ yield session
+ await engine.dispose()
+
+
+# ---------------------------------------------------------------------------
+# Seed helpers
+# ---------------------------------------------------------------------------
+
+
+async def _seed_project(session: AsyncSession):
+ org = await make_organization(session)
+ team = await make_team(session, organization=org)
+ user = await make_user(session)
+ await make_membership(session, user=user, team=team, role="team_admin")
+ project = await make_project(session, team=team)
+ return org, team, project
+
+
+async def _make_component(
+ session: AsyncSession,
+ *,
+ malicious_state: str | None,
+ advisory_id: str | None = None,
+):
+ from models import Component, ComponentVersion
+
+ suffix = unique_suffix()
+ purl = f"pkg:npm/gate-mal-{suffix}"
+ component = Component(purl=purl, package_type="npm", name=f"gate-mal-{suffix}")
+ session.add(component)
+ await session.commit()
+ await session.refresh(component)
+
+ cv = ComponentVersion(
+ component_id=component.id,
+ version="1.0.0",
+ purl_with_version=f"{purl}@1.0.0",
+ malicious_state=malicious_state,
+ malicious_id=advisory_id,
+ malicious_source="osv.dev@seed" if malicious_state else None,
+ malicious_evaluated_at=datetime.now(tz=UTC) if malicious_state else None,
+ )
+ session.add(cv)
+ await session.commit()
+ await session.refresh(cv)
+ return purl, cv
+
+
+async def _attach(session: AsyncSession, *, scan_id, cv_id):
+ from models import ScanComponent
+
+ session.add(ScanComponent(scan_id=scan_id, component_version_id=cv_id, direct=True))
+ await session.commit()
+
+
+async def _make_policy(
+ session: AsyncSession,
+ *,
+ org_id: uuid.UUID,
+ team_id: uuid.UUID,
+ malicious_exceptions: list | None = None,
+):
+ from models import LicensePolicy
+
+ policy = LicensePolicy(
+ organization_id=org_id,
+ team_id=team_id,
+ name="gate-mal-policy",
+ category_overrides={},
+ license_exceptions=[],
+ malicious_exceptions=malicious_exceptions or [],
+ unknown_license_category="conditional",
+ enabled=True,
+ )
+ session.add(policy)
+ await session.commit()
+ await session.refresh(policy)
+ return policy
+
+
+async def _seeded_scan(session: AsyncSession, project):
+ scan = await make_scan(session, project=project, status="succeeded")
+ return scan
+
+
+# ---------------------------------------------------------------------------
+# The axis itself
+# ---------------------------------------------------------------------------
+
+
+async def test_a_flagged_component_fails_the_gate(db_session: AsyncSession) -> None:
+ from services.policy_gate import evaluate_gate
+
+ _, _, project = await _seed_project(db_session)
+ scan = await _seeded_scan(db_session, project)
+ _, cv = await _make_component(
+ db_session, malicious_state="flagged", advisory_id="MAL-0000-SEED"
+ )
+ await _attach(db_session, scan_id=scan.id, cv_id=cv.id)
+
+ result = await evaluate_gate(db_session, project.id, scan_id=scan.id)
+
+ assert result.gate == "fail"
+ assert result.malicious_component_count == 1
+ assert result.malicious_gate_enforced is True
+ assert result.reason is not None and "known-malicious" in result.reason
+ # It is not a vulnerability: the CVE axis stays untouched.
+ assert result.critical_cve_count == 0
+
+
+async def test_clear_and_unassessed_components_do_not_block(
+ db_session: AsyncSession,
+) -> None:
+ from services.policy_gate import evaluate_gate
+
+ _, _, project = await _seed_project(db_session)
+ scan = await _seeded_scan(db_session, project)
+ _, clear_cv = await _make_component(db_session, malicious_state="clear")
+ _, unassessed_cv = await _make_component(db_session, malicious_state=None)
+ await _attach(db_session, scan_id=scan.id, cv_id=clear_cv.id)
+ await _attach(db_session, scan_id=scan.id, cv_id=unassessed_cv.id)
+
+ result = await evaluate_gate(db_session, project.id, scan_id=scan.id)
+
+ assert result.gate == "pass"
+ assert result.malicious_component_count == 0
+
+
+async def test_disabled_axis_reports_zero_but_says_it_was_not_checked(
+ db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ """0 with the axis off must be distinguishable from 0 with it on.
+
+ Both render as "no malicious packages" if a consumer reads only the count,
+ which is why the flag rides alongside it.
+ """
+ from services.policy_gate import evaluate_gate
+
+ monkeypatch.setenv("GATE_MALICIOUS_ENABLED", "false")
+
+ _, _, project = await _seed_project(db_session)
+ scan = await _seeded_scan(db_session, project)
+ _, cv = await _make_component(db_session, malicious_state="flagged")
+ await _attach(db_session, scan_id=scan.id, cv_id=cv.id)
+
+ result = await evaluate_gate(db_session, project.id, scan_id=scan.id)
+
+ assert result.gate == "pass"
+ assert result.malicious_component_count == 0
+ assert result.malicious_gate_enforced is False
+
+
+# ---------------------------------------------------------------------------
+# Waiver lifecycle
+# ---------------------------------------------------------------------------
+
+
+async def test_waiver_lifts_the_block_then_expiry_puts_it_back(
+ db_session: AsyncSession,
+) -> None:
+ """The sequence the waiver exists for, start to finish.
+
+ A single-state test would pass on an implementation that never expires
+ anything — the expiry is the whole reason the field is mandatory, so the
+ return to blocking is what needs pinning.
+ """
+ from services.policy_gate import evaluate_gate
+
+ org, team, project = await _seed_project(db_session)
+ scan = await _seeded_scan(db_session, project)
+ purl, cv = await _make_component(db_session, malicious_state="flagged")
+ await _attach(db_session, scan_id=scan.id, cv_id=cv.id)
+
+ # 1. Flagged, no waiver → blocked.
+ first = await evaluate_gate(db_session, project.id, scan_id=scan.id)
+ assert first.gate == "fail"
+
+ # 2. Waiver with a future expiry → the build moves again.
+ future = (datetime.now(tz=UTC) + timedelta(days=7)).isoformat()
+ policy = await _make_policy(
+ db_session,
+ org_id=org.id,
+ team_id=team.id,
+ malicious_exceptions=[
+ {
+ "component_purl": purl,
+ "reason": "advisory challenged upstream, awaiting retraction",
+ "expires_at": future,
+ }
+ ],
+ )
+ second = await evaluate_gate(db_session, project.id, scan_id=scan.id)
+ assert second.gate == "pass"
+ assert second.malicious_component_count == 0
+
+ # 3. The waiver expires → blocked again, with no action from anyone.
+ past = (datetime.now(tz=UTC) - timedelta(minutes=1)).isoformat()
+ policy.malicious_exceptions = [
+ {
+ "component_purl": purl,
+ "reason": "advisory challenged upstream, awaiting retraction",
+ "expires_at": past,
+ }
+ ]
+ db_session.add(policy)
+ await db_session.commit()
+
+ third = await evaluate_gate(db_session, project.id, scan_id=scan.id)
+ assert third.gate == "fail"
+ assert third.malicious_component_count == 1
+
+
+async def test_a_waiver_for_another_package_does_not_lift_the_block(
+ db_session: AsyncSession,
+) -> None:
+ from services.policy_gate import evaluate_gate
+
+ org, team, project = await _seed_project(db_session)
+ scan = await _seeded_scan(db_session, project)
+ _, cv = await _make_component(db_session, malicious_state="flagged")
+ await _attach(db_session, scan_id=scan.id, cv_id=cv.id)
+
+ future = (datetime.now(tz=UTC) + timedelta(days=7)).isoformat()
+ await _make_policy(
+ db_session,
+ org_id=org.id,
+ team_id=team.id,
+ malicious_exceptions=[
+ {
+ "component_purl": "pkg:npm/some-other-package",
+ "reason": "unrelated",
+ "expires_at": future,
+ }
+ ],
+ )
+
+ result = await evaluate_gate(db_session, project.id, scan_id=scan.id)
+ assert result.gate == "fail"
+ assert result.malicious_component_count == 1
+
+
+async def test_waiver_written_with_a_version_still_matches(
+ db_session: AsyncSession,
+) -> None:
+ """Waivers match on the base purl.
+
+ An operator reaching for this during an incident will paste whatever the
+ UI showed them, which carries a version. Matching only the exact string
+ would silently fail to lift the block.
+ """
+ from services.policy_gate import evaluate_gate
+
+ org, team, project = await _seed_project(db_session)
+ scan = await _seeded_scan(db_session, project)
+ purl, cv = await _make_component(db_session, malicious_state="flagged")
+ await _attach(db_session, scan_id=scan.id, cv_id=cv.id)
+
+ future = (datetime.now(tz=UTC) + timedelta(days=7)).isoformat()
+ await _make_policy(
+ db_session,
+ org_id=org.id,
+ team_id=team.id,
+ malicious_exceptions=[
+ {
+ "component_purl": f"{purl}@1.0.0",
+ "reason": "pasted from the drawer",
+ "expires_at": future,
+ }
+ ],
+ )
+
+ result = await evaluate_gate(db_session, project.id, scan_id=scan.id)
+ assert result.gate == "pass"
+
+
+async def test_a_partly_evaluated_scan_does_not_claim_to_be_assessed(
+ db_session: AsyncSession,
+) -> None:
+ """One evaluated row must not vouch for the ones nobody looked at.
+
+ The persist hook turns its evaluator off on the first exception and leaves
+ the rest of the scan unstamped, so "some rows have verdicts" is the shape
+ a failed enrichment leaves behind — not the shape of a healthy scan. An
+ `assessed > 0` test would call this assessed and report a clean zero.
+ """
+ from services.policy_gate import evaluate_gate
+
+ _, _, project = await _seed_project(db_session)
+ scan = await _seeded_scan(db_session, project)
+ _, evaluated = await _make_component(db_session, malicious_state="clear")
+ _, never_evaluated = await _make_component(db_session, malicious_state=None)
+ await _attach(db_session, scan_id=scan.id, cv_id=evaluated.id)
+ await _attach(db_session, scan_id=scan.id, cv_id=never_evaluated.id)
+
+ result = await evaluate_gate(db_session, project.id, scan_id=scan.id)
+
+ assert result.gate == "pass"
+ assert result.malicious_component_count == 0
+ assert result.malicious_scan_assessed is False
+
+
+async def test_a_fully_evaluated_scan_reports_assessed(
+ db_session: AsyncSession,
+) -> None:
+ from services.policy_gate import evaluate_gate
+
+ _, _, project = await _seed_project(db_session)
+ scan = await _seeded_scan(db_session, project)
+ for _ in range(2):
+ _, cv = await _make_component(db_session, malicious_state="clear")
+ await _attach(db_session, scan_id=scan.id, cv_id=cv.id)
+
+ result = await evaluate_gate(db_session, project.id, scan_id=scan.id)
+
+ assert result.malicious_scan_assessed is True
+
+
+async def test_a_malformed_waiver_array_leaves_the_block_in_place(
+ db_session: AsyncSession,
+) -> None:
+ """JSONB holds whatever was written to it.
+
+ A shape the reader cannot parse is not a reason to 500 every gate read for
+ the team, and not a reason to honour waivers it cannot read either.
+ """
+ from models import LicensePolicy
+ from services.policy_gate import evaluate_gate
+
+ org, team, project = await _seed_project(db_session)
+ scan = await _seeded_scan(db_session, project)
+ _, cv = await _make_component(db_session, malicious_state="flagged")
+ await _attach(db_session, scan_id=scan.id, cv_id=cv.id)
+
+ policy = LicensePolicy(
+ organization_id=org.id,
+ team_id=team.id,
+ name="malformed",
+ category_overrides={},
+ license_exceptions=[],
+ malicious_exceptions={"not": "a list"},
+ unknown_license_category="conditional",
+ enabled=True,
+ )
+ db_session.add(policy)
+ await db_session.commit()
+
+ result = await evaluate_gate(db_session, project.id, scan_id=scan.id)
+
+ assert result.gate == "fail"
+ assert result.malicious_component_count == 1
+
diff --git a/apps/frontend/src/features/projects/api/governance.ts b/apps/frontend/src/features/projects/api/governance.ts
index 9ada8b9..eb0f8ed 100644
--- a/apps/frontend/src/features/projects/api/governance.ts
+++ b/apps/frontend/src/features/projects/api/governance.ts
@@ -22,6 +22,7 @@ export interface GovernanceGate {
critical_cve_count: number;
forbidden_license_count: number;
epss_gate_count: number;
+ malicious_component_count: number;
scan_id: string | null;
}
diff --git a/apps/frontend/src/features/projects/api/projectDetailApi.ts b/apps/frontend/src/features/projects/api/projectDetailApi.ts
index b09fa83..da82cee 100644
--- a/apps/frontend/src/features/projects/api/projectDetailApi.ts
+++ b/apps/frontend/src/features/projects/api/projectDetailApi.ts
@@ -497,6 +497,10 @@ export interface GateResultResponse {
* when the EPSS gate is disabled (`epss_threshold === null`).
*/
epss_gate_count: number;
+ /** #26 — components the malicious snapshot flags on the evaluated scan. */
+ malicious_component_count: number;
+ /** Whether the malicious axis ran at all (false → the count means "not checked"). */
+ malicious_scan_assessed: boolean;
/** Active EPSS gate threshold in [0, 1], or `null` when the EPSS gate is off. */
epss_threshold: number | null;
project_id: string;
diff --git a/apps/frontend/src/features/projects/components/GateResultCard.tsx b/apps/frontend/src/features/projects/components/GateResultCard.tsx
index bc67884..ad38d5b 100644
--- a/apps/frontend/src/features/projects/components/GateResultCard.tsx
+++ b/apps/frontend/src/features/projects/components/GateResultCard.tsx
@@ -201,6 +201,26 @@ export function GateResultCard({ projectId, scanId }: GateResultCardProps) {
emphasize={data.forbidden_license_count > 0}
testid="gate-metric-forbidden"
/>
+ {data.malicious_component_count > 0 ? (
+