Control Tower環境下でSecurityHubのセキュリティ基準・コントロール無効化を自動化する

AWSTemplateFormatVersion: "2010-09-09"
Parameters:
  DisableList:
    Type: CommaDelimitedList
    Default: ""
  EnableList:
    Type: CommaDelimitedList
    Default: ""

Conditions:
  IsDisableListCondition:
    Fn::Not: [!Equals [!Join [",", !Ref DisableList], ""]]
  IsEnableListCondition:
    Fn::Not: [!Equals [!Join [",", !Ref EnableList], ""]]

Resources:
  DisableSecurityHubLambda:
    Type: Custom::DisableSecurityHubLambda
    Condition: IsDisableListCondition
    Properties:
      ServiceToken: !GetAtt "DisableLambdaFunction.Arn"
      DisableList: !Ref DisableList

  EnableSecurityHubLambda:
    Type: Custom::DisableSecurityHubLambda
    Condition: IsEnableListCondition
    Properties:
      ServiceToken: !GetAtt "EnableLambdaFunction.Arn"
      EnableList: !Ref EnableList

  DisableLambdaFunction:
    Type: AWS::Lambda::Function
    Properties:
      Role: !GetAtt "LambdaExecutionRole.Arn"
      Runtime: "python3.8"
      Handler: index.lambda_handler
      Timeout: "180"
      Code:
        ZipFile: |
          import sys
          import boto3
          import cfnresponse
          from logging import getLogger, INFO

          logger = getLogger()
          logger.setLevel(INFO)

          def disable_securityhub(event, context):
              # params
              logger.info('[START] disable_securityhub')
              standard_name = "cis-aws-foundations-benchmark/v/1.2.0"
              disable_list = event['ResourceProperties']['DisableList']
              disable_controls = {"aws-foundational-security-best-practices/v/1.0.0": disable_list}
              target_regions = ["us-east-1", "ap-northeast-1", "ap-southeast-1"]
              disable_reason = "init disable"

              try:
                  account_id = boto3.client('sts').get_caller_identity()['Account']

                  ec2_client = boto3.client('ec2')
                  regions = ec2_client.describe_regions()['Regions']

                  for region in regions:
                      region_name = region['RegionName']
                      logger.info("# " + region_name)
                      if region_name not in target_regions:
                          logger.info("skip region.")
                          continue
                      securityhub = boto3.client('securityhub', region_name=region_name)
                      # check enable security hub
                      try:
                          securityhub.get_enabled_standards()
                      except securityhub.exceptions.InvalidAccessException as e:
                          logger.info("Security Hub is disabled.")
                          continue
                      # disable standard
                      std_subsc_arn = "arn:aws:securityhub:{}:{}:subscription/{}".format(
                          region_name, account_id, standard_name)
                      res = securityhub.batch_disable_standards(
                          StandardsSubscriptionArns=[std_subsc_arn])
                      if res['ResponseMetadata']['HTTPStatusCode'] == 200:
                          logger.info('Disable Standard Success.')
                      else:
                          logger.info('Disable Standard Failed.')
                      # disable control
                      for standard in disable_controls:
                          logger.info('Disable Controls in ' + standard)
                          for control in disable_controls[standard]:
                              logger.info('  Target Control: ' + control)
                              ctl_arn = "arn:aws:securityhub:{}:{}:control/{}/{}".format(
                                  region_name, account_id, standard, control)
                              res = securityhub.update_standards_control(
                                  StandardsControlArn=ctl_arn,
                                  ControlStatus='DISABLED',
                                  DisabledReason=disable_reason)
                              if res['ResponseMetadata']['HTTPStatusCode'] == 200:
                                  logger.info('  Disable Success.')
                              else:
                                  logger.info('  Disable Failed.')
                  logger.info('[END] disable_securityhub')
              except Exception as e:
                  logger.error(e)
                  cfnresponse.send(event, context, cfnresponse.FAILED, {'Response': 'Failure'})
                  exit()


          def lambda_handler(event, context):
              logger.info('[START] lambda_handler')
              if event['RequestType'] == 'Create':
                  disable_securityhub(event, context)
                  cfnresponse.send(event, context, cfnresponse.SUCCESS,
                                              {'Response': 'Success'})
              if event['RequestType'] == 'Delete':
                  cfnresponse.send(event, context, cfnresponse.SUCCESS,
                                      {'Response': 'Success'})
              if event['RequestType'] == 'Update':
                  disable_securityhub(event, context)
                  cfnresponse.send(event, context, cfnresponse.SUCCESS,
                                      {'Response': 'Success'})
              logger.info('[END] lambda_handler')

  EnableLambdaFunction:
    Type: AWS::Lambda::Function
    Properties:
      Role: !GetAtt "LambdaExecutionRole.Arn"
      Runtime: "python3.8"
      Handler: index.lambda_handler
      Timeout: "180"
      Code:
        ZipFile: |
          import sys
          import boto3
          import cfnresponse
          from logging import getLogger, INFO

          logger = getLogger()
          logger.setLevel(INFO)

          def enable_securityhub(event, context):
              # params
              logger.info('[START] enable_securityhub_control')
              enable_list = event['ResourceProperties']['EnableList']
              enable_controls = {"aws-foundational-security-best-practices/v/1.0.0": enable_list}
              target_regions = ["ap-northeast-1"]

              try:
                  account_id = boto3.client('sts').get_caller_identity()['Account']

                  ec2_client = boto3.client('ec2')
                  regions = ec2_client.describe_regions()['Regions']

                  for region in regions:
                      region_name = region['RegionName']
                      logger.info("# " + region_name)
                      if region_name not in target_regions:
                          logger.info("skip region.")
                          continue
                      securityhub = boto3.client('securityhub', region_name=region_name)

                      # enable control
                      for standard in enable_controls:
                          logger.info('Enable Controls in ' + standard)
                          for control in enable_controls[standard]:
                              logger.info('  Target Control: ' + control)
                              ctl_arn = "arn:aws:securityhub:{}:{}:control/{}/{}".format(
                                  region_name, account_id, standard, control)
                              res = securityhub.update_standards_control(
                                  StandardsControlArn=ctl_arn,
                                  ControlStatus='ENABLED')
                              if res['ResponseMetadata']['HTTPStatusCode'] == 200:
                                  logger.info('  Enable Success.')
                              else:
                                  logger.info('  Enable Failed.')
                  logger.info('[END] enable_securityhub_control')
              except Exception as e:
                  logger.error(e)
                  cfnresponse.send(event, context, cfnresponse.FAILED, {'Response': 'Failure'})
                  exit()


          def lambda_handler(event, context):
              logger.info('[START] lambda_handler')
              if event['RequestType'] == 'Create':
                  enable_securityhub(event, context)
                  cfnresponse.send(event, context, cfnresponse.SUCCESS,
                                              {'Response': 'Success'})
              if event['RequestType'] == 'Delete':
                  cfnresponse.send(event, context, cfnresponse.SUCCESS,
                                      {'Response': 'Success'})
              if event['RequestType'] == 'Update':
                  enable_securityhub(event, context)
                  cfnresponse.send(event, context, cfnresponse.SUCCESS,
                                      {'Response': 'Success'})
              logger.info('[END] lambda_handler')
  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
            Action:
              - sts:AssumeRole
      Path: /
      Policies:
        - PolicyName: lambda-disable-securityhub-policy
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: arn:aws:logs:*:*:*
              - Effect: Allow
                Action:
                  - securityhub:GetEnabledStandards
                  - securityhub:BatchDisableStandards
                  - securityhub:UpdateStandardsControl
                  - ec2:DescribeRegions
                  - sts:GetCallerIdentity
                Resource: "*"