Merge branch 'pre-v1-extract-pragmas' into pre-v1-launch

Author: Aaron Loo

detect_secrets/audit/common.py

@@ -10,6 +10,8 @@
 from ..core.potential_secret import PotentialSecret
 from ..exceptions import InvalidBaselineError
 from ..exceptions import SecretNotFoundOnSpecifiedLineError
+from ..util.inject import get_injectable_variables
+from ..util.inject import inject_variables_into_function
 
 
 def get_baseline_from_file(filename: str) -> Dict[str, Any]:
@@ -42,10 +44,18 @@ def get_raw_secret_from_file(secret: PotentialSecret) -> str:
     except IndexError:
         raise SecretNotFoundOnSpecifiedLineError(secret.line_number)
 
-    identified_secrets = plugin.analyze_line(
+    function = plugin.__class__.analyze_line
+    if not hasattr(function, 'injectable_variables'):
+        function.injectable_variables = set(get_injectable_variables(plugin.analyze_line))
+        function.path = f'{plugin.__class__.__name__}.analyze_line'
+
+    identified_secrets = inject_variables_into_function(
+        function,
+        self=plugin,
         filename=secret.filename,
         line=target_line,
         line_number=secret.line_number,     # TODO: this will be optional
+        enable_eager_search=True,
     )
 
     for identified_secret in identified_secrets:

detect_secrets/core/baseline.py

@@ -1,64 +1,31 @@
 import json
-import os
 import time
 from typing import Any
 from typing import Callable
 from typing import Dict
-from typing import List
 from typing import Union
 
 from . import upgrades
 from ..__version__ import VERSION
 from ..exceptions import UnableToReadBaselineError
 from ..settings import configure_settings_from_baseline
 from ..settings import get_settings
-from ..util import git
 from ..util.importlib import import_modules_from_package
-from ..util.path import get_relative_path_if_in_cwd
 from ..util.semver import Version
-from .log import log
+from .scan import get_files_to_scan
 from .secrets_collection import SecretsCollection
 
 
 def create(*paths: str, should_scan_all_files: bool = False) -> SecretsCollection:
     """Scans all the files recursively in path to initialize a baseline."""
     secrets = SecretsCollection()
 
-    for path in paths:
-        files = _get_files_to_scan(path, should_scan_all_files)
-        for filename in files:
-            secrets.scan_file(filename)
+    for filename in get_files_to_scan(*paths, should_scan_all_files=should_scan_all_files):
+        secrets.scan_file(filename)
 
     return secrets
 
 
-def _get_files_to_scan(root: str, should_scan_all_files: bool) -> List[str]:
-    output: List[str] = []
-    if not should_scan_all_files:
-        valid_paths = git.get_tracked_files(root)
-        if not valid_paths:
-            log.warning('Did not detect git repository. Try scanning all files instead.')
-            return output
-
-    for path_root, _, filenames in os.walk(root):
-        for filename in filenames:
-            path = get_relative_path_if_in_cwd(os.path.join(path_root, filename))
-            if not path:
-                # e.g. symbolic links may be pointing outside the root directory
-                continue
-
-            if (
-                not should_scan_all_files
-                and path not in valid_paths
-            ):
-                # Not a git-tracked file
-                continue
-
-            output.append(path)
-
-    return output
-
-
 def load(baseline: Dict[str, Any], filename: str) -> SecretsCollection:
     """
     With a given baseline file, load all settings and discovered secrets from it.

detect_secrets/core/scan.py

@@ -1,30 +1,64 @@
+import os
+import subprocess
 from functools import lru_cache
 from importlib import import_module
 from typing import Generator
 from typing import IO
+from typing import Iterable
 from typing import List
 from typing import Optional
 from typing import Tuple
 
 from . import plugins
+from ..filters.allowlist import is_line_allowlisted
 from ..settings import get_settings
 from ..transformers import get_transformers
 from ..transformers import ParsingError
 from ..types import SelfAwareCallable
+from ..util import git
 from ..util.code_snippet import get_code_snippet
 from ..util.inject import get_injectable_variables
 from ..util.inject import inject_variables_into_function
+from ..util.path import get_relative_path_if_in_cwd
 from .log import log
 from .plugins import Plugin
 from .potential_secret import PotentialSecret
 
 
+def get_files_to_scan(*paths: str, should_scan_all_files: bool) -> Generator[str, None, None]:
+    if not should_scan_all_files:
+        try:
+            valid_paths = git.get_tracked_files(git.get_root_directory())
+        except subprocess.CalledProcessError:
+            log.warning('Did not detect git repository. Try scanning all files instead.')
+            return []
+
+    for path in paths:
+        iterator = [(os.getcwd(), None, [path])] if os.path.isfile(path) else os.walk(path)
+        for path_root, _, filenames in iterator:
+            for filename in filenames:
+                path = get_relative_path_if_in_cwd(os.path.join(path_root, filename))
+                if not path:
+                    # e.g. symbolic links may be pointing outside the root directory
+                    continue
+
+                if (
+                    not should_scan_all_files
+                    and path not in valid_paths
+                ):
+                    # Not a git-tracked file
+                    continue
+
+                yield path
+
+
 def scan_line(line: str) -> Generator[PotentialSecret, None, None]:
     """Used for adhoc string scanning."""
     # Disable this, since it doesn't make sense to run this for adhoc usage.
     get_settings().disable_filters(
         'detect_secrets.filters.common.is_invalid_file',
     )
+    get_filters.cache_clear()
 
     for plugin in get_plugins():
         for secret in _scan_line(
@@ -60,69 +94,149 @@ def scan_file(filename: str) -> Generator[PotentialSecret, None, None]:
         return
 
     try:
-        with open(filename) as f:
-            log.info(f'Checking file: {filename}')
-
-            try:
-                lines = _get_transformed_file(f)
-                if not lines:
-                    lines = f.readlines()
-            except UnicodeDecodeError:
-                # We flat out ignore binary files.
-                return
-
-            has_secret = False
+        has_secret = False
+        for lines in _get_lines_from_file(filename):
             for secret in _process_line_based_plugins(
                 lines=list(enumerate(lines, 1)),
-                filename=f.name,
+                filename=filename,
             ):
                 has_secret = True
                 yield secret
 
             if has_secret:
-                return
-
-            # Only if no secrets, then use eager transformers
-            f.seek(0)
-            lines = _get_transformed_file(f, use_eager_transformers=True)
-            if not lines:
-                return
-
-            yield from _process_line_based_plugins(
-                lines=list(enumerate(lines, 1)),
-                filename=f.name,
-            )
+                break
     except IOError:
         log.warning(f'Unable to open file: {filename}')
+        return
 
 
 def scan_diff(diff: str) -> Generator[PotentialSecret, None, None]:
     """
     :raises: ImportError
     """
-    # Local imports, so that we don't need to require unidiff for versions of
-    # detect-secrets that don't use it.
-    from unidiff import PatchSet
+    if not get_plugins():   # pragma: no cover
+        log.warning('No plugins to scan with!')
+        return
+
+    for filename, lines in _get_lines_from_diff(diff):
+        yield from _process_line_based_plugins(lines, filename=filename)
 
+
+def scan_for_allowlisted_secrets_in_file(filename: str) -> Generator[PotentialSecret, None, None]:
+    """
+    Developers are able to add individual lines to the allowlist using
+    `detect_secrets.filters.allowlist.is_line_allowlisted`. However, there are
+    times when we want to verify that no *actual* secrets are added to the codebase
+    via this feature.
+
+    This scans specifically for these lines, and ignores everything else.
+    """
     if not get_plugins():   # pragma: no cover
         log.warning('No plugins to scan with!')
         return
 
+    if _filter_files(filename):
+        return
+
+    # NOTE: Unlike `scan_file`, we don't ever have to use eager file transfomers, since we already
+    # know which lines we want to scan.
+    try:
+        for lines in _get_lines_from_file(filename):
+            yield from _scan_for_allowlisted_secrets_in_lines(enumerate(lines, 1), filename)
+            break
+    except IOError:
+        log.warning(f'Unable to open file: {filename}')
+        return
+
+
+def scan_for_allowlisted_secrets_in_diff(diff: str) -> Generator[PotentialSecret, None, None]:
+    if not get_plugins():   # pragma: no cover
+        log.warning('No plugins to scan with!')
+        return
+
+    for filename, lines in _get_lines_from_diff(diff):
+        yield from _scan_for_allowlisted_secrets_in_lines(lines, filename)
+
+
+def _scan_for_allowlisted_secrets_in_lines(
+    lines: Iterable[Tuple[int, str]],
+    filename: str,
+) -> Generator[PotentialSecret, None, None]:
+    # We control the setting here because it makes more sense than requiring the caller
+    # to set this setting before calling this function.
+    get_settings().disable_filters('detect_secrets.filters.allowlist.is_line_allowlisted')
+    get_filters.cache_clear()
+
+    for line_number, line in lines:
+        line = line.rstrip()
+
+        if not is_line_allowlisted(filename, line):
+            continue
+
+        if any([
+            inject_variables_into_function(filter_fn, filename=filename, line=line)
+            for filter_fn in get_filters_with_parameter('line')
+        ]):
+            continue
+
+        for plugin in get_plugins():
+            yield from _scan_line(plugin, filename, line, line_number)
+
+
+def _get_lines_from_file(filename: str) -> Generator[List[str], None, None]:
+    """
+    This attempts to get lines in a given file. If no more lines are needed, the caller
+    is responsible for breaking out of this loop.
+
+    :raises: IOError
+    :raises: FileNotFoundError
+    """
+    with open(filename) as f:
+        log.info(f'Checking file: {filename}')
+
+        try:
+            lines = _get_transformed_file(f)
+            if not lines:
+                lines = f.readlines()
+        except UnicodeDecodeError:
+            # We flat out ignore binary files
+            return
+
+        yield lines
+
+        # If the above lines don't prove to be useful to the caller, try using eager transformers.
+        f.seek(0)
+        lines = _get_transformed_file(f, use_eager_transformers=True)
+        if not lines:
+            return
+
+        yield lines
+
+
+def _get_lines_from_diff(diff: str) -> Generator[Tuple[str, List[Tuple[int, str]]], None, None]:
+    """
+    :raises: ImportError
+    """
+    # Local imports, so that we don't need to require unidiff for versions of
+    # detect-secrets that don't use it.
+    from unidiff import PatchSet
+
     patch_set = PatchSet.from_string(diff)
     for patch_file in patch_set:
         filename = patch_file.path
         if _filter_files(filename):
             continue
 
-        lines = [
-            (line.target_line_no, line.value)
-            for chunk in patch_file
-            # target_lines refers to incoming (new) changes
-            for line in chunk.target_lines()
-            if line.is_added
-        ]
-
-        yield from _process_line_based_plugins(lines, filename=filename)
+        yield (
+            filename,
+            [
+                (line.target_line_no, line.value)
+                for chunk in patch_file
+                # target_lines refers to incoming (new) changes
+                for line in chunk.target_lines()
+                if line.is_added
+            ],
+        )
 
 
 def _filter_files(filename: str) -> bool:

detect_secrets/core/upgrades/v1_0.py

@@ -22,6 +22,9 @@ def _migrate_filters(baseline: Dict[str, Any]) -> None:
     contain the default filters used before this version upgrade.
     """
     baseline['filters_used'] = [
+        {
+            'path': 'detect_secrets.filters.allowlist.is_line_allowlisted',
+        },
         {
             'path': 'detect_secrets.filters.heuristic.is_sequential_string',
         },

detect_secrets/core/usage/scan.py

@@ -17,6 +17,7 @@ def add_scan_action(parent: argparse._SubParsersAction) -> argparse.ArgumentPars
     )
 
     _add_adhoc_scanning(parser)
+    _add_pragma_scanning(parser)
     _add_initialize_baseline_options(parser)
 
     return parser
@@ -33,6 +34,17 @@ def _add_adhoc_scanning(parser: argparse.ArgumentParser) -> None:
     )
 
 
+def _add_pragma_scanning(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        '--only-allowlisted',
+        action='store_true',
+        help=(
+            'Only scans the lines that are flagged with `allowlist secret`. This helps '
+            'verify that individual exceptions are indeed non-secrets.'
+        ),
+    )
+
+
 def _add_initialize_baseline_options(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         'path',