Advisory References: commit

makes sure that only commits refrenced in the advisory with a commit hash are treated as 'fixing commits', and not if the references is eg. 'commit::master'

prospector/core/prospector.py

@@ -113,7 +113,9 @@ def prospector(  # noqa: C901
                     )
                     sys.exit(1)
 
-    fixing_commit = advisory_record.get_fixing_commit()
+    commits_in_advisory_references = (
+        advisory_record.get_commits_in_advisory_references()
+    )
     # print(advisory_record.references)
     # obtain a repository object
     repository = Git(repository_url, git_cache)
@@ -131,10 +133,12 @@ def prospector(  # noqa: C901
 
     candidates: Dict[str, RawCommit] = dict()
 
-    if len(fixing_commit) > 0:
-        candidates = get_commits_no_tags(repository, fixing_commit)
+    if len(commits_in_advisory_references) > 0:
+        candidates = get_commits_no_tags(
+            repository, commits_in_advisory_references
+        )
         if len(candidates) > 0 and any(
-            [c for c in candidates if c in fixing_commit]
+            [c for c in candidates if c in commits_in_advisory_references]
         ):
             console.print("Fixing commit found in the advisory references\n")
             advisory_record.has_fixing_commit = True

prospector/datamodel/advisory.py

@@ -131,7 +131,9 @@ def fetch_references(self):
 
     def parse_references_from_third_party(self):
         """Parse the references from third party sites"""
-        for ref in self.search_references_debian() + self.search_references_redhat():
+        for ref in (
+            self.search_references_debian() + self.search_references_redhat()
+        ):
             # self.references[ref] += 2
             self.references[self.extract_hashes(ref)] += 2
 
@@ -167,7 +169,9 @@ def parse_advisory(self, data):
         # )
         self.versions = {
             "affected": [
-                item.get("versionEndIncluding", item.get("versionStartIncluding"))
+                item.get(
+                    "versionEndIncluding", item.get("versionStartIncluding")
+                )
                 for item in data["configurations"][0]["nodes"][0]["cpeMatch"]
             ],  # TODO: can return to tuples
             "fixed": [
@@ -178,22 +182,40 @@ def parse_advisory(self, data):
         self.versions["affected"] = [
             v for v in self.versions["affected"] if v is not None
         ]
-        self.versions["fixed"] = [v for v in self.versions["fixed"] if v is not None]
+        self.versions["fixed"] = [
+            v for v in self.versions["fixed"] if v is not None
+        ]
+
+    def get_commits_in_advisory_references(self) -> List[str]:
+        """Processes the advisory's references to extract commit IDs if
+        present. Only keeps the five most important ones.
 
-    def get_fixing_commit(self) -> List[str]:
+        Returns:
+            A list of references to a commit.
+        """
         self.references = dict(
-            sorted(self.references.items(), key=lambda item: item[1], reverse=True)
+            sorted(
+                self.references.items(), key=lambda item: item[1], reverse=True
+            )
         )
         limit = 0
-        while len([r for r in self.references.keys() if r.startswith("commit::")]) > 5:
+        while (
+            len([r for r in self.references.keys() if r.startswith("commit::")])
+            > 5
+        ):
             self.references = {
                 k: v
                 for k, v in self.references.items()
                 if ("commit" in k and v > limit) or ("commit" not in k)
             }
             limit += 1
 
-        return [ref.split("::")[1] for ref in self.references if "commit::" in ref]
+        return [
+            ref.split("::")[1]
+            for ref in self.references
+            if "commit::" in ref
+            and ref.split("::")[1] not in ["master", "main"]
+        ]
 
     def search_references_debian(self) -> List[str]:
         url = "https://security-tracker.debian.org/tracker/"
@@ -221,7 +243,9 @@ def search_references_redhat(self) -> List[str]:
 
         return []
 
-    def extract_hashes(self, reference: str, filter: bool = False) -> str | None:
+    def extract_hashes(
+        self, reference: str, filter: bool = False
+    ) -> str | None:
         if bool(re.search(r"a=commit;", reference)):
             return "commit::" + re.search(r";h=(\w{6,40})", reference).group(1)
 
@@ -258,12 +282,15 @@ def parse_advisory_2(self, details, metadata):
         for field, key in timestamp_fields.items():
             timestamp = metadata.get(key)
             setattr(
-                self, field, int(isoparse(timestamp).timestamp()) if timestamp else None
+                self,
+                field,
+                int(isoparse(timestamp).timestamp()) if timestamp else None,
             )
         if not self.description:
             self.description = details["descriptions"][0]["value"]
         self.references = defaultdict(
-            int, {self.extract_hashes(r["url"]): 2 for r in details["references"]}
+            int,
+            {self.extract_hashes(r["url"]): 2 for r in details["references"]},
         )
 
 
@@ -290,7 +317,9 @@ def get_from_nvd(cve_id: str):
         headers = {"apiKey": NVD_API_KEY} if NVD_API_KEY else None
         params = {"cveId": cve_id}
 
-        response = requests.get(NVD_REST_ENDPOINT, headers=headers, params=params)
+        response = requests.get(
+            NVD_REST_ENDPOINT, headers=headers, params=params
+        )
 
         if response.status_code != 200:
             return None
@@ -314,7 +343,9 @@ def is_url_allowed(url: str) -> bool:
     return False
 
 
-def get_from_local(vuln_id: str, nvd_rest_endpoint: str = LOCAL_NVD_REST_ENDPOINT):
+def get_from_local(
+    vuln_id: str, nvd_rest_endpoint: str = LOCAL_NVD_REST_ENDPOINT
+):
     try:
         response = requests.get(nvd_rest_endpoint + vuln_id)
         if response.status_code != 200: