Utility to expand a YAML template into a cloud-custodian policy for taggable AWS resources

expand-tagging-policy-template.py

#!/usr/bin/env python
# encoding: utf-8
"""
Given a YAML template, expand the `policy_templates` in the provided YAML
template and generate a YAML file containing each policy expanded for every
taggable AWS resource type.
"""

from __future__ import absolute_import, division, print_function

import argparse

import yaml

# FIXME: there should be some way to scrape this from the schema rather than
# copying the list from
# https://gist.github.com/jtroberts83/56c1abe81e273f495c6754020409b480 and
# removing the ones which don't support tagging. The c7n resources hierarchy
# doesn't obviously expose it but schema validation gets it from somewhere.

AWS_TAGGABLE_RESOURCES = [
    "acm-certificate",
    "ami",
    "app-elb",
    "app-elb-target-group",
    "asg",
    "cache-cluster",
    "cache-snapshot",
    # Untaggable: "cache-subnet-group",
    "cfn",
    "cloud-directory",
    "cloudhsm-cluster",
    # Untaggable: "cloudsearch",
    "codebuild",
    # Untaggable: "codecommit",
    # Untaggable: "codepipeline",
    # Untaggable: "customer-gateway",
    # Untaggable: "directconnect",
    "directory",
    # Untaggable: "dynamodb-backup",
    # Untaggable: "dynamodb-stream",
    "dynamodb-table",
    "ebs",
    "ebs-snapshot",
    "ec2",
    "ec2-reserved",
    "ecr",
    "ecs",
    "ecs-container-instance",
    "ecs-service",
    "ecs-task",
    "ecs-task-definition",
    "efs",
    # Untaggable: "efs-mount-target",
    # Untaggable: "eks",
    # Untaggable: "elasticbeanstalk",
    # Untaggable: "elasticbeanstalk-environment",
    "elasticsearch",
    "elb",
    # Untaggable: "emr",
    "eni",
    # Untaggable: "firehose",
    "fsx",
    "fsx-backup",
    "glacier",
    "hostedzone",
    # Untaggable: "hsm",
    # Untaggable: "hsm-client",
    # Untaggable: "hsm-hapg",
    # Untaggable: "iam-certificate",
    # Untaggable: "iam-group",
    # Untaggable: "iam-policy",
    # Untaggable: "iam-profile",
    "iam-role",
    "iam-user",
    # Untaggable: "identity-pool",
    "internet-gateway",
    "key-pair",
    "kinesis",
    # Untaggable:  "kinesis-analytics",
    # Untaggable:  "kms",
    "kms-key",
    "lambda",
    # Untaggable: "lightsail-db",
    # Untaggable:  "lightsail-elb",
    # Untaggable: "lightsail-instance",
    "message-broker",
    "nat-gateway",
    "network-acl",
    "network-addr",
    # Untaggable: "opswork-cm",
    # Untaggable: "opswork-stack",
    "peering-connection",
    "r53domain",
    "rds",
    "rds-snapshot",
    # Untaggable: "rds-subnet-group",
    "redshift",
    "redshift-snapshot",
    # Untaggable: "redshift-subnet-group",
    "route-table",
    # Untaggable: "rrset",
    "s3",
    "secrets-manager",
    "security-group",
    # Untaggable: "simpledb",
    # Untaggable: "sns",
    "sqs",
    # Untaggable: "storage-gateway",
    "subnet",
    "transit-attachment",
    "transit-gateway",
    "vpc",
    "vpc-endpoint",
    "vpn-connection",
    "vpn-gateway",
    # Untaggable: "waf",
    # Untaggable: "waf-regional",
]


def expand_template(template_file):
    template = yaml.safe_load(template_file)

    template_policies = template.get("policy_templates", None)

    if not template_policies or not isinstance(template_policies, (list, tuple)):
        raise ValueError(
            "Expected %s to have a policy_templates key containing a list of policies"
            % template_file.name
        )

    policies = []
    expanded = {"policies": policies}

    for template in template_policies:
        for resource in AWS_TAGGABLE_RESOURCES:
            policy = template.copy()
            policy["resource"] = "aws.%s" % resource
            policy["name"] = "%s-%s" % (policy["name"], resource)
            policies.append(policy)

    with open(template_file.name.replace("-template.yaml", ".yaml"), "w") as output:
        yaml.dump(expanded, output)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__.strip())
    parser.add_argument("template_file", type=argparse.FileType("r"))
    args = parser.parse_args()

    if not args.template_file.name.endswith("-template.yaml"):
        parser.error("Template filename must end in “-template.yaml”")

    expand_template(args.template_file)

tagging-requirements-template.yaml

policy_templates:
    - name: tagging-compliance
      description: Identify resources which lack our accounting tags
      filters:
          - 'tag:Environment': absent
          - 'tag:Service': absent
          - or:
                - 'tag:Owner': absent
                - 'tag:ResponsibleParty': absent
                - 'tag:Contact': absent
                - 'tag:Creator': absent
      actions:
          - type: tag
            key: Warning
            value: Untagged Instance