Add skeleton for audit results view

detect_secrets/core/audit.py

@@ -7,9 +7,11 @@
 import sys
 from builtins import input
 from collections import defaultdict
+from copy import deepcopy
 
 from ..plugins.common import initialize
 from ..plugins.common.filetype import determine_file_type
+from ..plugins.common.util import get_mapping_from_secret_type_to_class_name
 from ..plugins.high_entropy_strings import HighEntropyStringsPlugin
 from .baseline import merge_results
 from .bidirectional_iterator import BidirectionalIterator
@@ -32,6 +34,71 @@ class RedundantComparisonError(Exception):
     pass
 
 
+AUDIT_RESULT_TO_STRING = {
+    True: 'positive',
+    False: 'negative',
+    None: 'unknown',
+}
+
+EMPTY_PLUGIN_AUDIT_RESULT = {
+    'results': {
+        'positive': [],
+        'negative': [],
+        'unknown': [],
+    },
+    'config': {},
+}
+
+
+def determine_audit_results(baseline):
+    """
+    Given a baseline which has been audited, returns
+    a dictionary describing the results of each plugin in the following form:
+    {
+        "plugin_name1": {
+            "results": {
+                "positive": [list of secrets with is_secret: true caught by this plugin],
+                "negative": [list of secrets with is_secret: false caught by this plugin],
+                "unknown": [list of secrets with no is_secret entry caught by this plugin]
+            },
+            "config": {configuration used for the plugin}
+        },
+        ...
+    }
+    """
+    all_secrets = _secret_generator(baseline)
+
+    audit_results = defaultdict(lambda: deepcopy(EMPTY_PLUGIN_AUDIT_RESULT))
+    secret_type_mapping = get_mapping_from_secret_type_to_class_name()
+
+    for filename, secret in all_secrets:
+        plugin_name = secret_type_mapping[secret['type']]
+        audit_result = AUDIT_RESULT_TO_STRING[secret.get('is_secret')]
+
+        # TODO: figure out how to plaintext-ify
+        audit_results[plugin_name]['results'][audit_result].append(secret['hashed_secret'])
+
+    for plugin_config in baseline['plugins_used']:
+        plugin_name = plugin_config['name']
+        if plugin_name not in audit_results:
+            continue
+
+        audit_results[plugin_name]['config'].update(plugin_config)
+
+    # TODO: pull in git repo and commit info
+
+    return audit_results
+
+
+def print_audit_results(baseline_filename):
+    baseline = _get_baseline_from_file(baseline_filename)
+    if not baseline:
+        print('Failed to retrieve baseline from {filename}'.format(filename=baseline_filename))
+        return
+
+    print(json.dumps(determine_audit_results(baseline)))
+
+
 def audit_baseline(baseline_filename):
     original_baseline = _get_baseline_from_file(baseline_filename)
     if not original_baseline:

tests/core/audit_test.py

@@ -475,6 +475,119 @@ def new_baseline(self):
         }
 
 
+class TestDetermineAuditResults(object):
+    def get_audited_baseline(self, plugin_config={}, is_secret=None):
+        """
+        Returns a baseline in dict form with 1 plugin and 1 secret.
+        :param plugin_config: An optional dict for the plugin's config.
+        :param is_secret: An optional bool for whether the secret has been
+        audited.
+        """
+        baseline_fixture = {
+            'plugins_used': [
+                {
+                    'name': 'HexHighEntropyString',
+                },
+            ],
+            'results': {
+                'file': [
+                    {
+                        'hashed_secret': 'a837eb90d815a852f68f56f70b1b3fab24c46c84',
+                        'line_number': 1,
+                        'type': 'Hex High Entropy String',
+                    },
+                ],
+            },
+        }
+
+        if plugin_config:
+            baseline_fixture['plugins_used'][0].update(plugin_config)
+
+        if is_secret is not None:
+            baseline_fixture['results']['file'][0]['is_secret'] = is_secret
+
+        return baseline_fixture
+
+    @pytest.mark.parametrize(
+        'plugin_config',
+        [
+            {},
+            {'hex_limit': 2},
+        ],
+    )
+    @pytest.mark.parametrize(
+        'is_secret, expected_audited_result',
+        [
+            (True, 'positive'),
+            (False, 'negative'),
+            (None, 'unknown'),
+        ],
+    )
+    def test_determine_audit_results(
+        self,
+        plugin_config,
+        is_secret,
+        expected_audited_result,
+    ):
+        baseline = self.get_audited_baseline(plugin_config, is_secret)
+        results = audit.determine_audit_results(baseline)
+
+        if plugin_config:
+            assert results['HexHighEntropyString']['config'].items() >= plugin_config.items()
+
+        for audited_result, list_of_secrets in results['HexHighEntropyString']['results'].items():
+            expected_num_secrets = 1 if audited_result == expected_audited_result else 0
+            assert len(list_of_secrets) == expected_num_secrets
+
+
+class TestPrintAuditResults():
+
+    @contextmanager
+    def mock_env(self, baseline):
+        with mock.patch.object(
+            # We mock this, so we don't need to do any file I/O.
+            audit,
+            '_get_baseline_from_file',
+            return_value=baseline,
+        ) as _mock:
+            yield _mock
+
+    @pytest.mark.parametrize(
+        'mock_baseline, expected_message',
+        [
+            (
+                {},
+                'Failed to retrieve baseline',
+            ),
+            (
+                None,
+                'Failed to retrieve baseline',
+            ),
+            (
+                {'plugins_used': {'name': 'MyFakePlugin'}, 'results': {}},
+                '{}',
+            ),
+        ],
+    )
+    def test_print_audit_results_none(
+        self, mock_printer, mock_baseline, expected_message,
+    ):
+        """
+        This doesn't actually test for correctness; we rely on
+        good tests for determine_audit_results.
+        """
+        with self.mock_env(
+            baseline=mock_baseline,
+        ), mock.patch.object(
+            audit,
+            'determine_audit_results',
+            return_value={},
+        ):
+            audit.print_audit_results('somefilename')
+
+        assert expected_message in mock_printer.message
+
+
 class TestPrintContext(object):
 
     def run_logic(self, secret=None, secret_lineno=15, settings=None):