Merge pull request #261 from GeekMasher/codescanning-analyses

Code Scanning Analyses Updates

Author: GeekMasher

examples/codescanning.py

@@ -18,12 +18,14 @@
     print("Code Scanning is not enabled :(")
     exit()
 
-# Get list of the delta alerts in a PR
 if GitHub.repository.isInPullRequest():
-    alerts = cs.getAlertsInPR("refs/heads/main")
+    # Get list of the delta alerts in a PR
+    print(f"Alerts from PR :: {GitHub.repository.getPullRequestNumber()}")
+    alerts = cs.getAlertsInPR(GitHub.repository.reference or "")
 
-# Get all alerts
 else:
+    # Get all alerts
+    print("Alerts from default Branch")
     alerts = cs.getAlerts("open")
 
 print(f"Alert Count :: {len(alerts)}")

src/ghastoolkit/octokit/codescanning.py

@@ -4,7 +4,7 @@
 import json
 import time
 import logging
-from typing import Any, List, Optional
+from typing import Any, List, Optional, Union
 from ghastoolkit.errors import GHASToolkitError, GHASToolkitTypeError
 from ghastoolkit.octokit.github import GitHub, Repository
 from ghastoolkit.octokit.octokit import OctoItem, RestRequest, loadOctoItem
@@ -69,14 +69,149 @@ def __str__(self) -> str:
         return f"CodeAlert({self.number}, '{self.state}', '{self.tool_name}', '{self.rule_id}')"
 
 
+@dataclass
+class CodeScanningTool(OctoItem):
+    """Code Scanning Tool.
+
+    https://docs.github.com/rest/code-scanning/code-scanning#list-code-scanning-analyses-for-a-repository
+    """
+
+    name: str
+    """Tool Name"""
+    guid: Optional[str] = None
+    """Tool GUID"""
+    version: Optional[str] = None
+    """Tool Version"""
+
+    def __str__(self) -> str:
+        """To String."""
+        if self.version:
+            return f"CodeScanningTool({self.name}, '{self.version}')"
+        else:
+            return f"CodeScanningTool({self.name})"
+
+    def __repr__(self) -> str:
+        return self.__str__()
+
+
+@dataclass
+class CodeScanningAnalysisEnvironment(OctoItem):
+    """Code Scanning Analysis Environment.
+
+    https://docs.github.com/rest/code-scanning/code-scanning
+    """
+
+    language: Optional[str] = None
+    """Language"""
+
+    build_mode: Optional[str] = None
+    """CodeQL Build Mode"""
+
+    def __str__(self) -> str:
+        """To String."""
+        if self.language:
+            return f"CodeScanningAnalysisEnvironment({self.language})"
+        return "CodeScanningAnalysisEnvironment()"
+
+
+@dataclass
+class CodeScanningAnalysis(OctoItem):
+    """Code Scanning Analysis.
+
+    https://docs.github.com/rest/code-scanning/code-scanning#list-code-scanning-analyses-for-a-repository
+    """
+
+    id: int
+    """Unique Identifier"""
+    ref: str
+    """Reference (branch, tag, etc)"""
+    commit_sha: str
+    """Commit SHA"""
+    analysis_key: str
+    """Analysis Key"""
+    environment: CodeScanningAnalysisEnvironment
+    """Environment"""
+    error: str
+    """Error"""
+    created_at: str
+    """Created At"""
+    results_count: int
+    """Results Count"""
+    rules_count: int
+    """Rules Count"""
+    url: str
+    """URL"""
+    sarif_id: str
+    """SARIF ID"""
+    tool: CodeScanningTool
+    """Tool Information"""
+    deletable: bool
+    """Deletable"""
+    warning: str
+    """Warning"""
+
+    category: Optional[str] = None
+    """Category"""
+
+    def __post_init__(self) -> None:
+        if isinstance(self.environment, str):
+            # Load the environment as JSON
+            self.environment = loadOctoItem(
+                CodeScanningAnalysisEnvironment, json.loads(self.environment)
+            )
+        if isinstance(self.tool, dict):
+            # Load the tool as JSON
+            self.tool = loadOctoItem(CodeScanningTool, self.tool)
+
+    @property
+    def language(self) -> Optional[str]:
+        """Language from the Environment."""
+        return self.environment.language
+
+    @property
+    def build_mode(self) -> Optional[str]:
+        """Build Mode from the Environment."""
+        return self.environment.build_mode
+
+    def __str__(self) -> str:
+        """To String."""
+        return f"CodeScanningAnalysis({self.id}, '{self.ref}', '{self.tool.name}')"
+
+
+@dataclass
+class CodeScanningConfiguration(OctoItem):
+    """Code Scanning Configuration for Default Setup.
+
+    https://docs.github.com/rest/code-scanning/code-scanning#get-a-code-scanning-default-setup-configuration--parameters
+    """
+
+    state: str
+    """State of the Configuration"""
+    query_suite: str
+    """Query Suite"""
+    languages: list[str]
+    """Languages"""
+    updated_at: str
+    """Updated At"""
+    schedule: str
+    """Scheduled (weekly)"""
+
+    def __str__(self) -> str:
+        """To String."""
+        return f"CodeScanningConfiguration('{self.state}', '{self.query_suite}', '{self.languages}')"
+
+    def __repr__(self) -> str:
+        return self.__str__()
+
+
 class CodeScanning:
     """Code Scanning."""
 
     def __init__(
         self,
         repository: Optional[Repository] = None,
         retry_count: int = 1,
-        retry_sleep: int = 15,
+        retry_sleep: Union[int, float] = 15,
     ) -> None:
         """Code Scanning REST API.
 
@@ -90,7 +225,7 @@ def __init__(
         self.repository = repository or GitHub.repository
         self.tools: List[str] = []
 
-        self.setup: Optional[dict] = None
+        self.setup: Optional[CodeScanningConfiguration] = None
 
         if not self.repository:
             raise GHASToolkitError("CodeScanning requires Repository to be set")
@@ -125,7 +260,7 @@ def isCodeQLDefaultSetup(self) -> bool:
         if not self.setup:
             self.setup = self.getDefaultConfiguration()
 
-        return self.setup.get("state", "not-configured") == "configured"
+        return self.setup.state == "configured"
 
     def enableDefaultSetup(
         self,
@@ -167,7 +302,7 @@ def getOrganizationAlerts(self, state: str = "open") -> list[CodeAlert]:
             docs="https://docs.github.com/en/rest/code-scanning#list-code-scanning-alerts-for-an-organization",
         )
 
-    def getDefaultConfiguration(self) -> dict:
+    def getDefaultConfiguration(self) -> CodeScanningConfiguration:
         """Get Default Code Scanning Configuration.
 
         Permissions:
