detecting-aws-guardduty-findings-automation

name	detecting-aws-guardduty-findings-automation
description	使用 EventBridge 和 Lambda 自动化处理 AWS GuardDuty 威胁检测发现，实现实时事件响应、自动隔离受损资源和安全通知工作流。
domain	cybersecurity
subdomain	cloud-security
tags	["aws","guardduty","eventbridge","lambda","threat-detection","automation","incident-response","siem"]
version	1.0
author	mahipal
license	Apache-2.0

检测 AWS GuardDuty 发现自动化

概述

Amazon GuardDuty 是一种威胁检测服务，持续监控 AWS 账户的恶意活动和未授权行为。通过将 GuardDuty 与 Amazon EventBridge 和 AWS Lambda 集成，安全团队实现自动化、实时的威胁响应，将平均响应时间（MTTR）从数小时缩短到数秒。GuardDuty 分析 VPC 流日志、CloudTrail 管理和数据事件、DNS 日志、EKS 审计日志和 S3 数据事件。

前置条件

• 已启用 GuardDuty 的 AWS 账户
• Lambda 执行的 IAM 角色
• 为 GuardDuty 事件配置的 EventBridge
• 用于安全通知的 SNS 主题
• Security Hub 集成（推荐）

启用 GuardDuty

# 启用 GuardDuty
aws guardduty create-detector --enable --finding-publishing-frequency FIFTEEN_MINUTES

# 启用附加数据源
aws guardduty update-detector \
  --detector-id DETECTOR_ID \
  --data-sources '{
    "S3Logs": {"Enable": true},
    "Kubernetes": {"AuditLogs": {"Enable": true}},
    "MalwareProtection": {"ScanEc2InstanceWithFindings": {"EbsVolumes": true}},
    "RuntimeMonitoring": {"Enable": true}
  }'

EventBridge 规则配置

高严重性发现的规则

{
  "source": ["aws.guardduty"],
  "detail-type": ["GuardDuty Finding"],
  "detail": {
    "severity": [{"numeric": [">=", 7.0]}]
  }
}

通过 CLI 创建 EventBridge 规则

aws events put-rule \
  --name "guardduty-high-severity" \
  --event-pattern '{
    "source": ["aws.guardduty"],
    "detail-type": ["GuardDuty Finding"],
    "detail": {
      "severity": [{"numeric": [">=", 7.0]}]
    }
  }'

aws events put-targets \
  --rule "guardduty-high-severity" \
  --targets "Id"="lambda-handler","Arn"="arn:aws:lambda:us-east-1:123456789012:function:guardduty-response"

Lambda 自动响应函数

EC2 实例隔离

import boto3
import json
import os

ec2 = boto3.client('ec2')
sns = boto3.client('sns')

QUARANTINE_SG = os.environ.get('QUARANTINE_SECURITY_GROUP')
SNS_TOPIC = os.environ.get('SNS_TOPIC_ARN')

def lambda_handler(event, context):
    finding = event['detail']
    finding_type = finding['type']
    severity = finding['severity']
    account_id = finding['accountId']
    region = finding['region']

    # 提取资源信息
    resource = finding.get('resource', {})
    resource_type = resource.get('resourceType', '')

    if resource_type == 'Instance':
        instance_id = resource['instanceDetails']['instanceId']
        instance_tags = {t['key']: t['value']
                        for t in resource['instanceDetails'].get('tags', [])}

        # 如果已隔离则跳过
        if instance_tags.get('SecurityStatus') == 'Quarantined':
            return {'statusCode': 200, 'body': 'Already quarantined'}

        # 获取当前安全组用于取证
        instance = ec2.describe_instances(InstanceIds=[instance_id])
        current_sgs = [sg['GroupId'] for sg in
                       instance['Reservations'][0][][][]]

        
        ec2.create_tags(
            Resources=[instance_id],
            Tags=[
                {: , : },
                {: , : finding_type},
                {: , : .join(current_sgs)},
                {: , : finding[]}
            ]
        )

        
         QUARANTINE_SG:
            ec2.modify_instance_attribute(
                InstanceId=instance_id,
                Groups=[QUARANTINE_SG]
            )

        
        volumes = ec2.describe_volumes(
            Filters=[{: , : [instance_id]}]
        )
         vol  volumes[]:
            ec2.create_snapshot(
                VolumeId=vol[],
                Description=,
                TagSpecifications=[{
                    : ,
                    : [
                        {: , : },
                        {: , : instance_id},
                        {: , : finding_type}
                    ]
                }]
            )

        
        sns.publish(
            TopicArn=SNS_TOPIC,
            Subject=,
            Message=json.dumps({
                : ,
                : instance_id,
                : finding_type,
                : severity,
                : account_id,
                : region,
                : current_sgs,
                : finding.get(, )
            }, indent=)
        )

         {
            : ,
            : 
        }

     {: , : }

