github scm passing tests; time for refactor

Author: flowstate

pyproject.toml

@@ -110,8 +110,14 @@ exclude = [
 # Enable Pyflakes (`F`) and a subset of the pycodestyle (`E`)  codes by default.
 # Unlike Flake8, Ruff doesn't enable pycodestyle warnings (`W`) or
 # McCabe complexity (`C901`) by default.
-select = ["E4", "E7", "E9", "F"]
-ignore = []
+select = [
+    "E4", "E7", "E9", "F",  # Current rules
+    "I",    # isort
+    "F401", # Unused imports
+    "F403", # Star imports
+    "F405", # Star imports undefined
+    "F821", # Undefined names
+]
 
 # Allow fix for all enabled rules (when `--fix`) is provided.
 fixable = ["ALL"]
@@ -120,6 +126,9 @@ unfixable = []
 # Allow unused variables when underscore-prefixed.
 dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
 
+[tool.ruff.lint.isort]
+known-first-party = ["socketsecurity"]
+
 [tool.ruff.format]
 # Like Black, use double quotes for strings.
 quote-style = "double"

socketsecurity/core/__init__.py

@@ -1,32 +1,30 @@
+import base64
+import json
 import logging
+import platform
+import time
+from glob import glob
 from pathlib import PurePath
-from .utils import socket_globs
-from .config import SocketConfig
-from .client import CliClient
-import requests
 from urllib.parse import urlencode
-import base64
-import json
+
+import requests
+
+from socketsecurity import __version__
+from socketsecurity.core.classes import Alert, Diff, FullScan, FullScanParams, Issue, Package, Purl, Report, Repository
 from socketsecurity.core.exceptions import (
-    APIFailure, APIKeyMissing, APIAccessDenied, APIInsufficientQuota, APIResourceNotFound, APICloudflareError
+    APIAccessDenied,
+    APICloudflareError,
+    APIFailure,
+    APIInsufficientQuota,
+    APIKeyMissing,
+    APIResourceNotFound,
 )
-from socketsecurity import __version__
-from socketsecurity.core.licenses import Licenses
 from socketsecurity.core.issues import AllIssues
-from socketsecurity.core.classes import (
-    Report,
-    Issue,
-    Package,
-    Alert,
-    FullScan,
-    FullScanParams,
-    Repository,
-    Diff,
-    Purl
-)
-import platform
-from glob import glob
-import time
+from socketsecurity.core.licenses import Licenses
+
+from .cli_client import CliClient
+from .config import SocketConfig
+from .utils import socket_globs
 
 __all__ = [
     "Core",
@@ -270,6 +268,11 @@ def create_sbom_output(self, diff: Diff) -> dict:
     # TODO: verify what this does. It looks like it should be named "all_files_unsupported"
     @staticmethod
     def match_supported_files(files: list) -> bool:
+        """
+        Checks if any of the files in the list match the supported file patterns
+        Returns True if NO files match (meaning no changes to manifest files)
+        Returns False if ANY files match (meaning there are manifest changes)
+        """
         matched_files = []
         not_matched = False
         for ecosystem in socket_globs:
@@ -724,3 +727,21 @@ def save_file(file_name: str, content: str) -> None:
     #     output = []
     #     for package_id in diff.packages:
     #         purl =  Core.create_purl(package_id, diff.packages)
+
+    @staticmethod
+    def has_manifest_files(files: list) -> bool:
+        """
+        Checks if any files in the list are supported manifest files.
+        Returns True if ANY files match our manifest patterns (meaning we need to scan)
+        Returns False if NO files match (meaning we can skip scanning)
+        """
+        for ecosystem in socket_globs:
+            patterns = socket_globs[ecosystem]
+            for file_name in patterns:
+                pattern = patterns[file_name]["pattern"]
+                for file in files:
+                    if "\\" in file:
+                        file = file.replace("\\", "/")
+                    if PurePath(file).match(pattern):
+                        return True  # Found a manifest file, no need to check further
+        return False  # No manifest files found

socketsecurity/core/client.py renamed to socketsecurity/core/cli_client.py

@@ -0,0 +1,37 @@
+from abc import ABC, abstractmethod
+from typing import Dict
+
+from ..classes import Comment
+from .client import ScmClient
+
+
+class SCM(ABC):
+    def __init__(self, client: ScmClient):
+        self.client = client
+
+    @abstractmethod
+    def check_event_type(self) -> str:
+        """Determine the type of event (push, pr, comment)"""
+        pass
+
+    @abstractmethod
+    def add_socket_comments(
+        self,
+        security_comment: str,
+        overview_comment: str,
+        comments: Dict[str, Comment],
+        new_security_comment: bool = True,
+        new_overview_comment: bool = True
+    ) -> None:
+        """Add or update comments on PR"""
+        pass
+
+    @abstractmethod
+    def get_comments_for_pr(self, repo: str, pr: str) -> Dict[str, Comment]:
+        """Get existing comments for PR"""
+        pass
+
+    @abstractmethod
+    def remove_comment_alerts(self, comments: Dict[str, Comment]) -> None:
+        """Process and remove alerts from comments"""
+        pass

socketsecurity/core/scm/__init__.py

@@ -0,0 +1,41 @@
+from abc import abstractmethod
+from typing import Dict
+
+from ..cli_client import CliClient
+
+
+class ScmClient(CliClient):
+    def __init__(self, token: str, api_url: str):
+        self.token = token
+        self.api_url = api_url
+
+    @abstractmethod
+    def get_headers(self) -> Dict:
+        """Each SCM implements its own auth headers"""
+        pass
+
+    def request(self, path: str, **kwargs):
+        """Override base request to use SCM-specific headers and base_url"""
+        headers = kwargs.pop('headers', None) or self.get_headers()
+        return super().request(
+            path=path,
+            headers=headers,
+            base_url=self.api_url,
+            **kwargs
+        )
+
+class GithubClient(ScmClient):
+    def get_headers(self) -> Dict:
+        return {
+            'Authorization': f"Bearer {self.token}",
+            'User-Agent': 'SocketPythonScript/0.0.1',
+            "accept": "application/json"
+        }
+
+class GitlabClient(ScmClient):
+    def get_headers(self) -> Dict:
+        return {
+            'Authorization': f"Bearer {self.token}",
+            'User-Agent': 'SocketPythonScript/0.0.1',
+            "accept": "application/json"
+        }

socketsecurity/core/scm/base.py

@@ -1,30 +1,33 @@
 import json
 import os
-from socketsecurity.core import log, do_request
-import requests
-from socketsecurity.core.classes import Comment
-from socketsecurity.core.scm_comments import Comments
 import sys
+from dataclasses import dataclass
 
+from git import Optional
+
+from socketsecurity.core import do_request, log
+from socketsecurity.core.classes import Comment
+from socketsecurity.core.scm_comments import Comments
 
-global github_sha
-global github_api_url
-global github_ref_type
-global github_event_name
-global github_workspace
-global github_repository
-global github_ref_name
-global github_actor
-global default_branch
-global github_env
-global pr_number
-global pr_name
-global is_default_branch
-global commit_message
-global committer
-global gh_api_token
-global github_repository_owner
-global event_action
+# Declare all globals with initial None values
+github_sha: Optional[str] = None
+github_api_url: Optional[str] = None
+github_ref_type: Optional[str] = None
+github_event_name: Optional[str] = None
+github_workspace: Optional[str] = None
+github_repository: Optional[str] = None
+github_ref_name: Optional[str] = None
+github_actor: Optional[str] = None
+default_branch: Optional[str] = None
+github_env: Optional[str] = None
+pr_number: Optional[str] = None
+pr_name: Optional[str] = None
+is_default_branch: bool = False
+commit_message: Optional[str] = None
+committer: Optional[str] = None
+gh_api_token: Optional[str] = None
+github_repository_owner: Optional[str] = None
+event_action: Optional[str] = None
 
 github_variables = [
     "GITHUB_SHA",
@@ -45,11 +48,58 @@
     "EVENT_ACTION"
 ]
 
+@dataclass
+class GithubConfig:
+    """Configuration from GitHub environment variables"""
+    sha: str
+    api_url: str
+    ref_type: str
+    event_name: str
+    workspace: str
+    repository: str
+    ref_name: str
+    default_branch: bool
+    pr_number: Optional[str]
+    pr_name: Optional[str]
+    commit_message: Optional[str]
+    actor: str
+    env: str
+    token: str
+    owner: str
+    event_action: Optional[str]
+
+    @classmethod
+    def from_env(cls) -> 'GithubConfig':
+        """Create config from environment variables"""
+        token = os.getenv('GH_API_TOKEN')
+        if not token:
+            log.error("Unable to get Github API Token from GH_API_TOKEN")
+            sys.exit(2)
+
+        return cls(
+            sha=os.getenv('GITHUB_SHA', ''),
+            api_url=os.getenv('GITHUB_API_URL', ''),
+            ref_type=os.getenv('GITHUB_REF_TYPE', ''),
+            event_name=os.getenv('GITHUB_EVENT_NAME', ''),
+            workspace=os.getenv('GITHUB_WORKSPACE', ''),
+            repository=os.getenv('GITHUB_REPOSITORY', '').split('/')[-1],
+            ref_name=os.getenv('GITHUB_REF_NAME', ''),
+            default_branch=os.getenv('DEFAULT_BRANCH', '').lower() == 'true',
+            pr_number=os.getenv('PR_NUMBER'),
+            pr_name=os.getenv('PR_NAME'),
+            commit_message=os.getenv('COMMIT_MESSAGE'),
+            actor=os.getenv('GITHUB_ACTOR', ''),
+            env=os.getenv('GITHUB_ENV', ''),
+            token=token,
+            owner=os.getenv('GITHUB_REPOSITORY_OWNER', ''),
+            event_action=os.getenv('EVENT_ACTION')
+        )
+
+
 for env in github_variables:
     var_name = env.lower()
     globals()[var_name] = os.getenv(env) or None
     if var_name == "default_branch":
-        global is_default_branch
         if default_branch is None or default_branch.lower() == "false":
             is_default_branch = False
         else:
@@ -233,15 +283,14 @@ def post_reaction(comment_id: int) -> None:
     def comment_reaction_exists(comment_id: int) -> bool:
         repo = github_repository.rsplit("/", 1)[1]
         path = f"repos/{github_repository_owner}/{repo}/issues/comments/{comment_id}/reactions"
-        response = do_request(path, headers=headers, base_url=github_api_url)
-        exists = False
         try:
+            response = do_request(path, headers=headers, base_url=github_api_url)
             data = response.json()
             for reaction in data:
                 content = reaction.get("content")
                 if content is not None and content == ":thumbsup:":
-                    exists = True
+                    return True
         except Exception as error:
             log.error(f"Unable to get reaction for {comment_id} for PR {pr_number}")
             log.error(error)
-        return  exists
+        return False