Implement uploading SARIF results for OpenScanHub

packit_service/worker/handlers/open_scan_hub.py

@@ -2,8 +2,9 @@
 # SPDX-License-Identifier: MIT
 
 import logging
-from typing import Union
+from typing import Optional, Union
 
+from ogr.services.github import GithubProject
 from packit.config import (
     JobType,
 )
@@ -47,6 +48,7 @@ def __init__(self, **kwargs):
         self.event: Union[OpenScanHubTaskFinishedEvent | OpenScanHubTaskStartedEvent] = (
             self.data.to_event()
         )
+        self._openscanhub_helper: Optional[OpenScanHubHelper] = None
 
     @staticmethod
     def get_checkers() -> tuple[type[Checker], ...]:
@@ -68,6 +70,12 @@ def get_helper(self) -> OpenScanHubHelper:
             build=self.event.build,
         )
 
+    @property
+    def openscanhub_helper(self):
+        if not self._openscanhub_helper:
+            self._openscanhub_helper = self.get_helper()
+        return self._openscanhub_helper
+
     def check_scan_and_build(self):
         task_id = self.data.event_dict["task_id"]
         if not self.event.scan or not self.event.build:
@@ -114,6 +122,12 @@ def run(self) -> TaskResults:
             self.event.scan.set_issues_added_url(self.event.issues_added_url)
             self.event.scan.set_issues_fixed_url(self.event.issues_fixed_url)
             self.event.scan.set_scan_results_url(self.event.scan_results_url)
+            if isinstance(self.project, GithubProject):
+                issues_sarif = self.openscanhub_helper.get_sarif_to_upload(
+                    self.event.issues_added_url
+                )
+                if issues_sarif:
+                    self.openscanhub_helper.upload_sarif(issues_sarif)
         else:
             state = BaseCommitStatus.neutral
             description = f"Scan in OpenScanHub is finished in a {self.event.status} state."
@@ -123,7 +137,7 @@ def run(self) -> TaskResults:
             else:
                 self.event.scan.set_status(OSHScanStatus.failed)
 
-        self.get_helper().report(
+        self.openscanhub_helper.report(
             state=state,
             description=description,
             url=self.event.scan.url,
@@ -154,7 +168,7 @@ def run(self) -> TaskResults:
         description = "Scan in OpenScanHub has started."
         self.event.scan.set_status(OSHScanStatus.running)
 
-        self.get_helper().report(
+        self.openscanhub_helper.report(
             state=state,
             description=description,
             url=self.event.scan.url,

packit_service/worker/helpers/open_scan_hub.py

@@ -1,6 +1,7 @@
 # Copyright Contributors to the Packit project.
 # SPDX-License-Identifier: MIT
-
+import base64
+import gzip
 import json
 import logging
 import re
@@ -10,11 +11,13 @@
 from pathlib import Path
 from typing import Optional
 
+from ogr.services.github import GithubProject
 from packit.config import (
     JobConfig,
     JobConfigTriggerType,
     JobType,
 )
+from packit.utils import run_command
 
 from packit_service.constants import (
     OPEN_SCAN_HUB_FEATURE_DESCRIPTION,
@@ -271,3 +274,68 @@ def download_srpm(srpm_model: SRPMBuildModel) -> Optional[Path]:
             return None
 
         return base_srpm_path, srpm_path
+
+    @staticmethod
+    def get_sarif_to_upload(url: str) -> Optional[str]:
+        """
+        Fetch the file content, convert to SARIF using csgrep,
+        compress and encode it so that it can be uploaded to GitHub.
+        """
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory).joinpath(basename(url))
+            if not download_file(url, path):
+                logger.info(f"Downloading of file {url} was not successful.")
+                return None
+
+            # run `csgrep` to convert to SARIF format
+            result = run_command(["csgrep", "--mode=sarif", str(path)], fail=False, output=True)
+
+        if not result.success:
+            logger.info(f"Conversion to SARIF was not successful: {result.stderr}")
+            return None
+
+        logger.info("Conversion to SARIF was successful, about to compress and encode.")
+
+        try:
+            # TODO replace csmock with OpenScanHub/ [Packit] OpenScanHub where needed
+            #  so that this name is displayed later in GitHub UI
+            sarif_data = result.stdout.encode("utf-8")
+            compressed_data = gzip.compress(sarif_data)
+            base64_encoded_data = base64.b64encode(compressed_data).decode("utf-8")
+
+            logger.info("SARIF file successfully compressed and encoded.")
+            return base64_encoded_data
+
+        except Exception as e:
+            logger.error(f"An error occurred during compression and encoding: {e}")
+            return None
+
+    def upload_sarif(self, data: str):
+        """
+        Upload the encoded SARIF to GitHub.
+        """
+        if self.copr_build_helper.job_build.merge_pr_in_ci:
+            # TODO this is not really correct and we need to discuss it
+            commit_sha = self.copr_build_helper.pull_request_object.merge_commit_sha
+            ref = f"refs/pull/{self.copr_build_helper.pr_id}/merge"
+        else:
+            commit_sha = self.copr_build_helper.db_project_event.commit_sha
+            ref = f"refs/pull/{self.copr_build_helper.pr_id}/head"
+
+        # there is no PyGithub support yet, API docs:
+        # https://docs.github.com/en/rest/code-scanning/code-scanning?
+        # apiVersion=2022-11-28#upload-an-analysis-as-sarif-data--parameters
+        payload = {
+            "commit_sha": commit_sha,
+            "ref": ref,
+            "sarif": data,
+        }
+        if not isinstance(self.copr_build_helper.project, GithubProject):
+            return
+
+        pygithub_repo = self.copr_build_helper.project.github_repo
+        pygithub_repo._requester.requestJsonAndCheck(
+            "POST",
+            f"{pygithub_repo.url}/code-scanning/sarifs",
+            input=payload,
+        )

tests/unit/test_open_scan_hub.py

@@ -269,6 +269,8 @@ def test_handle_scan_task_finished(
                     "0.7.5-1.20241007054606793155.pr405.23.g829aafd6/scan-results.js?format=raw"
                 ),
             }
+            flexmock(OpenScanHubHelper).should_receive("get_sarif_to_upload").and_return(flexmock())
+            flexmock(OpenScanHubHelper).should_receive("upload_sarif")
         elif scan_status == OpenScanHubTaskFinishedEvent.Status.cancel:
             state = BaseCommitStatus.neutral
             description = f"Scan in OpenScanHub is finished in a {scan_status} state."