Merge pull request #56 from Yelp/fix_audit_crash_on_missing_file

Fix --audit Crash on Missing File

detect_secrets/core/audit.py

@@ -1,6 +1,7 @@
 from __future__ import print_function
 
 import json
+import os
 import subprocess
 import sys
 import textwrap
@@ -24,13 +25,16 @@ def audit_baseline(baseline_filename):
     if not original_baseline:
         return
 
+    files_removed = _remove_nonexistent_files_from_baseline(original_baseline)
+
     current_secret_index = 0
     results = defaultdict(list)
     for filename, secret, total in _secret_generator(original_baseline):
         _clear_screen()
 
         if 'is_secret' not in secret:
             current_secret_index += 1
+
             try:
                 _print_context(
                     filename,
@@ -54,7 +58,7 @@ def audit_baseline(baseline_filename):
         _handle_user_decision(decision, secret)
         results[filename].append(secret)
 
-    if current_secret_index == 0:
+    if current_secret_index == 0 and not files_removed:
         print('Nothing to audit!')
         return
 
@@ -66,6 +70,45 @@ def audit_baseline(baseline_filename):
     _save_baseline_to_file(baseline_filename, original_baseline)
 
 
+def _get_baseline_from_file(filename):  # pragma: no cover
+    try:
+        with open(filename) as f:
+            return json.loads(f.read())
+    except (IOError, json.decoder.JSONDecodeError):
+        print('Not a valid baseline file!', file=sys.stderr)
+        return
+
+
+def _remove_nonexistent_files_from_baseline(baseline):
+    files_removed = False
+    for filename in baseline['results'].copy():
+        if not os.path.exists(filename):
+            del baseline['results'][filename]
+            files_removed = True
+    return files_removed
+
+
+def _secret_generator(baseline):
+    """Generates secrets to audit, from the baseline"""
+    num_secrets_to_parse = sum(
+        map(
+            lambda filename: len(
+                list(
+                    filter(
+                        lambda secret: 'is_secret' not in secret,
+                        baseline['results'][filename],
+                    ),
+                ),
+            ),
+            baseline['results'],
+        ),
+    )
+
+    for filename, secrets in baseline['results'].items():
+        for secret in secrets:
+            yield filename, secret, num_secrets_to_parse
+
+
 def _clear_screen():    # pragma: no cover
     subprocess.call(['clear'])
 
@@ -131,22 +174,40 @@ def _print_context(filename, secret, count, total, plugin_settings):   # pragma:
         raise error_obj
 
 
+def _get_user_decision(prompt_secret_decision=True):
+    """
+    :type prompt_secret_decision: bool
+    :param prompt_secret_decision: if False, won't ask to label secret.
+    """
+    allowable_user_input = ['s', 'q']
+    if prompt_secret_decision:
+        allowable_user_input.extend(['y', 'n'])
+
+    user_input = None
+    while user_input not in allowable_user_input:
+        if user_input:
+            print('Invalid input.')
+
+        if 'y' in allowable_user_input:
+            user_input_string = 'Is this a valid secret? (y)es, (n)o, '
+        else:
+            user_input_string = 'What would you like to do? '
+        user_input_string += '(s)kip, (q)uit: '
+
+        user_input = input(user_input_string)
+        if user_input:
+            user_input = user_input[0].lower()
+
+    return user_input
+
+
 def _handle_user_decision(decision, secret):
     if decision == 'y':
         secret['is_secret'] = True
     elif decision == 'n':
         secret['is_secret'] = False
 
 
-def _get_baseline_from_file(filename):  # pragma: no cover
-    try:
-        with open(filename) as f:
-            return json.loads(f.read())
-    except (IOError, json.decoder.JSONDecodeError):
-        print('Not a valid baseline file!', file=sys.stderr)
-        return
-
-
 def _save_baseline_to_file(filename, data):  # pragma: no cover
     with open(filename, 'w') as f:
         f.write(json.dumps(
@@ -156,27 +217,6 @@ def _save_baseline_to_file(filename, data):  # pragma: no cover
         ))
 
 
-def _secret_generator(baseline):
-    """Generates secrets to audit, from the baseline"""
-    num_secrets_to_parse = sum(
-        map(
-            lambda filename: len(
-                list(
-                    filter(
-                        lambda secret: 'is_secret' not in secret,
-                        baseline['results'][filename],
-                    ),
-                ),
-            ),
-            baseline['results'],
-        ),
-    )
-
-    for filename, secrets in baseline['results'].items():
-        for secret in secrets:
-            yield filename, secret, num_secrets_to_parse
-
-
 def _get_secret_with_context(
     filename,
     secret,
@@ -317,30 +357,3 @@ def _raw_secret_generator(plugin, secret_line):
         with plugin.non_quoted_string_regex(strict=False):
             for raw_secret in plugin.secret_generator(secret_line):
                 yield raw_secret
-
-
-def _get_user_decision(prompt_secret_decision=True):
-    """
-    :type prompt_secret_decision: bool
-    :param prompt_secret_decision: if False, won't ask to label secret.
-    """
-    allowable_user_input = ['s', 'q']
-    if prompt_secret_decision:
-        allowable_user_input.extend(['y', 'n'])
-
-    user_input = None
-    while user_input not in allowable_user_input:
-        if user_input:
-            print('Invalid input.')
-
-        if 'y' in allowable_user_input:
-            user_input_string = 'Is this a valid secret? (y)es, (n)o, '
-        else:
-            user_input_string = 'What would you like to do? '
-        user_input_string += '(s)kip, (q)uit: '
-
-        user_input = input(user_input_string)
-        if user_input:
-            user_input = user_input[0].lower()
-
-    return user_input

detect_secrets/core/baseline.py

@@ -160,7 +160,6 @@ def merge_baseline(old_baseline, new_baseline):
     :type new_baseline: dict
     :param new_baseline: most recent scan
     """
-    # TODO: merge_results does not take into account `is_secret`
     new_baseline['results'] = merge_results(
         old_baseline['results'],
         new_baseline['results'],

tests/core/audit_test.py

@@ -101,7 +101,10 @@ def test_leapfrog_decision(self, mock_printer):
 
     @contextmanager
     def run_logic(self, inputs, modified_baseline=None, input_baseline=None):
-        with self.mock_env(inputs, baseline=input_baseline) as m:
+        with self.mock_env(
+            inputs,
+            baseline=input_baseline,
+        ) as m:
             audit.audit_baseline('will_be_mocked')
 
             if not modified_baseline:
@@ -123,7 +126,7 @@ def mock_env(self, user_inputs=None, baseline=None):
             '_get_baseline_from_file',
             return_value=baseline,
         ), mock.patch.object(
-            # We mock this, because we don't really care about clearing
+            # We mock this because we don't really care about clearing
             # screens for test cases.
             audit,
             '_clear_screen',
@@ -134,7 +137,12 @@ def mock_env(self, user_inputs=None, baseline=None):
         ), mock_user_input(
             user_inputs,
         ), mock.patch.object(
-            # We mock this, so we don't need to do any file I/O.
+            # We mock this so we don't modify the baseline.
+            audit,
+            '_remove_nonexistent_files_from_baseline',
+            return_value=False,
+        ), mock.patch.object(
+            # We mock this so we don't need to do any file I/O.
             audit,
             '_save_baseline_to_file',
         ) as m: