Menu
使用 EventBridge 和 Lambda 自动化处理 AWS GuardDuty 威胁检测发现，实现实时事件响应、自动隔离受损资源和安全通知工作流。
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
Based on SOC occupation classification
通过分析 Zeek dns.log 中的高熵子域名查询、超量查询量、超长查询长度以及异常 DNS 记录类型，检测 DNS 隧道和数据外泄中的隐蔽通道通信。适用于：当需要狩猎基于 DNS 的 C2 或数据外泄通道、调查异常 DNS 查询模式、或响应涉及 DNS 隧道工具（iodine、dnscat2、DNSExfiltrator）的威胁情报时使用。
实施 Google 的 BeyondCorp 零信任访问模型，通过 IAP、Access Context Manager 和 Chrome Enterprise Premium，消除网络边界的隐式信任，强制执行基于身份的访问控制，实现无 VPN 的安全应用访问。适用于将传统 VPN 替换为零信任架构、部署 Identity-Aware Proxy、配置设备信任策略、或为远程办公实施上下文感知访问控制时使用。
在授权的安全评估过程中，使用 Burp Suite 的扫描器、Intruder 和 Repeater 工具识别和验证跨站脚本（XSS）漏洞。适用于 Web 应用渗透测试中检测反射型、存储型和 DOM 型 XSS，验证自动化扫描器报告的 XSS 发现，以及评估 CSP 和 XSS 过滤器的有效性时使用。
攻击活动溯源归因分析涉及系统性地评估证据，以确定哪个威胁行为者或组织对某次网络行动负责。本技能涵盖使用 Diamond Model 和 ACH（竞争假设分析）收集并加权溯源归因指标、分析基础设施重叠、TTP 一致性、恶意软件代码相似性、操作时序模式和语言痕迹，以构建置信度加权的溯源归因评估。
从 PE 文件和内存转储中提取并分析 Cobalt Strike beacon 配置，以识别 C2 基础设施、Malleable C2 配置文件和攻击者操作惯例。
使用 Ghidra 及专用脚本对 Go 编译的恶意软件进行逆向工程，包括函数恢复、字符串提取和去符号表 Go 二进制文件的类型重建。
| 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 |
Amazon GuardDuty 是一种威胁检测服务，持续监控 AWS 账户的恶意活动和未授权行为。通过将 GuardDuty 与 Amazon EventBridge 和 AWS Lambda 集成，安全团队实现自动化、实时的威胁响应，将平均响应时间（MTTR）从数小时缩短到数秒。GuardDuty 分析 VPC 流日志、CloudTrail 管理和数据事件、DNS 日志、EKS 审计日志和 S3 数据事件。
# 启用 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}
}'
{
"source": ["aws.guardduty"],
"detail-type": ["GuardDuty Finding"],
"detail": {
"severity": [{"numeric": [">=", 7.0]}]
}
}
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"
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]['Instances'][0]['SecurityGroups']]
# 使用发现信息和原始安全组标记实例
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']}
]
)
# 移至隔离安全组（阻断所有流量）
if QUARANTINE_SG:
ec2.modify_instance_attribute(
InstanceId=instance_id,
Groups=[QUARANTINE_SG]
)
# 创建 EBS 快照用于取证
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}
]
}]
)
# 通知安全团队
sns.publish(
TopicArn=SNS_TOPIC,
Subject=f'[GuardDuty] {finding_type} - 实例 {instance_id} 已隔离',
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_id} 已隔离并创建快照'
}
return {'statusCode': 200, 'body': '非 EC2 发现已处理'}
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}')
except Exception as e:
actions_taken.append(f'停用密钥失败: {str(e)}')
# 为用户附加拒绝所有策略
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'已为 {user_name} 应用拒绝所有策略')
except Exception as e:
actions_taken.append(f'应用拒绝策略失败: {str(e)}')
# 通知
sns.publish(
TopicArn=SNS_TOPIC,
Subject=f'[GuardDuty] IAM 入侵 - {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)}
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