feat(general): Add resource code filter to all checkov loggers (#5356)

* Created template code for using custom logger which removes code string templates from the log

* Added other logging methods and moved env variable to constructor

* replaced logger with logger adapter

* Added tests

* mypy

* Renamed to indicate it's general for templates

* Renamed files and qfixed mypy

* Instead of using a replacement code template, just log empty message

* Eventually decided to use logging.Filter as the best option

* Add resource code filter to all checkov logs

* flake8

checkov/ansible/utils.py

@@ -7,6 +7,7 @@
 
 from checkov.ansible.graph_builder.graph_components.resource_types import ResourceType
 from checkov.common.parsers.yaml.parser import parse
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.util.consts import START_LINE, END_LINE
 from checkov.common.util.file_utils import read_file_with_any_encoding
 from checkov.common.util.suppression import collect_suppressions_for_context
@@ -59,6 +60,7 @@
 }
 
 logger = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(logger)
 
 
 def get_scannable_file_paths(root_folder: str | Path) -> set[Path]:

checkov/arm/parser/parser.py

@@ -9,9 +9,11 @@
 
 from checkov.common.parsers.json import parse as json_parse
 from checkov.common.parsers.yaml import loader
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.util.file_utils import read_file_with_any_encoding
 
 LOGGER = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(LOGGER)
 
 
 def parse(filename: str) -> tuple[dict[str, Any], list[tuple[int, str]]] | tuple[None, None]:

checkov/cloudformation/parser/__init__.py

@@ -9,7 +9,10 @@
 from yaml.scanner import ScannerError
 from yaml import YAMLError
 
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
+
 LOGGER = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(LOGGER)
 
 
 def parse(

checkov/cloudformation/parser/cfn_yaml.py

@@ -25,6 +25,7 @@
 
 from checkov.common.parsers.json.decoder import SimpleDecoder
 from checkov.common.parsers.node import StrNode, DictNode, ListNode
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.util.consts import MAX_IAC_FILE_SIZE
 from checkov.common.util.file_utils import read_file_with_any_encoding
 
@@ -44,6 +45,7 @@
 FN_PREFIX = 'Fn::'
 
 LOGGER = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(LOGGER)
 
 
 class ContentType(str, Enum):

checkov/common/checks/base_check.py

@@ -6,6 +6,7 @@
 from collections.abc import Iterable
 from typing import List, Dict, Any, Callable, Optional
 
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.typing import _SkippedCheck, _CheckResult
 from checkov.common.util.type_forcers import force_list
 from checkov.common.models.enums import CheckResult, CheckCategories, CheckFailLevel
@@ -31,6 +32,7 @@ def __init__(
         self.path: str | None = None
         self.supported_entities = supported_entities
         self.logger = logging.getLogger("{}".format(self.__module__))
+        add_resource_code_filter_to_logger(self.logger)
         self.evaluated_keys: List[str] = []
         self.entity_path = ""
         self.entity_type = ""

checkov/common/checks/base_check_registry.py

@@ -12,6 +12,7 @@
 from typing import Generator, Tuple, Dict, List, Optional, Any, TYPE_CHECKING
 
 from checkov.common.models.enums import CheckResult
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.typing import _SkippedCheck, _CheckResult
 from checkov.runner_filter import RunnerFilter
 
@@ -27,6 +28,7 @@ class BaseCheckRegistry:
 
     def __init__(self, report_type: str) -> None:
         self.logger = logging.getLogger(__name__)
+        add_resource_code_filter_to_logger(self.logger)
         # IMPLEMENTATION NOTE: Checks is used to directly access checks based on an specific entity
         self.checks: Dict[str, List[BaseCheck]] = defaultdict(list)
         # IMPLEMENTATION NOTE: When using a wildcard, every pattern needs to be checked. To reduce the

checkov/common/checks_infra/registry.py

@@ -11,6 +11,7 @@
 from checkov.common.checks_infra.checks_parser import GraphCheckParser
 from checkov.common.graph.checks_infra.base_parser import BaseGraphCheckParser
 from checkov.common.graph.checks_infra.registry import BaseRegistry
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.runner_filter import RunnerFilter
 from checkov.common.checks_infra.resources_types import resources_types
 
@@ -28,6 +29,7 @@ def __init__(self, checks_dir: str, parser: BaseGraphCheckParser | None = None)
         self.checks: list[BaseGraphCheck] = []
         self.checks_dir = checks_dir
         self.logger = logging.getLogger(__name__)
+        add_resource_code_filter_to_logger(self.logger)
 
     def load_checks(self) -> None:
         if self.checks:

checkov/common/checks_infra/solvers/attribute_solvers/contains_attribute_solver.py

@@ -4,8 +4,10 @@
 
 from checkov.common.graph.checks_infra.enums import Operators
 from checkov.common.checks_infra.solvers.attribute_solvers.base_attribute_solver import BaseAttributeSolver
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 
 logger = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(logger)
 
 
 class ContainsAttributeSolver(BaseAttributeSolver):

checkov/common/goget/github/get_git.py

@@ -5,6 +5,7 @@
 import shutil
 
 from checkov.common.goget.base_getter import BaseGetter
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.util.contextmanagers import temp_environ
 
 try:
@@ -21,6 +22,7 @@
 class GitGetter(BaseGetter):
     def __init__(self, url: str, create_clone_and_result_dirs: bool = True) -> None:
         self.logger = logging.getLogger(__name__)
+        add_resource_code_filter_to_logger(self.logger)
         self.create_clone_and_res_dirs = create_clone_and_result_dirs
         self.tag = ''
         self.commit_id: str | None = None

checkov/common/goget/registry/get_registry.py

@@ -3,6 +3,7 @@
 import os
 
 from checkov.common.goget.base_getter import BaseGetter
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.util.file_utils import extract_tar_archive
 from checkov.common.util.file_utils import extract_zip_archive
 from checkov.common.util.http_utils import DEFAULT_TIMEOUT
@@ -11,6 +12,7 @@
 class RegistryGetter(BaseGetter):
     def __init__(self, url: str, extension: str, create_clone_and_result_dirs: bool = False) -> None:
         self.logger = logging.getLogger(__name__)
+        add_resource_code_filter_to_logger(self.logger)
         self.extension = extension
         self.create_clone_and_res_dirs = create_clone_and_result_dirs
         super().__init__(url)

checkov/common/graph/checks_infra/debug.py

@@ -9,12 +9,14 @@
 from termcolor import colored
 
 from checkov.common.graph.graph_builder import CustomAttributes
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.util.env_vars_config import env_vars_config
 
 if TYPE_CHECKING:
     from checkov.common.graph.checks_infra.solvers.base_solver import BaseSolver
 
 logger = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(logger)
 
 
 def graph_check(check_id: str, check_name: str) -> None:

checkov/common/parsers/json/__init__.py

@@ -11,9 +11,11 @@
 
 from checkov.common.parsers.json.decoder import Decoder
 from checkov.common.parsers.json.errors import DecodeError
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.util.file_utils import read_file_with_any_encoding
 
 LOGGER = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(LOGGER)
 
 
 def load(

checkov/common/parsers/node.py

@@ -4,11 +4,14 @@
 from copy import deepcopy
 from typing import TYPE_CHECKING, Any, Type, Generator
 
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
+
 if TYPE_CHECKING:
     from checkov.common.parsers.json.decoder import Mark
 
 
 LOGGER = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(LOGGER)
 
 
 class TemplateAttributeError(AttributeError):

checkov/common/parsers/yaml/parser.py

@@ -6,8 +6,10 @@
 from yaml import YAMLError
 
 import checkov.common.parsers.yaml.loader as loader
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 
 logger = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(logger)
 
 
 def parse(

checkov/common/runners/runner_registry.py

@@ -34,6 +34,7 @@
 from checkov.common.output.sarif import Sarif
 from checkov.common.output.spdx import SPDX
 from checkov.common.parallelizer.parallel_runner import parallel_runner
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.typing import _ExitCodeThresholds, _BaseRunner, _ScaExitCodeThresholds
 from checkov.common.util import data_structures_utils
 from checkov.common.util.banner import tool as tool_name
@@ -82,6 +83,7 @@ def __init__(
         secrets_omitter_class: Type[SecretsOmitter] = SecretsOmitter,
     ) -> None:
         self.logger = logging.getLogger(__name__)
+        add_resource_code_filter_to_logger(self.logger)
         self.runner_filter = runner_filter
         self.runners = list(runners)
         self.banner = banner

checkov/common/util/file_utils.py

@@ -11,7 +11,10 @@
 
 from charset_normalizer import from_path
 
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
+
 logger = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(logger)
 
 
 def convert_to_unix_path(path: str) -> str:

checkov/common/util/http_utils.py

@@ -14,6 +14,7 @@
 from urllib3.response import HTTPResponse
 from urllib3.util import parse_url
 
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.util.consts import DEV_API_GET_HEADERS, DEV_API_POST_HEADERS, PRISMA_API_GET_HEADERS, \
     PRISMA_PLATFORM, BRIDGECREW_PLATFORM
 from checkov.common.util.data_structures_utils import merge_dicts
@@ -27,6 +28,7 @@
 DEFAULT_TIMEOUT = (3.1, 30)
 
 logger = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(logger)
 
 
 @overload

checkov/common/util/runner_dependency_handler.py

@@ -3,10 +3,13 @@
 import logging
 from typing import TYPE_CHECKING
 
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
+
 if TYPE_CHECKING:
     from checkov.common.runners.runner_registry import RunnerRegistry
 
 logger = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(logger)
 
 
 class RunnerDependencyHandler():

checkov/common/util/stopit/utils.py

@@ -16,6 +16,8 @@
 
 from typing_extensions import ParamSpec, Self
 
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
+
 if TYPE_CHECKING:
     from types import TracebackType
 
@@ -25,6 +27,7 @@
 # Custom logger
 LOG = logging.getLogger(name='stopit')
 LOG.addHandler(NullHandler())
+add_resource_code_filter_to_logger(LOG)
 
 
 class TimeoutException(Exception):

checkov/common/util/type_forcers.py

@@ -2,6 +2,7 @@
 
 import json
 import logging
+import typing
 from json import JSONDecodeError
 from typing import TypeVar, overload, Any, Dict
 
@@ -44,15 +45,16 @@ def force_float(var: Any) -> float | None:
         return None
 
 
-def convert_str_to_bool(bool_str: bool | str) -> bool | str:
+def convert_str_to_bool(bool_str: bool | str) -> bool:
     if isinstance(bool_str, str):
         bool_str_lower = bool_str.lower()
         if bool_str_lower in ("true", '"true"'):
             return True
         elif bool_str_lower in ("false", '"false"'):
             return False
 
-    return bool_str
+    # If we got here it must be a boolean, mypy doesn't understand it, so we use cast
+    return typing.cast(bool, bool_str)
 
 
 def force_dict(obj: Any) -> dict[str, Any] | None:

checkov/contributor_metrics.py

@@ -4,11 +4,14 @@
 import logging
 import json
 import subprocess  # nosec
+
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.util.http_utils import request_wrapper
 from checkov.common.bridgecrew.platform_integration import BcPlatformIntegration
 from typing import Any
 
 logger = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(logger)
 
 
 def report_contributor_metrics(repository: str, source: str,

checkov/kubernetes/parser/k8_json.py

@@ -8,12 +8,14 @@
 import yaml
 from yaml.loader import SafeLoader
 
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.util.file_utils import read_file_with_any_encoding
 
 if TYPE_CHECKING:
     from yaml import MappingNode
 
 logger = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(logger)
 
 
 def loads(content: str) -> list[dict[str, Any]]:

checkov/kubernetes/parser/k8_yaml.py

@@ -8,12 +8,14 @@
 import yaml
 from yaml.loader import SafeLoader
 
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.util.file_utils import read_file_with_any_encoding
 
 if TYPE_CHECKING:
     from yaml import MappingNode
 
 logger = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(logger)
 
 
 def loads(content: str) -> List[Dict[str, Any]]:

checkov/kubernetes/parser/parser.py

@@ -6,10 +6,12 @@
 
 from yaml import YAMLError
 
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.kubernetes.parser import k8_yaml, k8_json
 from checkov.kubernetes.parser.validatior import K8sValidator
 
 logger = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(logger)
 
 
 def parse(filename: str) -> tuple[list[dict[str, Any]], list[tuple[int, str]]] | None:

checkov/logging_init.py

@@ -2,10 +2,13 @@
 import os
 from io import StringIO
 
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
+
 LOG_LEVEL = os.getenv('LOG_LEVEL', 'WARNING').upper()
 logging.basicConfig(level=LOG_LEVEL)
 log_formatter = logging.Formatter("%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s]  %(message)s")
 root_logger = logging.getLogger()
+add_resource_code_filter_to_logger(root_logger)
 stream_handler = root_logger.handlers[0]
 stream_handler.setFormatter(log_formatter)
 stream_handler.setLevel(LOG_LEVEL)

checkov/main.py

@@ -43,6 +43,7 @@
 from checkov.common.goget.github.get_git import GitGetter
 from checkov.common.output.baseline import Baseline
 from checkov.common.bridgecrew.check_type import checkov_runners, CheckType
+from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
 from checkov.common.runners.runner_registry import RunnerRegistry
 from checkov.common.util import prompt
 from checkov.common.util.banner import banner as checkov_banner, tool as checkov_tool
@@ -90,6 +91,7 @@
 outer_registry = None
 
 logger = logging.getLogger(__name__)
+add_resource_code_filter_to_logger(logger)
 
 # sca package runner added during the run method
 DEFAULT_RUNNERS = [