AWS Config custom rule to evaluate AWS account tags

template.yml

# aws cloudformation deploy \
#   --profile mgmt \
#   --template-file ./template.yml \
#   --stack-name ConfigRuleAccountTags \
#   --capabilities CAPABILITY_IAM

Resources:
  ConfigRule:
    Type: AWS::Config::ConfigRule
    DependsOn: EvaluationFunctionConfigPermission
    Properties:
      ConfigRuleName: aws-accounts-required-tags
      Description: AWS Accounts must have required tags
      Source:
        Owner: CUSTOM_LAMBDA
        SourceIdentifier: !Sub "${EvaluationFunction.Arn}"
        SourceDetails:
          - MessageType: ScheduledNotification
            MaximumExecutionFrequency: One_Hour # One_Hour Three_Hours Six_Hours Twelve_Hours TwentyFour_Hours
            EventSource: aws.config

  EvaluationFunction:
    Type: AWS::Lambda::Function
    Properties:
      Role: !Sub "${EvaluationFunctionRole.Arn}"
      Handler: index.handler
      Timeout: 300
      Runtime: python3.11
      Tags:
        - Key: CfnStackId
          Value: !Ref AWS::StackId
      Code:
        ZipFile: |
          import boto3
          import os
          import json
          import re

          config = boto3.client("config", region_name=os.environ["AWS_REGION"])
          orgs = boto3.client("organizations", region_name=os.environ["AWS_REGION"])

          # Mapping of required tag keys and optionally tag value requirements
          required_tags = {
              "Environment": {"AllowedValues": ["prod", "nonprod"]},
              "Application": {},
              "Owner": {"Regex": "^.+@example.com$"},
          }

          def handler(event, context):
              # print(json.dumps(event))

              # print("Listing accounts")
              accts = []
              for pg in orgs.get_paginator("list_accounts").paginate():
                  accts += pg["Accounts"]
              # print(json.dumps(accts, default=str))

              evaluations = []
              orderingtime = json.loads(event["invokingEvent"])["notificationCreationTime"]

              for acct in accts:
                  print("Evaluating account:", acct["Id"])
                  tags = {}
                  for pg in orgs.get_paginator("list_tags_for_resource").paginate(ResourceId=acct['Id']):
                      for t in pg["Tags"]:
                          tags[t["Key"]] = t["Value"]

                  missing_requireds = []
                  invalid_values = []
                  evaluation = {
                      "ComplianceResourceType": "AWS::::Account",
                      "ComplianceResourceId": acct["Id"],
                      "ComplianceType": "COMPLIANT",
                      "Annotation": "",
                      "OrderingTimestamp": orderingtime,
                  }

                  for required_key, value_spec in required_tags.items():
                      if required_key not in tags:
                          evaluation["ComplianceType"] = "NON_COMPLIANT"
                          missing_requireds.append(required_key)
                          continue
                      if "AllowedValues" in value_spec:
                          if tags[required_key] not in value_spec["AllowedValues"]:
                              evaluation["ComplianceType"] = "NON_COMPLIANT"
                              invalid_values.append(required_key)
                      if "Regex" in value_spec:
                          if not re.search(value_spec["Regex"], tags[required_key]):
                              evaluation["ComplianceType"] = "NON_COMPLIANT"
                              invalid_values.append(required_key)

                  evaluation["Annotation"] = f"Missing required tags: {missing_requireds}. Invalid value tags: {invalid_values}."

                  evaluations.append(evaluation)

              print("Submitting evaluations:")
              print(json.dumps(evaluations, default=str))

              response = config.put_evaluations(Evaluations=evaluations, ResultToken=event["resultToken"])

              print("PutEvaluations response:")
              print(json.dumps(response, default=str))


  EvaluationFunctionConfigPermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Sub "${EvaluationFunction.Arn}"
      Action: lambda:InvokeFunction
      Principal: config.amazonaws.com
      SourceAccount: !Ref AWS::AccountId

  EvaluationFunctionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: ["sts:AssumeRole"]
      ManagedPolicyArns:
        - !Sub "arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
      Policies:
        - PolicyName: Inline
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Resource: "*"
                Action:
                  - config:PutEvaluations
                  - organizations:ListTagsForResource
                  - organizations:ListAccounts

  EvaluationFunctionLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub '/aws/lambda/${EvaluationFunction}'
      RetentionInDays: 14
      Tags:
        - Key: CfnStackId
          Value: !Ref AWS::StackId