Drop AdvisoryDataV2

Add constraint to make sure we have at least one field to create a valid Patch obj.
Update patch_text only if patch_text field is empty.
Return multiple objects for classify_patch_source function
Add patch in AdviosryData.from_dict()

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

Author: ziadhany

vulnerabilities/importer.py

@@ -633,6 +633,7 @@ def from_dict(cls, advisory_data):
             "affected_packages": [
                 affected_package_cls.from_dict(pkg) for pkg in affected_packages if pkg is not None
             ],
+            "patches": [PatchData.from_dict(patch) for patch in advisory_data.get("patches", [])],
             "references": [Reference.from_dict(ref) for ref in advisory_data["references"]],
             "date_published": datetime.datetime.fromisoformat(date_published)
             if date_published
@@ -643,77 +644,6 @@ def from_dict(cls, advisory_data):
         return cls(**transformed)
 
 
-@dataclasses.dataclass(order=True)
-class AdvisoryDataV2:
-    """
-    This data class expresses the contract between data sources and the import runner.
-
-    If a vulnerability_id is present then:
-        summary or affected_packages or references must be present
-    otherwise
-        either affected_package or references should be present
-
-    date_published must be aware datetime
-    """
-
-    advisory_id: str = ""
-    aliases: List[str] = dataclasses.field(default_factory=list)
-    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
-
-    def __post_init__(self):
-        if self.date_published and not self.date_published.tzinfo:
-            logger.warning(f"AdvisoryData with no tzinfo: {self!r}")
-        if self.summary:
-            self.summary = self.clean_summary(self.summary)
-
-    def clean_summary(self, summary):
-        # https://nvd.nist.gov/vuln/detail/CVE-2013-4314
-        # https://github.com/cms-dev/cms/issues/888#issuecomment-516977572
-        summary = summary.strip()
-        if summary:
-            summary = summary.replace("\x00", "\uFFFD")
-        return summary
-
-    def to_dict(self):
-        return {
-            "aliases": self.aliases,
-            "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 "",
-        }
-
-    @classmethod
-    def from_dict(cls, advisory_data):
-        date_published = advisory_data["date_published"]
-        transformed = {
-            "aliases": advisory_data["aliases"],
-            "summary": advisory_data["summary"],
-            "affected_packages": [
-                AffectedPackage.from_dict(pkg)
-                for pkg in advisory_data["affected_packages"]
-                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,
-            "weaknesses": advisory_data["weaknesses"],
-            "url": advisory_data.get("url") or None,
-        }
-        return cls(**transformed)
-
-
 class NoLicenseError(Exception):
     pass
 

vulnerabilities/models.py

@@ -2773,7 +2773,13 @@ class Patch(models.Model):
         help_text="SHA512 checksum of the patch content.",
     )
 
+    def clean(self):
+        if not self.patch_url and not self.patch_text:
+            raise ValidationError("Either patch_url or patch_text must be provided.")
+
     def save(self, *args, **kwargs):
+        # https://docs.djangoproject.com/en/4.2/ref/models/instances/#django.db.models.Model.clean
+        self.full_clean()
         if self.patch_text:
             self.patch_checksum = compute_patch_checksum(self.patch_text)
         super().save(*args, **kwargs)

vulnerabilities/pipelines/v2_importers/aosp_importer.py

@@ -90,23 +90,23 @@ def collect_advisories(self):
                     patch_url = commit_data.get("patchUrl")
                     commit_id = commit_data.get("commitId")
 
-                    base_purl, patch_obj = classify_patch_source(
+                    base_purl, patch_objs = classify_patch_source(
                         url=patch_url,
                         commit_hash=commit_id,
                         patch_text=None,
                     )
-
-                    if isinstance(patch_obj, PackageCommitPatchData):
-                        fixed_commit = patch_obj
-                        affected_package = AffectedPackageV2(
-                            package=base_purl,
-                            fixed_by_commit_patches=[fixed_commit],
-                        )
-                        affected_packages.append(affected_package)
-                    elif isinstance(patch_obj, PatchData):
-                        patches.append(patch_obj)
-                    elif isinstance(patch_obj, ReferenceV2):
-                        references.append(patch_obj)
+                    for patch_obj in patch_objs:
+                        if isinstance(patch_obj, PackageCommitPatchData):
+                            fixed_commit = patch_obj
+                            affected_package = AffectedPackageV2(
+                                package=base_purl,
+                                fixed_by_commit_patches=[fixed_commit],
+                            )
+                            affected_packages.append(affected_package)
+                        elif isinstance(patch_obj, PatchData):
+                            patches.append(patch_obj)
+                        elif isinstance(patch_obj, ReferenceV2):
+                            references.append(patch_obj)
 
                 url = (
                     "https://raw.githubusercontent.com/quarkslab/aosp_dataset/refs/heads/master/cves/"

vulnerabilities/pipes/advisory.py

@@ -16,6 +16,7 @@
 from typing import List
 from typing import Union
 
+from django.core.exceptions import ValidationError
 from django.db import transaction
 from django.db.models import Q
 from django.db.models.query import QuerySet
@@ -128,13 +129,19 @@ def get_or_create_advisory_package_commit_patches(
         key = (commit_obj.commit_hash, commit_obj.vcs_url)
         input_data = data_map[key]
 
-        if (
-            commit_obj.patch_checksum != input_data.patch_checksum
-            or commit_obj.patch_text != input_data.patch_text
-        ):
+        if not commit_obj.patch_text and input_data.patch_text:
             commit_obj.patch_checksum = input_data.patch_checksum
             commit_obj.patch_text = input_data.patch_text
             to_update.append(commit_obj)
+        elif (
+            commit_obj.patch_text
+            and input_data.patch_text
+            and (commit_obj.patch_text != input_data.patch_text)
+        ):
+            raise ValidationError(
+                f"Patch text conflict detected: existing record: {commit_obj.vcs_url} - {commit_obj.commit_hash} has different patch text"
+                f"than {input_data.vcs_url} - {input_data.commit_hash}"
+            )
 
     if to_update:
         PackageCommitPatch.objects.bulk_update(to_update, fields=["patch_checksum", "patch_text"])
@@ -204,25 +211,39 @@ def classify_patch_source(url, commit_hash, patch_text):
         if not patch_text:
             return
 
-        return None, PatchData(patch_text=patch_text)
+        return None, [PatchData(patch_text=patch_text)]
 
     purl = url2purl(url)
     if not purl or (purl.type not in VCS_URLS_SUPPORTED_TYPES):
         if commit_hash:
-            return None, ReferenceV2(
-                reference_id=commit_hash, reference_type=AdvisoryReference.COMMIT, url=url
-            )
-        return None, PatchData(patch_url=url, patch_text=patch_text)
+            if not patch_text:
+                return None, [
+                    ReferenceV2(
+                        reference_id=commit_hash, reference_type=AdvisoryReference.COMMIT, url=url
+                    )
+                ]
+
+            return None, [
+                ReferenceV2(
+                    reference_id=commit_hash, reference_type=AdvisoryReference.COMMIT, url=url
+                ),
+                PatchData(patch_url=url, patch_text=patch_text),
+            ]
+
+        return None, [PatchData(patch_url=url, patch_text=patch_text)]
 
     if not commit_hash and not purl.version:
-        return None, PatchData(patch_url=url, patch_text=patch_text or None)
+        return None, [PatchData(patch_url=url, patch_text=patch_text)]
 
     base_purl = get_core_purl(purl)
     base_purl_str = base_purl.to_string()
     base_url = purl2url(base_purl_str)
-    return base_purl, PackageCommitPatchData(
-        vcs_url=base_url, commit_hash=purl.version or commit_hash, patch_text=patch_text or None
-    )
+    package_commit_hash = purl.version or commit_hash
+    return base_purl, [
+        PackageCommitPatchData(
+            vcs_url=base_url, commit_hash=package_commit_hash, patch_text=patch_text
+        )
+    ]
 
 
 def insert_advisory(advisory: AdvisoryData, pipeline_id: str, logger: Callable = None):

vulnerabilities/tests/pipes/test_vulnerablecode_importer_pipeline_v2.py

@@ -191,7 +191,7 @@ def patch_source_samples():
             "url": "https://unknown.com/abc/def",
             "commit_hash": "8eb1b04ca4ae6fc0a0ef46f1b0c042f64db28ff9",
             "patch_text": "+1-2",
-        },  # ReferenceV2
+        },  # ReferenceV2, PatchData
         {
             "url": "https://unknown.com/abc/def/be891173be2fbdc897116bf5aa4fc9fdc8dc4f3d",
             "commit_hash": None,
@@ -225,28 +225,28 @@ def dumpy_patch_advisory(patch_source_samples):
         commit_hash = entry["commit_hash"]
         patch_text = entry["patch_text"]
 
-        base_purl, patch_obj = classify_patch_source(
+        base_purl, patch_objs = classify_patch_source(
             url=url, commit_hash=commit_hash, patch_text=patch_text
         )
-
-        if isinstance(patch_obj, PackageCommitPatchData):
-            # For testing only: commit hashes starting with "a" are treated as introduced_by_commit_patches,
-            # all others are treated as fixed_by_commit_patches.
-            if patch_obj.commit_hash.startswith("a"):
-                affected_package = AffectedPackageV2(
-                    package=base_purl,
-                    introduced_by_commit_patches=[patch_obj],
-                )
-            else:
-                affected_package = AffectedPackageV2(
-                    package=base_purl,
-                    fixed_by_commit_patches=[patch_obj],
-                )
-            affected_packages.append(affected_package)
-        elif isinstance(patch_obj, PatchData):
-            patches.append(patch_obj)
-        elif isinstance(patch_obj, ReferenceV2):
-            references.append(patch_obj)
+        for patch_obj in patch_objs:
+            if isinstance(patch_obj, PackageCommitPatchData):
+                # For testing only: commit hashes starting with "a" are treated as introduced_by_commit_patches,
+                # all others are treated as fixed_by_commit_patches.
+                if patch_obj.commit_hash.startswith("a"):
+                    affected_package = AffectedPackageV2(
+                        package=base_purl,
+                        introduced_by_commit_patches=[patch_obj],
+                    )
+                else:
+                    affected_package = AffectedPackageV2(
+                        package=base_purl,
+                        fixed_by_commit_patches=[patch_obj],
+                    )
+                affected_packages.append(affected_package)
+            elif isinstance(patch_obj, PatchData):
+                patches.append(patch_obj)
+            elif isinstance(patch_obj, ReferenceV2):
+                references.append(patch_obj)
 
     return AdvisoryData(
         summary="Test patch advisory",

vulnerabilities/tests/test_data/ruby/parse-advisory-ruby-expected.json

@@ -56,6 +56,7 @@
       ]
     }
   ],
+  "patches": [],
   "date_published": "2018-01-09T00:00:00+00:00",
   "weaknesses": []
 }