minor changes

Author: sacca97

prospector/client/cli/prospector_client.py

@@ -33,6 +33,40 @@
 core_statistics = execution_statistics.sub_collection("core")
 
 
+def prospector_find_twins(
+    vulnerability_id: str,
+    repository_url: str,
+    commit_id: str,
+    git_cache: str = "/tmp/git_cache",
+):
+    advisory_record = build_advisory_record(
+        vulnerability_id,
+    )
+    repository = Git(repository_url, git_cache)
+    repository.clone()
+
+    # tags = repository.get_tags()
+
+    commits = repository.find_commits_for_twin_lookups(commit_id=commit_id)
+    preprocessed_commits = list()
+    pbar = tqdm(
+        list(commits.values()),
+        desc="Preprocessing commits",
+        unit="commit",
+    )
+    for raw_commit in pbar:
+        preprocessed_commits.append(make_from_raw_commit(raw_commit))
+
+    ranked_candidates = evaluate_commits(preprocessed_commits, advisory_record, None)
+
+    ConsoleWriter.print("Commit ranking and aggregation...")
+    # I NEED TO GET THE FIRST REACHABLE TAG OR NO-TAG
+    ranked_candidates = tag_and_aggregate_commits(ranked_candidates, None)
+    ConsoleWriter.print_(MessageStatus.OK)
+
+    return ranked_candidates, advisory_record
+
+
 # @profile
 @measure_execution_time(execution_statistics, name="core")
 def prospector(  # noqa: C901
@@ -73,6 +107,9 @@ def prospector(  # noqa: C901
             set(modified_files),
         )
 
+    fixing_commit = advisory_record.get_fixing_commit()
+    if fixing_commit is not None:
+        pass
     # obtain a repository object
     repository = Git(repository_url, git_cache)
 
@@ -186,7 +223,7 @@ def evaluate_commits(commits: List[Commit], advisory: AdvisoryRecord, rules: Lis
 
 
 def tag_and_aggregate_commits(commits: List[Commit], next_tag: str) -> List[Commit]:
-    if next_tag is None:
+    if next_tag is None or next_tag == "":
         return commits
 
     twin_tags_map = {commit.commit_id: commit.get_tag() for commit in commits}

prospector/datamodel/advisory.py

@@ -139,6 +139,12 @@ def parse_advisory(self, data):
         ]
         self.versions["fixed"] = [v for v in self.versions["fixed"] if v is not None]
 
+    def get_fixing_commit(self) -> Optional[str]:
+        for reference in self.references.keys():
+            if "github.com" in reference and "commit" in reference:
+                return reference.split("/")[-1]
+        return None
+
 
 def get_from_nvd(cve_id: str):
     """Get an advisory from the NVD dtabase"""

prospector/git/git.py

@@ -21,6 +21,7 @@
 GIT_SEPARATOR = "-@-@-@-@-"
 
 TEN_DAYS_TIME_DELTA = 14 * 24 * 60 * 60
+ONE_MONTH_TIME_DELTA = 30 * 24 * 60 * 60
 
 
 def do_clone(url, output_folder, shallow=False, skip_existing=False):
@@ -207,12 +208,14 @@ def create_commits(
 
         if next_tag:
             until = self.extract_tag_timestamp(next_tag) + TEN_DAYS_TIME_DELTA
-            cmd += f" --until={until}"
+            if until:
+                cmd += f" --until={until}"
 
         # TODO: if find twins is true, we dont need the ancestors, only the timestamps
         if prev_tag:
             since = self.extract_tag_timestamp(prev_tag) - TEN_DAYS_TIME_DELTA
-            cmd += f" --since={since}"
+            if since:
+                cmd += f" --since={since}"
 
         if filter_extension:
             cmd += " *." + " *.".join(filter_extension)
@@ -255,6 +258,13 @@ def parse_git_output(self, raw: List[str]) -> Dict[str, RawCommit]:
 
         return commits
 
+    def find_commits_for_twin_lookups(self, commit_id):
+        commit_timestamp = self.extract_tag_timestamp(commit_id)
+        return self.create_commits(
+            since=commit_timestamp - ONE_MONTH_TIME_DELTA,
+            until=commit_timestamp + ONE_MONTH_TIME_DELTA,
+        )
+
     @measure_execution_time(execution_statistics.sub_collection("core"))
     def get_commit(self, id):
         return RawCommit(self, id)

prospector/git/version_to_tag.py

@@ -52,7 +52,8 @@ def get_possible_tags(tags: list[str], versions: str):
             for tag in tags
             if next_version == clean_tag(tag) and not is_rc_or_date(tag)
         ]
-
+    # print(tags)
+    # print(prev_tag, next_tag)
     if len(prev_tag) == 1 and len(next_tag) == 1:
         return prev_tag[0], next_tag[0]
     elif len(prev_tag) == 1 and len(next_tag) > 1: