Add GitHub action for validating IAM policies in CloudFormation templates

Author: mponaws

Dockerfile

@@ -0,0 +1,8 @@
+# FROM public.ecr.aws/lambda/python:3.10
+FROM python:3.10
+# Install cfn-policy-validator
+RUN  pip install cfn-policy-validator==0.0.31
+
+COPY main.py /main.py
+
+ENTRYPOINT ["python3", "/main.py"]

action.yaml

@@ -0,0 +1,59 @@
+name: 'Policy checks to validate AWS IAM policies in CloudFormation templates" Action For GitHub Actions'
+description: "Uses ValidatePolicy, CheckAccessNotGranted, CheckNoNewAccess APIs from AWS Access Analyzer for policy checks - https://docs.aws.amazon.com/access-analyzer/latest/APIReference/"
+branding:
+  icon: "cloud"
+  color: "orange"
+inputs:
+  policy-check-type:
+    description: "Type of the policy check. Valid values: VALIDATE_POLICY, CHECK_NO_NEW_ACCESS, CHECK_ACCESS_NOT_GRANTED"
+    required: true
+  template-path:
+    description: "The path to the CloudFormation template."
+    required: true
+  region:
+    description: "The destination region the resources will be deployed to."
+    required: true
+  parameters:
+    description: "Keys and values for CloudFormation template parameters. Only parameters that are referenced by IAM policies in the template are required. Example format - KEY=VALUE [KEY=VALUE ...]"
+  template-configuration-file:
+    description: "A JSON formatted file that specifies template parameter values, a stack policy, and tags. Only parameters are used from this file. Everything else is ignored. Identical values passed in the --parameters flag override parameters in this file. See CloudFormation documentation for file format: https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/continuous-delivery-codepipeline-cfn-artifacts.html"
+  ignore-finding:
+    description: 'Allow validation failures to be ignored. Specify as a comma separated list of findings to be ignored. Can be individual finding codes (e.g. "PASS_ROLE_WITH_STAR_IN_RESOURCE"), a specific resource name (e.g. "MyResource"), or a combination of both separated by a period.(e.g. "MyResource.PASS_ROLE_WITH_STAR_IN_RESOURCE"). Names of finding codes may change in IAM Access Analyzer over time. Valid options: FINDING_CODE,RESOURCE_NAME,RESOURCE_NAME.FINDING_CODE'
+  actions:
+    description: 'List of comma-separated actions. Example format - ACTION,ACTION,ACTION. This attribute is considered and required when policy-check-type is "CHECK_ACCESS_NOT_GRANTED"'
+  reference-policy:
+    description: 'A JSON formatted file that specifies the path to the reference policy that is used for a permissions comparison. This attribute is considered and required when policy-check-type is "CHECK_NO_NEW_ACCESS"'
+  reference-policy-type:
+    description: 'The policy type associated with the IAM policy under analysis and the reference policy. Valid values: IDENTITY, RESOURCE. This attribute is considered and required when policy-check-type is "CHECK_NO_NEW_ACCESS"'
+  treat-finding-type-as-blocking:
+    description: 'Specify which finding types should be treated as blocking. Other finding types are treated as non blocking. If the tool detects any blocking finding types, it will exit with a non-zero exit code. If all findings are non blocking or there are no findings, the tool exits with an exit code of 0. Defaults to "ERROR" and "SECURITY_WARNING". Specify as a comma separated list of finding types that should be blocking. Pass "NONE" to ignore all findings. This attribute is considered only when policy-check-type is "VALIDATE_POLICY"'
+  treat-findings-as-non-blocking:
+    description: 'When not specified, the tool detects any findings, it will exit with a non-zero exit code. When specified, the tool exits with an exit code of 0. This attribute is considered only when policy-check-type is "CHECK_NO_NEW_ACCESS" or "CHECK_ACCESS_NOT_GRANTED"'
+    default: "False"
+  allow-external-principals:
+    description: 'A comma separated list of external principals that should be ignored. Specify as a comma separated list of a 12 digit AWS account ID, a federated web identity user, a federated SAML user, or an ARN. Specify \"*\" to allow anonymous access. (e.g. 123456789123,arn:aws:iam::111111111111:role/MyOtherRole,graph.facebook.com). Valid options: ACCOUNT,ARN". This attribute is considered only when policy-check-type is "VALIDATE_POLICY"'
+  allow-dynamic-ref-without-version:
+    description: "Override the default behavior and allow dynamic SSM references without version numbers. The version number ensures that the SSM parameter value that was validated is the one that is deployed."
+  exclude-resource-types:
+    description: "List of comma-separated resource types. Resource types should be the same as Cloudformation template resource names such as AWS::IAM::Role, AWS::S3::Bucket. Valid option syntax: AWS::SERVICE::RESOURCE"
+outputs:
+  result:
+    description: "Result of the policy checks"
+runs:
+  using: "docker"
+  image: Dockerfile
+  args:
+    - ${{ inputs.policy-check-type}}
+    - ${{ inputs.template-path }}
+    - ${{ inputs.region }}
+    - ${{ inputs.parameters }}
+    - ${{ inputs.template-configuration-file }}
+    - ${{ inputs.ignore-finding }}
+    - ${{ inputs.actions }}
+    - ${{ inputs.reference-policy }}
+    - ${{ inputs.reference-policy-type }}
+    - ${{ inputs.treat-finding-type-as-blocking }}
+    - ${{ inputs.treat-findings-as-non-blocking }}
+    - ${{ inputs.allow-external-principals }}
+    - ${{ inputs.allow-dynamic-ref-without-version }}
+    - ${{ inputs.exclude-resource-types }}

