Doppelte Kosten vermeiden: Automatische EFS-Backups und AWS-Backup-Pläne

yaml

Code kopiert!

AWSTemplateFormatVersion: '2010-09-09'
Description: >
  CloudFormation Template to create a Lambda function, IAM role, and AWS Config Rule to check if duplicate EFS backups exist.

Resources:
  # IAM Role for Lambda
  DuplicateEFSBackupCheckRole:
    Type: 'AWS::IAM::Role'
    Properties: 
      AssumeRolePolicyDocument: 
        Version: '2012-10-17'
        Statement: 
          - Effect: "Allow"
            Principal: 
              Service: 
                - 'lambda.amazonaws.com'
            Action: 
              - 'sts:AssumeRole'
      Policies:
        - PolicyName: 'DuplicateEFSBackupCheckPolicy'
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: 'Allow'
                Action:
                  - 'logs:CreateLogGroup'
                  - 'logs:CreateLogStream'
                  - 'logs:PutLogEvents'
                Resource: '*'
              - Effect: 'Allow'
                Action:
                  - 'elasticfilesystem:DescribeFileSystems'
                Resource: '*'
              - Effect: 'Allow'
                Action:
                  - 'backup:ListBackupVaults'
                  - 'backup:ListRecoveryPointsByBackupVault'
                Resource: '*'
              - Effect: 'Allow'
                Action:
                  - 'config:PutEvaluations'
                Resource: '*'

  # Lambda Function
  DuplicateEFSBackupCheckLambda:
    Type: 'AWS::Lambda::Function'
    Properties:
      Handler: 'index.lambda_handler'
      Role: !GetAtt 'DuplicateEFSBackupCheckRole.Arn'
      Code:
        ZipFile: |
          import boto3
          from botocore.exceptions import ClientError
          import json

          def lambda_handler(event, context):
              message = json.loads(event['invokingEvent'])
              
              if message['messageType'] == 'ScheduledNotification':
                  handle_scheduled_config(event, message, context)
              else:
                  print(f'Unexpected event {event}.')

          def handle_scheduled_config(event, message, context):
              efs_client = boto3.client('efs')
              backup_client = boto3.client('backup')
              
              # Get List of all EFS File Systems
              file_systems = efs_client.describe_file_systems()
              
              # Iterate over the EFS File Systems
              for fs in file_systems['FileSystems']:
                  compliance_status = 'COMPLIANT'
                  file_system_id = fs['FileSystemId']
                  region = boto3.session.Session().region_name
                  account_id = context.invoked_function_arn.split(":")[4]
                  
                  # Construct the File System ARN
                  file_system_arn = f'arn:aws:elasticfilesystem:{region}:{account_id}:file-system/{file_system_id}'
                  
                  # Get list of all backup vaults
                  backup_vaults = backup_client.list_backup_vaults()
                  
                  recovery_point_count = 0
                  vaults_with_recovery_points = []
                  
                  # Iterate over each backup vault
                  for vault in backup_vaults['BackupVaultList']:
                      vault_name = vault['BackupVaultName']
                      
                      # List recovery points for the file system in each vault
                      try:
                          recovery_points = backup_client.list_recovery_points_by_backup_vault(
                              BackupVaultName=vault_name,
                              ByResourceArn=file_system_arn
                          )
                          if 'RecoveryPoints' in recovery_points and recovery_points['RecoveryPoints']:
                              recovery_point_count += 1
                              vaults_with_recovery_points.append(vault_name)
                      except ClientError as e:
                          print(f"Error describing recovery points for vault {vault_name} and file system {file_system_id}: {e}")
                          continue
                  
                  # Print the vaults with recovery points for the current file system
                  if vaults_with_recovery_points:
                      print(f"File System {file_system_id} has recovery points in the following vaults: {', '.join(vaults_with_recovery_points)}")
                  
                  # If the file system has recovery points in more than one backup vault, mark as NON_COMPLIANT
                  if recovery_point_count > 1:
                      print(f"File System {file_system_id} is NON_COMPLIANT")
                      compliance_status = 'NON_COMPLIANT'
                  else:
                      print(f"File System {file_system_id} is COMPLIANT")
                  
                  # Send evaluation result to AWS Config
                  send_evaluation_result(file_system_id, compliance_status, message['notificationCreationTime'], event['resultToken'])

          def send_evaluation_result(resource_id, compliance_status, timestamp, result_token):
              # Prepare evaluation result
              evaluation_result = {
                  'ComplianceResourceType': 'AWS::EFS::FileSystem',
                  'ComplianceResourceId': resource_id,
                  'ComplianceType': compliance_status,
                  'OrderingTimestamp': timestamp
              }
              
              # Create AWS Config client
              config_client = boto3.client('config')
              
              try:
                  # Put the evaluation result to AWS Config
                  config_client.put_evaluations(
                      Evaluations=[evaluation_result],
                      ResultToken=result_token
                  )
              except ClientError as e:
                  print(f"Error sending evaluation result to AWS Config: {e}")
      Runtime: 'python3.12'
      Timeout: 300

  # Lambda Permission
  DuplicateEFSBackupCheckLambdaPermission:
    Type: 'AWS::Lambda::Permission'
    Properties:
      FunctionName: !GetAtt 'DuplicateEFSBackupCheckLambda.Arn'
      Action: 'lambda:InvokeFunction'
      Principal: 'config.amazonaws.com'
      SourceAccount: !Sub ${AWS::AccountId}

  # AWS Config Rule
  DuplicateEFSBackupCheckConfigRule:
    Type: 'AWS::Config::ConfigRule'
    Properties: 
      ConfigRuleName: 'duplicate-efs-backup-check'
      Description: 'Checks if an EFS file system has backups in more than one backup vault.'
      Scope: 
        ComplianceResourceTypes: 
          - 'AWS::EFS::FileSystem'
      Source: 
        Owner: 'CUSTOM_LAMBDA'
        SourceIdentifier: !GetAtt 'DuplicateEFSBackupCheckLambda.Arn'
        SourceDetails: 
          - EventSource: 'aws.config'
            MessageType: 'ScheduledNotification'
            MaximumExecutionFrequency: 'Twelve_Hours'