Remove get_code_commit function

Add all the fields in keys for comparison CodeCommitData

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

Author: ziadhany

vulnerabilities/importer.py

@@ -221,7 +221,13 @@ def __lt__(self, other):
 
     # TODO: Add cache
     def _cmp_key(self):
-        return (self.commit_hash, self.vcs_url, self.commit_author, self.commit_message)
+        return (
+            self.commit_hash,
+            self.vcs_url,
+            self.commit_author,
+            self.commit_message,
+            self.commit_date,
+        )
 
     def to_dict(self) -> dict:
         """Return a normalized dictionary representation of the commit."""

vulnerabilities/importers/osv.py

@@ -12,6 +12,7 @@
 from typing import Iterable
 from typing import List
 from typing import Optional
+from typing import Tuple
 
 import dateparser
 from cvss.exceptions import CVSS3MalformedError
@@ -84,8 +85,8 @@ def parse_advisory_data(
         )
 
         for fixed_range in affected_pkg.get("ranges") or []:
-            fixed_version = get_fixed_versions(
-                fixed_range=fixed_range, raw_id=raw_id, supported_ecosystem=purl.type
+            fixed_version, _ = get_fixed_versions_and_commits(
+                ranges=fixed_range, raw_id=raw_id, supported_ecosystem=purl.type
             )
 
             for version in fixed_version:
@@ -150,12 +151,11 @@ def parse_advisory_data_v2(
         fixed_versions = []
         fixed_version_range = None
         for fixed_range in affected_pkg.get("ranges") or []:
-            fixed_version = get_fixed_versions(
-                fixed_range=fixed_range, raw_id=advisory_id, supported_ecosystem=purl.type
+            fixed_version, (introduced_commits, fixed_commits) = get_fixed_versions_and_commits(
+                ranges=fixed_range, raw_id=advisory_id, supported_ecosystem=purl.type
             )
             fixed_versions.extend([v.string for v in fixed_version])
 
-            introduced_commits, fixed_commits = get_code_commit(fixed_range, raw_id=advisory_id)
             fixed_by_commits.extend(fixed_commits)
             affected_by_commits.extend(introduced_commits)
 
@@ -195,32 +195,23 @@ def parse_advisory_data_v2(
     )
 
 
-def extract_fixed_versions(fixed_range) -> Iterable[str]:
+def extract_introduced_and_fixed(ranges) -> Tuple[List[str], List[str]]:
     """
-    Return a list of fixed version strings given a ``fixed_range`` mapping of
-    OSV data.
+    Return pairs of introduced and fixed versions or commit hashes given a ``ranges``
+    mapping of OSV data.
 
-    >>> list(extract_fixed_versions(
-    ... {"type": "SEMVER", "events": [{"introduced": "0"},{"fixed": "1.6.0"}]}))
-    ['1.6.0']
-
-    >>> list(extract_fixed_versions(
-    ... {"type": "ECOSYSTEM","events":[{"introduced": "0"},
-    ... {"fixed": "1.0.0"},{"fixed": "9.0.0"}]}))
-    ['1.0.0', '9.0.0']
-    """
-    for event in fixed_range.get("events") or []:
-        fixed = event.get("fixed")
-        if fixed:
-            yield fixed
+    Both introduced and fixed fields may represent semantic versions or commit hashes.
 
+    >>> list(extract_introduced_and_fixed(
+    ... {"type": "SEMVER", "events": [{"introduced": "0"}, {"fixed": "1.6.0"}]}))
+    [('0', None), (None, '1.6.0')]
 
-def extract_commits(introduced_range) -> Iterable[str]:
+    >>> list(extract_introduced_and_fixed(
+    ... {"type": "GIT", "events": [{"introduced": "abc123"},
+    ... {"fixed": "def456"}]}))
+    [('abc123', None), (None, 'def456')]
     """
-    Return a list of fixed version strings given a ``fixed_range`` mapping of
-    OSV data.
-    """
-    for event in introduced_range.get("events") or []:
+    for event in ranges.get("events") or []:
         introduced = event.get("introduced")
         fixed = event.get("fixed")
         yield introduced, fixed
@@ -369,91 +360,81 @@ def get_fixed_version_range(versions, ecosystem):
         logger.error(f"Failed to create VersionRange from: {versions}: error:{e!r}")
 
 
-def get_fixed_versions(fixed_range, raw_id, supported_ecosystem) -> List[Version]:
+def get_fixed_versions_and_commits(
+    ranges, raw_id, supported_ecosystem=None
+) -> Tuple[List[Version], Tuple]:
     """
-    Return a list of unique fixed univers Versions given a ``fixed_range``
-    univers VersionRange and a ``raw_id``.
+    Extract and return all unique fixed versions and related commit data
+    from a given OSV vulnerability range.
+
     For example::
-    >>> get_fixed_versions(fixed_range={}, raw_id="GHSA-j3f7-7rmc-6wqj", supported_ecosystem="pypi",)
-    []
-    >>> get_fixed_versions(
-    ...   fixed_range={"type": "ECOSYSTEM", "events": [{"fixed": "1.7.0"}], },
+    >>> get_fixed_versions_and_commits(ranges={}, raw_id="GHSA-j3f7-7rmc-6wqj", supported_ecosystem="pypi",)
+    ([], ([], []))
+    >>> get_fixed_versions_and_commits(
+    ...   ranges={"type": "ECOSYSTEM", "events": [{"fixed": "1.7.0"}], },
     ...   raw_id="GHSA-j3f7-7rmc-6wqj",
     ...   supported_ecosystem="pypi",
     ... )
-    [PypiVersion(string='1.7.0')]
+    ([PypiVersion(string='1.7.0')], ([], []))
     """
     fixed_versions = []
-    if "type" not in fixed_range:
-        logger.error(f"Invalid fixed_range type for: {fixed_range} for OSV id: {raw_id!r}")
-        return []
+    introduced_commits = []
+    fixed_commits = []
 
-    fixed_range_type = fixed_range["type"]
+    if "type" not in ranges:
+        logger.error(f"Invalid range type for: {ranges} for OSV id: {raw_id!r}")
+        return [], ([], [])
+
+    fixed_range_type = ranges["type"]
 
     version_range_class = RANGE_CLASS_BY_SCHEMES.get(supported_ecosystem)
     version_class = version_range_class.version_class if version_range_class else None
 
-    for version in extract_fixed_versions(fixed_range):
-        if fixed_range_type == "ECOSYSTEM":
+    for introduced, fixed in extract_introduced_and_fixed(ranges):
+        if fixed_range_type == "ECOSYSTEM" and fixed:
             try:
                 if not version_class:
                     raise InvalidVersion(
                         f"Unsupported version for ecosystem: {supported_ecosystem}"
                     )
-                fixed_versions.append(version_class(version))
+                fixed_versions.append(version_class(fixed))
             except InvalidVersion:
                 logger.error(
-                    f"Invalid version class: {version_class} - {version!r} for OSV id: {raw_id!r}"
+                    f"Invalid version class: {version_class} - {fixed!r} for OSV id: {raw_id!r}"
                 )
 
-        elif fixed_range_type == "SEMVER":
+        elif fixed_range_type == "SEMVER" and fixed:
             try:
-                fixed_versions.append(SemverVersion(version))
+                fixed_versions.append(SemverVersion(fixed))
             except InvalidVersion:
-                logger.error(f"Invalid SemverVersion: {version!r} for OSV id: {raw_id!r}")
-
-        if fixed_range_type == "GIT":
-            # We process this in the get_code_commit function.
-            continue
-        else:
-            logger.error(f"Unsupported fixed version type: {version!r} for OSV id: {raw_id!r}")
+                logger.error(f"Invalid SemverVersion: {fixed!r} for OSV id: {raw_id!r}")
 
-    return dedupe(fixed_versions)
+        elif fixed_range_type == "GIT" and (fixed or introduced):
+            repo = ranges.get("repo")
+            if not repo:
+                logger.error(f"Missing 'repo' field in ranges: {ranges} (OSV id: {raw_id!r})")
+                continue
 
+            # Git uses this magic hash for the empty tree
+            if introduced == "0":
+                introduced = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"
 
-def get_code_commit(ranges, raw_id):
-    """
-    Return two lists of unique code commits (introduced and fixed) extracted from a
-    given vulnerability `ranges` dictionary.
-    """
-    if ranges.get("type") != "GIT":
-        logger.debug(f"Skipping non-GIT range for OSV id: {raw_id!r}")
-        return [], []
-
-    repo = ranges.get("repo")
-    if not repo:
-        logger.error(f"Missing 'repo' field in range: {ranges} (OSV id: {raw_id!r})")
-        return [], []
-
-    repo = ranges.get("repo")
-    introduced_commits, fixed_commits = [], []
-    for introduced, fixed in extract_commits(ranges):
-        # Git uses this magic hash for the empty tree
-        if introduced == "0":
-            introduced = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"
-
-        try:
             if introduced:
-                introduced_commit = CodeCommitData(commit_hash=introduced, vcs_url=repo)
-                introduced_commits.append(introduced_commit)
-        except ValueError as e:
-            logger.error(f"Failed to extract introduced commits: {e!r}")
+                try:
+                    introduced_commit = CodeCommitData(commit_hash=introduced, vcs_url=repo)
+                    introduced_commits.append(introduced_commit)
+                except ValueError as e:
+                    logger.error(f"Failed to extract introduced commits: {e!r}")
 
-        try:
             if fixed:
-                fixed_commit = CodeCommitData(commit_hash=fixed, vcs_url=repo)
-                fixed_commits.append(fixed_commit)
-        except ValueError as e:
-            logger.error(f"Failed to extract fixed commits: {e!r}")
+                try:
+                    fixed_commit = CodeCommitData(commit_hash=fixed, vcs_url=repo)
+                    fixed_commits.append(fixed_commit)
+                except ValueError as e:
+                    logger.error(f"Failed to extract fixed commits: {e!r}")
+
+        else:
+            if fixed:
+                logger.error(f"Unsupported fixed version type: {ranges!r} for OSV id: {raw_id!r}")
 
-    return introduced_commits, fixed_commits
+    return dedupe(fixed_versions), (introduced_commits, fixed_commits)