Before logic changes to K8 registry, which has some gnarly logic
MODIFY
checkov/arm/base_registry.py
checkov/arm/base_registry.py
@@ -16,30 +16,16 @@ class Registry(BaseCheckRegistry):
entity_type, entity_configuration = self.extract_entity_details(entity)
results = {}
checks = self.get_checks(entity_type)
- check_id_allowlist = runner_filter.checks
- check_id_denylist = runner_filter.skip_checks
for check in checks:
skip_info = {}
if skipped_checks:
if check.id in [x['id'] for x in skipped_checks]:
skip_info = [x for x in skipped_checks if x['id'] == check.id][0]
- if check_id_allowlist:
- if check.id in check_id_allowlist:
- self.logger.debug("Running check: {} on file {}".format(check.name, scanned_file))
- result = check.run(scanned_file=scanned_file, entity_configuration=entity_configuration,
- entity_name=entity_type, entity_type=entity_type, skip_info=skip_info)
- results[check] = result
- elif check_id_denylist:
- if check.id not in check_id_denylist:
- result = check.run(scanned_file=scanned_file, entity_configuration=entity_configuration,
- entity_name=entity_type, entity_type=entity_type, skip_info=skip_info)
- results[check] = result
- else:
+ if runner_filter.should_run_check(check.id):
self.logger.debug("Running check: {} on file {}".format(check.name, scanned_file))
result = check.run(scanned_file=scanned_file, entity_configuration=entity_configuration,
entity_name=entity_type, entity_type=entity_type, skip_info=skip_info)
-
results[check] = result
return results
'''
\ No newline at end of file

MODIFY
checkov/arm/runner.py
checkov/arm/runner.py
@@ -24,7 +24,7 @@ class Runner(BaseRunner):
files_list = []
if external_checks_dir:
for directory in external_checks_dir:
- arm_registry.load_external_checks(directory)
+ arm_registry.load_external_checks(directory, runner_filter)
if files:
for file in files:

MODIFY
checkov/cloudformation/runner.py
checkov/cloudformation/runner.py
@@ -24,7 +24,7 @@ class Runner(BaseRunner):
files_list = []
if external_checks_dir:
for directory in external_checks_dir:
- cfn_registry.load_external_checks(directory)
+ cfn_registry.load_external_checks(directory, runner_filter)
if files:
for file in files:

MODIFY
checkov/common/checks/base_check_registry.py
checkov/common/checks/base_check_registry.py
@@ -13,8 +13,19 @@ class BaseCheckRegistry(object):
self.logger = logging.getLogger(__name__)
self.checks = {}
self.check_id_allowlist = None
+ self.external_checks_runner_filter = None
def register(self, check):
+ # IMPLEMENTATION NOTE: Checks are registered when the script is loaded
+ # (see BaseResourceCheck.__init__() for the various frameworks). The only
+ # difficultly with this process is that external checks need to be specially
+ # identified for filter handling. That's why you'll see stateful setting of
+ # RunnerFilters during load_external_checks.
+ # Built-in checks are registered immediately at script start, before
+ # external checks.
+ if self.external_checks_runner_filter:
+ self.external_checks_runner_filter.notify_external_check(check.id)
+
for entity in check.supported_entities:
if entity not in self.checks.keys():
self.checks[entity] = []
@@ -45,22 +56,13 @@ class BaseCheckRegistry(object):
(entity_type, entity_name, entity_configuration) = self.extract_entity_details(entity)
results = {}
checks = self.get_checks(entity_type)
- check_id_allowlist = runner_filter.checks
- check_id_denylist = runner_filter.skip_checks
for check in checks:
skip_info = {}
if skipped_checks:
if check.id in [x['id'] for x in skipped_checks]:
skip_info = [x for x in skipped_checks if x['id'] == check.id][0]
- if check_id_allowlist:
- if check.id in check_id_allowlist:
- result = self.run_check(check, entity_configuration, entity_name, entity_type, scanned_file, skip_info)
- results[check] = result
- elif check_id_denylist:
- if check.id not in check_id_denylist:
- result = self.run_check(check, entity_configuration, entity_name, entity_type, scanned_file, skip_info)
- results[check] = result
- else:
+
+ if runner_filter.should_run_check(check.id):
result = self.run_check(check, entity_configuration, entity_name, entity_type, scanned_file, skip_info)
results[check] = result
return results
@@ -89,7 +91,7 @@ class BaseCheckRegistry(object):
return True
return False
- def load_external_checks(self, directory):
+ def load_external_checks(self, directory, runner_filter):
""" Browse a directory looking for .py files to import.
Log an error when the directory does not contains an __init__.py or
@@ -106,7 +108,12 @@ class BaseCheckRegistry(object):
for entry in directory_content:
if self._file_can_be_imported(entry):
check_name = entry.name.replace('.py', '')
+
+ # Filter is set while loading external checks so the filter can be informed
+ # of the checks, which need to be handled specially.
+ prior_external_checks_runner_filter = self.external_checks_runner_filter
try:
+ self.external_checks_runner_filter = runner_filter
self.logger.debug("Importing external check '{}'".format(check_name))
importlib.import_module(check_name)
except SyntaxError as e:
@@ -121,3 +128,6 @@ class BaseCheckRegistry(object):
error_column=e.args[1][2]
)
)
+ finally:
+ self.external_checks_runner_filter = prior_external_checks_runner_filter
+

MODIFY
checkov/kubernetes/base_registry.py
checkov/kubernetes/base_registry.py
@@ -22,6 +22,17 @@ class Registry(BaseCheckRegistry):
if skipped_checks:
if check.id in [x['id'] for x in skipped_checks]:
skip_info = [x for x in skipped_checks if x['id'] == check.id][0]
+
+ if self._should_run_scan(check.id, entity_configuration, check_id_allowlist, check_id_denylist):
+ self.logger.debug("Running check: {} on file {}".format(check.name, scanned_file))
+
+ result = check.run(scanned_file=scanned_file, entity_configuration=entity_configuration,
+ entity_name=entity_type, entity_type=entity_type, skip_info=skip_info)
+ results[check] = result
+ return results
+
+ @staticmethod
+ def _should_run_scan(check_id, entity_configuration, check_id_allowlist, check_id_denylist):
if check_id_allowlist:
# Allow list provides namespace-only allows, check-only allows, or both
# If namespaces not specified, all namespaces are scanned
@@ -39,7 +50,7 @@ class Registry(BaseCheckRegistry):
if "default" in allowed_namespaces:
run_check = True
else:
- if check.id in check_id_allowlist:
+ if check_id in check_id_allowlist:
if allowed_namespaces:
# Check if namespace in allowed namespaces
if "metadata" in entity_configuration and "namespace" in entity_configuration["metadata"]:
@@ -55,11 +66,7 @@ class Registry(BaseCheckRegistry):
# No namespaces to filter
run_check = True
if run_check:
- self.logger.debug("Running check: {} on file {}".format(check.name, scanned_file))
-
- result = check.run(scanned_file=scanned_file, entity_configuration=entity_configuration,
- entity_name=entity_type, entity_type=entity_type, skip_info=skip_info)
- results[check] = result
+ return True
elif check_id_denylist:
namespace_skip = False
if "metadata" in entity_configuration and "namespace" in entity_configuration["metadata"]:
@@ -71,14 +78,8 @@ class Registry(BaseCheckRegistry):
else:
if "default" in check_id_denylist:
namespace_skip = True
- if check.id not in check_id_denylist and namespace_skip == False:
- result = check.run(scanned_file=scanned_file, entity_configuration=entity_configuration,
- entity_name=entity_type, entity_type=entity_type, skip_info=skip_info)
- results[check] = result
+ if check_id not in check_id_denylist and namespace_skip == False:
+ return True
else:
- self.logger.debug("Running check: {} on file {}".format(check.name, scanned_file))
- result = check.run(scanned_file=scanned_file, entity_configuration=entity_configuration,
- entity_name=entity_type, entity_type=entity_type, skip_info=skip_info)
-
- results[check] = result
- return results
+ return True
+ return False

MODIFY
checkov/kubernetes/runner.py
checkov/kubernetes/runner.py
@@ -24,7 +24,7 @@ class Runner(BaseRunner):
files_list = []
if external_checks_dir:
for directory in external_checks_dir:
- registry.load_external_checks(directory)
+ registry.load_external_checks(directory, runner_filter)
if files:
for file in files:

MODIFY
checkov/runner_filter.py
checkov/runner_filter.py
@@ -2,6 +2,7 @@ class RunnerFilter(object):
framework = 'all'
checks = []
skip_checks = []
+ external_check_ids = set()
def __init__(self, framework='all', checks=None, skip_checks=None):
if checks is None:
@@ -18,3 +19,18 @@ class RunnerFilter(object):
else:
self.skip_checks = skip_checks
self.framework = framework
+
+ def notify_external_check(self, check_id):
+ self.external_check_ids.add(check_id)
+
+ def should_run_check(self, check_id):
+ if check_id in self.external_check_ids:
+ pass # enabled unless kipped
+ elif self.checks:
+ if check_id in self.checks:
+ return True
+ else:
+ return False
+ if self.skip_checks and check_id in self.skip_checks:
+ return False
+ return True

MODIFY
checkov/serverless/base_registry.py
checkov/serverless/base_registry.py
@@ -15,29 +15,15 @@ class Registry(BaseCheckRegistry):
entity_type, entity_configuration = self.extract_entity_details(entity)
results = {}
checks = self.get_checks(entity_type)
- check_id_allowlist = runner_filter.checks
- check_id_denylist = runner_filter.skip_checks
for check in checks:
skip_info = {}
if skipped_checks:
if check.id in [x['id'] for x in skipped_checks]:
skip_info = [x for x in skipped_checks if x['id'] == check.id][0]
- if check_id_allowlist:
- if check.id in check_id_allowlist:
- self.logger.debug("Running check: {} on file {}".format(check.name, scanned_file))
- result = check.run(scanned_file=scanned_file, entity_configuration=entity_configuration,
- entity_name=entity_type, entity_type=entity_type, skip_info=skip_info)
- results[check] = result
- elif check_id_denylist:
- if check.id not in check_id_denylist:
- result = check.run(scanned_file=scanned_file, entity_configuration=entity_configuration,
- entity_name=entity_type, entity_type=entity_type, skip_info=skip_info)
- results[check] = result
- else:
+ if runner_filter.should_run_check(check.id):
self.logger.debug("Running check: {} on file {}".format(check.name, scanned_file))
result = check.run(scanned_file=scanned_file, entity_configuration=entity_configuration,
entity_name=entity_type, entity_type=entity_type, skip_info=skip_info)
-
results[check] = result
return results

MODIFY
checkov/serverless/runner.py
checkov/serverless/runner.py
@@ -26,7 +26,7 @@ class Runner(BaseRunner):
files_list = []
if external_checks_dir:
for directory in external_checks_dir:
- sls_registry.load_external_checks(directory)
+ sls_registry.load_external_checks(directory, runner_filter)
if files:
for file in files:

MODIFY
checkov/terraform/runner.py
checkov/terraform/runner.py
@@ -43,7 +43,7 @@ class Runner(BaseRunner):
parsing_errors = {}
if external_checks_dir:
for directory in external_checks_dir:
- resource_registry.load_external_checks(directory)
+ resource_registry.load_external_checks(directory, runner_filter)
if root_folder:
root_folder = os.path.abspath(root_folder)
self.parser.hcl2(directory=root_folder, tf_definitions=self.tf_definitions, parsing_errors=parsing_errors)

ADD
tests/common/test_runner_filter.py
+import os
+import unittest
+from unittest import mock
+
+from checkov.runner_filter import RunnerFilter
+
+
+class TestRunnerFilter(unittest.TestCase):
+
+ # Expected pseudo-code for when checks should run:
+ # if has_check_flag_specified():
+ # checks_to_run = checks_specifically_included
+ # else:
+ # checks_to_run = all_built_in_checks
+ # if has_checks_dir_specified():
+ # checks_to_run += checks_from_external_dir
+ # for skipped_check in skip_check_flags():
+ # checks_to_run.remove(skipped_check)
+
+ def test_should_run_default(self):
+ instance = RunnerFilter()
+ self.assertTrue(instance.should_run_check("CHECK_1"))
+
+ def test_should_run_specific_enable(self):
+ instance = RunnerFilter(checks=["CHECK_1"])
+ self.assertTrue(instance.should_run_check("CHECK_1"))
+
+ def test_should_run_omitted_specific_enable(self):
+ instance = RunnerFilter(checks=["CHECK_1"])
+ self.assertFalse(instance.should_run_check("CHECK_999"))
+
+ def test_should_run_specific_disable(self):
+ instance = RunnerFilter(skip_checks=["CHECK_1"])
+ self.assertFalse(instance.should_run_check("CHECK_1"))
+
+ def test_should_run_omitted_specific_disable(self):
+ instance = RunnerFilter(skip_checks=["CHECK_1"])
+ self.assertTrue(instance.should_run_check("CHECK_999"))
+
+ def test_should_run_external(self):
+ instance = RunnerFilter(skip_checks=["CHECK_1"])
+ instance.notify_external_check("EXT_CHECK_999")
+ self.assertTrue(instance.should_run_check("EXT_CHECK_999"))
+
+ def test_should_run_external2(self):
+ instance = RunnerFilter(checks=["CHECK_1"], skip_checks=["CHECK_2"])
+ instance.notify_external_check("EXT_CHECK_999")
+ self.assertTrue(instance.should_run_check("EXT_CHECK_999"))
+
+ def test_should_run_external3(self):
+ instance = RunnerFilter(checks=["EXT_CHECK_999"])
+ instance.notify_external_check("EXT_CHECK_999")
+ self.assertTrue(instance.should_run_check("EXT_CHECK_999"))
+
+ def test_should_run_external_disabled(self):
+ instance = RunnerFilter(skip_checks=["CHECK_1", "EXT_CHECK_999"])
+ instance.notify_external_check("EXT_CHECK_999")
+ self.assertFalse(instance.should_run_check("EXT_CHECK_999"))
+
+ def test_should_run_specific_disable_AND_enable(self):
+ instance = RunnerFilter(checks=["CHECK_1"], skip_checks=["CHECK_1"])
+ self.assertTrue(instance.should_run_check("CHECK_1"))
+
+
+if __name__ == '__main__':
+ unittest.main()

ADD
tests/kubernetes/test_base_registry.py
+import os
+import unittest
+from unittest import mock
+
+from checkov.kubernetes.base_registry import Registry
+from checkov.runner_filter import RunnerFilter
+
+
+class TestRunnerFilter(unittest.TestCase):
+
+ def test_run_by_id_default(self):
+ instance = Registry()
+ self.assertTrue(instance._should_run_scan("CKV_1", {}, [], []))
+
+ def test_run_by_id_specific_enable(self):
+ instance = Registry()
+ self.assertTrue(instance._should_run_scan("CKV_1", {}, ["CKV_1"], []))
+
+ def test_run_by_id_omitted_specific_enable(self):
+ instance = Registry()
+ self.assertFalse(instance._should_run_scan("CKV_999", {}, ["CKV_1"], []))
+
+ def test_run_by_id_specific_disable(self):
+ instance = Registry()
+ self.assertFalse(instance._should_run_scan("CKV_1", {}, [], ["CKV_1"]))
+
+ def test_run_by_id_omitted_specific_disable(self):
+ instance = Registry()
+ self.assertTrue(instance._should_run_scan("CKV_999", {}, [], ["CKV_1"]))
+
+ def test_run_by_id_external(self):
+ instance = Registry()
+ self.assertTrue(instance._should_run_scan("CKV_EXT_999", {}, [], ["CKV_1"]))
+
+ def test_run_by_id_external2(self):
+ instance = Registry()
+ self.assertTrue(instance._should_run_scan("CKV_EXT_999", {}, ["CKV_1"], ["CKV_2"]))
+
+ def test_run_by_id_external3(self):
+ instance = Registry()
+ self.assertTrue(instance._should_run_scan("CKV_EXT_999", {}, ["CKV_EXT_999"], []))
+
+ def test_run_by_id_external_disabled(self):
+ instance = Registry()
+ self.assertFalse(instance._should_run_scan("CKV_EXT_999", {}, [], ["CKV_1", "CKV_EXT_999"]))
+
+ def test_run_by_id_specific_disable_AND_enable(self):
+ instance = Registry()
+ self.assertTrue(instance._should_run_scan("CKV_1", {}, ["CKV_1"], ["CKV_1"]))
+
+ # Namespace filtering
+
+ def test_namespace_allow_default(self):
+ instance = Registry()
+ config = {"metadata": {"namespace": "not_matched"}}
+ # TODO: Maybe this case SHOULD run? Matching old state (false) for now. - @robeden
+ self.assertFalse(instance._should_run_scan("CKV_1", config, ["default"], []))
+
+ def test_namespace_deny_default(self):
+ instance = Registry()
+ config = {"metadata": {"namespace": "not_matched"}}
+ # TODO: Maybe this case SHOULDN'T run? Matching old state (true) for now. - @robeden
+ self.assertTrue(instance._should_run_scan("CKV_1", config, [], ["default"]))
+
+ def test_namespace_allow_specific(self):
+ instance = Registry()
+ config = {"metadata": {"namespace": "matched"}}
+ self.assertTrue(instance._should_run_scan("CKV_1", config, ["matched"], []))
+
+ def test_namespace_deny_specific(self):
+ instance = Registry()
+ config = {"metadata": {"namespace": "matched"}}
+ self.assertFalse(instance._should_run_scan("CKV_1", config, [], ["matched"]))
+
+ def test_namespace_allow_specific_other(self):
+ instance = Registry()
+ config = {"metadata": {"namespace": "not_matched"}}
+ self.assertFalse(instance._should_run_scan("CKV_1", config, ["something_else"], []))
+
+ def test_namespace_deny_specific_other(self):
+ instance = Registry()
+ config = {"metadata": {"namespace": "not_matched"}}
+ self.assertTrue(instance._should_run_scan("CKV_1", config, [], ["something_else"]))
+
+
+if __name__ == '__main__':
+ unittest.main()

MODIFY
tests/terraform/checks/data/test_registry.py
tests/terraform/checks/data/test_registry.py
@@ -2,6 +2,8 @@ import os
import unittest
from unittest.mock import patch
+from checkov.runner_filter import RunnerFilter
+
class TestRegistry(unittest.TestCase):
@@ -23,7 +25,7 @@ class TestRegistry(unittest.TestCase):
def test_registry_external_check_load(self):
current_dir = os.path.dirname(os.path.realpath(__file__))
external_dir = current_dir + "/example_external_dir/extra_checks"
- self.registry.load_external_checks(external_dir)
+ self.registry.load_external_checks(external_dir, RunnerFilter())
external_check_loaded = False
external_check = None

MODIFY
tests/terraform/checks/resource/registry/test_registry.py
tests/terraform/checks/resource/registry/test_registry.py
@@ -2,6 +2,8 @@ import os
import unittest
from unittest.mock import patch
+from checkov.runner_filter import RunnerFilter
+
class TestRegistry(unittest.TestCase):
@@ -22,7 +24,7 @@ class TestRegistry(unittest.TestCase):
def test_registry_external_check_load(self):
current_dir = os.path.dirname(os.path.realpath(__file__))
external_dir = current_dir + "/example_external_dir/extra_checks"
- self.registry.load_external_checks(external_dir)
+ self.registry.load_external_checks(external_dir, RunnerFilter())
external_check_loaded = False
external_check = None

Before logic changes to K8 registry, which has some gnarly logic