Add support for introduced and fixed commits in AdvisoryData

aboutcode/federated/__init__.py

@@ -1687,7 +1687,7 @@ def package_path_elements(
  _pkg, _, core_path = core_purl.partition(":")
  purl_hash = _compute_hash(core_purl=core_purl, max_value=max_value)
 
- version = normalize_version(purl.version)
+ version = normalize_version(purl.version, purl.type)
  if version:
  version = percent_quote_more(version)
 

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

@@ -36,7 +36,9 @@
 from vulnerabilities.severity_systems import SCORING_SYSTEMS
 from vulnerabilities.severity_systems import ScoringSystem
 from vulnerabilities.utils import classproperty
+from vulnerabilities.utils import compute_patch_checksum
 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,103 @@ 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
+ patch_checksum: Optional[str] = dataclasses.field(init=False, default=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.")
+
+ if self.patch_text:
+ self.patch_checksum = compute_patch_checksum(self.patch_text)
+
+ 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,
+ self.patch_checksum,
+ )
+
+ 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,
+ "patch_checksum": self.patch_checksum,
+ }
+
+ @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 = compute_patch_checksum(self.patch_text)
+
+ 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 +443,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 +480,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 +495,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 +512,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 +537,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 +567,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 +600,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,
@@ -505,74 +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
  ],
- "references": [Reference.from_dict(ref) for ref in advisory_data["references"]],
- "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)
-
-
-@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)
- 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],
- "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
- ],
+ "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

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 []