add --base-scan-id / --base-commit-sha diff baseline overrides

Signed-off-by: lelia <2418071+lelia@users.noreply.github.com>

Author: lelia

socketsecurity/config.py

@@ -119,6 +119,8 @@ class CliConfig:
     scm: str = "api"
     sbom_file: Optional[str] = None
     commit_sha: str = ""
+    base_scan_id: Optional[str] = None
+    base_commit_sha: Optional[str] = None
     generate_license: bool = False
     enable_debug: bool = False
     allow_unverified: bool = False
@@ -265,6 +267,8 @@ def from_args(cls, args_list: Optional[List[str]] = None) -> 'CliConfig':
             'scm': args.scm,
             'sbom_file': args.sbom_file,
             'commit_sha': args.commit_sha,
+            'base_scan_id': args.base_scan_id,
+            'base_commit_sha': args.base_commit_sha,
             'generate_license': args.generate_license,
             'enable_debug': args.enable_debug,
             'enable_diff': args.enable_diff,
@@ -406,6 +410,12 @@ def from_args(cls, args_list: Optional[List[str]] = None) -> 'CliConfig':
             logging.error("--workspace-name requires --sub-path to be specified")
             exit(1)
 
+        # argparse only enforces the mutually exclusive group for real CLI args;
+        # this also catches both values arriving via a --config file.
+        if args.base_scan_id and args.base_commit_sha:
+            logging.error("--base-scan-id and --base-commit-sha are mutually exclusive")
+            exit(1)
+
         if args.sarif_scope == "full" and not args.reach:
             logging.error("--sarif-scope full requires --reach to be specified")
             exit(1)
@@ -560,6 +570,25 @@ def create_argument_parser() -> argparse.ArgumentParser:
         help="Committer for the commit (comma separated)",
         nargs="*"
     )
+    base_scan_group = pr_group.add_mutually_exclusive_group()
+    base_scan_group.add_argument(
+        "--base-scan-id",
+        dest="base_scan_id",
+        metavar="<id>",
+        default=None,
+        help="Full scan ID to diff the new scan against, overriding the repository's "
+             "head scan as the baseline. Mutually exclusive with --base-commit-sha."
+    )
+    base_scan_group.add_argument(
+        "--base-commit-sha",
+        dest="base_commit_sha",
+        metavar="<sha>",
+        default=None,
+        help="Commit SHA to diff the new scan against, overriding the repository's head "
+             "scan as the baseline. The most recent full scan matching this commit (e.g. "
+             "the merge base from 'git merge-base origin/main HEAD') is used; the CLI "
+             "errors if no scan exists for it. Mutually exclusive with --base-scan-id."
+    )
 
     # Path and File options
     path_group = parser.add_argument_group('Path and File')

socketsecurity/core/__init__.py

@@ -1154,6 +1154,94 @@ def get_head_scan_for_repo(self, repo_slug: str) -> str:
         repo_info = self.get_repo_info(repo_slug)
         return repo_info.head_full_scan_id if repo_info.head_full_scan_id else None
 
+    def get_full_scan_id_by_commit(
+            self,
+            repo_slug: str,
+            commit_sha: str,
+            workspace: Optional[str] = None,
+            scan_type: Optional[str] = None
+    ) -> Optional[str]:
+        """
+        Finds the most recent full scan for a repository + commit SHA.
+
+        Used by --base-commit-sha to resolve the diff baseline (e.g. the merge-base
+        commit of a PR). When the same commit was scanned more than once, the newest
+        scan wins.
+
+        Args:
+            repo_slug: Repository slug the scan belongs to
+            commit_sha: Commit SHA the scan was created from
+            workspace: Socket workspace the scan belongs to, if any
+            scan_type: Socket scan type to match, if any
+
+        Returns:
+            Full scan ID if one exists for that commit, None otherwise
+        """
+        query_params = {
+            "repo": repo_slug,
+            "commit_hash": commit_sha,
+            "sort": "created_at",
+            "direction": "desc",
+            "per_page": 1,
+        }
+        if workspace:
+            query_params["workspace"] = workspace
+        if scan_type:
+            query_params["scan_type"] = scan_type
+
+        response = self.sdk.fullscans.get(
+            self.config.org_slug,
+            query_params,
+        )
+        results = response.get("results") if isinstance(response, dict) else None
+        if not results:
+            return None
+        return results[0].get("id")
+
+    def resolve_base_full_scan_id(self, params: FullScanParams) -> Optional[str]:
+        """
+        Resolves the baseline full scan ID to diff a new scan against.
+
+        Priority: --base-scan-id (used verbatim), then --base-commit-sha (newest
+        full scan for that commit), then the repository's current head scan. A
+        --base-commit-sha with no matching full scan is a hard error rather than a
+        silent fallback to the head scan, because diffing against the wrong
+        baseline silently misreports which alerts a PR introduces.
+
+        Returns:
+            Full scan ID to use as the diff baseline, or None when the repository
+            has no head scan yet (caller creates an empty baseline scan).
+        """
+        if self.cli_config and self.cli_config.base_scan_id:
+            log.info(f"Using full scan {self.cli_config.base_scan_id} as diff baseline (--base-scan-id)")
+            return self.cli_config.base_scan_id
+
+        if self.cli_config and self.cli_config.base_commit_sha:
+            commit_sha = self.cli_config.base_commit_sha
+            scan_id = self.get_full_scan_id_by_commit(
+                params.repo,
+                commit_sha,
+                workspace=params.workspace,
+                scan_type=params.scan_type,
+            )
+            if scan_id is None:
+                log.error(
+                    f"No full scan found for commit {commit_sha} in repo {params.repo} "
+                    "(--base-commit-sha). Ensure a scan was created for that commit "
+                    "(e.g. the CLI runs on default-branch pushes), or pass "
+                    "--base-scan-id instead."
+                )
+                if self.cli_config.disable_blocking:
+                    sys.exit(0)
+                sys.exit(self.cli_config.exit_code_on_api_error)
+            log.info(f"Using full scan {scan_id} (commit {commit_sha}) as diff baseline (--base-commit-sha)")
+            return scan_id
+
+        try:
+            return self.get_head_scan_for_repo(params.repo)
+        except APIResourceNotFound:
+            return None
+
     @staticmethod
     def update_package_values(pkg: Package) -> Package:
         pkg.purl = f"{pkg.name}@{pkg.version}"
@@ -1242,8 +1330,8 @@ def get_added_and_removed_packages(
                     OVERWRITES whatever the diff embedded, before anything reads it.
                 Either way the embedded license payload is dead weight, and on
                 large dependency trees it inflated the diff response past ~2.3MB
-                and truncated it mid-string, crashing ``response.json()``
-                (CE-224, customer: Tremendous). Defaulting to ``False`` keeps the
+                and truncated it mid-string, crashing ``response.json()``.
+                Defaulting to ``False`` keeps the
                 diff lean with zero change to any output artifact. The parameter
                 is retained as an explicit override seam, not wired to the
                 ``--exclude-license-details`` user flag (which still governs the
@@ -1388,11 +1476,9 @@ def create_new_diff(
             log.info("No supported manifest files found - creating empty scan for diff comparison")
             scan_files = Core.empty_head_scan_file()
 
-        try:
-            # Get head scan ID
-            head_full_scan_id = self.get_head_scan_for_repo(params.repo)
-        except APIResourceNotFound:
-            head_full_scan_id = None
+        # Resolve the baseline scan: --base-scan-id / --base-commit-sha override
+        # the repository's head scan (None only when the repo has no head scan yet).
+        head_full_scan_id = self.resolve_base_full_scan_id(params)
 
         # If no head scan exists, create an empty baseline scan
         if head_full_scan_id is None:
@@ -1475,7 +1561,7 @@ def create_new_diff(
         # (the --exclude-license-details user flag) into the diff request. The
         # diff path never consumes embedded license data (see
         # get_added_and_removed_packages docstring), so requesting it only bloats
-        # the response and risks the CE-224 truncation crash on large repos. The
+        # the response and risks the truncation crash on large repos. The
         # user flag still controls the dashboard report URL below; it just no
         # longer gates this internal diff payload.
         (

tests/core/test_sdk_methods.py

@@ -1,6 +1,7 @@
 import pytest
 from socketdev.fullscans import FullScanParams
 
+from socketsecurity.config import CliConfig
 from socketsecurity.core import Core
 from socketsecurity.core.socket_config import SocketConfig
 
@@ -10,6 +11,23 @@ def core(mock_sdk_with_responses):
     config = SocketConfig(api_key="test_key")
     return Core(config=config, sdk=mock_sdk_with_responses)
 
+
+def make_cli_config(*extra_args):
+    return CliConfig.from_args(["--api-token", "test-token", "--repo", "test", *extra_args])
+
+
+def make_full_scan_params(**overrides):
+    values = {
+        "repo": "test",
+        "branch": "main",
+        "commit_hash": "head123",
+        "scan_type": "socket",
+        "workspace": None,
+    }
+    values.update(overrides)
+    return FullScanParams(**values)
+
+
 def test_get_repo_info(core, mock_sdk_with_responses):
     """Test getting repository information"""
     repo_info = core.get_repo_info("test")
@@ -44,6 +62,119 @@ def test_get_head_scan_for_repo_no_head(core, mock_sdk_with_responses):
     head_scan_id = core.get_head_scan_for_repo("no-head")
     assert head_scan_id is None
 
+def test_get_full_scan_id_by_commit(core, mock_sdk_with_responses):
+    """Looks up the newest full scan for a repo + commit via the list endpoint"""
+    mock_sdk_with_responses.fullscans.get.return_value = {
+        "results": [{"id": "base-scan-id", "commit_hash": "abc123"}],
+        "nextPage": None,
+    }
+
+    scan_id = core.get_full_scan_id_by_commit("test", "abc123")
+
+    assert scan_id == "base-scan-id"
+    mock_sdk_with_responses.fullscans.get.assert_called_once_with(
+        core.config.org_slug,
+        {
+            "repo": "test",
+            "commit_hash": "abc123",
+            "sort": "created_at",
+            "direction": "desc",
+            "per_page": 1,
+        },
+    )
+
+
+def test_get_full_scan_id_by_commit_scopes_to_workspace_and_scan_type(core, mock_sdk_with_responses):
+    """Looks up a baseline scan from the same workspace and scan type as the new scan"""
+    mock_sdk_with_responses.fullscans.get.return_value = {
+        "results": [{"id": "workspace-reach-base", "commit_hash": "abc123"}],
+        "nextPage": None,
+    }
+
+    scan_id = core.get_full_scan_id_by_commit(
+        "test",
+        "abc123",
+        workspace="customer-a",
+        scan_type="socket_tier1",
+    )
+
+    assert scan_id == "workspace-reach-base"
+    mock_sdk_with_responses.fullscans.get.assert_called_once_with(
+        core.config.org_slug,
+        {
+            "repo": "test",
+            "commit_hash": "abc123",
+            "sort": "created_at",
+            "direction": "desc",
+            "per_page": 1,
+            "workspace": "customer-a",
+            "scan_type": "socket_tier1",
+        },
+    )
+
+
+def test_get_full_scan_id_by_commit_not_found(core, mock_sdk_with_responses):
+    """No scan for the commit returns None (empty results and SDK error dict)"""
+    mock_sdk_with_responses.fullscans.get.return_value = {"results": [], "nextPage": None}
+    assert core.get_full_scan_id_by_commit("test", "abc123") is None
+
+    mock_sdk_with_responses.fullscans.get.return_value = {}
+    assert core.get_full_scan_id_by_commit("test", "abc123") is None
+
+def test_resolve_base_full_scan_id_defaults_to_head_scan(core):
+    """Without base overrides the repository head scan is the baseline"""
+    assert core.resolve_base_full_scan_id(make_full_scan_params()) == "head"
+
+def test_resolve_base_full_scan_id_uses_base_scan_id(core):
+    """--base-scan-id is used verbatim, without touching the repo endpoint"""
+    core.cli_config = make_cli_config("--base-scan-id", "explicit-base")
+
+    assert core.resolve_base_full_scan_id(make_full_scan_params()) == "explicit-base"
+    core.sdk.repos.repo.assert_not_called()
+
+def test_resolve_base_full_scan_id_uses_base_commit_sha(core):
+    """--base-commit-sha resolves through the full-scans list endpoint"""
+    core.cli_config = make_cli_config("--base-commit-sha", "abc123")
+    core.sdk.fullscans.get.return_value = {
+        "results": [{"id": "merge-base-scan"}],
+        "nextPage": None,
+    }
+
+    params = make_full_scan_params(workspace="customer-a", scan_type="socket_tier1")
+
+    assert core.resolve_base_full_scan_id(params) == "merge-base-scan"
+    core.sdk.repos.repo.assert_not_called()
+    core.sdk.fullscans.get.assert_called_once_with(
+        core.config.org_slug,
+        {
+            "repo": "test",
+            "commit_hash": "abc123",
+            "sort": "created_at",
+            "direction": "desc",
+            "per_page": 1,
+            "workspace": "customer-a",
+            "scan_type": "socket_tier1",
+        },
+    )
+
+def test_resolve_base_full_scan_id_commit_sha_not_found_exits(core):
+    """A --base-commit-sha with no scan is a hard error (exit_code_on_api_error)"""
+    core.cli_config = make_cli_config("--base-commit-sha", "abc123")
+    core.sdk.fullscans.get.return_value = {"results": [], "nextPage": None}
+
+    with pytest.raises(SystemExit) as exc_info:
+        core.resolve_base_full_scan_id(make_full_scan_params())
+    assert exc_info.value.code == core.cli_config.exit_code_on_api_error
+
+def test_resolve_base_full_scan_id_commit_sha_not_found_disable_blocking(core):
+    """--disable-blocking keeps the missing-base error from failing the build"""
+    core.cli_config = make_cli_config("--base-commit-sha", "abc123", "--disable-blocking")
+    core.sdk.fullscans.get.return_value = {"results": [], "nextPage": None}
+
+    with pytest.raises(SystemExit) as exc_info:
+        core.resolve_base_full_scan_id(make_full_scan_params())
+    assert exc_info.value.code == 0
+
 def test_get_full_scan(core, mock_sdk_with_responses, head_scan_metadata, head_scan_stream):
     """Test getting an existing full scan"""
     full_scan = core.get_full_scan("head")
@@ -98,7 +229,7 @@ def test_get_added_and_removed_packages(core):
     # Verify SDK was called correctly.
     # include_license_details defaults to "false": the diff path never consumes
     # embedded license data (license artifacts come from the PURL endpoint), so
-    # requesting it only bloats the response and risks the CE-224 truncation
+    # requesting it only bloats the response and risks the truncation
     # crash on large repos.
     core.sdk.fullscans.stream_diff.assert_called_once_with(
         core.config.org_slug,