about to big refactor socketcli

Author: flowstate

socketsecurity/config.py

@@ -0,0 +1,186 @@
+from dataclasses import dataclass
+from typing import List, Optional
+import argparse
+import os
+
+@dataclass
+class CliConfig:
+    api_token: str
+    repo: Optional[str]
+    branch: str = ""
+    committer: Optional[List[str]] = None
+    pr_number: str = "0"
+    commit_message: Optional[str] = None
+    default_branch: bool = False
+    target_path: str = "./"
+    scm: str = "api"
+    sbom_file: Optional[str] = None
+    commit_sha: str = ""
+    generate_license: bool = False
+    enable_debug: bool = False
+    allow_unverified: bool = False
+    enable_json: bool = False
+    disable_overview: bool = False
+    disable_security_issue: bool = False
+    files: str = "[]"
+    ignore_commit_files: bool = False
+    disable_blocking: bool = False
+
+    @classmethod
+    def from_args(cls, args_list: Optional[List[str]] = None) -> 'CliConfig':
+        parser = create_argument_parser()
+        args = parser.parse_args(args_list)
+
+        # Get API token from env or args
+        api_token = os.getenv("SOCKET_SECURITY_API_KEY") or args.api_token
+
+        return cls(
+            api_token=api_token,
+            repo=args.repo,
+            branch=args.branch,
+            committer=args.committer,
+            pr_number=args.pr_number,
+            commit_message=args.commit_message,
+            default_branch=args.default_branch,
+            target_path=args.target_path,
+            scm=args.scm,
+            sbom_file=args.sbom_file,
+            commit_sha=args.commit_sha,
+            generate_license=args.generate_license,
+            enable_debug=args.enable_debug,
+            allow_unverified=args.allow_unverified,
+            enable_json=args.enable_json,
+            disable_overview=args.disable_overview,
+            disable_security_issue=args.disable_security_issue,
+            files=args.files,
+            ignore_commit_files=args.ignore_commit_files,
+            disable_blocking=args.disable_blocking
+        )
+
+def create_argument_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="socketcli",
+        description="Socket Security CLI"
+    )
+
+    parser.add_argument(
+        "--api-token",
+        help="Socket Security API token (can also be set via SOCKET_SECURITY_API_KEY env var)",
+        required=False
+    )
+
+    parser.add_argument(
+        "--repo",
+        help="Repository name in owner/repo format",
+        required=False
+    )
+
+    parser.add_argument(
+        "--branch",
+        help="Branch name",
+        default=""
+    )
+
+    parser.add_argument(
+        "--committer",
+        help="Committer(s) to filter by",
+        nargs="*"
+    )
+
+    parser.add_argument(
+        "--pr-number",
+        help="Pull request number",
+        default="0"
+    )
+
+    parser.add_argument(
+        "--commit-message",
+        help="Commit message"
+    )
+
+    # Boolean flags
+    parser.add_argument(
+        "--default-branch",
+        action="store_true",
+        help="Use default branch"
+    )
+
+    parser.add_argument(
+        "--generate-license",
+        action="store_true",
+        help="Generate license information"
+    )
+
+    parser.add_argument(
+        "--enable-debug",
+        action="store_true",
+        help="Enable debug logging"
+    )
+
+    parser.add_argument(
+        "--allow-unverified",
+        action="store_true",
+        help="Allow unverified packages"
+    )
+
+    parser.add_argument(
+        "--enable-json",
+        action="store_true",
+        help="Output in JSON format"
+    )
+
+    parser.add_argument(
+        "--disable-overview",
+        action="store_true",
+        help="Disable overview output"
+    )
+
+    parser.add_argument(
+        "--disable-security-issue",
+        action="store_true",
+        help="Disable security issue checks"
+    )
+
+    parser.add_argument(
+        "--ignore-commit-files",
+        action="store_true",
+        help="Ignore commit files"
+    )
+
+    parser.add_argument(
+        "--disable-blocking",
+        action="store_true",
+        help="Disable blocking mode"
+    )
+
+    # Path and file related arguments
+    parser.add_argument(
+        "--target-path",
+        default="./",
+        help="Target path for analysis"
+    )
+
+    parser.add_argument(
+        "--scm",
+        default="api",
+        help="Source control management type"
+    )
+
+    parser.add_argument(
+        "--sbom-file",
+        help="SBOM file path"
+    )
+
+    parser.add_argument(
+        "--commit-sha",
+        default="",
+        help="Commit SHA"
+    )
+
+    parser.add_argument(
+        "--files",
+        default="[]",
+        help="Files to analyze (JSON array string)"
+    )
+
+    return parser

socketsecurity/core/__init__.py