@@ -177,7 +312,7 @@ def getDefaultConfiguration(self) -> dict:
         """
         result = self.rest.get("/repos/{owner}/{repo}/code-scanning/default-setup")
         if isinstance(result, dict):
-            self.setup = result
+            self.setup = loadOctoItem(CodeScanningConfiguration, result)
             return self.setup
 
         raise GHASToolkitTypeError(
@@ -239,11 +374,7 @@ def getAlertsInPR(self, base: str) -> list[CodeAlert]:
         # Try merge and then head
         analysis = self.getAnalyses(reference=self.repository.reference)
         if len(analysis) == 0:
-            analysis = self.getAnalyses(
-                reference=self.repository.reference.replace("/merge", "/head")
-            )
-            if len(analysis) == 0:
-                raise GHASToolkitError("No analyses found for the PR")
+            raise GHASToolkitError("No analyses found for the PR")
 
         # For CodeQL results using Default Setup
         reference = analysis[0].get("ref")
@@ -299,7 +430,7 @@ def getAlertInstances(
 
     def getAnalyses(
         self, reference: Optional[str] = None, tool: Optional[str] = None
-    ) -> list[dict]:
+    ) -> list[CodeScanningAnalysis]:
         """Get a list of all the analyses for a given repository.
 
         This function will retry X times with a Y second sleep between each retry to
