Merge pull request #103 from advanced-security/geekmasher-workflow

feat: Fix errors issues and add better error handling

.github/workflows/main.yml

@@ -64,5 +64,4 @@ jobs:
     - name: Run action
       uses: ./
       with:
-        token: ${{ secrets.GITHUB_TOKEN }}
-
+        token: "${{ secrets.POLICY_AS_CODE_TESTING_TOKEN }}"

Pipfile

@@ -5,14 +5,15 @@ verify_ssl = true
 
 [packages]
 pyyaml = "*"
-ghastoolkit = "==0.11.9"
 semantic-version = "*"
+ghastoolkit = "==0.12.2"
 
 [dev-packages]
 sphinx = "*"
 myst-parser = "*"
 black = "*"
 typing_extensions = "*"
+tomli = "*"
 
 [scripts]
 main = "python -m ghascompliance"

ghascompliance/__main__.py

@@ -165,7 +165,8 @@
 
     Octokit.info("Finished loading policy")
 
-    if arguments.display and policy.policy:
+    if policy.policy:
+        Octokit.info("Policy as Code Summary:")
         Octokit.info("```")
         for plcy, data in policy.policy.items():
             if plcy == "name":
@@ -189,37 +190,41 @@
 
     errors = 0
 
-    try:
-        if not arguments.disable_code_scanning:
-            errors += checks.checkCodeScanning()
-
-        if not arguments.disable_dependabot:
-            errors += checks.checkDependabot()
-
-        # Dependency Graph
-        if not arguments.disable_dependencies:
-            errors += checks.checkDependencies()
+    checks = [
+        ("code_scanning", checks.checkCodeScanning),
+        ("dependabot", checks.checkDependabot),
+        ("dependencies", checks.checkDependencies),
+        ("dependency_licensing", checks.checkDependencyLicensing),
+        ("secret_scanning", checks.checkSecretScanning),
+    ]
+
+    for check in checks:
+        try:
+            if not getattr(arguments, f"disable_{check[0]}"):
+                errors += check[1]()
+
+        except Exception as err:
+            Octokit.error("Unknown Exception was hit, please repo this to " + __url__)
+            Octokit.error(str(err))
+
+            errors += 1  # add to error count
+
+            # Add to summary
+            Summary.addHeader(f"{Summary.__ICONS__['cross']} :: Error Encountered", 2)
+            Summary.addLine(
+                f"An unexpected exception was encountered while performing policy checks. Please report this to {__url__}"
+            )
+            Summary.addLine(Summary.formatItalics(str(err)))
 
-        # Dependency Graph Licensing
-        if not arguments.disable_dependency_licensing:
-            errors += checks.checkDependencyLicensing()
+            if arguments.debug:
+                raise err
 
-        if not arguments.disable_secret_scanning:
-            errors += checks.checkSecretScanning()
+    Octokit.endGroup()
 
-    except Exception as err:
-        Octokit.error("Unknown Exception was hit, please repo this to " + __url__)
-        Octokit.error(str(err))
-        Summary.addHeader(f"{Summary.__ICONS__['cross']} :: Error Encountered", 2)
-        Summary.addLine(
-            f"An unexpected exception was encountered while performing policy checks. Please report this to {__url__}"
-        )
-        Summary.addLine(Summary.formatItalics(str(err)))
+    Octokit.createGroup("Summary")
 
-        if arguments.debug:
-            raise err
-    finally:
-        # Summary and PR comment
+    # Summary and PR comment
+    if GitHub.repository.isInPullRequest():
         Summary.outputJobSummary()
         PullRequest.addPrComment(policy.name)
 

ghascompliance/checks.py

@@ -79,6 +79,10 @@ def checkCodeScanning(self):
 
         codescanning = CodeScanning()
 