@@ -140,10 +140,6 @@ def do_request(
 
 
 class Core:
-    # token: str
-    # base_api_url: str
-    # request_timeout: int
-    # reports: list
 
     client: CliClient
     config: SocketConfig
@@ -153,30 +149,6 @@ def __init__(self, config: SocketConfig, client: CliClient):
         self.client = client
         self.set_org_vars()
 
-    # def __init__(
-    #         self,
-    #         token: str,
-    #         base_api_url: str = None,
-    #         request_timeout: int = None,
-    #         enable_all_alerts: bool = False,
-    #         allow_unverified: bool = False
-    # ):
-    #     global allow_unverified_ssl
-    #     allow_unverified_ssl = allow_unverified
-    #     self.token = token + ":"
-    #     encode_key(self.token)
-    #     self.socket_date_format = "%Y-%m-%dT%H:%M:%S.%fZ"
-    #     self.base_api_url = base_api_url
-    #     if self.base_api_url is not None:
-    #         Core.set_api_url(self.base_api_url)
-    #     self.request_timeout = request_timeout
-    #     if self.request_timeout is not None:
-    #         Core.set_timeout(self.request_timeout)
-    #     if enable_all_alerts:
-    #         global all_new_alerts
-    #         all_new_alerts = True
-    #     Core.set_org_vars()
-
 
     def set_org_vars(self) -> None:
         """Sets the main shared configuration variables"""
@@ -231,26 +203,6 @@ def get_sbom_data(self, full_scan_id: str) -> list:
 
         return results
 
-    # ORIGINAL - remove after verification
-    # def get_sbom_data(self, full_scan_id: str) -> list:
-    #     path = f"orgs/{self.config.org_slug}/full-scans/{full_scan_id}"
-    #     response = self.client.request(path)
-    #     results = []
-    #     try:
-    #         data = response.json()
-    #         results = data.get("sbom_artifacts") or []
-    #     except Exception as error:
-    #         log.debug("Failed with old style full-scan API using new format")
-    #         log.debug(error)
-    #         data = response.text
-    #         data.strip('"')
-    #         data.strip()
-    #         for line in data.split("\n"):
-    #             if line != '"' and line != "" and line is not None:
-    #                 item = json.loads(line)
-    #                 results.append(item)
-    #     return results
-
     def get_security_policy(self) -> dict:
         """Get the Security policy and determine the effective Org security policy"""
         payload = [{"organization": self.config.org_id}]

socketsecurity/output.py

@@ -0,0 +1,89 @@
+from typing import Optional, Dict, Any
+import json
+import sys
+import logging
+from pathlib import Path
+from .core.classes import Diff, Issue
+
+
+class OutputHandler:
+    blocking_disabled: bool
+    logger: logging.Logger
+
+    def __init__(self, blocking_disabled: bool):
+        self.blocking_disabled = blocking_disabled
+        self.logger = logging.getLogger("socketcli")
+
+    def handle_output(self, diff_report: Diff, sbom_file_name: Optional[str] = None, json_output: bool = False) -> int:
+        """Main output handler that determines output format and returns exit code"""
+        if json_output:
+            self.output_console_json(diff_report, sbom_file_name)
+        else:
+            self.output_console_comments(diff_report, sbom_file_name)
+
+        self.save_sbom_file(diff_report, sbom_file_name)
+        return 0 if self.report_pass(diff_report) else 1
+
+    def output_console_comments(self, diff_report: Diff, sbom_file_name: Optional[str] = None) -> None:
+        """Outputs formatted console comments"""
+        if not diff_report.issues:
+            self.logger.info("No issues found")
+            return
+
+        for issue in diff_report.issues:
+            self._output_issue(issue)
+
+    def output_console_json(self, diff_report: Diff, sbom_file_name: Optional[str] = None) -> None:
+        """Outputs JSON formatted results"""
+        output = {
+            "issues": [self._format_issue(issue) for issue in diff_report.issues],
+            "pass": self.report_pass(diff_report)
+        }
+        if sbom_file_name:
+            output["sbom_file"] = sbom_file_name
+
+        json.dump(output, sys.stdout, indent=2)
+        sys.stdout.write("\n")
+
+    def report_pass(self, diff_report: Diff) -> bool:
+        """Determines if the report passes security checks"""
+        if not diff_report.issues:
+            return True
+
+        if self.blocking_disabled:
+            return True
+
+        return not any(issue.blocking for issue in diff_report.issues)
+
+    def save_sbom_file(self, diff_report: Diff, sbom_file_name: Optional[str] = None) -> None:
+        """Saves SBOM file if filename is provided"""
+        if not sbom_file_name or not diff_report.sbom:
+            return
+
+        sbom_path = Path(sbom_file_name)
+        sbom_path.parent.mkdir(parents=True, exist_ok=True)
+
+        with open(sbom_path, "w") as f:
+            json.dump(diff_report.sbom, f, indent=2)
+
+    def _output_issue(self, issue: Issue) -> None:
+        """Helper method to format and output a single issue"""
+        severity = issue.severity.upper() if issue.severity else "UNKNOWN"
+        status = "🚫 Blocking" if issue.blocking else "⚠️ Warning"
+
+        self.logger.warning(f"\n{status} - Severity: {severity}")
+        self.logger.warning(f"Title: {issue.title}")
+        if issue.description:
+            self.logger.warning(f"Description: {issue.description}")
+        if issue.recommendation:
+            self.logger.warning(f"Recommendation: {issue.recommendation}")
+
+    def _format_issue(self, issue: Issue) -> Dict[str, Any]:
+        """Helper method to format an issue for JSON output"""
+        return {
+            "title": issue.title,
+            "description": issue.description,
+            "severity": issue.severity,
+            "blocking": issue.blocking,
+            "recommendation": issue.recommendation
+        }

socketsecurity/socketcli.py

@@ -1,7 +1,6 @@
 import argparse
 import json
 
-import socketsecurity.core
 from socketsecurity.core import Core, __version__
 from socketsecurity.logging import initialize_logging, set_debug_mode
 from socketsecurity.core.classes import FullScanParams, Diff, Package, Issue
@@ -11,13 +10,8 @@
 from git import InvalidGitRepositoryError, NoSuchPathError
 import os
 import sys
-import logging
 
 socket_logger, cli_logger = initialize_logging()
-
-log_format = "%(asctime)s: %(message)s"
-logging.basicConfig(level=logging.INFO, format=log_format)
-socketsecurity.core.log.setLevel(level=logging.INFO)
 log = cli_logger
 blocking_disabled = False