optimized and bool flag args fixed

Author: flowstate

socketsecurity/config.py

@@ -215,6 +215,7 @@ def create_argument_parser() -> argparse.ArgumentParser:
     config_group.add_argument(
         "--default_branch",
         dest="default_branch",
+        action="store_true",
         help=argparse.SUPPRESS
     )
     config_group.add_argument(
@@ -226,6 +227,7 @@ def create_argument_parser() -> argparse.ArgumentParser:
     config_group.add_argument(
         "--pending_head",
         dest="pending_head",
+        action="store_true",
         help=argparse.SUPPRESS
     )
 
@@ -240,6 +242,7 @@ def create_argument_parser() -> argparse.ArgumentParser:
     output_group.add_argument(
         "--generate_license",
         dest="generate_license",
+        action="store_true",
         help=argparse.SUPPRESS
     )
     output_group.add_argument(
@@ -251,6 +254,7 @@ def create_argument_parser() -> argparse.ArgumentParser:
     output_group.add_argument(
         "--enable_debug",
         dest="enable_debug",
+        action="store_true",
         help=argparse.SUPPRESS
     )
     output_group.add_argument(
@@ -262,6 +266,7 @@ def create_argument_parser() -> argparse.ArgumentParser:
     output_group.add_argument(
         "--enable_json",
         dest="enable_json",
+        action="store_true",
         help=argparse.SUPPRESS
     )
     output_group.add_argument(
@@ -273,6 +278,7 @@ def create_argument_parser() -> argparse.ArgumentParser:
     output_group.add_argument(
         "--disable_overview",
         dest="disable_overview",
+        action="store_true",
         help=argparse.SUPPRESS
     )
 
@@ -292,6 +298,7 @@ def create_argument_parser() -> argparse.ArgumentParser:
     security_group.add_argument(
         "--disable_security_issue",
         dest="disable_security_issue",
+        action="store_true",
         help=argparse.SUPPRESS
     )
 
@@ -306,6 +313,7 @@ def create_argument_parser() -> argparse.ArgumentParser:
     advanced_group.add_argument(
         "--ignore_commit_files",
         dest="ignore_commit_files",
+        action="store_true",
         help=argparse.SUPPRESS
     )
     advanced_group.add_argument(
@@ -317,6 +325,7 @@ def create_argument_parser() -> argparse.ArgumentParser:
     advanced_group.add_argument(
         "--disable_blocking",
         dest="disable_blocking",
+        action="store_true",
         help=argparse.SUPPRESS
     )
     advanced_group.add_argument(

socketsecurity/core/__init__.py

@@ -6,6 +6,7 @@
 from glob import glob
 from pathlib import PurePath
 from typing import BinaryIO, Dict, List, Optional, Tuple
+from itertools import chain
 
 from socketdev import socketdev
 from socketdev.fullscans import (
@@ -405,41 +406,26 @@ def get_added_and_removed_packages(self, head_full_scan_id: Optional[str], new_f
         log.info(f"Replaced: {len(diff_report.artifacts.replaced)}")
         log.info(f"Updated: {len(diff_report.artifacts.updated)}")
 
-        added_artifacts = diff_report.artifacts.added + diff_report.artifacts.updated
-        removed_artifacts = diff_report.artifacts.removed + diff_report.artifacts.replaced
-
         added_packages: Dict[str, Package] = {}
         removed_packages: Dict[str, Package] = {}
 
-        for artifact in added_artifacts:
+        # Process added and updated artifacts
+        for artifact in chain(diff_report.artifacts.added, diff_report.artifacts.updated):
             try:
-                pkg = Package.from_diff_artifact(artifact)
+                pkg = Package.from_socket_artifact(artifact)
                 added_packages[artifact.id] = pkg
-            except KeyError:
-                log.error(f"KeyError: Could not create package from added artifact {artifact.id}")
-                log.error(f"Artifact details - name: {artifact.name}, version: {artifact.version}")
-                matches = [p for p in added_artifacts.values() if p.name == artifact.name and p.version == artifact.version]
-                if matches:
-                    log.error(f"Found {len(matches)} packages with matching name/version:")
-                    for m in matches:
-                        log.error(f"  ID: {m.id}, name: {m.name}, version: {m.version}")
-                else:
-                    log.error("No matching packages found in new_full_scan")
+            except KeyError as e:
+                log.error(f"KeyError creating package from added artifact {artifact.id}: {e}")
+                log.error(f"Artifact: name={artifact.name}, version={artifact.version}")
 
-        for artifact in removed_artifacts:
+        # Process removed and replaced artifacts
+        for artifact in chain(diff_report.artifacts.removed, diff_report.artifacts.replaced):
             try:
-                pkg = Package.from_diff_artifact(asdict(artifact))
+                pkg = Package.from_diff_artifact(artifact)
                 removed_packages[artifact.id] = pkg
-            except KeyError:
-                log.error(f"KeyError: Could not create package from removed artifact {artifact.id}")
-                log.error(f"Artifact details - name: {artifact.name}, version: {artifact.version}")
-                matches = [p for p in removed_artifacts.values() if p.name == artifact.name and p.version == artifact.version]
-                if matches:
-                    log.error(f"Found {len(matches)} packages with matching name/version:")
-                    for m in matches:
-                        log.error(f"  ID: {m.id}, name: {m.name}, version: {m.version}")
-                else:
-                    log.error("No matching packages found in head_full_scan")
+            except KeyError as e:
+                log.error(f"KeyError creating package from removed artifact {artifact.id}: {e}")
+                log.error(f"Artifact: name={artifact.name}, version={artifact.version}")
 
         return added_packages, removed_packages
 
@@ -518,32 +504,38 @@ def create_diff_report(
         seen_new_packages = set()
         seen_removed_packages = set()
 
+        # Process added packages
         for package_id, package in added_packages.items():
-            purl = Core.create_purl(package_id, added_packages)
-            base_purl = f"{purl.ecosystem}/{purl.name}@{purl.version}"
-
-            if (not direct_only or package.direct) and base_purl not in seen_new_packages:
-                diff.new_packages.append(purl)
-                seen_new_packages.add(base_purl)
+            # Calculate source data once per package
+            package.introduced_by = self.get_source_data(package, added_packages)
+            
+            if not direct_only or package.direct:
+                base_purl = f"{package.type}/{package.name}@{package.version}"
+                if base_purl not in seen_new_packages:
+                    purl = Core.create_purl(package_id, added_packages)
+                    diff.new_packages.append(purl)
+                    seen_new_packages.add(base_purl)
 
             self.add_package_alerts_to_collection(
                 package=package,
-                alerts_collection=alerts_in_added_packages,
-                packages=added_packages
+                alerts_collection=alerts_in_added_packages
             )
 
+        # Process removed packages
         for package_id, package in removed_packages.items():
-            purl = Core.create_purl(package_id, removed_packages)
-            base_purl = f"{purl.ecosystem}/{purl.name}@{purl.version}"
-
-            if (not direct_only or package.direct) and base_purl not in seen_removed_packages:
-                diff.removed_packages.append(purl)
-                seen_removed_packages.add(base_purl)
+            # Calculate source data once per package
+            package.introduced_by = self.get_source_data(package, removed_packages)
+            
+            if not direct_only or package.direct:
+                base_purl = f"{package.type}/{package.name}@{package.version}"
+                if base_purl not in seen_removed_packages:
+                    purl = Core.create_purl(package_id, removed_packages)
+                    diff.removed_packages.append(purl)
+                    seen_removed_packages.add(base_purl)
 
             self.add_package_alerts_to_collection(
                 package=package,
-                alerts_collection=alerts_in_removed_packages,
-                packages=removed_packages
+                alerts_collection=alerts_in_removed_packages
             )
 
         diff.new_alerts = Core.get_new_alerts(
@@ -552,7 +544,6 @@ def create_diff_report(
         )
 
         diff.new_capabilities = Core.get_capabilities_for_added_packages(added_packages)
-
         Core.add_purl_capabilities(diff)
 
         return diff
@@ -647,29 +638,20 @@ def add_purl_capabilities(diff: Diff) -> None:
 
         diff.new_packages = new_packages
 
-    def add_package_alerts_to_collection(self, package: Package, alerts_collection: dict, packages: dict) -> dict:
-        """
-        Processes alerts from a package and adds them to a shared alerts collection.
-        
-        Args:
-            package: Package to process alerts from
-            alerts_collection: Dictionary to store processed alerts
-            packages: Dictionary of all packages for dependency lookup
-            
-        Returns:
-            Updated alerts collection dictionary
-        """
+    def add_package_alerts_to_collection(self, package: Package, alerts_collection: dict) -> None:
+        """Processes alerts from a package and adds them to a shared alerts collection."""
         default_props = type('EmptyProps', (), {
             'description': "",
             'title': "",
             'suggestion': "",
             'nextStepTitle': ""
         })()
 
-        for alert_item in package.alerts:
-            alert = Alert(**alert_item)
+        for alert in package.alerts:
+            if alert.type == 'licenseSpdxDisj':
+                continue
+
             props = getattr(self.config.all_issues, alert.type, default_props)
-            introduced_by = self.get_source_data(package, packages)
 
             issue_alert = Issue(
                 pkg_type=package.type,
@@ -684,7 +666,7 @@ def add_package_alerts_to_collection(self, package: Package, alerts_collection:
                 title=props.title,
                 suggestion=props.suggestion,
                 next_step_title=props.nextStepTitle,
-                introduced_by=introduced_by,
+                introduced_by=package.introduced_by,  
                 purl=package.purl,
                 url=package.url
             )
@@ -693,13 +675,10 @@ def add_package_alerts_to_collection(self, package: Package, alerts_collection:
                 action = self.config.security_policy[alert.type]['action']
                 setattr(issue_alert, action, True)
 
-            if issue_alert.type != 'licenseSpdxDisj':
-                if issue_alert.key not in alerts_collection:
-                    alerts_collection[issue_alert.key] = [issue_alert]
-                else:
-                    alerts_collection[issue_alert.key].append(issue_alert)
-
-        return alerts_collection
+            if alert.key not in alerts_collection:
+                alerts_collection[alert.key] = [issue_alert]
+            else:
+                alerts_collection[alert.key].append(issue_alert)
 
     @staticmethod
     def save_file(file_name: str, content: str) -> None:

socketsecurity/core/classes.py

@@ -1,6 +1,6 @@
 import json
 from dataclasses import dataclass, field
-from typing import Dict, List, TypedDict, Any, Optional
+from typing import Dict, List, TypedDict, Any, Optional, Tuple
 
 from socketdev.fullscans import FullScanMetadata, SocketArtifact, SocketArtifactLink, SocketScore, SocketAlert, DiffArtifact
 
@@ -121,6 +121,7 @@ class Package(SocketArtifactLink):
     purl: str = ""
     transitives: int = 0
     url: str = ""
+    introduced_by: List[Tuple[str, str]] = field(default_factory=list)
 
     @classmethod
     def from_socket_artifact(cls, artifact: SocketArtifact) -> "Package":