detecting-aws-guardduty-findings-automation

Name: detecting-aws-guardduty-findings-automation
Author: mukul975

github.com/mukul975安全35.2k68 次安装8 周前更新

代码仓库

关于 detecting-aws-guardduty-findings-automation

这项安全技能面向 AWS 安全团队，介绍 GuardDuty 告警的自动化检测与响应。内容包括 EventBridge 规则、Lambda 响应、Terraform 部署和多账户配置。

主要功能

01启用 GuardDuty 与数据源
02按严重程度配置 EventBridge 规则
03Lambda 隔离 EC2 与处理 IAM 凭证
04Terraform 部署示例
05多账户组织配置

适用场景

01自动响应高严重性发现
02隔离疑似受侵的 EC2 实例
03管理组织内账户的 GuardDuty

安装

复制下面的命令，在终端中运行。

npx skills add https://github.com/mukul975/anthropic-cybersecurity-skills --skill detecting-aws-guardduty-findings-automation

安装后在 Claude Code 中输入/detecting-aws-guardduty-findings-automation确认已加载。

包含的文件

除 SKILL.md 外，安装时一起下载的参考资料和脚本。

LICENSE

assets/

template.md

references/

api-reference.md

standards.md

workflows.md

scripts/

agent.py

process.py

详细介绍

检测 AWS GuardDuty 调查结果自动化

概述

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

适用场景

调查需要检测 AWS GuardDuty 调查结果自动化的安全事件时
为此领域构建检测规则或威胁狩猎查询时
SOC 分析师需要此类分析的结构化操作流程时
验证相关攻击技术的安全监控覆盖范围时

先决条件

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

启用 GuardDuty

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

# Enable additional data sources
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']

    # Extract resource information
    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', [])}

        # Skip if already quarantined
        if instance_tags.get('SecurityStatus') == 'Quarantined':
            return {'statusCode': 200, 'body': 'Already quarantined'}

        # Get current security groups for forensics
        instance = ec2.describe_instances(InstanceIds=[instance_id])
        current_sgs = [sg['GroupId'] for sg in
                       instance['Reservations'][0]['Instances'][0]['SecurityGroups']]

        # Tag instance with finding info and original SGs
        ec2.create_tags(
            Resources=[instance_id],
            Tags=[
                {'Key': 'SecurityStatus', 'Value': 'Quarantined'},
                {'Key': 'GuardDutyFinding', 'Value': finding_type},
                {'Key': 'OriginalSecurityGroups', 'Value': ','.join(current_sgs)},
                {'Key': 'QuarantineTime', 'Value': finding['updatedAt']}
            ]
        )

        # Move to quarantine security group (blocks all traffic)
        if QUARANTINE_SG:
            ec2.modify_instance_attribute(
                InstanceId=instance_id,
                Groups=[QUARANTINE_SG]
            )

        # Create EBS snapshots for forensics
        volumes = ec2.describe_volumes(
            Filters=[{'Name': 'attachment.instance-id', 'Values': [instance_id]}]
        )
        for vol in volumes['Volumes']:
            ec2.create_snapshot(
                VolumeId=vol['VolumeId'],
                Description=f'GuardDuty forensic snapshot - {finding_type}',
                TagSpecifications=[{
                    'ResourceType': 'snapshot',
                    'Tags': [
                        {'Key': 'Purpose', 'Value': 'ForensicCapture'},
                        {'Key': 'SourceInstance', 'Value': instance_id},
                        {'Key': 'FindingType', 'Value': finding_type}
                    ]
                }]
            )

        # Notify security team
        sns.publish(
            TopicArn=SNS_TOPIC,
            Subject=f'[GuardDuty] {finding_type} - Instance {instance_id} Quarantined',
            Message=json.dumps({
                'action': 'instance_quarantined',
                'instance_id': instance_id,
                'finding_type': finding_type,
                'severity': severity,
                'account': account_id,
                'region': region,
                'original_security_groups': current_sgs,
                'description': finding.get('description', '')
            }, indent=2)
        )

        return {
            'statusCode': 200,
            'body': f'Instance {instance_id} quarantined and snapshots created'
        }

    return {'statusCode': 200, 'body': 'Non-EC2 finding processed'}

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 = []

    # Deactivate the compromised access key
    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'Deactivated access key {access_key_id}')
        except Exception as e:
            actions_taken.append(f'Failed to deactivate key: {str(e)}')

    # Attach deny-all policy to user
    deny_policy = {
        "Version": "2012-10-17",
        "Statement": [{
            "Effect": "Deny",
            "Action": "*",
            "Resource": "*"
        }]
    }

    try:
        iam.put_user_policy(
            UserName=user_name,
            PolicyName='GuardDuty-DenyAll-Quarantine',
            PolicyDocument=json.dumps(deny_policy)
        )
        actions_taken.append(f'Applied deny-all policy to {user_name}')
    except Exception as e:
        actions_taken.append(f'Failed to apply deny policy: {str(e)}')

    # Notify
    sns.publish(
        TopicArn=SNS_TOPIC,
        Subject=f'[GuardDuty] IAM Compromise - {user_name}',
        Message=json.dumps({
            'finding_type': finding_type,
            'user': user_name,
            'access_key': access_key_id,
            'actions_taken': actions_taken,
            'severity': finding['severity']
        }, indent=2)
    )

    return {'statusCode': 200, 'body': 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 high severity findings"

  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
}

调查结果类别

类别	严重性范围	示例
后门	5.0 - 8.0	Backdoor:EC2/C&CActivity
加密货币	5.0 - 8.0	CryptoCurrency:EC2/BitcoinTool
木马	5.0 - 8.0	Trojan:EC2/BlackholeTraffic
未经授权的访问	5.0 - 8.0	UnauthorizedAccess:IAMUser/ConsoleLogin
侦察	2.0 - 5.0	Recon:EC2/PortProbeUnprotected
持久化	5.0 - 8.0	Persistence:IAMUser/AnomalousBehavior

多账户设置

# Designate GuardDuty administrator
aws guardduty enable-organization-admin-account \
  --admin-account-id 111111111111

# Auto-enable for new accounts
aws guardduty update-organization-configuration \
  --detector-id DETECTOR_ID \
  --auto-enable

参考资料

AWS GuardDuty 最佳实践：https://aws.github.io/aws-security-services-best-practices/guides/guardduty/
EventBridge 集成：https://docs.aws.amazon.com/guardduty/latest/ug/guardduty_findings_eventbridge.html
GuardDuty 调查结果类型参考

支持 Markdown

登录后评论

基本信息

作者: mukul975
仓库: mukul975/anthropic-cybersecurity-skills
分类: 安全
安装次数: 68
GitHub Stars: 35,191

报告问题