IAM 凭据入侵响应

import boto3
import json
import os

iam = boto3.client('iam')
sns = boto3.client('sns')

SNS_TOPIC = os.environ.get('SNS_TOPIC_ARN')

def lambda_handler(event, context):
    finding = event['detail']
    finding_type = finding['type']

    if 'IAMUser' not in finding_type and 'UnauthorizedAccess' not in finding_type:
        return {'statusCode': 200, 'body': 'Not an IAM finding'}

    resource = finding.get('resource', {})
    access_key_details = resource.get('accessKeyDetails', {})
    user_name = access_key_details.get('userName', '')
    access_key_id = access_key_details.get('accessKeyId', '')

    if not user_name:
        return {'statusCode': 200, 'body': 'No user identified'}

    actions_taken = []

    # 停用受损的访问密钥
    if access_key_id and access_key_id != 'GeneratedFindingAccessKeyId':
        try:
            iam.update_access_key(
                UserName=user_name,
                AccessKeyId=access_key_id,
                Status='Inactive'
            )
            actions_taken.append(f'已停用访问密钥 {access_key_id}')
         Exception  e:
            actions_taken.append()

    
    deny_policy = {
        : ,
        : [{
            : ,
            : ,
            : 
        }]
    }

    :
        iam.put_user_policy(
            UserName=user_name,
            PolicyName=,
            PolicyDocument=json.dumps(deny_policy)
        )
        actions_taken.append()
     Exception  e:
        actions_taken.append()

    
    sns.publish(
        TopicArn=SNS_TOPIC,
        Subject=,
        Message=json.dumps({
            : finding_type,
            : user_name,
            : access_key_id,
            : actions_taken,
            : finding[]
        }, indent=)
    )

     {: , : json.dumps(actions_taken)}

Terraform 部署

resource "aws_guardduty_detector" "main" {
  enable = true
  finding_publishing_frequency = "FIFTEEN_MINUTES"

  datasources {
    s3_logs { enable = true }
    kubernetes { audit_logs { enable = true } }
    malware_protection {
      scan_ec2_instance_with_findings {
        ebs_volumes { enable = true }
      }
    }
  }
}

resource "aws_cloudwatch_event_rule" "guardduty_high" {
  name        = "guardduty-high-severity"
  description = "GuardDuty 高严重性发现"

  event_pattern = jsonencode({
    source      = ["aws.guardduty"]
    detail-type = ["GuardDuty Finding"]
    detail = {
      severity = [{ numeric = [">=", 7.0] }]
    }
  })
}

resource "aws_cloudwatch_event_target" "lambda" {
  rule = aws_cloudwatch_event_rule.guardduty_high.name
  arn  = aws_lambda_function.guardduty_response.arn
}

发现类别

类别	严重性范围	示例
Backdoor	5.0 - 8.0	Backdoor:EC2/C&CActivity
CryptoCurrency	5.0 - 8.0	CryptoCurrency:EC2/BitcoinTool
Trojan	5.0 - 8.0	Trojan:EC2/BlackholeTraffic
UnauthorizedAccess	5.0 - 8.0	UnauthorizedAccess:IAMUser/ConsoleLogin
Recon	2.0 - 5.0	Recon:EC2/PortProbeUnprotected
Persistence	5.0 - 8.0	Persistence:IAMUser/AnomalousBehavior

多账户设置

# 指定 GuardDuty 管理员
aws guardduty enable-organization-admin-account \
  --admin-account-id 111111111111

# 为新账户自动启用
aws guardduty update-organization-configuration \
  --detector-id DETECTOR_ID \
  --auto-enable

参考资料

• AWS GuardDuty Best Practices: https://aws.github.io/aws-security-services-best-practices/guides/guardduty/
• EventBridge Integration: https://docs.aws.amazon.com/guardduty/latest/ug/guardduty_findings_eventbridge.html
• GuardDuty Finding Types Reference