break(terraform): return UNKNOWN for unrendered values in graph checks (

#3689)

* return UNKNOWN for unrendered values in graph attribute solvers

* fix lint

* use is_value_attribute_check to check if to skip checks for unrendered values
add None option to complex solvers as well

* add comment

* fix

* add unknown list results

* fix lint

* fix lint

* fix import

* revert import changes

* revert import changes

* add type annotations

* add unknown to connection solvers, fix running connections solvers from and/or connection solvers

* remove unknown results from passed in tests

* handle unknown for WILDCARD_PATTERN and jsonpath checks

* fix test

* fix test

* PR fixes

* Apply suggestions from code review

Co-authored-by: Anton Grübel <[email protected]>

* Update checkov/common/checks_infra/solvers/attribute_solvers/base_attribute_solver.py

Co-authored-by: Anton Grübel <[email protected]>

* fix import

Co-authored-by: Anton Grübel <[email protected]>

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

@@ -6,6 +6,7 @@
 
 class AnyResourceSolver(BaseAttributeSolver):
     operator = Operators.ANY  # noqa: CCE003  # a static attribute
+    is_value_attribute_check = False  # noqa: CCE003  # a static attribute
 
     def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
         return vertex is not None

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

@@ -22,14 +22,10 @@
 SUPPORTED_BLOCK_TYPES = {BlockType.RESOURCE, TerraformBlockType.DATA}
 WILDCARD_PATTERN = re.compile(r"(\S+[.][*][.]*)+")
 
-OPERATION_TO_FUNC = {
-    'all': all,
-    'any': any
-}
-
 
 class BaseAttributeSolver(BaseSolver):
     operator = ""  # noqa: CCE003  # a static attribute
+    is_value_attribute_check = True    # noqa: CCE003  # a static attribute
 
     def __init__(self, resource_types: List[str], attribute: Optional[str], value: Any, is_jsonpath_check: bool = False
                  ) -> None:
@@ -40,28 +36,50 @@ def __init__(self, resource_types: List[str], attribute: Optional[str], value: A
         self.is_jsonpath_check = is_jsonpath_check
         self.parsed_attributes: Dict[Optional[str], Any] = {}
 
-    def run(self, graph_connector: DiGraph) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
+    def run(self, graph_connector: DiGraph) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, Any]]]:
         executer = ThreadPoolExecutor()
         jobs = []
         passed_vertices: List[Dict[str, Any]] = []
         failed_vertices: List[Dict[str, Any]] = []
+        unknown_vertices: List[Dict[str, Any]] = []
         for _, data in graph_connector.nodes(data=True):
             if (not self.resource_types or data.get(CustomAttributes.RESOURCE_TYPE) in self.resource_types) \
                     and data.get(CustomAttributes.BLOCK_TYPE) in SUPPORTED_BLOCK_TYPES:
-                jobs.append(executer.submit(self._process_node, data, passed_vertices, failed_vertices))
+                jobs.append(executer.submit(
+                    self._process_node, data, passed_vertices, failed_vertices, unknown_vertices))
 
         concurrent.futures.wait(jobs)
-        return passed_vertices, failed_vertices
+        return passed_vertices, failed_vertices, unknown_vertices
+
+    def get_operation(self, vertex: Dict[str, Any]) -> Optional[bool]:
+        attr_val = vertex.get(self.attribute)   # type:ignore[arg-type]  # due to attribute can be None
+        # if this value contains an underendered variable, then we cannot evaluate value checks,
+        # and will return None (for UNKNOWN)
+        # handle edge cases in some policies that explicitly look for blank values
+        if self.is_value_attribute_check and self._is_variable_dependant(attr_val, vertex['source_']) \
+                and self.value != '':
+            return None
 
-    def get_operation(self, vertex: Dict[str, Any]) -> bool:
         if self.attribute and (self.is_jsonpath_check or re.match(WILDCARD_PATTERN, self.attribute)):
             attribute_matches = self.get_attribute_matches(vertex)
