From 867a1b5e3bc884ea6b17c85ffe46491d726a01d2 Mon Sep 17 00:00:00 2001 From: Public Source Publisher Date: Wed, 14 Oct 2026 09:42:42 +0000 Subject: [PATCH] Publish curated ITWorx_MCP_Hub source from fb03947ffb946bfeaca89437fbb5f90751a14993 --- PUBLIC_SOURCE_EXPORT.md | 2 +- PUBLIC_SOURCE_MANIFEST.json | 7 +- tests/test_scanner_cve_diagnostic.py | 176 +++++++++++++++++++++++++++ 3 files changed, 183 insertions(+), 2 deletions(-) create mode 100644 tests/test_scanner_cve_diagnostic.py diff --git a/PUBLIC_SOURCE_EXPORT.md b/PUBLIC_SOURCE_EXPORT.md index dc160b7..d28e78a 100644 --- a/PUBLIC_SOURCE_EXPORT.md +++ b/PUBLIC_SOURCE_EXPORT.md @@ -1,3 +1,3 @@ # Public source export -This source snapshot was generated from the private canonical repository at revision `ae713b53e32933192b7446ac25fd6394e69eb133`. It contains no private Git history or operational evidence. Changes are published from the canonical repository. +This source snapshot was generated from the private canonical repository at revision `fb03947ffb946bfeaca89437fbb5f90751a14993`. It contains no private Git history or operational evidence. Changes are published from the canonical repository. diff --git a/PUBLIC_SOURCE_MANIFEST.json b/PUBLIC_SOURCE_MANIFEST.json index 9b0c653..a2c0cad 100644 --- a/PUBLIC_SOURCE_MANIFEST.json +++ b/PUBLIC_SOURCE_MANIFEST.json @@ -1,6 +1,6 @@ { "schemaVersion": 1, - "sourceRevision": "ae713b53e32933192b7446ac25fd6394e69eb133", + "sourceRevision": "fb03947ffb946bfeaca89437fbb5f90751a14993", "files": [ { "path": ".dockerignore", @@ -3462,6 +3462,11 @@ "sha256": "3a3a6674c9e3146827cc34aa493e0445505067b6e90d5618c3552c7dc2861118", "bytes": 3902 }, + { + "path": "tests/test_scanner_cve_diagnostic.py", + "sha256": "d6d36338f22459a2bb53547933e6dea0863ff669c96f7133edd008052cdff1b6", + "bytes": 6864 + }, { "path": "tests/test_scanner_private_files.py", "sha256": "772b87fafb5a877a32234ab10b6ed418445575ac9dc0e13380491d294c55330c", diff --git a/tests/test_scanner_cve_diagnostic.py b/tests/test_scanner_cve_diagnostic.py new file mode 100644 index 0000000..0fdecbc --- /dev/null +++ b/tests/test_scanner_cve_diagnostic.py @@ -0,0 +1,176 @@ +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()