Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 24 additions & 30 deletions vulnerabilities/pipelines/v2_importers/debian_importer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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:
Expand All @@ -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(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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),
}