feat(terraform_plan): Expose field changes to python checks

checkov/terraform/plan_parser.py

@@ -13,6 +13,7 @@
 SIMPLE_TYPES = (str, int, float, bool)
 TF_PLAN_RESOURCE_ADDRESS = "__address__"
 TF_PLAN_RESOURCE_CHANGE_ACTIONS = "__change_actions__"
+TF_PLAN_RESOURCE_CHANGE_KEYS = "__change_keys__"
 
 RESOURCE_TYPES_JSONIFY = {
     "aws_batch_job_definition": "container_properties",
@@ -129,7 +130,6 @@ def _prepare_resource_block(
     :param resource: tf planned_values resource block
     :param conf: tf configuration resource block
     :param resource_changes: tf resource_changes block
-
     :returns:
         - resource_block: a list of strings representing the header columns
         - prepared: whether conditions met to prepare data
@@ -161,6 +161,7 @@ def _prepare_resource_block(
         changes = resource_changes.get(resource_address)  # type:ignore[arg-type]  # becaus eit 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["changed_keys"]
 
         resource_block[resource_type][resource.get("name", "default")] = resource_conf
         prepared = True
@@ -186,7 +187,7 @@ def _find_child_modules(
             nested_blocks = _find_child_modules(
                 child_modules=nested_child_modules,
                 resource_changes=resource_changes,
-                root_module_conf=root_module_conf
+                root_module_conf=root_module_conf,
             )
             for block_type, resource_blocks in nested_blocks.items():
                 blocks[block_type].extend(resource_blocks)
@@ -238,14 +239,24 @@ def _get_resource_changes(template: dict[str, Any]) -> dict[str, dict[str, Any]]
     """Returns a resource address to resource changes dict"""
 
     resource_changes_map = {}
-
     resource_changes = template.get("resource_changes")
-    if resource_changes and isinstance(resource_changes, list):
-        resource_changes_map = {
-            change.get("address", ""): change
-            for change in resource_changes
-        }
 
+    if resource_changes and isinstance(resource_changes, list):
+        for each in resource_changes:
+            resource_changes_map[each["address"]] = each
+            changes = []
+
+            # before + after are None when resources are created/destroyed, so make them safe
+            if not each["change"]["before"]:
+                each["change"]["before"] = {}
+            if not each["change"]["after"]:
+                each["change"]["after"] = {}
+
+            for field in each["change"]["before"]:
+                if each["change"]["before"][field] != each["change"]["after"].get(field):
+                    changes.append(field)
+
+            resource_changes_map[each["address"]]["changed_keys"] = changes
     return resource_changes_map
 
 

tests/terraform/runner/extra_tf_plan_checks/security_group_rule_protocol_changed.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from typing import Any
+
+from checkov.common.models.enums import CheckResult, CheckCategories
+from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
+from checkov.terraform.plan_parser import TF_PLAN_RESOURCE_CHANGE_ACTIONS, TF_PLAN_RESOURCE_CHANGE_KEYS
+
+
+class SecurityGroupRuleProtocolChanged(BaseResourceCheck):
+    def __init__(self) -> None:
+        name = "Ensure security group rule protocol is not being changed"
+        id = "CUSTOM_CHANGE_1"
+        supported_resources = ("aws_security_group_rule",)
+        categories = (CheckCategories.GENERAL_SECURITY,)
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+    def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
+        actions = conf.get(TF_PLAN_RESOURCE_CHANGE_ACTIONS)
+        if isinstance(actions, list) and "update" in actions:
+            if "protocol" in conf.get(TF_PLAN_RESOURCE_CHANGE_KEYS):
+                self.details.append("some great details")
+                return CheckResult.FAILED
+        return CheckResult.PASSED
+
+
+scanner = SecurityGroupRuleProtocolChanged()