Enhance code commit collection capabilities #2022

Signed-off-by: ziad hany <ziadhany2016@gmail.com>

Author: ziadhany

requirements.txt

@@ -64,7 +64,7 @@ MarkupSafe==2.1.1
 matplotlib-inline==0.1.3
 multidict==6.0.2
 mypy-extensions==0.4.3
-packageurl-python==0.15.6
+packageurl-python==0.17.6
 packaging==21.3
 paramiko==3.4.0
 parso==0.8.3

setup.cfg

@@ -71,7 +71,7 @@ install_requires =
     drf-spectacular[sidecar]>=0.24.2
 
     #essentials
-    packageurl-python>=0.15
+    packageurl-python>=0.17
     univers>=30.12.0
     license-expression>=30.0.0
 

vulnerabilities/importer.py

@@ -10,6 +10,7 @@
 import dataclasses
 import datetime
 import functools
+import hashlib
 import logging
 import traceback
 import xml.etree.ElementTree as ET
@@ -37,6 +38,7 @@
 from vulnerabilities.severity_systems import ScoringSystem
 from vulnerabilities.utils import classproperty
 from vulnerabilities.utils import get_reference_id
+from vulnerabilities.utils import is_commit
 from vulnerabilities.utils import is_cve
 from vulnerabilities.utils import nearest_patched_package
 from vulnerabilities.utils import purl_to_dict
@@ -194,6 +196,97 @@ def from_url(cls, url):
         return cls(url=url)
 
 