@@ -315,13 +446,23 @@ def getAnalyses(
 
         https://docs.github.com/en/enterprise-cloud@latest/rest/code-scanning#list-code-scanning-analyses-for-a-repository
         """
+        ref = reference or self.repository.reference
+        logger.debug(f"Getting Analyses for {ref}")
+        if ref is None:
+            raise GHASToolkitError("Reference is required for getting analyses")
+
         counter = 0
+
+        logger.debug(
+            f"Fetching Analyses (retries {self.retry_count} every {self.retry_sleep}s)"
+        )
+
         while counter < self.retry_count:
             counter += 1
 
             results = self.rest.get(
                 "/repos/{org}/{repo}/code-scanning/analyses",
-                {"tool_name": tool, "ref": reference or self.repository.reference},
+                {"tool_name": tool, "ref": ref},
             )
             if not isinstance(results, list):
                 raise GHASToolkitTypeError(
@@ -332,13 +473,37 @@ def getAnalyses(
                     docs="https://docs.github.com/en/enterprise-cloud@latest/rest/code-scanning#list-code-scanning-analyses-for-a-repository",
                 )
 
-            if len(results) > 0 and self.retry_count > 1:
-                logger.info(
-                    f"No analyses found, retrying {counter}/{self.retry_count})"
+            # Try default setup `head` if no results (required for default setup)
+            if (
+                len(results) == 0
+                and self.repository.isInPullRequest()
+                and (ref.endswith("/merge") or ref.endswith("/head"))
+            ):
+                logger.debug("No analyses found for `merge`, trying `head`")
+                results = self.rest.get(
+                    "/repos/{org}/{repo}/code-scanning/analyses",
+                    {"tool_name": tool, "ref": ref.replace("/merge", "/head")},
                 )
-                time.sleep(self.retry_sleep)
+                if not isinstance(results, list):
+                    raise GHASToolkitTypeError(
+                        "Error getting analyses from Repository",
+                        permissions=[
+                            '"Code scanning alerts" repository permissions (read)'
+                        ],
+                        docs="https://docs.github.com/en/enterprise-cloud@latest/rest/code-scanning#list-code-scanning-analyses-for-a-repository",
+                    )
+
+            if len(results) < 0:
+                # If the retry count is less than 1, we don't retry
+                if self.retry_count < 1:
+                    logger.debug(
+                        f"No analyses found, retrying {counter}/{self.retry_count})"
+                    )
+                    time.sleep(self.retry_sleep)
             else:
-                return results
+                return [
+                    loadOctoItem(CodeScanningAnalysis, analysis) for analysis in results
+                ]
 
         # If we get here, we have retried the max number of times and still no results
         raise GHASToolkitError(
@@ -349,7 +514,7 @@ def getAnalyses(
 
     def getLatestAnalyses(
         self, reference: Optional[str] = None, tool: Optional[str] = None
-    ) -> list[dict]:
+    ) -> list[CodeScanningAnalysis]:
         """Get Latest Analyses for every tool.
 
         Permissions:
@@ -361,16 +526,31 @@ def getLatestAnalyses(
         results = []
 
         for analysis in self.getAnalyses(reference, tool):
-            name = analysis.get("tool", {}).get("name")
-            if name in tools:
+            if analysis.tool.name in tools:
                 continue
-            tools.add(name)
+            tools.add(analysis.tool.name)
             results.append(analysis)
 
         self.tools = list(tools)
 
         return results
 
+    def getFailedAnalyses(
+        self, reference: Optional[str] = None
+    ) -> list[CodeScanningAnalysis]:
+        """Get Failed Analyses for a given reference. This will return all analyses with errors or warnings.
+
+        Permissions:
+        - "Code scanning alerts" repository permissions (read)
+
+        https://docs.github.com/en/rest/code-scanning/code-scanning
+        """
+        return [
+            analysis
+            for analysis in self.getAnalyses(reference)
+            if analysis.error != "" or analysis.warning != ""
+        ]
+
     def getTools(self, reference: Optional[str] = None) -> List[str]:
         """Get list of tools from the latest analyses.
 

src/ghastoolkit/utils/cli.py

@@ -119,6 +119,7 @@ def parse_args(self) -> Namespace:
         # GitHub Init
         GitHub.init(
             repository=arguments.repository,
+            reference=arguments.ref,
             owner=arguments.owner,
             instance=arguments.instance,
             token=arguments.token,