feat(terraform_plan): support tf_plan after_unknown enrichment (#6517)

* add first code content to run tests

* remove flag for testing

* fix S3BucketObject check to check for dict

* change code to reflect after_unknown

* add flag for eval of tf_plan after_unknown

* fix according to comments

* fix mypy issue

---------

Co-authored-by: Max Amelchenko <mamelchenko@paloaltonetworks.com>

checkov/common/util/consts.py

@@ -5,6 +5,7 @@
 START_LINE = '__startline__'
 END_LINE = '__endline__'
 LINE_FIELD_NAMES = {START_LINE, END_LINE}
+TRUE_AFTER_UNKNOWN = 'true_after_unknown'
 
 DEV_API_GET_HEADERS = {
     'Accept': 'application/json'

checkov/terraform/checks/resource/aws/S3BucketObjectLock.py

@@ -14,7 +14,7 @@ def __init__(self) -> None:
 
     def scan_resource_conf(self, conf: Dict[str, List[Any]]) -> CheckResult:
         lock_conf = conf.get("object_lock_configuration")
-        if lock_conf and lock_conf[0]:
+        if lock_conf and lock_conf[0] and isinstance(lock_conf[0], dict):
             lock_enabled = lock_conf[0].get("object_lock_enabled")
             if lock_enabled in ["Enabled", ["Enabled"]]:
                 return CheckResult.PASSED

checkov/terraform/plan_parser.py

@@ -3,11 +3,12 @@
 import itertools
 import json
 import logging
+import os
 from typing import Any, Dict, List, Optional, Tuple, cast
 
 from checkov.common.graph.graph_builder import CustomAttributes
 from checkov.common.parsers.node import ListNode
-from checkov.common.util.consts import LINE_FIELD_NAMES
+from checkov.common.util.consts import LINE_FIELD_NAMES, TRUE_AFTER_UNKNOWN
 from checkov.common.util.type_forcers import force_list
 from checkov.terraform.context_parsers.tf_plan import parse
 
@@ -16,6 +17,7 @@
 TF_PLAN_RESOURCE_CHANGE_ACTIONS = "__change_actions__"
 TF_PLAN_RESOURCE_CHANGE_KEYS = "__change_keys__"
 TF_PLAN_RESOURCE_PROVISIONERS = "provisioners"
+TF_PLAN_RESOURCE_AFTER_UNKNOWN = 'after_unknown'
 
 RESOURCE_TYPES_JSONIFY = {
     "aws_batch_job_definition": "container_properties",
@@ -199,10 +201,12 @@ def _prepare_resource_block(
         resource_address: str | None = resource.get("address")
         resource_conf[TF_PLAN_RESOURCE_ADDRESS] = resource_address  # type:ignore[assignment]  # special field
 
-        changes = resource_changes.get(resource_address)  # type:ignore[arg-type]  # becaus eit can be None
+        changes = resource_changes.get(resource_address)  # type:ignore[arg-type]  # because it can be None
         if changes:
             resource_conf[TF_PLAN_RESOURCE_CHANGE_ACTIONS] = changes.get("change", {}).get("actions") or []
             resource_conf[TF_PLAN_RESOURCE_CHANGE_KEYS] = changes.get(TF_PLAN_RESOURCE_CHANGE_KEYS) or []
+            # enrich conf with after_unknown values
+            _eval_after_unknown(changes, resource_conf)
 
         provisioners = conf.get(TF_PLAN_RESOURCE_PROVISIONERS) if conf else None
         if provisioners:
@@ -213,6 +217,22 @@ def _prepare_resource_block(
     return resource_block, block_type, prepared
 
 
+def _eval_after_unknown(changes: dict[str, Any], resource_conf: dict[str, Any]) -> None:
+    after_unknown = changes.get("change", {}).get(TF_PLAN_RESOURCE_AFTER_UNKNOWN)
+    if os.getenv('EVAL_TF_PLAN_AFTER_UNKNOWN') and after_unknown and isinstance(after_unknown, dict):
+        for k, v in after_unknown.items():
+            # We check if the value of the field is True. That would mean its value is known after the apply
+            # We also check whether the field is not already present in the conf since we do not want to
+            # override it. Overriding can actually cause losing its value
+            if v is True and k not in resource_conf:
+                # We set the value to 'true_after_unknown' and not its original value
+                # We need to set a constant other than a boolean (True/"true"),
+                # so it will not collide with actual possible values of those attributes
+                # In these cases, policies checking the existence of a value will succeed,
+                # but policies checking for concrete values will fail
+                resource_conf[k] = _clean_simple_type_list([TRUE_AFTER_UNKNOWN])
+
+
 def _find_child_modules(
     child_modules: ListNode, resource_changes: dict[str, dict[str, Any]], root_module_conf: dict[str, Any]
 ) -> dict[str, list[dict[str, dict[str, Any]]]]:

tests/terraform/runner/resources/plan_and_tf_combine_graph_with_missing_resources/tfplan.json

@@ -186,7 +186,6 @@
                "grant":true,
                "hosted_zone_id":true,
                "id":true,
-               "lifecycle_rule":true,
                "logging":true,
                "object_lock_configuration":true,
                "object_lock_enabled":true,
@@ -264,7 +263,6 @@
                "grant":true,
                "hosted_zone_id":true,
                "id":true,
-               "lifecycle_rule":true,
                "logging":true,
                "object_lock_configuration":true,
                "object_lock_enabled":true,

tests/terraform/runner/test_plan_runner.py

@@ -816,6 +816,7 @@ def test_plan_and_tf_combine_graph(self):
         assert report.passed_checks[0].file_path.endswith('.json')
         assert report.passed_checks[1].file_path.endswith('.json')
 
+    @mock.patch.dict(os.environ, {'EVAL_TF_PLAN_AFTER_UNKNOWN': 'True'})
     def test_plan_and_tf_combine_graph_with_missing_resources(self):
         tf_file_path = Path(__file__).parent / "resources/plan_and_tf_combine_graph_with_missing_resources/tfplan.json"
         repo_path = Path(__file__).parent / "resources/plan_and_tf_combine_graph_with_missing_resources"