Merge pull request #245 from GeekMasher/errors-as-code

feat: Update Octokit errors

Author: GeekMasher

examples/secrets.py

@@ -1,4 +1,5 @@
 """Secret Scanning API example."""
+
 import os
 from ghastoolkit import GitHub, SecretScanning
 
@@ -11,8 +12,6 @@
     print("Secret Scanning is disabled :(")
     exit(1)
 
-print(f"Push Protection :: {secret_scanning.isPushProtectionEnabled()}")
-
 try:
     alerts = secret_scanning.getAlerts("open")
 except:

src/ghastoolkit/octokit/advisories.py

@@ -1,6 +1,7 @@
 """GitHub Security Advisories API."""
 
 from typing import Dict, Optional
+from ghastoolkit.errors import GHASToolkitError, GHASToolkitTypeError
 from ghastoolkit.octokit.github import GitHub, Repository
 from ghastoolkit.octokit.octokit import RestRequest
 from ghastoolkit.supplychain.advisories import Advisories, Advisory, AdvisoryAffect
@@ -20,7 +21,10 @@ def __init__(self, repository: Optional[Repository] = None) -> None:
         self.rest = RestRequest(self.repository)
 
     def getAdvisories(self) -> Advisories:
-        """Get list of security advisories from a repository."""
+        """Get list of security advisories from a repository.
+
+        https://docs.github.com/en/rest/security-advisories/repository-advisories#list-repository-security-advisories
+        """
         results = self.rest.get(
             "/repos/{owner}/{repo}/security-advisories", authenticated=True
         )
@@ -29,24 +33,38 @@ def getAdvisories(self) -> Advisories:
             for advisory in results:
                 advisories.append(self.loadAdvisoryData(advisory))
             return advisories
-        raise Exception(f"Error getting advisories from repository")
+
+        raise GHASToolkitTypeError(
+            f"Error getting advisories from repository",
+            docs="https://docs.github.com/en/rest/security-advisories/repository-advisories#list-repository-security-advisories",
+        )
 
     def getAdvisory(self, ghsa_id: str) -> Advisory:
-        """Get advisory by ghsa id."""
+        """Get advisory by ghsa id.
+
+        https://docs.github.com/en/rest/security-advisories/repository-advisories#get-a-repository-security-advisory
+        """
         result = self.rest.get(
             "/repos/{owner}/{repo}/security-advisories/{ghsa_id}",
             {"ghsa_id": ghsa_id},
             authenticated=True,
         )
         if isinstance(result, dict):
             return self.loadAdvisoryData(result)
-        raise Exception(f"Error getting advisory by id")
+
+        raise GHASToolkitTypeError(
+            f"Error getting advisory by id",
+            docs="https://docs.github.com/en/rest/security-advisories/repository-advisories#get-a-repository-security-advisory",
+        )
 
     def createAdvisory(
         self, advisory: Advisory, repository: Optional[Repository] = None
     ):
