Support custom policy rule

rdk/rdk.py

@@ -269,6 +269,7 @@ def get_rule_parser(is_required, command):
             "python3.8-lib",
             "python3.9",
             "python3.9-lib",
+            "cloudformation-guard2.0"
             "python3.10",
             "python3.10-lib",
             "python3.11",
@@ -1224,6 +1225,7 @@ def create(self):
                 "python3.8-lib": ".py",
                 "python3.9": ".py",
                 "python3.9-lib": ".py",
+                "cloudformation-guard2.0": "guard",
                 "python3.10": ".py",
                 "python3.10-lib": ".py",
                 "python3.11": ".py",
@@ -1249,6 +1251,8 @@ def create(self):
                 # copy rule template into rule directory
                 if self.args.runtime == "java8":
                     self.__create_java_rule()
+                elif self.args.runtime == "cloudformation-guard2.0":
+                    self.__create_cloudformation_guard_rule()
                 else:
                     src = os.path.join(
                         path.dirname(__file__),
@@ -1928,8 +1932,53 @@ def deploy(self):
 
                 continue
 
+            try:
+                rule_description = rule_params["Description"]
+            except KeyError:
+                rule_description = rule_name
+
+            source_runtime = rule_params["SourceRuntime"]
+
             print(f"[{my_session.region_name}]: Found Custom Rule.")
 
+            if source_runtime == "cloudformation-guard2.0":
+                print(f"[{my_session.region_name}]: Updating policy rule for " + rule_name)
+                policy = os.path.join(os.getcwd(), rules_dir, rule_name, "rule_code.guard")
+                my_cfg = my_session.client("config")
+                response = my_cfg.put_config_rule(
+                    ConfigRule={
+                        'ConfigRuleName': rule_name,
+                        'Description': rule_description,
+                        'Scope': {
+                            'ComplianceResourceTypes': source_events.split(',')
+                        },
+                        'Source': {
+                            'Owner': 'CUSTOM_POLICY',
+                            'SourceDetails': [
+                                {
+                                    'EventSource': 'aws.config',
+                                    'MessageType': 'ConfigurationItemChangeNotification'
+                                },
+                                {
+                                    'EventSource': 'aws.config',
+                                    'MessageType': 'OversizedConfigurationItemChangeNotification'
+                                },
+                            ],
+                            'CustomPolicyDetails': {
+                                'PolicyRuntime': 'guard-2.x.x',
+                                'PolicyText': open(policy).read(),
+                                'EnableDebugLogDelivery': False
+                            }
+                        },
+                        'InputParameters': json.dumps(combined_input_parameters),
+                    },
+                    Tags=cfn_tags
+                )
+                if response['ResponseMetadata']['HTTPStatusCode'] != 200:
+                    print(f"[{my_session.region_name}]: API status error: " + response.__repr__())
+                print(f"[{my_session.region_name}]: Update done.")
+                return
+
             s3_src = ""
             s3_dst = self.__upload_function_code(rule_name, rule_params, account_id, my_session, code_bucket_name)
 
@@ -1949,11 +1998,6 @@ def deploy(self):
             else:
                 boundaryPolicyArn = ""
 
-            try:
-                rule_description = rule_params["Description"]
-            except KeyError:
-                rule_description = rule_name
-
             my_params = [
                 {
                     "ParameterKey": "RuleName",
@@ -2151,9 +2195,6 @@ def deploy(self):
             if cfn_tags is not None and len(cfn_tags) > 0:
                 self.__tag_config_rule(rule_name, cfn_tags, my_session)
 
-        print(f"[{my_session.region_name}]: Config deploy complete.")
-
-        return 0
 
     def deploy_organization(self):
         self.__parse_deploy_organization_args()
@@ -3231,6 +3272,27 @@ def __create_java_rule(self):
         dst = os.path.join(os.getcwd(), rules_dir, self.args.rulename, "build.gradle")
         shutil.copyfile(src, dst)
 
+    def __create_cloudformation_guard_rule(self):
+        src = os.path.join(path.dirname(__file__), "template", "runtime", "cloudformation-guard2.0", "rule_code.guard")
+        dst = os.path.join(os.getcwd(), rules_dir, self.args.rulename, "rule_code.guard")
+        shutil.copyfile(src, dst)
+        f = fileinput.input(files=dst, inplace=True)
+        for line in f:
+            if self.args.resource_types:
+                rule_name = self.args.rulename.replace("-", "_")
+                resource_types = ",".join(["'" + typ + "'" for typ in self.args.resource_types.split(",")])
+                print(
+                    line.replace("<%RuleName%>", rule_name)
+                    .replace(
+                        "<%ApplicableResources%>",
+                        " when resourceType IN [" + resource_types + "]",
+                    ),
+                    end="",
+                )
+            else:
+                print(line.replace("<%RuleName%>", rule_name), end="")
+        f.close()
+
     def __print_log_event(self, event):
         time_string = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(event["timestamp"] / 1000))
 
@@ -3465,6 +3527,14 @@ def __parse_rule_args(self, is_required):
             print("You must specify either a resource type trigger or a maximum frequency.")
             sys.exit(1)
 
+        if is_required and self.args.runtime == "cloudformation-guard2.0":
+            if self.args.maximum_frequency:
+                print("maximum frequency can not be used on a custom policy rule.")
+                sys.exit(1)
+            if not self.args.resource_types:
+                print("You must specify a resource type for a custom policy rule.")
+                sys.exit(1)
+
         if self.args.input_parameters:
             try:
                 print(self.args.input_parameters)

rdk/template/runtime/cloudformation-guard2.0/rule_code.guard

@@ -0,0 +1,3 @@
+rule <%RuleName%><%ApplicableResources%> {
+    # Add your custom logic here
+}