+            filtered_attribute_matches = attribute_matches
+            if self.is_value_attribute_check and self.value != '':
+                filtered_attribute_matches = [
+                    a for a in attribute_matches
+                    if not self._is_variable_dependant(vertex.get(a), vertex['source_'])
+                ]
 
-            operator = OPERATION_TO_FUNC['all'] if self.is_jsonpath_check else OPERATION_TO_FUNC['any']
             if attribute_matches:
-                return self.resource_type_pred(vertex, self.resource_types) and operator(
-                    self._get_operation(vertex=vertex, attribute=attr) for attr in attribute_matches
-                )
+                if self.is_jsonpath_check:
+                    if self.resource_type_pred(vertex, self.resource_types) and all(
+                            self._get_operation(vertex=vertex, attribute=attr) for attr in filtered_attribute_matches):
+                        return True if len(attribute_matches) == len(filtered_attribute_matches) else None
+                    return False
+
+                if self.resource_type_pred(vertex, self.resource_types) and any(
+                        self._get_operation(vertex=vertex, attribute=attr) for attr in filtered_attribute_matches):
+                    return True
+                return False if len(attribute_matches) == len(filtered_attribute_matches) else None
 
         return self.resource_type_pred(vertex, self.resource_types) and self._get_operation(
             vertex=vertex, attribute=self.attribute
@@ -71,11 +89,16 @@ def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bo
         raise NotImplementedError
 
     def _process_node(
-        self, data: Dict[str, Any], passed_vartices: List[Dict[str, Any]], failed_vertices: List[Dict[str, Any]]
+        self, data: Dict[str, Any], passed_vartices: List[Dict[str, Any]], failed_vertices: List[Dict[str, Any]],
+            unknown_vertices: List[Dict[str, Any]]
     ) -> None:
         if not self.resource_type_pred(data, self.resource_types):
             return
-        if self.get_operation(vertex=data):
+        result = self.get_operation(vertex=data)
+        # A None indicate for UNKNOWN result - the vertex shouldn't be added to the passed or the failed vertices
+        if result is None:
+            unknown_vertices.append(data)
+        elif result:
             passed_vartices.append(data)
         else:
             failed_vertices.append(data)

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

@@ -9,10 +9,4 @@ class EndingWithAttributeSolver(BaseAttributeSolver):
 
     def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
         attr = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
-
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr, vertex["source_"]):
-            return True
-
         return isinstance(attr, str) and attr.endswith(self.value)

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

@@ -9,11 +9,6 @@ class EqualsAttributeSolver(BaseAttributeSolver):
 
     def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
         attr_val = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        # handle edge cases in some policies that explicitly look for blank values
-        if self.value != '' and self._is_variable_dependant(attr_val, vertex['source_']):
-            return True
         if type(attr_val) == bool or type(self.value) == bool:
             # handle cases like str(False) == "false"
             # generally self.value will be a string, but could be a bool if the policy was created straight from json

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

@@ -6,6 +6,7 @@
 
 class ExistsAttributeSolver(BaseAttributeSolver):
     operator = Operators.EXISTS  # noqa: CCE003  # a static attribute
+    is_value_attribute_check = False  # noqa: CCE003  # a static attribute
 
     def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
         return vertex.get(attribute) is not None  # type:ignore[arg-type]  # due to attribute can be None

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

@@ -10,10 +10,6 @@ class GreaterThanAttributeSolver(BaseAttributeSolver):
 
     def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
         vertex_attr = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(vertex_attr, vertex['source_']):
-            return True
         attr_float = force_float(vertex_attr)
         value_float = force_float(self.value)
 

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

@@ -11,10 +11,6 @@ class GreaterThanOrEqualAttributeSolver(BaseAttributeSolver):
     def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
 
         vertex_attr = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(vertex_attr, vertex['source_']):
