Sentinel policy to parse an AWS IAM Policy and fail based on not allowed actions

restrict-iam-policy.sentinel

import "tfplan/v2" as tfplan
import "json"
import "types"

// Parametized not allowed list
param not_allowed_actions default [
	"s3:*",
	"s3:GetObject",
	"s3:PutObject",
	"kms:*",
	"kms:Decrypt",
	"kms:Encrypt",
	# "ec2:Describe*",
]

# Function will filter all resource changes and only return those of the given type
# that are also being created or updated (i.e. ignore destroy)
find_resources = func(type) {
	resources = filter tfplan.resource_changes as address, rc {
		rc.type is type and
			rc.mode is "managed" and
			(rc.change.actions contains "create" or rc.change.actions contains "update")
	}
	return resources
}

# Function to take an input and normalize the result to be a list(string)
# If the input is a string, return as a single element list
# If its a list, return as list
normalize_as_list = func(data) {
	if types.type_of(data) == "list" {
		return data
	}
	if types.type_of(data) == "string" {
		return [data]
	}
	return error("Invalid type: ", data, types.type_of(data))
}

# Function to validate each IAM role in the plan
validate_iam_policy = func() {
	validated = true

	# Get all IAM Role Policies
	allRolePolicy = find_resources("aws_iam_role_policy")

	for allRolePolicy as address, rc {
		policy_json = rc.change.after.policy
		policy = json.unmarshal(policy_json)

		for policy.Statement as _, statement {
			# print("Statement", i, statement)
			effect = statement.Effect
			actions = normalize_as_list(statement.Action)
			resources = normalize_as_list(statement.Resource)

			# Debug prints
			# print("Actions", actions)
			# print("Resources", resources)
			# Only consider Allow effects
			if effect == "Allow" {
				# Filter through each action to see if it has an item in the not allowed list
				found_violation = filter actions as a {
					a in not_allowed_actions
				}

				# If a violation is found, print it
				if length(found_violation) > 0 {
					print("Invalid Allow action found", found_violation, "in", statement)
					validated = false
				}
			}
		}
	}
	return validated
}

# Validate every IAM policy in the plan
violatingIamPolicy = validate_iam_policy()

# Main rule
main = rule {
	violatingIamPolicy
}