ITWorx_MCP_Hub-Public/tests/test_scanner_cve_diagnostic.py

import json
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

import pytest

from scripts import build_mcp_hub_release_images as producer

IMAGE_ID = "sha256:" + "a" * 64


def report(identifier: object) -> dict[str, Any]:
    return {
        "SchemaVersion": 2,
        "CreatedAt": datetime.now(UTC).isoformat(),
        "Metadata": {"ImageID": IMAGE_ID, "ImageConfig": {"os": "linux", "architecture": "amd64"}},
        "Trivy": {"Version": producer.TRIVY_VERSION},
        "Results": [
            {
                "Packages": [{"Name": "private-package-do-not-echo", "Version": "1.0"}],
                "Vulnerabilities": [
                    {
                        "Severity": "HIGH",
                        "FixedVersion": "private-fixed-version-do-not-echo",
                        "VulnerabilityID": identifier,
                        "Description": "private-description-do-not-echo",
                    }
                ],
            }
        ],
    }


def diagnostic(value: dict[str, Any], service: str = "connector-appops") -> dict[str, Any]:
    with pytest.raises(producer.VulnerabilityRejected) as error:
        producer._validate_vulnerability(value, IMAGE_ID, service=service)
    assert "private-" not in str(error.value)
    text = str(error.value)
    return json.loads(text[text.index("{") :])


def test_closed_diagnostic_is_bound_to_exact_image_and_service() -> None:
    assert diagnostic(report("CVE-2026-12345")) == {
        "service": "connector-appops",
        "imageId": IMAGE_ID,
        "identifiers": ["CVE-2026-12345"],
        "blockingCount": 1,
        "omittedIdentifierCount": 0,
        "unavailableIdentifierCount": 0,
    }


@pytest.mark.parametrize(
    "identifier", [None, 123, "CVE-2026-12345\nsecret", "secret", "CVE-2026-" + "1" * 100]
)
def test_untrusted_identifiers_are_never_echoed(identifier: object) -> None:
    result = diagnostic(report(identifier))
    assert result["identifiers"] == []
    assert result["unavailableIdentifierCount"] == 1
    assert "secret" not in json.dumps(result)


def test_identifiers_are_deduplicated_sorted_and_bounded() -> None:
    value = report("GHSA-49wp-qq6x-g2rf")
    findings = value["Results"][0]["Vulnerabilities"]
    findings.extend(
        dict(findings[0], VulnerabilityID=f"CVE-2026-{number:04d}") for number in range(40, 0, -1)
    )
    findings.append(dict(findings[0]))
    result = diagnostic(value)
    assert result["blockingCount"] == 42
    assert result["identifiers"] == [f"CVE-2026-{number:04d}" for number in range(1, 33)]
    assert result["omittedIdentifierCount"] == 9
    assert diagnostic(report("GHSA-49wp-qq6x-g2rf"))["identifiers"] == ["GHSA-49wp-qq6x-g2rf"]


@pytest.mark.parametrize(
    "kind", ["service", "image", "binding", "stale", "schema", "empty", "malformed"]
)
def test_non_policy_rejections_cannot_be_caught_as_cve(kind: str) -> None:
    value = report("CVE-2026-12345")
    service, image_id = "connector-appops", IMAGE_ID
    if kind == "service":
        service = "private-secret"
    elif kind == "image":
        image_id = "private-secret"
    elif kind == "binding":
        value["Metadata"]["ImageID"] = "sha256:" + "b" * 64
    elif kind == "stale":
        value["CreatedAt"] = "2020-01-01T00:00:00Z"
    elif kind == "schema":
        value["SchemaVersion"] = 1
    elif kind == "empty":
        value["Results"][0]["Packages"] = []
    else:
        value["Results"][0]["Vulnerabilities"].append({"Severity": "invalid"})
    with pytest.raises(producer.BuildRejected) as error:
        producer._validate_vulnerability(value, image_id, service=service)
    assert type(error.value) is producer.BuildRejected
    assert "private-" not in str(error.value)


@pytest.mark.parametrize(("severity", "fixed"), [("HIGH", ""), ("MEDIUM", "1.1"), ("LOW", "1.1")])
def test_existing_nonblocking_policy_is_unchanged(severity: str, fixed: str) -> None:
    value = report("CVE-2026-12345")
    value["Results"][0]["Vulnerabilities"][0].update(Severity=severity, FixedVersion=fixed)
    producer._validate_vulnerability(value, IMAGE_ID, service="connector-appops")


@pytest.mark.parametrize("failure", [None, "stale", "secret"])
def test_scan_inventory_aggregates_only_policy_denials(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure: str | None
) -> None:
    calls: list[tuple[str, str]] = []
    images = {
        service: {"imageId": IMAGE_ID, "imageRef": service} for service in producer.ALL_SERVICES
    }

    def execute(*args: object, **kwargs: object) -> None:
        db = tmp_path / "trivy-cache" / "db"
        db.mkdir()
        producer._write_new(
            db / "metadata.json",
            json.dumps(
                {
                    "UpdatedAt": datetime.now(UTC).isoformat(),
                    "DownloadedAt": datetime.now(UTC).isoformat(),
                }
            ).encode(),
        )

    def capture(command: list[str], **kwargs: object) -> bytes:
        scanner, service = command[command.index("--scanners") + 1], command[-1]
        calls.append((service, scanner))
        if scanner == "vuln":
            value = report("CVE-2026-12345")
            if failure == "stale" and service == producer.ALL_SERVICES[1]:
                value["CreatedAt"] = "2020-01-01T00:00:00Z"
        else:
            value = {"Metadata": {"ImageID": IMAGE_ID}, "Results": []}
            if failure == "secret":
                value["Results"] = [{"Secrets": [{"Match": "private-secret-never-persist"}]}]
        return json.dumps(value).encode()

    monkeypatch.setattr(producer, "_execute", execute)
    monkeypatch.setattr(producer, "_private_db_metadata", lambda path: None)
    monkeypatch.setattr(producer, "_run_scanner_capture", capture)
    with pytest.raises(producer.BuildRejected) as error:
        producer._install_and_scan(
            images=images,
            temp_root=tmp_path,
            env={},
            cwd=tmp_path,
            runner=lambda *args, **kwargs: None,
            scanner_override=tmp_path / "scanner",
        )
    assert "private-" not in str(error.value)
    if failure is None:
        assert calls == [
            (service, scanner)
            for service in producer.ALL_SERVICES
            for scanner in ("vuln", "secret")
        ]
        lines = str(error.value).splitlines()
        assert len(lines) == 17
        assert [json.loads(line[line.index("{") :])["service"] for line in lines] == list(
            producer.ALL_SERVICES
        )
        assert {path.name for path in (tmp_path / "reports").iterdir()} == producer.REPORT_NAMES
    else:
        assert len(calls) < 34
        assert "secret-never-persist" not in "".join(
            path.read_text() for path in (tmp_path / "reports").iterdir()
        )
    assert not (tmp_path / "candidate.json").exists()