diff --git a/vulnerabilities/pipelines/v2_importers/debian_importer.py b/vulnerabilities/pipelines/v2_importers/debian_importer.py index 4f72aedc8..d39b38cb1 100644 --- a/vulnerabilities/pipelines/v2_importers/debian_importer.py +++ b/vulnerabilities/pipelines/v2_importers/debian_importer.py @@ -22,7 +22,6 @@ from vulnerabilities.importer import ReferenceV2 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2 from vulnerabilities.utils import create_weaknesses_list -from vulnerabilities.utils import dedupe from vulnerabilities.utils import fetch_response from vulnerabilities.utils import get_item @@ -115,8 +114,7 @@ def collect_advisories(self) -> Iterable[AdvisoryData]: def parse(self, pkg_name: str, records: Mapping[str, Any]) -> Iterable[AdvisoryData]: for record_identifier, record in records.items(): - affected_versions = [] - fixed_versions = [] + affected_packages = [] releases = record["releases"].items() for release_name, release_record in releases: @@ -128,41 +126,37 @@ def parse(self, pkg_name: str, records: Mapping[str, Any]) -> Iterable[AdvisoryD ) continue - purl = PackageURL( - name=pkg_name, - type="deb", - namespace="debian", - qualifiers={"distro": release_name}, - ) + # Debian uses "0" for releases that were never affected + fixed_version = release_record.get("fixed_version") + if fixed_version == "0": + continue - if release_record.get("status", "") == "resolved": - fixed_versions.append(version) - else: - affected_versions.append(version) + affected_version_range = None + if release_record.get("status", "") != "resolved": + affected_version_range = DebianVersionRange.from_versions([version]) - if release_record.get("fixed_version"): - fixed_versions.append(release_record["fixed_version"]) + fixed_version_range = None + if fixed_version: + fixed_version_range = DebianVersionRange.from_versions([fixed_version]) - references = [] - debianbug = record.get("debianbug") - if debianbug: - bug_url = f"https://bugs.debian.org/cgi-bin/bugreport.cgi?bug={debianbug}" - references.append(ReferenceV2(url=bug_url, reference_id=str(debianbug))) - affected_versions = dedupe(affected_versions) - fixed_versions = dedupe(fixed_versions) - if affected_versions: - affected_version_range = DebianVersionRange.from_versions(affected_versions) - else: - affected_version_range = None - affected_packages = [] - for fixed_version in fixed_versions: affected_packages.append( AffectedPackageV2( - package=purl, + package=PackageURL( + name=pkg_name, + type="deb", + namespace="debian", + qualifiers={"distro": release_name}, + ), affected_version_range=affected_version_range, - fixed_version_range=DebianVersionRange.from_versions([fixed_version]), + fixed_version_range=fixed_version_range, ) ) + + references = [] + debianbug = record.get("debianbug") + if debianbug: + bug_url = f"https://bugs.debian.org/cgi-bin/bugreport.cgi?bug={debianbug}" + references.append(ReferenceV2(url=bug_url, reference_id=str(debianbug))) weaknesses = get_cwe_from_debian_advisory(record) yield AdvisoryDataV2( diff --git a/vulnerabilities/tests/pipelines/v2_importers/test_debian_importer.py b/vulnerabilities/tests/pipelines/v2_importers/test_debian_importer.py index 4474a5efd..4c9221826 100644 --- a/vulnerabilities/tests/pipelines/v2_importers/test_debian_importer.py +++ b/vulnerabilities/tests/pipelines/v2_importers/test_debian_importer.py @@ -120,7 +120,12 @@ def test_affected_packages_generation(importer, sample_response): for pkg in affected_packages: assert isinstance(pkg, AffectedPackageV2) assert isinstance(pkg.package, PackageURL) - assert isinstance(pkg.fixed_version_range, DebianVersionRange) + + bullseye, bookworm = affected_packages + assert bullseye.package.qualifiers == {"distro": "bullseye"} + assert isinstance(bullseye.fixed_version_range, DebianVersionRange) + assert bookworm.package.qualifiers == {"distro": "bookworm"} + assert isinstance(bookworm.affected_version_range, DebianVersionRange) def test_debian_bug_reference(importer, sample_response): @@ -135,3 +140,47 @@ def test_debian_bug_reference(importer, sample_response): assert isinstance(ref, ReferenceV2) assert ref.reference_id == "123456" assert "bugs.debian.org" in ref.url + + +def test_affected_packages_are_built_per_release(importer): + records = { + "CVE-2024-0727": { + "releases": { + "bookworm": { + "status": "resolved", + "repositories": {"bookworm": "3.0.20-1~deb12u2"}, + "fixed_version": "3.0.13-1~deb12u1", + }, + "trixie": { + "status": "resolved", + "repositories": {"trixie": "3.5.7-1~deb13u2"}, + "fixed_version": "3.1.5-1", + }, + "bullseye": { + "status": "resolved", + "repositories": {"bullseye": "1.1.1w-0+deb11u1"}, + "fixed_version": "0", + }, + "sid": { + "status": "open", + "repositories": {"sid": "3.6.5-1"}, + }, + } + } + } + + advisory = next(importer.parse("openssl", records)) + result = { + pkg.package.qualifiers["distro"]: ( + str(pkg.affected_version_range) if pkg.affected_version_range else None, + str(pkg.fixed_version_range) if pkg.fixed_version_range else None, + ) + for pkg in advisory.affected_packages + } + + assert len(advisory.affected_packages) == 3 + assert result == { + "bookworm": (None, "vers:deb/3.0.13-1~deb12u1"), + "trixie": (None, "vers:deb/3.1.5-1"), + "sid": ("vers:deb/3.6.5-1", None), + }