Merge pull request #1109 from GitGuardian/alina/add-source-uuid-secret-scan

feat: add new parameter source_uuid to secret scan

Author: alina-tuholukova-gg

changelog.d/20250625_161705_alina.tuholukova.md

@@ -0,0 +1,4 @@
+### Changed
+
+- `ggshield secret scan` now provides an `--source-uuid` option. When this option is set, it will create the incidents on the GIM
+  dashboard on the corresponding source. Note that the token should have the scope `scan:create-incidents`.

ggshield/cmd/secret/scan/secret_scan_common_options.py

@@ -1,3 +1,4 @@
+import uuid
 from typing import Callable, List, Optional
 
 import click
@@ -149,6 +150,26 @@ def _banlist_detectors_callback(
 )
 
 
+def _source_uuid_callback(
+    ctx: click.Context, param: click.Parameter, value: Optional[str]
+) -> Optional[str]:
+    if value is not None:
+        try:
+            uuid.UUID(value)
+        except ValueError:
+            raise click.BadParameter("source-uuid must be a valid UUID")
+    return create_config_callback("secret", "source_uuid")(ctx, param, value)
+
+
+_source_uuid_option = click.option(
+    "--source-uuid",
+    help="Identifier of the custom source in GitGuardian. If used, incidents will be created and visible on the "
+    "dashboard. Requires the 'scan:create-incidents' scope.",
+    callback=_source_uuid_callback,
+    default=None,
+)
+
+
 def add_secret_scan_common_options() -> Callable[[AnyFunction], AnyFunction]:
     def decorator(cmd: AnyFunction) -> AnyFunction:
         add_common_options()(cmd)
@@ -163,6 +184,7 @@ def decorator(cmd: AnyFunction) -> AnyFunction:
         _with_incident_details_option(cmd)
         instance_option(cmd)
         _all_secrets(cmd)
+        _source_uuid_option(cmd)
         return cmd
 
     return decorator

ggshield/core/client.py

@@ -4,12 +4,14 @@
 import requests
 import urllib3
 from pygitguardian import GGClient, GGClientCallbacks
+from pygitguardian.models import APITokensResponse, Detail, TokenScope
 from requests import Session
 
 from .config import Config
 from .constants import DEFAULT_INSTANCE_URL
 from .errors import (
     APIKeyCheckError,
+    MissingScopesError,
     ServiceUnavailableError,
     UnexpectedError,
     UnknownInstanceError,
@@ -87,10 +89,12 @@ def create_session(allow_self_signed: bool = False) -> Session:
     return session
 
 
-def check_client_api_key(client: GGClient) -> None:
+def check_client_api_key(client: GGClient, required_scopes: set[TokenScope]) -> None:
     """
     Raises APIKeyCheckError if the API key configured for the client is not usable
     (either it is invalid or unset). Raises UnexpectedError if the API is down.
+
+    If required_scopes is not empty, also checks that the API key has the required scopes.
     """
     try:
         response = client.read_metadata()
@@ -102,9 +106,8 @@ def check_client_api_key(client: GGClient) -> None:
 
     if response is None:
         # None means success
-        return
-
-    if response.status_code == 401:
+        pass
+    elif response.status_code == 401:
         raise APIKeyCheckError(client.base_uri, "Invalid API key.")
     elif response.status_code == 404:
         raise UnexpectedError(
@@ -118,3 +121,18 @@ def check_client_api_key(client: GGClient) -> None:
         raise UnexpectedError(
             f"GitGuardian server is not responding as expected.\nDetails: {response.detail}"
         )
+
+    # Check token scopes if required_scopes is not empty
+    if required_scopes:
+        response = client.api_tokens()
+
+        if not isinstance(response, (Detail, APITokensResponse)):
+            raise UnexpectedError("Unexpected api_tokens response")
+        elif isinstance(response, Detail):
+            raise UnexpectedError(response.detail)
+
+        missing_scopes = required_scopes - set(
+            TokenScope(scope) for scope in response.scopes
+        )
+        if missing_scopes:
+            raise MissingScopesError(list(missing_scopes))

ggshield/core/config/user_config.py

@@ -3,6 +3,7 @@
 from dataclasses import field
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Set, Tuple
+from uuid import UUID
 
 import marshmallow_dataclass
 from marshmallow import ValidationError
@@ -44,6 +45,7 @@ class SecretConfig(FilteredConfig):
     # if configuration key is left unset the dashboard's remediation message is used.
     all_secrets: bool = False
     prereceive_remediation_message: str = ""
+    source_uuid: Optional[UUID] = None
 
     def add_ignored_match(self, secret: IgnoredMatch) -> None:
         """
@@ -70,6 +72,7 @@ def dump_for_monitoring(self) -> str:
                     self.prereceive_remediation_message
                 ),
                 "all_secrets": self.all_secrets,
+                "source_uuid": self.source_uuid,
             }
         )
 

ggshield/verticals/auth/oauth.py

@@ -327,7 +327,7 @@ def check_existing_token(self) -> bool:
         # Check our API key is valid, if not forget it
         client = create_client_from_config(self.config)
         try:
-            check_client_api_key(client)
+            check_client_api_key(client, set())
         except APIKeyCheckError:
             # Forget the account
             logger.debug(

ggshield/verticals/secret/repo.py

@@ -20,6 +20,9 @@
 from ggshield.core.text_utils import STYLE, format_text
 from ggshield.utils.git_shell import get_list_commit_SHA, is_git_dir
 from ggshield.utils.os import cd
+from ggshield.verticals.secret.secret_scanner import (
+    get_required_token_scopes_from_config,
+)
 
 from .output import SecretOutputHandler
 from .secret_scan_collection import Results, SecretScanCollection
@@ -176,7 +179,7 @@ def scan_commit_range(
     :param commit_list: List of commits sha to scan
     :param verbose: Display successful scan's message
     """
-    check_client_api_key(client)
+    check_client_api_key(client, get_required_token_scopes_from_config(secret_config))
     max_documents = client.secret_scan_preferences.maximum_documents_per_scan
 
     with ui.create_progress(len(commit_list) + int(include_staged)) as progress:

ggshield/verticals/secret/secret_scanner.py

@@ -7,14 +7,14 @@
 from typing import Dict, Iterable, List, Optional, Union
 
 from pygitguardian import GGClient
-from pygitguardian.models import APITokensResponse, Detail, MultiScanResult, TokenScope
+from pygitguardian.models import Detail, MultiScanResult, TokenScope
 
 from ggshield.core import ui
 from ggshield.core.cache import Cache
 from ggshield.core.client import check_client_api_key
 from ggshield.core.config.user_config import SecretConfig
 from ggshield.core.constants import MAX_WORKERS
-from ggshield.core.errors import MissingScopesError, UnexpectedError, handle_api_error
+from ggshield.core.errors import handle_api_error
 from ggshield.core.scan import DecodeError, ScanContext, Scannable
 from ggshield.core.scanner_ui.scanner_ui import ScannerUI
 from ggshield.core.text_utils import pluralize
@@ -50,7 +50,8 @@ def __init__(
         check_api_key: Optional[bool] = True,
     ):
         if check_api_key:
-            check_client_api_key(client)
+            scopes = get_required_token_scopes_from_config(secret_config)
+            check_client_api_key(client, scopes)
 
         self.client = client
         self.cache = cache
@@ -60,16 +61,6 @@ def __init__(
 
         self.command_id = scan_context.command_id
 
-        if secret_config.with_incident_details:
-            response = self.client.api_tokens()
-
-            if not isinstance(response, (Detail, APITokensResponse)):
-                raise UnexpectedError("Unexpected api_tokens response")
-            elif isinstance(response, Detail):
-                raise UnexpectedError(response.detail)
-            if TokenScope.INCIDENTS_READ not in response.scopes:
-                raise MissingScopesError([TokenScope.INCIDENTS_READ])
-
     def scan(
         self,
         files: Iterable[Scannable],
@@ -109,12 +100,21 @@ def _scan_chunk(
             for x in chunk
         ]
 
-        return executor.submit(
-            self.client.multi_content_scan,
-            documents,
-            self.headers,
-            all_secrets=True,
-        )
+        # Use scan_and_create_incidents if source_uuid is provided, otherwise use multi_content_scan
+        if self.secret_config.source_uuid:
+            return executor.submit(
+                self.client.scan_and_create_incidents,
+                documents,
+                self.secret_config.source_uuid,
+                extra_headers=self.headers,
+            )
+        else:
+            return executor.submit(
+                self.client.multi_content_scan,
+                documents,
+                self.headers,
+                all_secrets=True,
+            )
 
     def _start_scans(
         self,
@@ -230,6 +230,11 @@ def handle_scan_chunk_error(detail: Detail, chunk: List[Scannable]) -> None:
     handle_api_error(detail)
     details = None
 
+    # Handle source_uuid not found error specifically
+    if "Source not found" in detail.detail:
+        ui.display_error("The provided source was not found in GitGuardian.")
+        return
+
     ui.display_error("Scanning failed. Results may be incomplete.")
     try:
         # try to load as list of dicts to get per file details
@@ -251,5 +256,24 @@ def handle_scan_chunk_error(detail: Detail, chunk: List[Scannable]) -> None:
         # if the details had a request error
         filenames = "\n".join(f"- {file.filename}" for file in chunk)
         ui.display_error(f"The following chunk is affected:\n{filenames}")
-
         ui.display_error(str(detail))
+
+
+def get_required_token_scopes_from_config(
+    secret_config: SecretConfig,
+) -> set[TokenScope]:
+    """
+    Get the required token scopes based on the secret configuration.
+
+    Args:
+        secret_config: The secret configuration to analyze
+
+    Returns:
+        A set of TokenScope values required for the given configuration
+    """
+    scopes = {TokenScope.SCAN}
+    if secret_config.with_incident_details:
+        scopes.add(TokenScope.INCIDENTS_READ)
+    if secret_config.source_uuid:
+        scopes.add(TokenScope.SCAN_CREATE_INCIDENTS)
+    return scopes

pdm.lock

@@ -41,7 +41,7 @@ dependencies = [
     "marshmallow~=3.18.0",
     "marshmallow-dataclass~=8.5.8",
     "oauthlib~=3.2.1",
-    "pygitguardian~=1.23.0",
+    "pygitguardian @ git+https://github.com/GitGuardian/py-gitguardian.git",
     "pyjwt~=2.6.0",
     "python-dotenv~=0.21.0",
     "pyyaml~=6.0.1",