GeezerVisonThe World Wide Web, restored to 800 pixels of glory. |
| Home|About|Gaming cabinet|Archives|How it works|Starter sites |
GeezerVison is showing github.com — socket-python-cli/socketsecurity/socketcli.py at main · SocketDev/socket-python-cli Expand file tree/ Copy pathsocketcli.pyMore file actions 1117 lines (982 loc) · 52 KB / Copy pathsocketcli.pyFile metadata and controls1117 lines (982 loc) · 52 KB 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 import json import os import shutil import sys import traceback from datetime import datetime, timezone from typing import List, Optional, Tuple from uuid import uuid4 from dotenv import load_dotenv from git import InvalidGitRepositoryError, NoSuchPathError from socketdev import socketdev from socketdev.fullscans import FullScanParams from socketsecurity import USER_AGENT_NAME from socketsecurity.config import CliConfig, truncate_commit_message from socketsecurity.core import Core from socketsecurity.core.classes import Diff from socketsecurity.core.cli_client import CliClient from socketsecurity.core.git_interface import Git from socketsecurity.core.logging import initialize_logging, set_debug_mode from socketsecurity.core.messages import Messages from socketsecurity.core.pull_request import ( parse_pull_request_number, repository_owner, resolve_pull_request_context, resolve_repository_url, ) from socketsecurity.core.scm_comments import Comments from socketsecurity.core.socket_config import SocketConfig, module_folder_dirs from socketsecurity.core.streaming import StreamingLogs from socketsecurity.fossa_compat import build_fossa_attribution_payload from socketsecurity.output import OutputHandler socket_logger, log = initialize_logging() load_dotenv() # Buildkite sets BUILDKITE=true in every job environment. Used to gate log # section markers that would render as literal text on other CI platforms. IS_BUILDKITE = os.getenv("BUILDKITE") == "true" def _emit_infrastructure_error(message: str, include_traceback: bool = False) -> None: """Emit a structured error for infrastructure/API failures. When running in Buildkite, wraps the error in log-section markers (`^^^ +++` expands the section in the BK UI) and prints a soft_fail hint. On every other platform it's a plain log.error so the markers don't leak as literal text. This is presentation only -- it does not decide the exit code (the caller does that, honoring --disable-blocking and --exit-code-on-api-error). """ if IS_BUILDKITE: print("^^^ +++", flush=True) print("--- :warning: Socket infrastructure error", flush=True) log.error(message) if IS_BUILDKITE: log.error( "Tip: this is an infrastructure error, not a security finding. To keep it " "from blocking the build, add a soft_fail rule for the CLI's API-error exit " "code (default 3, or whatever you pass to --exit-code-on-api-error)." ) if include_traceback: traceback.print_exc() def _log_scan_mode_fallback(requested: str, effective: str, reason: str) -> None: log.info( f"Scan mode: requested={requested} effective={effective} reason={reason}" ) def build_license_artifact_payload( diff: Diff, legal_format: str = "socket", config: CliConfig | None = None, ) -> dict: """Build the license artifact payload from a diff, tolerating sparse scan paths.""" if legal_format == "fossa": if config is None: raise ValueError("config is required when building FOSSA-format legal artifacts") return build_fossa_attribution_payload(diff, config) all_packages = {} packages = getattr(diff, "packages", {}) or {} for purl in packages: package = packages[purl] output = { "id": package.id, "name": package.name, "version": package.version, "ecosystem": package.type, "direct": package.direct, "url": package.url, "license": package.license, "licenseDetails": package.licenseDetails, "licenseAttrib": package.licenseAttrib, "purl": package.purl, } all_packages[package.id] = output return all_packages def _write_attribution_file(config, payload: dict) -> None: Core.save_file(config.license_file_name, json.dumps(payload, indent=2)) DEFAULT_API_TIMEOUT = 1200 # Sentinel repo/branch names used when none can be detected from git or supplied via flags. # When the repo/branch are these defaults we skip forwarding SOCKET_REPO_NAME/SOCKET_BRANCH_NAME # to the coana CLI so unrelated default-named runs don't share reachability cache buckets. DEFAULT_REPO_NAME = "socket-default-repo" DEFAULT_BRANCH_NAME = "socket-default-branch" def get_api_request_timeout(config: CliConfig) -> int: return config.timeout if config.timeout is not None else DEFAULT_API_TIMEOUT def should_write_comment(disabled: bool, has_findings: bool, update_existing: bool) -> bool: """ Decides whether a Socket comment should be written for this run. :param disabled: bool - The comment type is switched off by flag. :param has_findings: bool - There is something to report this run. :param update_existing: bool - A comment of this type is already on the PR. :return: bool - True when the comment should be posted or updated. """ if disabled: # The flag means the CLI does not manage this comment at all. Checking # update_existing first only suppressed the very first post, so any # comment already on the pull request kept being updated. return False if not has_findings: # Nothing to report: refresh an existing comment so a resolved alerts # table is cleared, but do not open a new one. return update_existing return True def _match_ignored_alerts_for_telemetry( ignored_alerts: list, ignore_all: bool, ignore_commands: list[tuple[str, str]], ) -> list: """Return alerts attributable to one ignore comment.""" if ignore_all: return list(ignored_alerts) return [ alert for alert in ignored_alerts if any( Comments.is_ignore( alert.pkg_name, alert.pkg_version, name, version, alert.pkg_type, ) for name, version in ignore_commands ) ] def _select_pull_request_provider(integration_type: str, scm_type: str) -> str: """Prefer an active comment adapter when resolving pull request context.""" return scm_type if scm_type in ("github", "gitlab") else integration_type def create_scm_scan( core: Core, config: CliConfig, scm_event_type: Optional[str], *, scan_paths: List[str], params: FullScanParams, no_change: bool, base_paths: Optional[List[str]], explicit_files: Optional[List[str]], external_href: Optional[str], ) -> Tuple[Diff, bool]: """Create the scan for an SCM-integrated run. Only a pull request or merge request event has a baseline to compare against, so every other pipeline -- default-branch pushes included -- gets a full scan. The detected event type is authoritative: API-only diff flags cannot turn an ordinary branch pipeline into a comparison. Returns the diff and whether it came from a comparison. Callers need the second value because a full scan carries no "new alerts" category to comment on or to block a build with. """ scan_kwargs = { "no_change": no_change, "save_files_list_path": config.save_submitted_files_list, "save_manifest_tar_path": config.save_manifest_tar, "base_paths": base_paths, "explicit_files": explicit_files, } if scm_event_type == "diff": log.info("Starting comment logic for PR/MR event") diff = core.create_new_diff( scan_paths, params, external_href=external_href, **scan_kwargs ) return diff, True log.info("Starting non-PR/MR flow") # No before/after pair here, so there is nothing for external_href to hang off. diff = core.create_full_scan_with_report_url(scan_paths, params, **scan_kwargs) return diff, False def apply_git_context(config: CliConfig) -> Tuple[bool, Optional[Git]]: """ Fill in any repo details the caller did not pass from the checkout at target_path. Returns whether target_path is a git repository, along with the Git handle when it is. """ try: git_repo = Git(config.target_path, base_commit_sha=config.base_commit_sha) except InvalidGitRepositoryError: log.debug("Not a git repository, setting ignore_commit_files=True") config.ignore_commit_files = True return False, None except NoSuchPathError: raise Exception(f"Unable to find path {config.target_path}") if not config.repo: config.repo = git_repo.repo_name if not config.commit_sha: config.commit_sha = git_repo.commit_str if not config.branch: config.branch = git_repo.branch if not config.committers: config.committers = [git_repo.get_formatted_committer()] if not config.commit_message: # A repository's commit message is unbounded and ships in the query string. config.commit_message = truncate_commit_message(git_repo.commit_message) return True, git_repo def build_socket_sdk(config: CliConfig) -> socketdev: cli_user_agent_string = f"{USER_AGENT_NAME}/{config.version}" return socketdev( token=config.api_token, timeout=get_api_request_timeout(config), allow_unverified=config.allow_unverified, user_agent=cli_user_agent_string ) def cli(): try: main_code() except KeyboardInterrupt: log.info("Keyboard Interrupt detected, exiting") config = CliConfig.from_args() # Get current config if not config.disable_blocking: sys.exit(2) else: sys.exit(0) except Exception as error: config = CliConfig.from_args() # Get current config _emit_infrastructure_error( f"Unexpected error when running the CLI: {error}", include_traceback=True, ) # --disable-blocking forces a clean exit for ALL outcomes (it takes # precedence over --exit-code-on-api-error); otherwise infra/API errors # exit with the configurable code (default 3), keeping them distinct # from blocking security findings (exit 1). if not config.disable_blocking: sys.exit(config.exit_code_on_api_error) else: sys.exit(0) def main_code(): config = CliConfig.from_args() log.info(f"Starting Socket Security CLI version {config.version}") log.debug(f"config: {config.to_dict()}") # Warn if strict-blocking is used with disable-blocking if config.strict_blocking and config.disable_blocking: log.warning("Both --strict-blocking and --disable-blocking specified. " "--disable-blocking takes precedence and will always return exit code 0.") # Validate API token if not config.api_token: log.info("Socket API Token not found. Please set it using either:\n" "1. Command line: --api-token YOUR_TOKEN\n" "2. Environment variable: SOCKET_SECURITY_API_TOKEN") sys.exit(3) sdk = build_socket_sdk(config) # Suppress urllib3 InsecureRequestWarning when using --allow-unverified if config.allow_unverified: import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) output_handler = OutputHandler(config, sdk) log.debug("sdk loaded") if config.enable_debug: set_debug_mode(True) log.debug("Debug logging enabled") # Initialize Socket core components socket_config = SocketConfig( api_key=config.api_token, allow_unverified_ssl=config.allow_unverified, timeout=get_api_request_timeout(config) ) log.debug("loaded socket_config") client = CliClient(socket_config) sdk.api.api_url = socket_config.api_url log.debug("loaded client") with StreamingLogs( client=client, cli_logger=log, sdk_logger=socket_logger, client_version=config.version, upload_logs=config.upload_logs, enable_debug=config.enable_debug, ) as streaming: core = Core(socket_config, sdk, config) log.debug("loaded core") # Re-include directories that are excluded from manifest discovery by default # (e.g. build/dist). --include-dirs names them individually; --include-module-folders # re-includes the JS module folders as a group. Build a new set rather than mutating # the shared default_exclude_dirs in place. Applied here so it covers every find_files # call below, including the sub-path manifest pre-check. dirs_to_include = set(config.included_dirs or []) if config.include_module_folders: dirs_to_include |= module_folder_dirs if dirs_to_include: core.config.excluded_dirs = set(core.config.excluded_dirs) - dirs_to_include log.debug(f"Re-including normally-excluded directories in scan: {sorted(dirs_to_include)}") if config.excluded_ecosystems: core.config.excluded_ecosystems = list(config.excluded_ecosystems) # Check for required dependencies if reachability analysis is enabled if config.reach: log.info("Reachability analysis enabled, checking for required dependencies...") required_deps = ["node", "uv"] missing_deps = [] found_deps = [] for dep in required_deps: if shutil.which(dep): found_deps.append(dep) log.debug(f"Found required dependency: {dep}") else: missing_deps.append(dep) if missing_deps: log.error(f"Reachability analysis requires the following dependencies: {', '.join(required_deps)}") log.error(f"Missing dependencies: {', '.join(missing_deps)}") log.error("Please install the missing dependencies and try again.") sys.exit(3) log.info(f"All required dependencies found: {', '.join(found_deps)}") # Check if organization has an enterprise plan log.info("Checking organization plan for reachability analysis eligibility...") org_response = sdk.org.get(use_types=True) organizations = org_response.get("organizations", {}) if organizations: org_id = next(iter(organizations)) org_plan = organizations[org_id].get('plan', '') # Check if plan matches enterprise* pattern (enterprise, enterprise_trial, etc.) if not org_plan.startswith('enterprise'): log.error("Reachability analysis is only available for enterprise plans.") log.error(f"Your organization plan is: {org_plan}") log.error("Please upgrade to an enterprise plan to use reachability analysis.") sys.exit(3) log.info(f"Organization plan verified: {org_plan}") else: log.error("Unable to retrieve organization information for plan verification.") sys.exit(3) # Parse files argument try: if isinstance(config.files, list): # Already a list, use as-is specified_files = config.files elif isinstance(config.files, str): # Handle different string formats files_str = config.files.strip() # If the string is wrapped in extra quotes, strip them if ((files_str.startswith('"') and files_str.endswith('"')) or (files_str.startswith("'") and files_str.endswith("'"))): # Check if the inner content looks like JSON inner_str = files_str[1:-1] if inner_str.startswith('[') and inner_str.endswith(']'): files_str = inner_str # Try to parse as JSON try: specified_files = json.loads(files_str) except json.JSONDecodeError: # If JSON parsing fails, try replacing single quotes with double quotes files_str = files_str.replace("'", '"') specified_files = json.loads(files_str) else: # Default to empty list specified_files = [] except Exception as error: log.error(f"Unable to parse files argument: {config.files}") log.error(f"Error details: {error}") log.debug(f"Files type: {type(config.files)}") log.debug(f"Files repr: {repr(config.files)}") sys.exit(3) # Determine if files were explicitly specified files_explicitly_specified = config.files != "[]" and len(specified_files) > 0 # Variable to track if we need to override files with facts file facts_file_to_submit = None # Variable to track SBOM files to submit when using --reach-use-only-pregenerated-sboms sbom_files_to_submit = None # Manifest results retained from the --sub-path routing pre-check. Reusing # these avoids walking every selected sub-path again during scan creation. discovered_scan_files = None # Git setup is_repo, git_repo = apply_git_context(config) # Track whether repo/branch fell back to the default sentinels so reachability can skip # forwarding them as coana cache-bucket keys (computed before any workspace suffixing). repo_defaulted = not config.repo branch_defaulted = not config.branch if not config.repo: base_repo_name = DEFAULT_REPO_NAME if config.workspace_name: config.repo = f"{base_repo_name}-{config.workspace_name}" else: config.repo = base_repo_name log.debug(f"Using default repository name: {config.repo}") if not config.branch: config.branch = DEFAULT_BRANCH_NAME log.debug(f"Using default branch name: {config.branch}") # Calculate the scan paths - combine target_path with sub_paths if provided scan_paths = [] base_paths = [config.target_path] # Always use target_path as the single base path if config.sub_paths: for sub_path in config.sub_paths: full_scan_path = os.path.join(config.target_path, sub_path) log.debug(f"Using sub-path for scanning: {full_scan_path}") # Verify the scan path exists if not os.path.exists(full_scan_path): raise Exception(f"Sub-path does not exist: {full_scan_path}") scan_paths.append(full_scan_path) else: # Use the target path as the single scan path scan_paths = [config.target_path] # Modify repository name if workspace_name is provided if config.workspace_name and config.repo: config.repo = f"{config.repo}-{config.workspace_name}" log.debug(f"Modified repository name with workspace suffix: {config.repo}") elif config.workspace_name and not config.repo: # If no repo name was set but workspace_name is provided, we'll use it later log.debug(f"Workspace name provided: {config.workspace_name}") # Run reachability analysis if enabled if config.reach: from socketsecurity.core.tools.reachability import ReachabilityAnalyzer log.info("Starting reachability analysis...") # Find manifest files in scan paths (excluding .socket.facts.json to avoid circular dependency) log.info("Finding manifest files for reachability analysis...") manifest_files = [ # Always find all manifest files for the tar hash upload f for scan_path in scan_paths for f in core.find_files(scan_path) if not f.endswith('.socket.facts.json') ] if not manifest_files: log.warning("No manifest files found for reachability analysis") else: log.info(f"Found {len(manifest_files)} manifest files for reachability upload") # Upload manifests and get tar hash log.info("Uploading manifest files...") try: # Get org_slug early (we'll need it) org_slug = core.config.org_slug assert org_slug # Upload manifest files tar_hash = sdk.uploadmanifests.upload_manifest_files( org_slug=org_slug, file_paths=manifest_files, workspace=config.repo or "default-workspace", base_paths=[config.target_path], use_lazy_loading=False ) log.info(f"Manifest upload successful, tar hash: {tar_hash}") # Initialize and run reachability analyzer analyzer = ReachabilityAnalyzer(client) # Determine output path output_path = config.reach_output_file or ".socket.facts.json" # Run the analysis result = analyzer.run_reachability_analysis( org_slug=org_slug, target_directory=config.target_path, tar_hash=tar_hash, output_path=output_path, timeout=config.reach_analysis_timeout, memory_limit=config.reach_analysis_memory_limit, ecosystems=config.reach_ecosystems, # Union the deprecated --reach-exclude-paths with the unified --exclude-paths # and forward verbatim to coana's --exclude-dirs. Patterns are scan-root # relative; coana resolves --exclude-dirs relative to its `run` target, which # here is `.` == cwd == scan root, so passthrough is correct. If a nested # target is ever supported, re-anchor patterns to the target first (see Node's # pathRelativeToTarget in exclude-paths.mts). exclude_paths=( (config.reach_exclude_paths or []) + (config.exclude_paths or []) ) or None, min_severity=config.reach_min_severity, skip_cache=config.reach_skip_cache or False, disable_analytics=config.reach_disable_analytics or False, enable_analysis_splitting=config.reach_enable_analysis_splitting or False, detailed_analysis_log_file=config.reach_detailed_analysis_log_file or False, repo_name=None if repo_defaulted else config.repo, branch_name=None if branch_defaulted else config.branch, version=config.reach_version, concurrency=config.reach_concurrency, additional_params=config.reach_additional_params, enable_debug=config.enable_debug, use_only_pregenerated_sboms=config.reach_use_only_pregenerated_sboms, continue_on_analysis_errors=config.reach_continue_on_analysis_errors, continue_on_install_errors=config.reach_continue_on_install_errors, continue_on_missing_lock_files=config.reach_continue_on_missing_lock_files, continue_on_no_source_files=config.reach_continue_on_no_source_files, reach_debug=config.reach_debug, disable_external_tool_checks=config.reach_disable_external_tool_checks, disable_dependency_downloads=config.reach_disable_dependency_downloads, ) log.info("Reachability analysis completed successfully") log.info(f"Results written to: {result['report_path']}") if result.get('scan_id'): log.info(f"Reachability scan ID: {result['scan_id']}") # If only-facts-file mode, mark the facts file for submission if config.only_facts_file: facts_file_to_submit = os.path.abspath(result['report_path']) log.info(f"Only-facts-file mode: will submit only {facts_file_to_submit}") # If reach-use-only-pregenerated-sboms mode, submit CDX, SPDX, and facts file if config.reach_use_only_pregenerated_sboms: # Find only CDX and SPDX files for the final scan submission sbom_files_to_submit = [] for scan_path in scan_paths: sbom_files_to_submit.extend(core.find_sbom_files(scan_path)) if os.path.exists(result['report_path']): sbom_files_to_submit.append(result['report_path']) log.info(f"Pre-generated SBOMs mode: will submit {len(sbom_files_to_submit)} files (CDX, SPDX, and facts file)") except Exception as e: log.error(f"Reachability analysis failed: {str(e)}") if not config.disable_blocking: sys.exit(3) log.info("Continuing with normal scan flow...") # Canonicalize before any adapter reads it. Buildkite always sets # BUILDKITE_PULL_REQUEST -- to the string "false" on non-PR builds -- so the # documented --pr-number "$BUILDKITE_PULL_REQUEST" form delivers a truthy # non-numeric value that GithubConfig would otherwise treat as a real PR, # making a branch build look like a pull request event. config.pr_number = str(parse_pull_request_number(config.pr_number)) scm = None if config.scm == "github": from socketsecurity.core.scm.github import Github, GithubConfig # Only pass pr_number if it's not "0" (the default) pr_number = config.pr_number if config.pr_number != "0" else None github_config = GithubConfig.from_env(pr_number=pr_number) scm = Github(client=client, config=github_config, ignore_authorization=config.ignore_authorization) elif config.scm == 'gitlab': from socketsecurity.core.scm.gitlab import Gitlab, GitlabConfig gitlab_config = GitlabConfig.from_env() scm = Gitlab(client=client, config=gitlab_config, ignore_authorization=config.ignore_authorization) config.repository_url = resolve_repository_url( config.scm, config.repo, os.environ, remote_url=getattr(git_repo, "remote_url", None) if is_repo else None, ) # Don't override config.default_branch if it was explicitly set via --default-branch flag # Only use SCM detection if --default-branch wasn't provided if scm is not None and not config.default_branch: config.default_branch = scm.config.is_default_branch # Override files if only-facts-file mode is active if facts_file_to_submit: specified_files = [facts_file_to_submit] files_explicitly_specified = True log.debug(f"Overriding files to only submit facts file: {facts_file_to_submit}") # Override files if reach-use-only-pregenerated-sboms mode is active if sbom_files_to_submit: specified_files = sbom_files_to_submit files_explicitly_specified = True log.debug(f"Overriding files to submit only SBOM files (CDX, SPDX, and facts): {sbom_files_to_submit}") # Determine files to check based on the new logic files_to_check = [] force_api_mode = False force_diff_mode = False if files_explicitly_specified: # Case 2: Files are specified - use them and don't check commit details files_to_check = specified_files log.debug(f"Using explicitly specified files: {files_to_check}") elif not config.ignore_commit_files and is_repo: # Case 1: Files not specified and --ignore-commit-files not set - try to find changed files from commit files_to_check = git_repo.changed_files log.debug(f"Using changed files from commit: {files_to_check}") elif config.ignore_commit_files and is_repo: # Case 3: Git repo with --ignore-commit-files - force diff mode files_to_check = [] force_diff_mode = True log.debug("Git repo with --ignore-commit-files: forcing diff mode") else: # Case 4: Not a git repo (ignore_commit_files was auto-set to True) files_to_check = [] # If --enable-diff is set, force diff mode for non-git repos log.debug(f"Case 4: Non-git repo - config.enable_diff={config.enable_diff}, type={type(config.enable_diff)}") if config.enable_diff: force_diff_mode = True log.debug("Non-git repo with --enable-diff: forcing diff mode") else: log.debug("Non-git repo without --enable-diff: will use full scan mode") # Check if we have supported manifest files has_supported_files = files_to_check and core.has_manifest_files(files_to_check) # If using sub_paths, we need to check if manifest files exist in the scan paths if config.sub_paths and not files_explicitly_specified: # Override file checking to look in the scan paths instead # Get manifest files from all scan paths try: discovered_scan_files = [] for scan_path in scan_paths: scan_files = core.find_files(scan_path) discovered_scan_files.extend(scan_files) has_supported_files = len(discovered_scan_files) > 0 log.debug( f"Found {len(discovered_scan_files)} manifest files across " f"{len(scan_paths)} scan paths" ) except Exception as e: log.debug(f"Error finding files in scan paths: {e}") discovered_scan_files = None has_supported_files = False # Case 3: If no supported files or files are empty, force API mode (no PR comments) # BUT: Don't force API mode if we're in force_diff_mode log.debug(f"files_to_check={files_to_check}, has_supported_files={has_supported_files}, force_diff_mode={force_diff_mode}, config.enable_diff={config.enable_diff}") if not has_supported_files and not force_diff_mode: force_api_mode = True log.debug("No supported manifest files found, forcing API mode") log.debug(f"force_api_mode={force_api_mode}") # Determine scan behavior should_skip_scan = False # Always perform scan, but behavior changes based on supported files if not has_supported_files and not force_diff_mode: # No supported files and not forcing diff - still scan but in API mode should_skip_scan = False log.debug("No supported files but will scan in API mode") else: log.debug("Found supported manifest files or forcing diff mode, proceeding with normal scan") org_slug = core.config.org_slug if config.repo_is_public: core.config.repo_visibility = "public" integration_type = config.integration_type integration_org_slug = config.integration_org_slug or org_slug pr_provider = _select_pull_request_provider(integration_type, config.scm) pr_context = resolve_pull_request_context( pr_provider, config.pr_number, config.repo, configured_explicit=config.pr_number_explicit, env=os.environ, ) pr_number = pr_context.number if pr_number: config.pr_number = str(pr_number) if scm is not None: if hasattr(scm.config, "pr_number"): scm.config.pr_number = str(pr_number) elif hasattr(scm.config, "mr_iid"): scm.config.mr_iid = str(pr_number) log.debug( f"Resolved {pr_provider} pull request context: " f"number={pr_number}, url={pr_context.url or 'unavailable'}" ) # Determine if this should be treated as default branch # Priority order: # 1. If --default-branch flag is explicitly set to True, use that # 2. If SCM detected it's the default branch, use that # 3. If it's a git repo, use git_repo.is_default_branch # 4. Otherwise, default to False if config.default_branch: is_default_branch = True elif scm is not None and hasattr(scm.config, 'is_default_branch') and scm.config.is_default_branch: is_default_branch = True elif is_repo and git_repo.is_default_branch: is_default_branch = True else: is_default_branch = False params = FullScanParams( org_slug=org_slug, integration_type=integration_type, integration_org_slug=integration_org_slug, repo=config.repo, branch=config.branch, commit_message=config.commit_message, commit_hash=config.commit_sha, pull_request=pr_number, committers=config.committers, make_default_branch=is_default_branch, set_as_pending_head=is_default_branch, tmp=False, scan_type='socket_tier1' if config.reach else 'socket', workspace=config.workspace or None, ) params.include_license_details = not config.exclude_license_details # Initialize diff diff = Diff() diff.id = "NO_DIFF_RAN" diff.diff_url = "" diff.report_url = "" scan_explicit_files = ( sbom_files_to_submit if sbom_files_to_submit is not None else discovered_scan_files ) # Handle SCM-specific flows log.debug(f"Flow decision: scm={scm is not None}, force_diff_mode={force_diff_mode}, force_api_mode={force_api_mode}, enable_diff={config.enable_diff}") def _is_unprocessed(c): """Check if an ignore comment has not yet been marked with '+1' reaction. For GitHub, reactions['+1'] is already in the comment response (no extra call). For GitLab, has_thumbsup_reaction() makes a lazy API call per comment.""" if getattr(c, "reactions", {}).get("+1"): return False if hasattr(scm, "has_thumbsup_reaction") and scm.has_thumbsup_reaction(c.id): return False return True scm_event_type = scm.check_event_type() if scm is not None else None # Every branch below except the SCM full-scan one produces a comparison, or # is already covered by force_api_mode. See the blocking guard after the # scan for why this is tracked. comparison_ran = True if scm_event_type == "comment": # FIXME: This entire flow should be a separate command called "filter_ignored_alerts_in_comments" # It's not related to scanning or diff generation - it just: # 1. Triggers on comments in GitHub/GitLab # 2. If comment was from Socket, checks for ignore reactions # 3. Updates the comment to remove ignored alerts # This is completely separate from the main scanning functionality log.info("Comment initiated flow") if not config.disable_ignore: comments = scm.get_comments_for_pr() # Emit telemetry for ignore comments before +1 reaction is added. # The +1 reaction (added by remove_comment_alerts) serves as the "processed" marker. if "ignore" in comments: unprocessed = [c for c in comments["ignore"] if _is_unprocessed(c)] if unprocessed: try: events = [] for c in unprocessed: single = {"ignore": [c]} ignore_all, ignore_commands = Comments.get_ignore_options(single) user = getattr(c, "user", None) or getattr(c, "author", None) or {} now = datetime.now(timezone.utc).isoformat() shared_fields = { "event_kind": "user-action", "client_action": "ignore", "alert_action": "error", "event_sender_created_at": now, "vcs_provider": integration_type, "owner": repository_owner(config.repository_url), "repo": config.repo, "pr_number": pr_number, "ignore_all": ignore_all, "sender_name": user.get("login") or user.get("username", ""), "sender_id": str(user.get("id", "")), } if ignore_commands: for name, version in ignore_commands: events.append({**shared_fields, "event_id": str(uuid4()), "artifact_input": f"{name}@{version}"}) elif ignore_all: events.append({**shared_fields, "event_id": str(uuid4())}) if events: log.debug(f"Ignore telemetry: {len(events)} events to send") client.post_telemetry_events(org_slug, events) except Exception as e: log.warning(f"Failed to send ignore telemetry: {e}") log.debug("Removing comment alerts") scm.remove_comment_alerts(comments) else: log.info("Ignore commands disabled (--disable-ignore), skipping comment processing") elif scm is not None and not force_api_mode: log.info("Push initiated flow") diff, comparison_ran = create_scm_scan( core, config, scm_event_type, scan_paths=scan_paths, params=params, no_change=should_skip_scan, base_paths=base_paths, explicit_files=scan_explicit_files, external_href=pr_context.url, ) if comparison_ran: comments = scm.get_comments_for_pr() # FIXME: this overwrites diff.new_alerts, which was previously populated by Core.create_issue_alerts if not config.disable_ignore: log.debug("Removing comment alerts") alerts_before = list(diff.new_alerts) diff.new_alerts = Comments.remove_alerts(comments, diff.new_alerts) ignored_alerts = [a for a in alerts_before if a not in diff.new_alerts] # Emit telemetry per-comment so each event carries the comment author. unprocessed_ignore = [ c for c in comments.get("ignore", []) if _is_unprocessed(c) ] if ignored_alerts and unprocessed_ignore: try: events = [] now = datetime.now(timezone.utc).isoformat() for c in unprocessed_ignore: single = {"ignore": [c]} c_ignore_all, c_ignore_commands = Comments.get_ignore_options(single) user = getattr(c, "user", None) or getattr(c, "author", None) or {} sender_name = user.get("login") or user.get("username", "") sender_id = str(user.get("id", "")) # Match this comment's targets to the actual ignored alerts matched_alerts = _match_ignored_alerts_for_telemetry( ignored_alerts, c_ignore_all, c_ignore_commands, ) shared_fields = { "event_kind": "user-action", "client_action": "ignore", "event_sender_created_at": now, "vcs_provider": integration_type, "owner": repository_owner(config.repository_url), "repo": config.repo, "pr_number": pr_number, "ignore_all": c_ignore_all, "sender_name": sender_name, "sender_id": sender_id, } if matched_alerts: for alert in matched_alerts: # Derive alert_action from the alert's resolved action flags if getattr(alert, "error", False): alert_action = "error" elif getattr(alert, "warn", False): alert_action = "warn" elif getattr(alert, "monitor", False): alert_action = "monitor" else: alert_action = "error" events.append({**shared_fields, "alert_action": alert_action, "event_id": str(uuid4()), "artifact_purl": alert.purl}) elif c_ignore_all: events.append({**shared_fields, "event_id": str(uuid4())}) if events: client.post_telemetry_events(org_slug, events) # Mark ignore comments as processed with +1 reaction if hasattr(scm, "handle_ignore_reactions"): scm.handle_ignore_reactions(comments) except Exception as e: log.warning(f"Failed to send ignore telemetry: {e}") else: log.info("Ignore commands disabled (--disable-ignore), all alerts will be reported") log.debug("Creating Dependency Overview Comment") overview_comment = Messages.dependency_overview_template(diff) log.debug("Creating Security Issues Comment") security_comment = Messages.security_comment_template(diff, config) update_old_security_comment = ( security_comment is None or security_comment == "" or (len(comments) != 0 and comments.get("security") is not None) ) update_old_overview_comment = ( overview_comment is None or overview_comment == "" or (len(comments) != 0 and comments.get("overview") is not None) ) new_security_comment = should_write_comment( config.disable_security_issue, len(diff.new_alerts) > 0, update_old_security_comment, ) if not new_security_comment: log.debug("Security issue comment disabled, or no alerts and none to update") has_dependency_changes = any(( diff.new_packages, diff.updated_packages, diff.removed_packages, diff.replaced_packages, )) new_overview_comment = should_write_comment( config.disable_overview, has_dependency_changes, update_old_overview_comment, ) if not new_overview_comment: log.debug("Dependency Overview comment disabled, or no packages and none to update") log.debug(f"Adding comments for {config.scm}") scm.add_socket_comments( security_comment, overview_comment, comments, new_security_comment, new_overview_comment ) output_handler.handle_output(diff) elif (config.enable_diff or force_diff_mode) and not force_api_mode: # New logic: --enable-diff or force_diff_mode (from --ignore-commit-files in git repos) forces diff mode log.info("Diff mode enabled without SCM integration") diff = core.create_new_diff( scan_paths, params, no_change=should_skip_scan, save_files_list_path=config.save_submitted_files_list, save_manifest_tar_path=config.save_manifest_tar, base_paths=base_paths, explicit_files=scan_explicit_files, external_href=pr_context.url, ) output_handler.handle_output(diff) elif (config.enable_diff or force_diff_mode) and force_api_mode: # User requested diff mode but no manifest files were detected - this should not happen with new logic # but keeping as a safety net log.warning("--enable-diff was specified but no supported manifest files were detected in the changed files. Falling back to full scan mode.") _log_scan_mode_fallback( "diff", "full", "no-supported-manifest-in-changed-files", ) |
| You are visitor 01402503 — GeezerVison does not store the pages you look up. Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required. |