From ddefe2ed530da20c969da0d4dd01013a67050993 Mon Sep 17 00:00:00 2001 From: Marina Villarreal Date: Wed, 22 Jul 2026 14:14:46 -0700 Subject: [PATCH 1/2] Add Anchore V2 importer stub and register pipeline Signed-off-by: Marina Villarreal --- vulnerabilities/importers/__init__.py | 2 ++ .../v2_importers/anchore_importer.py | 23 +++++++++++++++++++ 2 files changed, 25 insertions(+) create mode 100644 vulnerabilities/pipelines/v2_importers/anchore_importer.py diff --git a/vulnerabilities/importers/__init__.py b/vulnerabilities/importers/__init__.py index a33e6d714..7df9e1351 100644 --- a/vulnerabilities/importers/__init__.py +++ b/vulnerabilities/importers/__init__.py @@ -9,6 +9,7 @@ from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2 from vulnerabilities.pipelines.v2_importers import alpine_linux_importer as alpine_linux_importer_v2 +from vulnerabilities.pipelines.v2_importers import anchore_importer as anchore_importer_v2 from vulnerabilities.pipelines.v2_importers import aosp_importer as aosp_importer_v2 from vulnerabilities.pipelines.v2_importers import apache_httpd_importer as apache_httpd_v2 from vulnerabilities.pipelines.v2_importers import apache_kafka_importer as apache_kafka_importer_v2 @@ -67,6 +68,7 @@ pysec_importer_v2.PyPIImporterPipeline, xen_importer_v2.XenImporterPipeline, curl_importer_v2.CurlImporterPipeline, + anchore_importer_v2.AnchoreImporterPipeline, oss_fuzz_v2.OSSFuzzImporterPipeline, istio_importer_v2.IstioImporterPipeline, postgresql_importer_v2.PostgreSQLImporterPipeline, diff --git a/vulnerabilities/pipelines/v2_importers/anchore_importer.py b/vulnerabilities/pipelines/v2_importers/anchore_importer.py new file mode 100644 index 000000000..eb6831ff6 --- /dev/null +++ b/vulnerabilities/pipelines/v2_importers/anchore_importer.py @@ -0,0 +1,23 @@ +# +# Copyright (c) nexB Inc. and others. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# + +import logging +from typing import Iterable + +from vulnerabilities.importer import AdvisoryDataV2 +from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2 + +logger = logging.getLogger(__name__) + + +class AnchoreImporterPipeline(VulnerableCodeBaseImporterPipelineV2): + """ + Importer for https://github.com/anchore/nvd-data-overrides + """ + + pipeline_id = "anchore_importer_v2" + datasource_id = "anchore" + spdx_license_expression = "Apache-2.0" + license_url = "https://github.com/anchore/nvd-data-overrides/blob/main/LICENSE" From b896a7d1dcc62c4da13d2227ef47c66644201688 Mon Sep 17 00:00:00 2001 From: K4TV Date: Tue, 11 Aug 2026 09:12:32 -0700 Subject: [PATCH 2/2] add tests for Anchore NVD Override advisory parsing and collection --- .../v2_importers/test_anchore_importer_v2.py | 141 ++++++++++++++++++ 1 file changed, 141 insertions(+) create mode 100644 vulnerabilities/tests/pipelines/v2_importers/test_anchore_importer_v2.py diff --git a/vulnerabilities/tests/pipelines/v2_importers/test_anchore_importer_v2.py b/vulnerabilities/tests/pipelines/v2_importers/test_anchore_importer_v2.py new file mode 100644 index 000000000..1bc7ae3cc --- /dev/null +++ b/vulnerabilities/tests/pipelines/v2_importers/test_anchore_importer_v2.py @@ -0,0 +1,141 @@ +# +# Copyright (c) nexB Inc. and others. All rights reserved. +# VulnerableCode is a trademark of nexB Inc. +# SPDX-License-Identifier: Apache-2.0 +# See http://www.apache.org/licenses/LICENSE-2.0 for the license text. +# See https://github.com/aboutcode-org/vulnerablecode for support or download. +# See https://aboutcode.org for more information about nexB OSS projects. + +import json +import logging +import pytest + +from vulnerabilities.importer import AdvisoryDataV2 +from vulnerabilities.pipelines.v2_importers.anchore_importer import AnchoreImporterPipeline +from vulnerabilities.pipelines.v2_importers.anchore_importer import parse_anchore_advisory + + +class DummyVCSResponse: + ''' A dummy class to simulate the response from fetch_via_vcs for testing purposes. + ''' + def __init__(self, dest_dir): + self.dest_dir = str(dest_dir) + +@pytest.fixture +def pipeline(): + return AnchoreImporterPipeline() # the actual pipeline instance used in tests + + +def test_parse_anchore_advisory_maps_basic_fields(tmp_path): + ''' + Parse an Anchore advisory, the basic information gets put into the right places + ''' + file_path = tmp_path / "data" / "nested" / "CVE-2024-0001.json" + file_path.parent.mkdir(parents=True) + + # fake raw data + raw_data = { + "description": "A sample advisory", + "_annotation": { + "description": "Ignored annotation description", + "references": ["https://example.com/annotation"], + }, + "cve": { + "references": { + "reference_data": [{"url": "https://example.com/cve-reference"}], + } + }, + } + + advisory = parse_anchore_advisory( + raw_data=raw_data, + file_path=file_path, + base_path=tmp_path, + logger=logging.getLogger(__name__), + ) + + assert isinstance(advisory, AdvisoryDataV2) + assert advisory.advisory_id == "anchore/CVE-2024-0001" + assert advisory.aliases == ["CVE-2024-0001"] + assert advisory.summary == "A sample advisory" + assert [ref.url for ref in advisory.references] == [ + "https://example.com/annotation", + "https://example.com/cve-reference", + ] + assert advisory.url == ( + "https://github.com/anchore/nvd-data-overrides/blob/main/data/nested/CVE-2024-0001.json" + ) + + +def test_collect_advisories_yields_advisories_from_data_directory(tmp_path, pipeline): + ''' + Test if it extracts advisories from the data directory + - Find the data directory + - Find the JSON files + - Open the files + - Read the JSON. + - Parse the data + - Create AdvisoryDataV2 objects + - Return those advisories + ''' + data_dir = tmp_path / "data" + data_dir.mkdir(parents=True) + + first = data_dir / "CVE-2024-0001.json" + first.write_text( + json.dumps( + { + "description": "First advisory", + "_annotation": {"references": ["https://example.com/one"]}, + "cve": {"references": {"reference_data": [{"url": "https://example.com/one-bis"}]}} + } + ) + ) + + second = data_dir / "CVE-2024-0002.json" + second.write_text( + json.dumps( + { + "description": "Second advisory", + "_annotation": {"references": ["https://example.com/two"]}, + "cve": {"references": {"reference_data": []}}, + } + ) + ) + + pipeline.vcs_response = DummyVCSResponse(tmp_path) + + advisories = list(pipeline.collect_advisories()) + advisory_ids = sorted(advisory.advisory_id for advisory in advisories) + + assert advisory_ids == ["anchore/CVE-2024-0001", "anchore/CVE-2024-0002"] + assert advisories[0].summary == "First advisory" + assert advisories[1].summary == "Second advisory" + + +def test_collect_advisories_returns_no_results_for_empty_data_directory(tmp_path, pipeline): + ''' + Test that if the data directory is empty, no advisories are returned. + ''' + data_dir = tmp_path / "data" + data_dir.mkdir(parents=True) + pipeline.vcs_response = DummyVCSResponse(tmp_path) + + assert list(pipeline.collect_advisories()) == [] + + +def test_collect_advisories_handles_malformed_json(tmp_path, pipeline): + ''' + Test a broken JSON file in the data directory, it should log a warning and continue. + ''' + data_dir = tmp_path / "data" + data_dir.mkdir(parents=True) + (data_dir / "CVE-2024-0003.json").write_text("{not valid json") + + pipeline.vcs_response = DummyVCSResponse(tmp_path) + messages = [] + pipeline.log = lambda message, level=logging.INFO: messages.append((message, level)) + + assert list(pipeline.collect_advisories()) == [] + assert any("Failed to process advisory" in message for message, _ in messages) +