main.py

@@ -0,0 +1,213 @@
+import os
+import re
+import subprocess
+import sys
+import traceback
+
+VALIDATE_POLICY = "VALIDATE_POLICY"
+CHECK_NO_NEW_ACCESS = "CHECK_NO_NEW_ACCESS"
+CHECK_ACCESS_NOT_GRANTED = "CHECK_ACCESS_NOT_GRANTED"
+
+CLI_POLICY_VALIDATOR = "cfn-policy-validator"
+
+TREAT_FINDINGS_AS_NON_BLOCKING = "INPUT_TREAT-FINDINGS-AS-NON-BLOCKING"
+
+POLICY_CHECK_TYPE = "INPUT_POLICY-CHECK-TYPE"
+
+# excluding the "INPUT_POLICY-CHECK-TYPE". Contains only other required inputs in cfn-policy-validator
+COMMON_REQUIRED_INPUTS = {"INPUT_TEMPLATE-PATH", "INPUT_REGION"}
+
+VALIDATE_POLICY_SPECIFIC_REQUIRED_INPUTS = set()
+
+CHECK_NO_NEW_ACCESS_SPECIFIC_REQUIRED_INPUTS = {
+    "INPUT_TEMPLATE-PATH",
+    "INPUT_REGION",
+    "INPUT_REFERENCE-POLICY",
+    "INPUT_REFERENCE-POLICY-TYPE",
+}
+
+CHECK_ACCESS_NOT_GRANTED_SPECIFIC_REQUIRED_INPUTS = {"INPUT_ACTIONS"}
+
+# excluding the "INPUT_POLICY-CHECK-TYPE". Contains only other required inputs in cfn-policy-validator
+COMMON_OPTIONAL_INPUTS = {
+    "INPUT_PARAMETERS",
+    "INPUT_TEMPLATE-CONFIGURATION-FILE",
+    "INPUT_IGNORE-FINDING",
+    "INPUT_ALLOW-DYNAMIC-REF-WITHOUT-VERSION",
+    "INPUT_EXCLUDE-RESOURCE-TYPES",
+}
+
+VALIDATE_POLICY_SPECIFIC_OPTIONAL_INPUTS = {
+    "INPUT_ALLOW-EXTERNAL-PRINCIPALS",
+    "INPUT_TREAT-FINDING-TYPE-AS-BLOCKING",
+}
+
+# Excluding the TREAT-FINDINGS-AS-NON-BLOCKING which is a flag and needs special handling
+CHECK_NO_NEW_ACCESS_SPECIFIC_OPTIONAL_INPUTS = set()
+
+# Excluding the TREAT-FINDINGS-AS-NON-BLOCKING which is a flag and needs special handling
+CHECK_ACCESS_NOT_GRANTED_SPECIFIC_OPTIONAL_INPUTS = set()
+
+
+VALID_POLICY_CHECK_TYPES = [
+    VALIDATE_POLICY,
+    CHECK_NO_NEW_ACCESS,
+    CHECK_ACCESS_NOT_GRANTED,
+]
+
+# Name of the output defined in the GitHub action schema
+ACTION_OUTPUT_RESULT = "result"
+
+
+def main():
+    policy_check = get_policy_check_type()
+    required_inputs = get_required_inputs(policy_check)
+    optional_inputs = get_optional_inputs(policy_check)
+    command_lst = build_command(
+        policy_check, required_inputs=required_inputs, optional_inputs=optional_inputs
+    )
+    result = execute_command(command_lst)
+    formatted_result = format_result(result)
+    set_github_action_output(ACTION_OUTPUT_RESULT, formatted_result)
+    return
+
+
+# Get the policy check name
+def get_policy_check_type():
+    policy_check = os.environ[POLICY_CHECK_TYPE]
+    if policy_check not in VALID_POLICY_CHECK_TYPES:
+        raise ValueError(
+            "Invalid value of policy-check-type: {}. Valid values are: {}".format(
+                policy_check, VALID_POLICY_CHECK_TYPES
+            )
+        )
+    return policy_check
+
+
+def get_flag_name(val):
+    return val.removeprefix("INPUT_").lower()
+
+
+def get_required_inputs(policy_check):
+    required_inputs = {}
+    check_specific_required_inputs = None
+    if policy_check == VALIDATE_POLICY:
+        check_specific_required_inputs = VALIDATE_POLICY_SPECIFIC_REQUIRED_INPUTS
+    elif policy_check == CHECK_NO_NEW_ACCESS:
+        check_specific_required_inputs = CHECK_NO_NEW_ACCESS_SPECIFIC_REQUIRED_INPUTS
+    elif policy_check == CHECK_ACCESS_NOT_GRANTED:
+        check_specific_required_inputs = (
+            CHECK_ACCESS_NOT_GRANTED_SPECIFIC_REQUIRED_INPUTS
+        )
+    required_inputs = COMMON_REQUIRED_INPUTS.union(check_specific_required_inputs)
+    return required_inputs
+
+
+def get_optional_inputs(policy_check):
+    optional_inputs = {}
+    check_specific_optional_inputs = None
+    if policy_check == VALIDATE_POLICY:
+        check_specific_optional_inputs = VALIDATE_POLICY_SPECIFIC_OPTIONAL_INPUTS
+    elif policy_check == CHECK_NO_NEW_ACCESS:
+        check_specific_optional_inputs = CHECK_NO_NEW_ACCESS_SPECIFIC_OPTIONAL_INPUTS
+    elif policy_check == CHECK_ACCESS_NOT_GRANTED:
+        check_specific_optional_inputs = (
+            CHECK_ACCESS_NOT_GRANTED_SPECIFIC_OPTIONAL_INPUTS
+        )
+    optional_inputs = check_specific_optional_inputs.union(COMMON_OPTIONAL_INPUTS)
+    return optional_inputs
+
+
+def build_command(policy_check_type, required_inputs, optional_inputs):
+    cli_tool_name = CLI_POLICY_VALIDATOR
+    command_lst = []
+    cli_operation_name = (
+        "validate"
+        if policy_check_type == VALIDATE_POLICY
+        else policy_check_type.replace("_", "-").lower()
+    )
+
+    sub_command_required_lst = get_sub_command(required_inputs, True)
+    sub_command_optional_lst = get_sub_command(optional_inputs, False)
+
+    command_lst.append(cli_tool_name)
+    command_lst.append(cli_operation_name)
+    command_lst.extend(sub_command_required_lst)
+    command_lst.extend(sub_command_optional_lst)
+
+    treat_findings_as_non_blocking_flag = get_treat_findings_as_non_blocking_flag(
+        policy_check_type
+    )
+    if len(treat_findings_as_non_blocking_flag) > 0:
+        command_lst.extend(get_treat_findings_as_non_blocking_flag(policy_check_type))
+    return command_lst
+
+
+def get_sub_command(inputFields, areRequiredFields):
+    flags = []
+
+    for input in inputFields:
+        # The default values to these environment variable when passed to docker is empty string through GitHub Actions
+        if os.environ[input] != "":
+            flag_name = get_flag_name(input)
+            flags.extend(["--{}".format(flag_name), os.environ[input]])
+        elif areRequiredFields:
+            raise ValueError("Missing value for required field: {}", input)
+
+    return flags
+
+
+def get_treat_findings_as_non_blocking_flag(policy_check):
+    # This is specific to custom checks - CheckAccessNotGranted & CheckNoNewAccess
+    if policy_check in (CHECK_ACCESS_NOT_GRANTED, CHECK_NO_NEW_ACCESS):
+        val = os.environ[TREAT_FINDINGS_AS_NON_BLOCKING]
+        if val == "True":
+            return ["--{}".format(get_flag_name(TREAT_FINDINGS_AS_NON_BLOCKING))]
+        elif val == "False":
+            return ""
+        else:
+            raise ValueError(
+                "Invalid value for {}: {}".format(TREAT_FINDINGS_AS_NON_BLOCKING, val)
+            )
+    return ""
+
+
+def execute_command(command):
+    try:
+        result = subprocess.run(
+            command, check=True, stdout=subprocess.PIPE, encoding="utf-8"
+        ).stdout
+        return result
+    except subprocess.CalledProcessError as err:
+        print(
+            "error code: {}, traceback: {}, output: {}".format(
+                err.returncode, err.with_traceback, err.output
+            )
+        )
+        raise
+    except Exception as err:
+        print(f"Unexpected {err=}, {type(err)=}")
+        raise
+
+
+def format_result(result):
+    result = re.sub(r"[\n\t\s]*", "", result)
+    print("result={}".format(result))
+    return result
+
+
+#  Output value should be set by writing to the outputs in the environment file
+#  https://docs.github.com/en/actions/using-workflows/workflow-commands-for-github-actions#setting-an-output-parameter
+def set_github_action_output(output_name, output_value):
+    with open(os.path.abspath(os.environ["GITHUB_OUTPUT"]), "a") as f:
+        f.write(f"{output_name}={output_value}")
+    return
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as e:
+        traceback.print_exc()
+        print(f"ERROR: Unexpected error occurred. {str(e)}", file=sys.stderr)
+        exit(1)