+        if not self.policy.checkTechnologyActive("codescanning"):
+            Octokit.info("Code Scanning is not active in the policy")
+            return 0
+
         if GitHub.repository.isInPullRequest():
             Octokit.info("Code Scanning Alerts from Pull Request (alert diff)")
             pr_base = (
@@ -302,6 +306,13 @@ def checkDependencyLicensing(self):
 
         # Dependencies
         depgraph = DependencyGraph()
+
+        if not self.policy.checkTechnologyActive("licensing"):
+            Octokit.debug("Skipping as licensing policy not set")
+            return 0
+
+        # TODO: Check if enabled
+
         if GitHub.repository.isInPullRequest():
             Octokit.info("Dependencies from Pull Request")
             pr_info = GitHub.repository.getPullRequestInfo()
@@ -439,6 +450,13 @@ def checkDependencies(self):
 
         # Dependencies
         depgraph = DependencyGraph()
+
+        if not self.policy.checkTechnologyActive("dependencygraph"):
+            Octokit.info("Skipping as DependencyGraph policy not set")
+            return 0
+
+        # TODO: Check if DependencyGraph is enabled in GitHub
+
         if GitHub.repository.isInPullRequest():
             Octokit.info("Dependencies from Pull Request")
             pr_info = GitHub.repository.getPullRequestInfo()
@@ -499,6 +517,15 @@ def checkSecretScanning(self):
         secret_violations = []
 
         secretscanning = SecretScanning()
+
+        if not self.policy.checkTechnologyActive("secretscanning"):
+            Octokit.info("Skipping as secret scanning policy not set")
+            return 0
+
+        if not secretscanning.isEnabled():
+            secret_violations.append(["Secret Scanning not enabled", ""])
+            return 1
+
         if GitHub.repository.isInPullRequest():
             Octokit.info("Secret Scanning Alerts from Pull Request")
             alerts = secretscanning.getAlertsInPR()

ghascompliance/octokit/octokit.py

@@ -1,13 +1,10 @@
 import os
-import json
 import yaml
 import logging
-import requests
 
-from ghastoolkit.octokit.octokit import GitHub
+from ghastoolkit import GitHub
 
 from ghascompliance.__version__ import __name__
-from ghascompliance.consts import API_ERRORS
 
 
 class Octokit:

ghascompliance/policy.py

@@ -1,14 +1,13 @@
 import os
 import json
-from ghastoolkit.octokit.octokit import GitHub, Repository
+from ghastoolkit.octokit.github import Dict
+from ghastoolkit.octokit.octokit import Repository
 import yaml
 import shutil
 import fnmatch
 import datetime
 import tempfile
-import subprocess
 from typing import List, Optional
-from urllib.parse import urlparse
 from ghascompliance.consts import SEVERITIES, TECHNOLOGIES, LICENSES
 from ghascompliance.octokit import Octokit
 
@@ -122,9 +121,12 @@ def loadPolicy(self, policy: dict):
 
         self.policy = policy
 
-    def loadPolicySection(self, name: str, data: dict):
+    def loadPolicySection(self, name: str, data: Optional[Dict] = None):
         time_to_remediate_policy = False
 
+        if not data:
+            data = {"level": "disabled"}
+
         for section, section_data in data.items():
             # check if only certain sections are present
             if section not in Policy.__SECTION_ITEMS__:
@@ -261,6 +263,9 @@ def matchContent(self, name: str, validators: List[str]):
                 return True
         return False
 
+    def checkTechnologyActive(self, technology: str):
+        return self.policy.get(technology, {}).get("level", "") != "disabled"
+
     def checkViolationRemediation(
         self,
         severity: str,

vendor/bin/normalizer

@@ -1,4 +1,4 @@
-#!/home/codespace/.local/share/virtualenvs/policy-as-code-tZI3uR7_/bin/python
+#!/home/geekmasher/.local/share/virtualenvs/policy-as-code-ys9TzZIz/bin/python
 # -*- coding: utf-8 -*-
 import re
 import sys

vendor/ghastoolkit/__init__.py

@@ -3,7 +3,7 @@
 __name__ = "ghastoolkit"
 __title__ = "GHAS Toolkit"
 
-__version__ = "0.11.9"
+__version__ = "0.12.2"
 
 __description__ = "GitHub Advanced Security Python Toolkit"
 __summary__ = """\

vendor/ghastoolkit/octokit/octokit.py

@@ -2,7 +2,7 @@
 import inspect
 import logging
 from string import Template
-from typing import Any, Optional, Union
+from typing import Any, Callable, Optional, Union
 from dataclasses import field, is_dataclass
 
 from requests import Session
@@ -178,6 +178,7 @@ def get(
         expected: Optional[int] = 200,
         authenticated: bool = False,
         display_errors: bool = True,
+        error_handler: Optional[Callable[[int, dict], Any]] = None,
     ) -> Union[dict, list[dict]]:
         """Get Request.
 
@@ -208,21 +209,33 @@ def get(
             params["page"] = page
 
             response = self.session.get(url, params=params)
+            # Every response should be a JSON (including errors)
             response_json = response.json()
 
             if expected and response.status_code != expected:
                 if display_errors:
                     logger.error(f"Error code from server :: {response.status_code}")
-                    logger.error(f"Content :: {response_json}")
 
                 known_error = __OCTOKIT_ERRORS__.get(response.status_code)
                 if known_error:
                     raise Exception(known_error)
-                raise Exception("REST Request failed :: non-expected server error")
 
-            if isinstance(response_json, dict) and response_json.get("errors"):
-                logger.error(response_json.get("message"))
-                raise Exception("REST Request failed :: error from server")
+            # Handle errors in the response
+            if isinstance(response_json, dict) and response_json.get("message"):
+                # Custom error handler callback
+                if error_handler:
+                    return error_handler(response.status_code, response_json)
+
+                # Default error handling
+                message = response_json.get("message", "No message provided")
+                docs = response_json.get(
+                    "documentation_url", "No documentation link provided"
+                )
+
+                logger.error(f"Error message from server :: {message}")
+                logger.error(f"Documentation Link :: {docs}")
+
+                raise Exception(f"REST Request failed :: {message}")
 
             if isinstance(response_json, dict):
                 return response_json

vendor/update.sh

@@ -12,7 +12,7 @@ if [ -f $PWD/Pipfile ]; then
 
     # pipenv clean
     # pipenv install --deploy
-    pipenv run pip freeze > "$VENDOR/requirements.txt"
+    pipenv requirements > "$VENDOR/requirements.txt"
     pip install -r "$VENDOR/requirements.txt" --target=$VENDOR --upgrade
 
     echo "[+] Clean up vendor folder"