-            return True
         attr_float = force_float(vertex_attr)
         value_float = force_float(self.value)
 

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

@@ -10,11 +10,6 @@ class IntersectsAttributeSolver(BaseAttributeSolver):
     def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
         attr = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
 
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr, vertex['source_']):
-            return True
-
         if isinstance(self.value, str) and isinstance(attr, Iterable):
             return self.value in attr
 

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

@@ -10,11 +10,6 @@ class IsEmptyAttributeSolver(BaseAttributeSolver):
     def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
         attr = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
 
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr, vertex["source_"]):
-            return True
-
         if isinstance(attr, (list, Sized)):
             return len(attr) == 0
 

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

@@ -12,9 +12,4 @@ def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bo
         if attr is None:
             return False
 
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr, vertex['source_']):
-            return True
-
         return attr is True

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

@@ -13,11 +13,6 @@ def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bo
             return False
 
         attr = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr, vertex['source_']):
-            return True
-
         if isinstance(attr, Sized):
             return len(attr) == force_int(self.value)
 

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

@@ -13,11 +13,6 @@ def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bo
         if attr is None:
             return False
 
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr, vertex['source_']):
-            return True
-
         value_int = force_int(self.value)
 
         if value_int is None:

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

@@ -13,11 +13,6 @@ def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bo
         if attr is None:
             return False
 
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr, vertex['source_']):
-            return True
-
         value_int = force_int(self.value)
 
         if value_int is None:

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

@@ -13,11 +13,6 @@ def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bo
         if attr is None:
             return False
 
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr, vertex['source_']):
-            return True
-
         value_int = force_int(self.value)
 
         if value_int is None:

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

@@ -13,11 +13,6 @@ def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bo
         if attr is None:
             return False
 
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr, vertex['source_']):
-            return True
-
         value_int = force_int(self.value)
 
         if value_int is None:

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

@@ -11,10 +11,4 @@ def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bo
         if vertex.get(attribute) is None:  # type:ignore[arg-type]  # due to attribute can be None
             return False
 
-        attr_val = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr_val, vertex['source_']):
-            return True
-
         return not super()._get_operation(vertex, attribute)

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

@@ -11,10 +11,4 @@ def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bo
         if vertex.get(attribute) is None:  # type:ignore[arg-type]  # due to attribute can be None
             return False
 
-        attr_val = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr_val, vertex['source_']):
-            return True
-
         return not super()._get_operation(vertex, attribute)

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

@@ -8,9 +8,4 @@ class NotEqualsAttributeSolver(EqualsAttributeSolver):
     operator = Operators.NOT_EQUALS  # noqa: CCE003  # a static attribute
 
     def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
-        attr_val = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr_val, vertex['source_']):
-            return True
         return not super()._get_operation(vertex, attribute)

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

@@ -9,10 +9,4 @@ class StartingWithAttributeSolver(BaseAttributeSolver):
 
     def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
         attr = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
-
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr, vertex["source_"]):
-            return True
-
         return isinstance(attr, str) and attr.startswith(self.value)

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

@@ -9,8 +9,4 @@ class WithinAttributeSolver(BaseAttributeSolver):
 
     def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
         attr = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
-        # if this value contains an underendered variable, then we cannot evaluate the check,
-        # so return True (since we cannot return UNKNOWN)
-        if self._is_variable_dependant(attr, vertex['source_']):
-            return True
         return attr in self.value

checkov/common/checks_infra/solvers/complex_solvers/and_solver.py

@@ -1,4 +1,4 @@
-from typing import List, Any, Dict
+from typing import List, Any, Dict, Optional
 
 from checkov.common.graph.checks_infra.enums import Operators
 from checkov.common.graph.checks_infra.solvers.base_solver import BaseSolver
@@ -16,8 +16,12 @@ def __init__(self, solvers: List[BaseSolver], resource_types: List[str]) -> None
     def _get_operation(self, *args: Any, **kwargs: Any) -> Any:
         return reduce(and_, args)
 