+@dataclasses.dataclass(eq=True)
+@functools.total_ordering
+class PackageCommitPatchData:
+    vcs_url: str
+    commit_hash: str
+    patch_text: Optional[str] = None
+
+    def __post_init__(self):
+        if not self.commit_hash:
+            raise ValueError("Commit must have a non-empty commit_hash.")
+
+        if not is_commit(self.commit_hash):
+            raise ValueError(f"Commit must be a valid a commit_hash: {self.commit_hash}.")
+
+        if not self.vcs_url:
+            raise ValueError("Commit must have a non-empty vcs_url.")
+
+    def __lt__(self, other):
+        if not isinstance(other, PackageCommitPatchData):
+            return NotImplemented
+        return self._cmp_key() < other._cmp_key()
+
+    # TODO: Add cache
+    def _cmp_key(self):
+        return (
+            self.vcs_url,
+            self.commit_hash,
+            self.patch_text,
+        )
+
+    def to_dict(self) -> dict:
+        """Return a normalized dictionary representation of the commit."""
+        return {
+            "vcs_url": self.vcs_url,
+            "commit_hash": self.commit_hash,
+            "patch_text": self.patch_text,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        """Create a PackageCommitPatchData instance from a dictionary."""
+        return cls(
+            vcs_url=data.get("vcs_url"),
+            commit_hash=data.get("commit_hash"),
+            patch_text=data.get("patch_text"),
+        )
+
+
+@dataclasses.dataclass(eq=True)
+@functools.total_ordering
+class PatchData:
+    patch_url: Optional[str] = None
+    patch_text: Optional[str] = None
+    patch_checksum: Optional[str] = dataclasses.field(init=False, default=None)
+
+    def __post_init__(self):
+        if not self.patch_url and not self.patch_text:
+            raise ValueError("A patch must include either patch_url or patch_text")
+
+        if self.patch_text:
+            self.patch_checksum = hashlib.sha512(self.patch_text.encode()).hexdigest()
+
+    def __lt__(self, other):
+        if not isinstance(other, PatchData):
+            return NotImplemented
+        return self._cmp_key() < other._cmp_key()
+
+    def _cmp_key(self):
+        return (
+            self.patch_url,
+            self.patch_text,
+            self.patch_checksum,
+        )
+
+    def to_dict(self) -> dict:
+        """Return a normalized dictionary representation of the commit."""
+        return {
+            "patch_url": self.patch_url,
+            "patch_text": self.patch_text,
+            "patch_checksum": self.patch_checksum,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        """Create a PatchData instance from a dictionary."""
+        return cls(
+            patch_url=data.get("patch_url"),
+            patch_text=data.get("patch_text"),
+        )
+
+
 class UnMergeablePackageError(Exception):
     """
     Raised when a package cannot be merged with another one.
@@ -344,21 +437,30 @@ class AffectedPackageV2:
     """
     Relate a Package URL with a range of affected versions and fixed versions.
     The Package URL must *not* have a version.
-    AffectedPackage must contain either ``affected_version_range`` or ``fixed_version_range``.
+    AffectedPackage must contain either ``affected_version_range`` or ``fixed_version_range`` or ``introduced_by_commits`` or ``fixed_by_commits``.
     """
 
     package: PackageURL
     affected_version_range: Optional[VersionRange] = None
     fixed_version_range: Optional[VersionRange] = None
+    introduced_by_commit_patches: List[PackageCommitPatchData] = dataclasses.field(
+        default_factory=list
+    )
+    fixed_by_commit_patches: List[PackageCommitPatchData] = dataclasses.field(default_factory=list)
 
     def __post_init__(self):
         if self.package.version:
             raise ValueError(f"Affected Package URL {self.package!r} cannot have a version.")
 
-        if not (self.affected_version_range or self.fixed_version_range):
+        if not (
+            self.affected_version_range
+            or self.fixed_version_range
+            or self.introduced_by_commit_patches
+            or self.fixed_by_commit_patches
+        ):
             raise ValueError(
-                f"Affected Package {self.package!r} should have either fixed version range or an "
-                "affected version range."
+                f"Affected package {self.package!r} must have either a fixed version range, "
+                "an affected version range, introduced commit patches, or fixed commit patches."
             )
 
     def __lt__(self, other):
@@ -372,6 +474,8 @@ def _cmp_key(self):
             str(self.package),
             str(self.affected_version_range or ""),
             str(self.fixed_version_range or ""),
+            str(self.introduced_by_commit_patches or []),
+            str(self.fixed_by_commit_patches or []),
         )
 
     def to_dict(self):
@@ -385,6 +489,12 @@ def to_dict(self):
             "package": purl_to_dict(self.package),
             "affected_version_range": affected_version_range,
             "fixed_version_range": fixed_version_range,
+            "introduced_by_commit_patches": [
+                commit.to_dict() for commit in self.introduced_by_commit_patches
+            ],
+            "fixed_by_commit_patches": [
+                commit.to_dict() for commit in self.fixed_by_commit_patches
+            ],
         }
 
     @classmethod
@@ -396,6 +506,10 @@ def from_dict(cls, affected_pkg: dict):
         fixed_version_range = None
         affected_range = affected_pkg["affected_version_range"]
         fixed_range = affected_pkg["fixed_version_range"]
+        introduced_by_commit_patches = (
+            affected_pkg.get("introduced_by_package_commit_patches") or []
+        )
+        fixed_by_commit_patches = affected_pkg.get("fixed_by_package_commit_patches") or []
 
         try:
             affected_version_range = VersionRange.from_string(affected_range)
@@ -417,6 +531,12 @@ def from_dict(cls, affected_pkg: dict):
             package=package,
             affected_version_range=affected_version_range,
             fixed_version_range=fixed_version_range,
+            introduced_by_commit_patches=[
+                PackageCommitPatchData.from_dict(commit) for commit in introduced_by_commit_patches
+            ],
+            fixed_by_commit_patches=[
+                PackageCommitPatchData.from_dict(commit) for commit in fixed_by_commit_patches
+            ],
         )
 
 
@@ -441,6 +561,7 @@ class AdvisoryData:
     )
     references: List[Reference] = dataclasses.field(default_factory=list)
     references_v2: List[ReferenceV2] = dataclasses.field(default_factory=list)
+    patches: List[PatchData] = dataclasses.field(default_factory=list)
     date_published: Optional[datetime.datetime] = None
     weaknesses: List[int] = dataclasses.field(default_factory=list)
     severities: List[VulnerabilitySeverity] = dataclasses.field(default_factory=list)
@@ -473,6 +594,7 @@ def to_dict(self):
                 "summary": self.summary,
                 "affected_packages": [pkg.to_dict() for pkg in self.affected_packages],
                 "references_v2": [ref.to_dict() for ref in self.references_v2],
+                "patches": [patch.to_dict() for patch in self.patches],
                 "severities": [sev.to_dict() for sev in self.severities],
                 "date_published": self.date_published.isoformat() if self.date_published else None,
                 "weaknesses": self.weaknesses,
@@ -533,6 +655,7 @@ class AdvisoryDataV2:
     summary: Optional[str] = ""
     affected_packages: List[AffectedPackage] = dataclasses.field(default_factory=list)
     references: List[ReferenceV2] = dataclasses.field(default_factory=list)
+    patches: List[PatchData] = dataclasses.field(default_factory=list)
     date_published: Optional[datetime.datetime] = None
     weaknesses: List[int] = dataclasses.field(default_factory=list)
     url: Optional[str] = None
@@ -557,6 +680,7 @@ def to_dict(self):
             "summary": self.summary,
             "affected_packages": [pkg.to_dict() for pkg in self.affected_packages],
             "references": [ref.to_dict() for ref in self.references],
+            "patches": [ref.to_dict() for ref in self.patches],
             "date_published": self.date_published.isoformat() if self.date_published else None,
             "weaknesses": self.weaknesses,
             "url": self.url if self.url else "",
@@ -574,6 +698,7 @@ def from_dict(cls, advisory_data):
                 if pkg is not None
             ],
             "references": [Reference.from_dict(ref) for ref in advisory_data["references"]],
+            "patches": [PatchData.from_dict(ref) for ref in advisory_data["patches"]],
             "date_published": datetime.datetime.fromisoformat(date_published)
             if date_published
             else None,

vulnerabilities/importers/__init__.py

@@ -41,6 +41,7 @@
 from vulnerabilities.pipelines import nvd_importer
 from vulnerabilities.pipelines import pypa_importer
 from vulnerabilities.pipelines import pysec_importer
+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 archlinux_importer as archlinux_importer_v2
 from vulnerabilities.pipelines.v2_importers import curl_importer as curl_importer_v2
@@ -81,6 +82,7 @@
         mozilla_importer_v2.MozillaImporterPipeline,
         github_osv_importer_v2.GithubOSVImporterPipeline,
         redhat_importer_v2.RedHatImporterPipeline,
+        aosp_importer_v2.AospImporterPipeline,
         nvd_importer.NVDImporterPipeline,
         github_importer.GitHubAPIImporterPipeline,
         gitlab_importer.GitLabImporterPipeline,

vulnerabilities/importers/curl.py

@@ -97,7 +97,7 @@ def parse_advisory_data(raw_data) -> AdvisoryData:
         ...     ]
         ... }
         >>> parse_advisory_data(raw_data)
-        AdvisoryData(advisory_id='', aliases=['CVE-2024-2379'], summary='QUIC certificate check bypass with wolfSSL', affected_packages=[AffectedPackage(package=PackageURL(type='generic', namespace='curl.se', name='curl', version=None, qualifiers={}, subpath=None), affected_version_range=GenericVersionRange(constraints=(VersionConstraint(comparator='=', version=SemverVersion(string='8.6.0')),)), fixed_version=SemverVersion(string='8.7.0'))], references=[Reference(reference_id='', reference_type='', url='https://curl.se/docs/CVE-2024-2379.html', severities=[VulnerabilitySeverity(system=Cvssv3ScoringSystem(identifier='cvssv3.1', name='CVSSv3.1 Base Score', url='https://www.first.org/cvss/v3-1/', notes='CVSSv3.1 base score and vector'), value='Low', scoring_elements='', published_at=None, url=None)]), Reference(reference_id='', reference_type='', url='https://hackerone.com/reports/2410774', severities=[])], references_v2=[], date_published=datetime.datetime(2024, 3, 27, 8, 0, tzinfo=datetime.timezone.utc), weaknesses=[297], severities=[], url='https://curl.se/docs/CVE-2024-2379.json', original_advisory_text=None)
+        AdvisoryData(advisory_id='', aliases=['CVE-2024-2379'], summary='QUIC certificate check bypass with wolfSSL', affected_packages=[AffectedPackage(package=PackageURL(type='generic', namespace='curl.se', name='curl', version=None, qualifiers={}, subpath=None), affected_version_range=GenericVersionRange(constraints=(VersionConstraint(comparator='=', version=SemverVersion(string='8.6.0')),)), fixed_version=SemverVersion(string='8.7.0'))], references=[Reference(reference_id='', reference_type='', url='https://curl.se/docs/CVE-2024-2379.html', severities=[VulnerabilitySeverity(system=Cvssv3ScoringSystem(identifier='cvssv3.1', name='CVSSv3.1 Base Score', url='https://www.first.org/cvss/v3-1/', notes='CVSSv3.1 base score and vector'), value='Low', scoring_elements='', published_at=None, url=None)]), Reference(reference_id='', reference_type='', url='https://hackerone.com/reports/2410774', severities=[])], references_v2=[], patches=[], date_published=datetime.datetime(2024, 3, 27, 8, 0, tzinfo=datetime.timezone.utc), weaknesses=[297], severities=[], url='https://curl.se/docs/CVE-2024-2379.json', original_advisory_text=None)
     """
 
     affected = get_item(raw_data, "affected")[0] if len(get_item(raw_data, "affected")) > 0 else []

vulnerabilities/migrations/0104_packagecommitpatch_patch_and_more.py

@@ -0,0 +1,114 @@
+# Generated by Django 4.2.25 on 2025-12-01 14:42
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0103_codecommit_impactedpackage_affecting_commits_and_more"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="PackageCommitPatch",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True, primary_key=True, serialize=False, verbose_name="ID"
+                    ),
+                ),
+                (
+                    "commit_hash",
+                    models.CharField(
+                        help_text="The commit hash of the patch in the VCS.", max_length=128
+                    ),
+                ),
+                (
+                    "vcs_url",
+                    models.URLField(
+                        help_text="The Version Control System URL (e.g., git repo URL).",
+                        max_length=1024,
+                    ),
+                ),
+                ("patch_text", models.TextField(blank=True, null=True)),
+                ("patch_checksum", models.CharField(blank=True, max_length=128, null=True)),
+            ],
+            options={
+                "unique_together": {("commit_hash", "vcs_url")},
+            },
+        ),
+        migrations.CreateModel(
+            name="Patch",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True, primary_key=True, serialize=False, verbose_name="ID"
+                    ),
+                ),
+                (
+                    "patch_url",
+                    models.URLField(
+                        blank=True, help_text="URL to the patch file or diff.", null=True
+                    ),
+                ),
+                (
+                    "patch_text",
+                    models.TextField(
+                        blank=True,
+                        help_text="The actual text content of the code patch/diff.",
+                        null=True,
+                    ),
+                ),
+                (
+                    "patch_checksum",
+                    models.CharField(
+                        help_text="SHA512 checksum of the patch content.", max_length=128
+                    ),
+                ),
+            ],
+            options={
+                "unique_together": {("patch_checksum", "patch_url")},
+            },
+        ),
+        migrations.RemoveField(
+            model_name="impactedpackage",
+            name="affecting_commits",
+        ),
+        migrations.RemoveField(
+            model_name="impactedpackage",
+            name="fixed_by_commits",
+        ),
+        migrations.DeleteModel(
+            name="CodeCommit",
+        ),
+        migrations.AddField(
+            model_name="advisoryv2",
+            name="patches",
+            field=models.ManyToManyField(
+                help_text="A list of patches associated with this advisory.",
+                related_name="advisories",
+                to="vulnerabilities.patch",
+            ),
+        ),
+        migrations.AddField(
+            model_name="impactedpackage",
+            name="fixed_by_package_commit_patches",
+            field=models.ManyToManyField(
+                help_text="PackageCommitPatches that fix this impact.",
+                related_name="fixed_in_impacts",
+                to="vulnerabilities.packagecommitpatch",
+            ),
+        ),
+        migrations.AddField(
+            model_name="impactedpackage",
+            name="introduced_by_package_commit_patches",
+            field=models.ManyToManyField(
+                help_text="PackageCommitPatches that introduce this impact.",
+                related_name="introduced_in_impacts",
+                to="vulnerabilities.packagecommitpatch",
+            ),
+        ),
+    ]