diff --git a/checkov/common/runners/base_runner.py b/checkov/common/runners/base_runner.py index b527aab568..a6ff520a70 100644 --- a/checkov/common/runners/base_runner.py +++ b/checkov/common/runners/base_runner.py @@ -44,6 +44,6 @@ def filter_ignored_paths(root_dir, names, excluded_paths: List[str]): # TODO this is not going to work well on Windows, because paths specified in the platform will use /, and # paths specified via the CLI argument will presumably use \\ if excluded_paths: - compiled = [re.compile(p) for p in excluded_paths] + compiled = [re.compile(p.replace('.terraform', '\.terraform')) for p in excluded_paths] [names.remove(path) for path in list(names) if any(pattern.search(os.path.join(root_dir, path)) for pattern in compiled)] diff --git a/checkov/secrets/plugins/entropy_keyword_combinator.py b/checkov/secrets/plugins/entropy_keyword_combinator.py index ab16c07874..4444178498 100644 --- a/checkov/secrets/plugins/entropy_keyword_combinator.py +++ b/checkov/secrets/plugins/entropy_keyword_combinator.py @@ -1,8 +1,8 @@ +from detect_secrets.core.potential_secret import PotentialSecret from detect_secrets.plugins.high_entropy_strings import Base64HighEntropyString, HexHighEntropyString from detect_secrets.plugins.keyword import KeywordDetector from detect_secrets.plugins.base import BasePlugin -from typing import Generator -from checkov.common.util.data_structures_utils import generator_reader_wrapper +from typing import Generator, Any, Set class EntropyKeywordCombinator(BasePlugin): @@ -12,18 +12,24 @@ def __init__(self, limit: float) -> None: self.high_entropy_scanners = (Base64HighEntropyString(limit=limit), HexHighEntropyString(limit=limit)) self.keyword_scanner = KeywordDetector() - def analyze_string(self, string: str) -> Generator[str, None, None]: - keyword_secrets_generator = self.keyword_scanner.analyze_string(string) - potential_entropy_secrets = [] - potential_kw_secret = generator_reader_wrapper(keyword_secrets_generator) - if not potential_kw_secret: - return - for entropy_scanner in self.high_entropy_scanners: - potential_entropy_secret = generator_reader_wrapper(entropy_scanner.analyze_string(string)) - if potential_entropy_secret: - self.secret_type = entropy_scanner.secret_type - potential_entropy_secrets.append(potential_entropy_secret) - if potential_kw_secret and len(potential_entropy_secrets) > 0: - yield string - + def analyze_line( + self, + filename: str, + line: str, + line_number: int = 0, + **kwargs: Any + ) -> Set[PotentialSecret]: + """ + This method first runs the keyword plugin. If it finds a match - it runs the entropy scanners, and if + one of the entropy scanners find a match (on a line which was already matched by keyword plugin) - it is returned. + """ + keyword_matches = self.keyword_scanner.analyze_line(filename, line, line_number, **kwargs) + if len(keyword_matches): + for entropy_scanner in self.high_entropy_scanners: + matches = entropy_scanner.analyze_line(filename, line, line_number, **kwargs) + if len(matches) > 0: + return matches + return set([]) + def analyze_string(self, string: str) -> Generator[str, None, None]: + raise NotImplementedError() diff --git a/checkov/secrets/runner.py b/checkov/secrets/runner.py index 96d837550a..9bf5992f81 100644 --- a/checkov/secrets/runner.py +++ b/checkov/secrets/runner.py @@ -108,8 +108,8 @@ def run(self, root_folder, external_checks_dir=None, files=None, runner_filter=R excluded_paths = (runner_filter.excluded_paths or []) + ignored_directories + [DEFAULT_EXTERNAL_MODULES_DIR] if root_folder: for root, d_names, f_names in os.walk(root_folder): - filter_ignored_paths(root, d_names, runner_filter.excluded_paths) - filter_ignored_paths(root, f_names, runner_filter.excluded_paths) + filter_ignored_paths(root, d_names, excluded_paths) + filter_ignored_paths(root, f_names, excluded_paths) for file in f_names: if file not in PROHIBITED_FILES and f".{file.split('.')[-1]}" in SUPPORTED_FILE_EXTENSIONS: files_to_scan.append(os.path.join(root, file)) diff --git a/tests/secrets/resources/terraform_failed/main.tf b/tests/secrets/resources/terraform_failed/main.tf index 58bb61e9d6..05a08627f2 100644 --- a/tests/secrets/resources/terraform_failed/main.tf +++ b/tests/secrets/resources/terraform_failed/main.tf @@ -1,4 +1,4 @@ provider "aws" { access_key = "AKIAIOSFODNN7EXAMPLE" - secret_key = "" + secret_key = "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" } \ No newline at end of file diff --git a/tests/secrets/test_plugin.py b/tests/secrets/test_plugin.py new file mode 100644 index 0000000000..60a51aa0a8 --- /dev/null +++ b/tests/secrets/test_plugin.py @@ -0,0 +1,22 @@ +import unittest +from checkov.secrets.plugins.entropy_keyword_combinator import EntropyKeywordCombinator + +class TestCombinatorPlugin(unittest.TestCase): + + def setUp(self) -> None: + self.plugin = EntropyKeywordCombinator(4.5) + + def test_positive_value(self): + result = self.plugin.analyze_line("mock.tf", "api_key = \"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMAAAKEY\"", 5) + self.assertEqual(1, len(result)) + secret = result.pop() + self.assertEqual('Base64 High Entropy String', secret.type) + self.assertEqual('c00f1a6e4b20aa64691d50781b810756d6254b8e', secret.secret_hash) + + def test_negative_keyword_value(self): + result = self.plugin.analyze_line("mock.tf", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMAAAKEY", 5) + self.assertEqual(0, len(result)) + + def test_negative_entropy_value(self): + result = self.plugin.analyze_line("mock.tf", "api_key = var.api_key", 5) + self.assertEqual(0, len(result)) diff --git a/tests/secrets/test_runner.py b/tests/secrets/test_runner.py index ec4e34b3eb..cd03b02a04 100644 --- a/tests/secrets/test_runner.py +++ b/tests/secrets/test_runner.py @@ -38,7 +38,7 @@ def test_runner_tf_failing_check(self): runner = Runner() report = runner.run(root_folder=valid_dir_path, external_checks_dir=None, runner_filter=RunnerFilter(framework='secrets')) - self.assertEqual(len(report.failed_checks), 1) + self.assertEqual(2, len(report.failed_checks)) self.assertEqual(report.parsing_errors, []) self.assertEqual(report.passed_checks, []) self.assertEqual(report.skipped_checks, []) @@ -60,7 +60,7 @@ def test_runner_multiple_files(self): runner = Runner() report = runner.run(root_folder=valid_dir_path, external_checks_dir=None, runner_filter=RunnerFilter(framework='secrets')) - self.assertEqual(len(report.failed_checks), 3) + self.assertEqual(4, len(report.failed_checks)) self.assertEqual(report.parsing_errors, []) self.assertEqual(report.passed_checks, []) self.assertEqual(report.skipped_checks, [])