-    def get_operation(self, vertex: Dict[str, Any]) -> bool:
+    def get_operation(self, vertex: Dict[str, Any]) -> Optional[bool]:
+        has_unrendered_attribute = False
         for solver in self.solvers:
-            if not solver.get_operation(vertex):
+            operation = solver.get_operation(vertex)
+            if operation is None:
+                has_unrendered_attribute = True
+            elif not operation:
                 return False
-        return True
+        return None if has_unrendered_attribute else True

checkov/common/checks_infra/solvers/complex_solvers/base_complex_solver.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from typing import List, Any, Tuple, Dict, TYPE_CHECKING
+from abc import abstractmethod
+from typing import List, Any, Tuple, Dict, TYPE_CHECKING, Optional
 
 from checkov.common.graph.checks_infra.enums import SolverType
 from checkov.common.graph.checks_infra.solvers.base_solver import BaseSolver
@@ -25,13 +26,21 @@ def _get_operation(self, *args: Any, **kwargs: Any) -> Any:
     def _get_negative_op(self, *args: Any) -> Any:
         return not self._get_operation(args)
 
-    def run(self, graph_connector: DiGraph) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
+    @abstractmethod
+    def get_operation(self, vertex: Dict[str, Any]) -> Optional[bool]:
+        raise NotImplementedError()
+
+    def run(self, graph_connector: DiGraph) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, Any]]]:
         passed_vertices = []
         failed_vertices = []
+        unknown_vertices = []
         for _, data in graph_connector.nodes(data=True):
             if self.resource_type_pred(data, self.resource_types):
-                if self.get_operation(data):
+                result = self.get_operation(data)
+                if result is None:
+                    unknown_vertices.append(data)
+                elif result:
                     passed_vertices.append(data)
                 else:
                     failed_vertices.append(data)
-        return passed_vertices, failed_vertices
+        return passed_vertices, failed_vertices, unknown_vertices

checkov/common/checks_infra/solvers/complex_solvers/not_solver.py

@@ -1,4 +1,4 @@
-from typing import List, Any, Dict
+from typing import List, Any, Dict, Optional
 
 from checkov.common.graph.checks_infra.enums import Operators
 from checkov.common.graph.checks_infra.solvers.base_solver import BaseSolver
@@ -18,5 +18,6 @@ def _get_operation(self, *args: Any, **kwargs: Any) -> Any:
             raise Exception('The "not" operator must have exactly one child')
         return not args[0]
 
-    def get_operation(self, vertex: Dict[str, Any]) -> bool:
-        return not self.solvers[0].get_operation(vertex)
+    def get_operation(self, vertex: Dict[str, Any]) -> Optional[bool]:
+        result = self.solvers[0].get_operation(vertex)
+        return None if result is None else not result

checkov/common/checks_infra/solvers/complex_solvers/or_solver.py

@@ -1,4 +1,4 @@
-from typing import List, Any, Dict
+from typing import List, Any, Dict, Optional
 
 from checkov.common.graph.checks_infra.enums import Operators
 from checkov.common.graph.checks_infra.solvers.base_solver import BaseSolver
@@ -16,8 +16,12 @@ def __init__(self, solvers: List[BaseSolver], resource_types: List[str]) -> None
     def _get_operation(self, *args: Any, **kwargs: Any) -> Any:
         return reduce(or_, args)
 
-    def get_operation(self, vertex: Dict[str, Any]) -> bool:
+    def get_operation(self, vertex: Dict[str, Any]) -> Optional[bool]:
+        has_unrendered_attribute = False
         for solver in self.solvers:
-            if solver.get_operation(vertex):
+            operation = solver.get_operation(vertex)
+            if operation:
                 return True
-        return False
+            if operation is None:
+                has_unrendered_attribute = True
+        return None if has_unrendered_attribute else False