-        """Create a GitHub Security Advisories for a repository."""
-        raise Exception("Unsupported feature")
+        """Create a GitHub Security Advisories for a repository.
+
+        https://docs.github.com/en/rest/security-advisories/repository-advisories#create-a-repository-security-advisory
+        """
+        raise GHASToolkitError("Unsupported feature")
 
     def createPrivateAdvisory(
         self, advisory: Advisory, repository: Optional[Repository] = None
@@ -57,8 +75,11 @@ def createPrivateAdvisory(
     def updateAdvisory(
         self, advisory: Advisory, repository: Optional[Repository] = None
     ):
-        """Update GitHub Security Advisory."""
-        raise Exception("Unsupported feature")
+        """Update GitHub Security Advisory.
+
+        https://docs.github.com/en/rest/security-advisories/repository-advisories#update-a-repository-security-advisory
+        """
+        raise GHASToolkitError("Unsupported feature")
 
     def loadAdvisoryData(self, data: Dict) -> Advisory:
         """Load Advisory from API data."""

src/ghastoolkit/octokit/clearlydefined.py

@@ -2,6 +2,7 @@
 from typing import Any, Optional
 from requests import Session
 
+from ghastoolkit.errors import GHASToolkitError
 from ghastoolkit.supplychain.dependencies import Dependency
 
 
@@ -41,7 +42,7 @@ def createCurationUrl(self, dependency: Dependency) -> Optional[str]:
 
     def getCurations(self, dependency: Dependency) -> dict[str, Any]:
         if not dependency.manager:
-            raise Exception(f"Dependency manager / type must be set")
+            raise GHASToolkitError(f"Dependency manager / type must be set")
 
         url = self.createCurationUrl(dependency)
         if not url:

src/ghastoolkit/octokit/dependabot.py

@@ -3,6 +3,7 @@
 import logging
 from typing import Optional
 
+from ghastoolkit.errors import GHASToolkitError, GHASToolkitTypeError
 from ghastoolkit.octokit.github import GitHub, Repository
 from ghastoolkit.octokit.octokit import GraphQLRequest, RestRequest
 from ghastoolkit.supplychain.advisories import Advisory
@@ -39,10 +40,17 @@ def isEnabled(self) -> bool:
         return False
 
     def isSecurityUpdatesEnabled(self) -> bool:
-        """Is Security Updates for Dependabot enabled."""
+        """Is Security Updates for Dependabot enabled.
+
+        https://docs.github.com/en/rest/reference/repos#get-a-repository
+        """
         result = self.rest.get("get/repos/{owner}/{repo}")
         if not isinstance(result, dict):
-            raise Exception(f"Unable to get repository info")
+            raise GHASToolkitTypeError(
+                "Unable to get repository info",
+                permissions=["Repository Administration (read)"],
+                docs="https://docs.github.com/en/rest/reference/repos#get-a-repository",
+            )
         saa = result.get("source", {}).get("security_and_analysis", {})
         status = saa.get("dependabot_security_updates", {}).get("status", "disabled")
         return status == "enabled"
@@ -58,10 +66,13 @@ def getAlerts(
     ) -> list[DependencyAlert]:
         """Get All Dependabot alerts from REST API.
 
-        https://docs.github.com/en/enterprise-cloud@latest/rest/dependabot/alerts?apiVersion=2022-11-28
+        https://docs.github.com/en/rest/dependabot/alerts
         """
         if state not in ["auto_dismissed", "dismissed", "fixed", "open"]:
-            raise Exception(f"Invalid state provided: {state}")
+            raise GHASToolkitError(
+                f"Invalid state provided: {state}",
+                docs="https://docs.github.com/en/rest/reference/repos#get-a-repository",
+            )
 
         logger.debug(f"Getting Dependabot alerts with state: {state}")
 
@@ -100,7 +111,10 @@ def getAlerts(
                 )
 
             return retval
-        raise Exception(f"Error getting Dependabot alerts")
+        raise GHASToolkitTypeError(
+            f"Error getting Dependabot alerts",
+            docs="https://docs.github.com/en/rest/dependabot/alerts",
+        )
 
     def getAlertsGraphQL(self) -> list[DependencyAlert]:
         """Get All Dependabot alerts from GraphQL API using the `GetDependencyAlerts` query."""
@@ -117,7 +131,8 @@ def getAlertsGraphQL(self) -> list[DependencyAlert]:
                 logger.error(
                     "This could be due to a lack of permissions or access token"
                 )
-                raise Exception(f"Failed to get GraphQL repository alerts")
+                raise GHASToolkitError(f"Failed to get GraphQL repository alerts")
+
             alerts = repo.get("vulnerabilityAlerts", {})
 
             for alert in alerts.get("edges", []):

src/ghastoolkit/octokit/dependencygraph.py

@@ -6,6 +6,7 @@
 
 from semantic_version import Version
 
+from ghastoolkit.errors import GHASToolkitError, GHASToolkitTypeError
 from ghastoolkit.octokit.github import GitHub, Repository
 from ghastoolkit.supplychain.advisories import Advisory
 from ghastoolkit.supplychain.dependencyalert import DependencyAlert
@@ -70,7 +71,7 @@ def getDependencies(self) -> Dependencies:
                 logger.warning("Using GraphQL API to resolve dependencies (GHES 3.6+)")
                 deps = self.getDependenciesGraphQL()
             else:
-                raise Exception("Enterprise Server version must be >= 3.6.0")
+                raise GHASToolkitError("Enterprise Server version must be >= 3.6.0")
         else:
             # cloud: download SBOM
             deps = self.getDependenciesSbom()
@@ -179,7 +180,7 @@ def getDependenciesInPR(self, base: str, head: str) -> Dependencies:
         """Get all the dependencies from a Pull Request."""
 
         if GitHub.isEnterpriseServer() and GitHub.server_version < Version("3.6.0"):
-            raise Exception("Enterprise Server version must be >= 3.6")
+            raise GHASToolkitError("Enterprise Server version must be >= 3.6")
 
         dependencies = Dependencies()
         base = urllib.parse.quote(base, safe="")
@@ -234,8 +235,18 @@ def getDependenciesInPR(self, base: str, head: str) -> Dependencies:
         return dependencies
 
     def exportBOM(self) -> Dependencies:
-        """Download / Export DependencyGraph SBOM."""
-        return self.rest.get("/repos/{owner}/{repo}/dependency-graph/sbom")
+        """Download / Export DependencyGraph SBOM.
+
+        https://docs.github.com/en/rest/dependency-graph/sboms#export-a-software-bill-of-materials-sbom-for-a-repository
+        """
+        result = self.rest.get("/repos/{owner}/{repo}/dependency-graph/sbom")
+        if result:
+            return result
+
+        raise GHASToolkitTypeError(
+            "Failed to download SBOM",
+            docs="https://docs.github.com/en/rest/dependency-graph/sboms#export-a-software-bill-of-materials-sbom-for-a-repository",
+        )
 
     def submitDependencies(
         self,

src/ghastoolkit/octokit/enterprise.py

@@ -3,6 +3,7 @@
 
 from semantic_version import Version
 
+from ghastoolkit.errors import GHASToolkitError
 from ghastoolkit.octokit.github import GitHub
 from ghastoolkit.octokit.octokit import Octokit, RestRequest
 from ghastoolkit.octokit.repository import Repository
@@ -23,12 +24,19 @@ def __init__(
         self.rest = RestRequest(GitHub.repository)
 
     def getRepositories(self) -> List[Repository]:
-        """Get Repositories."""
+        """Get Repositories.
+
+        https://docs.github.com/en/rest/repos/repos#list-organization-repositories
+        """
         repositories = []
         result = self.rest.get(f"/orgs/{self.name}/repos")
         if not isinstance(result, list):
             logger.error("Error getting repositories")
-            return []
+            raise GHASToolkitError(
+                "Error getting repositories",
+                permissions=["Metadata repository permissions (read)"],
+                docs="https://docs.github.com/en/rest/repos/repos#list-organization-repositories",
+            )
 
         for repository in result:
             repositories.append(Repository.parseRepository(repository.get("full_name")))
@@ -81,6 +89,10 @@ def enableDefaultSetup(self) -> bool:
                 logger.error(
                     "Enterprise Server 3.8 or lower does not support default setup"
                 )
+                raise GHASToolkitError(
+                    "Enterprise Server 3.8 or lower does not support default setup"
+                )
+
             elif GitHub.server_version and GitHub.server_version < Version("3.11.0"):
                 from ghastoolkit.octokit.codescanning import CodeScanning
 
@@ -136,13 +148,21 @@ def getOrganizations(self, include_github: bool = False) -> List[Organization]:
 
             if response.status_code != 200:
                 logger.error("Error getting organizations")
-                return []
+                raise GHASToolkitError(
+                    "Error getting organizations",
+                    permissions=["Metadata repository permissions (read)"],
+                    docs="https://docs.github.com/en/rest/orgs/orgs#list-organizations",
+                )
 
             result = response.json()
 
             if not isinstance(result, list):
                 logger.error("Error getting organizations")
-                return []
+                raise GHASToolkitError(
+                    "Error getting organizations",
+                    permissions=["Metadata repository permissions (read)"],
+                    docs="https://docs.github.com/en/rest/orgs/orgs#list-organizations",
+                )
 
             for org in result:
                 if not include_github and org.get("login") in github_orgs:

src/ghastoolkit/octokit/octokit.py

@@ -288,7 +288,7 @@ def patchJson(
     ) -> dict:
         repo = self.repository or GitHub.repository
         if not repo:
-            raise Exception("Repository needs to be set")
+            raise GHASToolkitError("Repository needs to be set")
 
         url = Octokit.route(path, repo, rtype="rest", **parameters)
         logger.debug(f"Patching content from URL :: {url}")
@@ -303,8 +303,8 @@ def patchJson(
                 logger.error(f"{response.content}")
                 known_error = __OCTOKIT_ERRORS__.get(response.status_code)
                 if known_error:
-                    raise Exception(known_error)
-                raise Exception("Failed to patch data")
+                    raise known_error
+                raise GHASToolkitError("Failed to patch data")
 
         return response.json()
 
@@ -333,7 +333,10 @@ def query(self, name: str, options: dict[str, Any] = {}) -> dict:
         query_content = self.queries.get(name)
 
         if not query_content:
-            raise Exception(f"Failed to load GraphQL query :: {name}")
+            raise GHASToolkitError(
+                f"Failed to load GraphQL query :: {name}",
+                docs="https://docs.github.com/en/enterprise-cloud@latest/graphql/overview/about-the-graphql-api",
+            )
 
         cursor = f'after: "{self.cursor}"' if self.cursor != "" else ""
 
@@ -345,7 +348,10 @@ def query(self, name: str, options: dict[str, Any] = {}) -> dict:
         if response.status_code != 200:
             logger.error(f"GraphQL API Status :: {response.status_code}")
             logger.error(f"GraphQL Content :: {response.content}")
-            raise Exception(f"Failed to get data from GraphQL API")
+            raise GHASToolkitError(
+                f"Failed to get data from GraphQL API",
+                docs="https://docs.github.com/en/enterprise-cloud@latest/graphql/overview/about-the-graphql-api",
+            )
 
         rjson = response.json()
         if rjson.get("errors"):

src/ghastoolkit/octokit/repository.py

@@ -8,6 +8,8 @@
 from typing import Optional, Union
 from urllib.parse import urlparse
 
+from ghastoolkit.errors import GHASToolkitError
+
 
 logger = logging.getLogger("ghastoolkit.octokit.repository")
 
@@ -29,6 +31,9 @@ class Repository:
     path: Optional[str] = None
     """Path inside the repository"""
 
+    pr_number: Optional[int] = None
+    """Pull Request Number (if in a PR)"""
+
     __prinfo__: Optional[dict] = None
 
     sha: Optional[str] = None
@@ -89,16 +94,17 @@ def getPullRequestInfo(self) -> dict:
         if not self.__prinfo__:
             from ghastoolkit.octokit.octokit import RestRequest
 
-            pull_number = self.getPullRequestNumber()
+            self.pr_number = self.getPullRequestNumber()
             self.__prinfo__ = RestRequest().get(
                 "/repos/{owner}/{repo}/pulls/{pull_number}",
-                {"pull_number": pull_number},
+                {"pull_number": self.pr_number},
             )
         return self.__prinfo__
 
     def getPullRequestCommits(self) -> list[str]:
         """Get list of Pull Request commits."""
         result = []
+
         if self.isInPullRequest():
             from ghastoolkit.octokit.octokit import RestRequest
 
@@ -224,7 +230,7 @@ def gitsha(self) -> str:
     def getFile(self, path: str) -> str:
         """Get a path relative from the base of the cloned repository."""
         if not self.clone_path:
-            raise Exception(f"Unknown clone path")
+            raise GHASToolkitError(f"Unknown clone path")
         return os.path.join(self.clone_path, path)
 
     def display(self) -> str: