Mmillet/spi 515 add all secrets option to ggshield secret scans

ggshield/cmd/secret/scan/secret_scan_common_options.py

@@ -140,6 +140,15 @@ def _banlist_detectors_callback(
 )
 
 
+_all_secrets = click.option(
+    "--all-secrets",
+    is_flag=True,
+    help=("Do not ignore any secret. Possible ignore-reason is shown as well."),
+    callback=create_config_callback("secret", "all_secrets"),
+    default=None,
+)
+
+
 def add_secret_scan_common_options() -> Callable[[AnyFunction], AnyFunction]:
     def decorator(cmd: AnyFunction) -> AnyFunction:
         add_common_options()(cmd)
@@ -153,6 +162,7 @@ def decorator(cmd: AnyFunction) -> AnyFunction:
         _banlist_detectors_option(cmd)
         _with_incident_details_option(cmd)
         instance_option(cmd)
+        _all_secrets(cmd)
         return cmd
 
     return decorator

ggshield/core/config/user_config.py

@@ -47,6 +47,7 @@ class SecretConfig(FilteredConfig):
     ignore_known_secrets: bool = False
     with_incident_details: bool = False
     # if configuration key is left unset the dashboard's remediation message is used.
+    all_secrets: bool = False
     prereceive_remediation_message: str = ""
 
     def add_ignored_match(self, secret: IgnoredMatch) -> None:
@@ -62,6 +63,11 @@ def add_ignored_match(self, secret: IgnoredMatch) -> None:
         self.ignored_matches.append(secret)
 
 
+SecretConfigTrackingSchema = marshmallow_dataclass.class_schema(SecretConfig)(
+    only=("show_secrets", "ignore_known_secrets", "with_incident_details")
+)
+
+
 def validate_policy_id(policy_id: str) -> bool:
     return bool(POLICY_ID_PATTERN.fullmatch(policy_id))
 

ggshield/verticals/secret/output/secret_text_output_handler.py

@@ -15,10 +15,9 @@
     pluralize,
     translate_validity,
 )
-from ggshield.verticals.secret.secret_scanner import IgnoreReason
 
 from ..extended_match import ExtendedMatch
-from ..secret_scan_collection import Result, SecretScanCollection
+from ..secret_scan_collection import IgnoreReason, Result, SecretScanCollection
 from .secret_output_handler import SecretOutputHandler
 
 

ggshield/verticals/secret/secret_scan_collection.py

@@ -1,35 +1,46 @@
+from collections import defaultdict
 from dataclasses import dataclass, field
 from enum import Enum
 from pathlib import Path
-from typing import (
-    Any,
-    Callable,
-    Dict,
-    Iterable,
-    List,
-    NamedTuple,
-    Optional,
-    Tuple,
-    Union,
-    cast,
-)
+from typing import Dict, Iterable, List, NamedTuple, Optional, Tuple, Union, cast
 
 from pygitguardian import GGClient
 from pygitguardian.models import Detail, Match, PolicyBreak, ScanResult, SecretIncident
 
+from ggshield.core.config.user_config import SecretConfig
 from ggshield.core.errors import UnexpectedError, handle_api_error
+from ggshield.core.filter import is_in_ignored_matches
 from ggshield.core.lines import Line, get_lines_from_content
-from ggshield.core.scan.scannable import Scannable
+from ggshield.core.scan import Scannable
 from ggshield.utils.git_shell import Filemode
 from ggshield.verticals.secret.extended_match import ExtendedMatch
 
 
-class IgnoreReason(Enum):
+class IgnoreReason(str, Enum):
     IGNORED_MATCH = "ignored_match"
     IGNORED_DETECTOR = "ignored_detector"
     KNOWN_SECRET = "known_secret"
+    BACKEND_EXCLUDED = "backend_excluded"
 
 
+def compute_ignore_reason(
+    policy_break: PolicyBreak, secret_config: SecretConfig
+) -> str | None:
+    """Computes the possible ignore reason associated with a PolicyBreak"""
+    ignore_reason = None
+    if policy_break.is_excluded:
+        ignore_reason = f"Excluded from backend ({policy_break.exclude_reason})"
+    elif is_in_ignored_matches(policy_break, secret_config.ignored_matches or []):
+        ignore_reason = IgnoreReason.IGNORED_MATCH
+    elif policy_break.break_type in secret_config.ignored_detectors:
+        ignore_reason = IgnoreReason.IGNORED_DETECTOR
+    elif secret_config.ignore_known_secrets and policy_break.known_secret:
+        ignore_reason = IgnoreReason.KNOWN_SECRET
+
+    return ignore_reason
+
+
+@dataclass
 class Result:
     """
     Return model for a scan which zips the information
@@ -41,28 +52,7 @@ class Result:
     path: Path
     url: str
     policy_breaks: List[PolicyBreak]
-    ignored_policy_breaks_count_by_reason: Dict[IgnoreReason, int]
-
-    def __init__(self, file: Scannable, scan: ScanResult):
-        self.filename = file.filename
-        self.filemode = file.filemode
-        self.path = file.path
-        self.url = file.url
-        self.policy_breaks = scan.policy_breaks
-        lines = get_lines_from_content(file.content, self.filemode)
-        self.enrich_matches(lines)
-        self.ignored_policy_breaks_count_by_reason = {}
-
-    def __eq__(self, other: Any) -> bool:
-        if not isinstance(other, Result):
-            return False
-        return (
-            self.filename == other.filename
-            and self.filemode == other.filemode
-            and self.path == other.path
-            and self.url == other.url
-            and self.policy_breaks == other.policy_breaks
-        )
+    ignored_policy_breaks_count_by_reason: Dict[str, int]
 
     @property
     def is_on_patch(self) -> bool:
@@ -89,21 +79,41 @@ def censor(self) -> None:
     def has_policy_breaks(self) -> bool:
         return len(self.policy_breaks) > 0
 
-    def apply_ignore_function(
-        self, reason: IgnoreReason, ignore_function: Callable[[PolicyBreak], bool]
+    @classmethod
+    def from_scan_result(
+        cls, file: Scannable, scan_result: ScanResult, secret_config: SecretConfig
     ):
-        assert (
-            reason not in self.ignored_policy_breaks_count_by_reason
-        ), f"Ignore was already computed for {IgnoreReason}"
+        """Creates a Result from a Scannable and a ScanResult.
+        - Removes ignored policy breaks
+        - replace matches by ExtendedMatches
+        """
+
         to_keep = []
-        ignored_count = 0
-        for policy_break in self.policy_breaks:
-            if ignore_function(policy_break):
-                ignored_count += 1
+        ignored_policy_breaks_count_by_reason = defaultdict(lambda: 0)
+        for policy_break in scan_result.policy_breaks:
+            ignore_reason = compute_ignore_reason(policy_break, secret_config)
+            if ignore_reason is not None:
+                if secret_config.all_secrets:
+                    policy_break.exclude_reason = ignore_reason
+                    policy_break.is_excluded = True
+                    to_keep.append(policy_break)
+                else:
+                    ignored_policy_breaks_count_by_reason[ignore_reason] += 1
             else:
                 to_keep.append(policy_break)
-        self.policy_breaks = to_keep
-        self.ignored_policy_breaks_count_by_reason[reason] = ignored_count
+
+        result = Result(
+            filename=file.filename,
+            filemode=file.filemode,
+            path=file.path,
+            url=file.url,
+            policy_breaks=to_keep,
+            ignored_policy_breaks_count_by_reason=ignored_policy_breaks_count_by_reason,
+        )
+
+        lines = get_lines_from_content(file.content, file.filemode)
+        result.enrich_matches(lines)
+        return result
 
 
 class Error(NamedTuple):

ggshield/verticals/secret/secret_scanner.py

@@ -12,15 +12,14 @@
 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.config.user_config import SecretConfig, SecretConfigTrackingSchema
 from ggshield.core.constants import MAX_WORKERS
 from ggshield.core.errors import MissingScopesError, UnexpectedError, handle_api_error
-from ggshield.core.filter import is_in_ignored_matches
 from ggshield.core.scan import DecodeError, ScanContext, Scannable
 from ggshield.core.text_utils import pluralize
 from ggshield.core.ui.scanner_ui import ScannerUI
 
-from .secret_scan_collection import Error, IgnoreReason, Result, Results
+from .secret_scan_collection import Error, Result, Results
 
 
 # GitGuardian API does not accept paths longer than this
@@ -56,9 +55,10 @@ def __init__(
         self.client = client
         self.cache = cache
         self.secret_config = secret_config
-        self.ignored_matches = secret_config.ignored_matches or []
-        self.ignored_detectors = secret_config.ignored_detectors
-        self.headers = scan_context.get_http_headers()
+        self.headers = scan_context.get_http_headers() | {
+            "scan_options": SecretConfigTrackingSchema.dumps(secret_config)
+        }
+
         self.command_id = scan_context.command_id
 
         if secret_config.with_incident_details:
@@ -114,7 +114,7 @@ def _scan_chunk(
             self.client.multi_content_scan,
             documents,
             self.headers,
-            ignore_known_secrets=True,
+            all_secrets=True,
         )
 
     def _start_scans(
@@ -214,28 +214,7 @@ def _collect_results(
 
             assert isinstance(scan, MultiScanResult)
             for file, scan_result in zip(chunk, scan.scan_results):
-                result = Result(
-                    file=file,
-                    scan=scan_result,
-                )
-                if not scan_result.has_policy_breaks:
-                    continue
-                result.apply_ignore_function(
-                    IgnoreReason.IGNORED_MATCH,
-                    lambda policy_break: is_in_ignored_matches(
-                        policy_break, self.ignored_matches
-                    ),
-                )
-                result.apply_ignore_function(
-                    IgnoreReason.IGNORED_DETECTOR,
-                    lambda policy_break: policy_break.break_type
-                    in self.ignored_detectors,
-                )
-                if self.secret_config.ignore_known_secrets:
-                    result.apply_ignore_function(
-                        IgnoreReason.KNOWN_SECRET,
-                        lambda policy_break: policy_break.known_secret,
-                    )
+                result = Result.from_scan_result(file, scan_result, self.secret_config)
                 for policy_break in result.policy_breaks:
                     self.cache.add_found_policy_break(policy_break, file.filename)
                 results.append(result)