radxzi-creator/aws-cloud-security-monitoring-Incident-response-pipeline-
GitHub: radxzi-creator/aws-cloud-security-monitoring-Incident-response-pipeline-
该项目在 AWS 上实现了一套无服务器安全管道，能够实时检测 IAM 提权行为并在约 90 秒内自动完成会话撤销与账户锁定。
Stars: 0 | Forks: 0
# aws-cloud-security-monitoring-Incident-response-pipeline-
🛡️ AWS 云安全监控与事件响应流水线
实时检测并自动响应 IAM 提权威胁。
遏制时间：约 90 秒
需要人工干预：无
📖 目录
项目概述
架构图
工作原理
使用的技术
安全检测
事件响应流程
测试与验证
截图
设置说明
部署
我的收获
未来增强
清理
许可证
🎯 项目概述
本项目在 AWS 上实现了一个无服务器安全监控与事件响应系统。
它可以实时检测高风险的 IAM 活动，并通过撤销会话和向受损用户附加 Deny-All 策略来自动响应。
它解决的问题：权限提升是云环境中最大的安全风险之一。
手动事件响应可能需要数小时，在此期间攻击者可能会造成重大破坏。
该系统将响应时间缩短至 2 分钟以内——完全自动化。
🏗️ 架构图
text
┌─────────────────────────────────────────────────────────────────┐
│ AWS Cloud Security Pipeline │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────────┐ ┌─────────────┐ │
│ │ CloudTrail │───▶│ CloudWatch │───▶│ Metric │ │
│ │ (Audit Log) │ │ Logs │ │ Filter │ │
│ └─────────────┘ └─────────────────┘ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ CloudWatch │ │
│ │ Alarm │ │
│ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ EventBridge│ │
│ │ Rule │ │
│ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ AWS Lambda │ │
│ │ (Responder)│ │
│ └─────────────┘ │
│ │ │
│ ┌────────────────────┼───── │
│ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────────┐ │
│ │ IAM │ │ SNS │ │
│ │ Deny-All │ │ Email Alert │ │
│ │ Policy │ │ │ │
│ └─────────────┘ └─────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
🛠️ 工作原理
CloudTrail 记录账户中的所有 IAM API 活动
CloudWatch Logs 集中存储日志
Metric Filter 扫描日志以查找 AttachUserPolicy 事件
当匹配到该模式时，CloudWatch Alarm 触发
EventBridge 将告警路由到 AWS Lambda
AWS Lambda（“紧急停止开关”）执行自动响应
Lambda 响应动作：
python
# 1. 查询 CloudWatch Logs 获取上下文 (用户名, IP)
# 2. 撤销所有活动会话 (如果 API 可用)
# 3. 附加 Deny-All 内联策略
# 4. 发送 SNS 邮件告警
🛠️ 使用的技术
服务 用途
AWS IAM 用户管理、权限、身份治理
AWS CloudTrail 所有 AWS API 调用的审计日志
Amazon CloudWatch Logs 集中式日志存储和查询
CloudWatch Metric Filters 基于模式的事件检测
CloudWatch Alarms 实时告警
Amazon EventBridge 事件路由和服务集成
AWS Lambda 自动化事件响应 (Python)
Amazon SNS 通知推送 (电子邮件)
🔍 安全检测
系统监控高风险 IAM 活动，包括：
Root 账户控制台登录 – 检测 root 用户活动
IAM 用户创建/删除 – 未经授权的用户管理
IAM 角色修改 – 对角色和权限的更改
策略附加 – 提权尝试
高风险 IAM 更改 – 任何可疑的 IAM 活动
🚨 事件响应流程
text
1. 用户为 IAM 用户附加 AdministratorAccess
│
▼
2. CloudTrail 记录 AttachUserPolicy 事件
│
▼
3. Metric Filter 检测到该事件
│
▼
4. CloudWatch Alarm 触发 (IN ALARM 状态)
│
▼
5. EventBridge 调用 Lambda
│
▼
6. Lambda 查询 CloudWatch Logs 获取上下文:
- Username: ScapegoatUser
- Source IP: 203.0.113.1
│
▼
7. Lambda 执行遏制动作:
- 撤销所有活动会话
- 附加 Deny-All 策略 (AUTO_LOCKDOWN_BY_SOC)
│
▼
8. Lambda 发送 SNS 告警:
┌─────────────────────────────────────────────┐
│ 🚨 AWS INCIDENT RESPONSE - AUTOMATED LOCKDOWN │
│ User: ScapegoatUser │
│ Source IP: 203.0.113.1 │
│ Action Taken: Revoked sessions + Deny-All │
│ Time: 2026-07-11T14:33:31Z │
└─────────────────────────────────────────────┘
🧪 测试与验证
测试设置：
创建了具有 ReadOnlyAccess 的测试用户 ScapegoatUser
模拟攻击：为该用户附加 AdministratorAccess
监控系统的响应
验证结果：
步骤 预期结果 状态
CloudTrail 记录 AttachUserPolicy 事件已记录 ✅ 通过
Metric Filter 匹配事件 模式已检测到 ✅ 通过
CloudWatch Alarm 触发 ALARM 状态 ✅ 通过
EventBridge 调用 Lambda Lambda 已触发 ✅ 通过
Lambda 附加 Deny-All 策略 策略已应用 ✅ 通过
发送 SNS 邮件告警 邮件已送达 ✅ 通过
用户被锁定 Deny-All 覆盖一切 ✅ 通过
截图：
[Lambda 日志显示 "LOCKING DOWN user"]
[IAM 内联策略 AUTO_LOCKDOWN_BY_SOC]
[处于 ALARM 状态的 CloudWatch Alarm]
[SNS 邮件通知]
1. IAM ScapegoatUser 权限
显示已附加 AUTO_LOCKDOWN_BY_SOC 内联策略
text
Permissions policies (4)
├── AdministratorAccess (Managed policy)
├── AUTO_LOCKDOWN_BY_SOC (Inline policy - Deny-All) ✅
├── IAMUserChangePassword (Managed policy)
└── ReadOnlyAccess (Managed policy)
2. Lambda 执行日志
text
🚨 LOCKING DOWN user: ScapegoatUser
✅ Deny-All policy attached to ScapegoatUser
Duration: 2459.82 ms
3. CloudWatch Alarm 状态
Privilege-Escalation-Alarm → ALARM 状态
IAM-Security-Detection-Alarm → ALARM 状态
4. SNS 邮件告警
text
Subject: Alarm "IAM-Security-Detection-Alarm" in ALARM state
Body: Threshold crossed: 1 datapoint >= threshold
Time: Saturday 11 July, 2026 05:07:14 UTC
🚀 设置说明
前置条件
具有管理员访问权限的 AWS 账户
已配置 AWS CLI
以下服务的 IAM 权限：CloudTrail、CloudWatch、IAM、Lambda、EventBridge、SNS
第 1 步：创建测试用户
bash
aws iam create-user --user-name ScapegoatUser
aws iam attach-user-policy --user-name ScapegoatUser --policy-arn arn:aws:iam::aws:policy/ReadOnlyAccess
第 2 步：创建 CloudTrail Trail
启用具有多区域支持的 CloudTrail
将日志转发到 CloudWatch Logs 组：cloudtrail-log-group
第 3 步：创建 Metric Filter
Filter pattern: { $.eventName = "AttachUserPolicy" }
Metric namespace: Security/SOC
Metric name: PrivilegeEscalationCount
第 4 步：创建 CloudWatch Alarm
指标：Security/SOC/PrivilegeEscalationCount
统计：Sum，周期：1 分钟
阈值：1 个数据点中 > 0
第 5 步：创建 EventBridge Rule
Event pattern: CloudWatch Alarm State Change
目标：Lambda 函数 SOCResponder
第 6 步：部署 Lambda 函数
运行环境：Python 3.12
超时：30 秒
具有以下权限的 IAM 角色：Logs、IAM、SNS
第 7 步：创建 SNS Topic
Topic 名称：SOC-PrivilegeEscalation-Alerts
订阅电子邮件 endpoint
📁 仓库结构
text
cloud-security-pipeline/
├── README.md # This file
├── architecture/
│ └── diagram.png # Architecture diagram
├── cloudformation/
│ └── pipeline.yaml # CloudFormation template
├── lambda/
│ └── responder.py # Lambda function code
├── cloudtrail/
│ └── metric-filter.json # Metric filter configuration
├── alarms/
│ └── alarm-config.json # CloudWatch Alarm config
├── eventbridge/
│ └── rule.json # EventBridge rule
└── screenshots/
├── lambda-logs.png
├── inline-policy.png
├── alarm-state.png
└── sns-email.png
💻 Lambda 函数代码
python
import boto3
import os
import json
import time
from datetime import datetime, timedelta
logs = boto3.client('logs')
iam = boto3.client('iam')
sns = boto3.client('sns')
EXEMPT_USERS = os.environ.get('EXEMPT_USERS', 'your-admin-username').split(',')
LOG_GROUP = os.environ.get('LOG_GROUP', 'cloudtrail-log-group')
def lambda_handler(event, context):
print(f"Event Triggered: {json.dumps(event)}")
```
query = f"""
fields @timestamp, userIdentity.userName, sourceIPAddress, eventName, requestParameters.userName
| filter eventName = "AttachUserPolicy"
| sort @timestamp desc
| limit 1
"""
try:
start_query_response = logs.start_query(
logGroupName=LOG_GROUP,
startTime=int((datetime.utcnow() - timedelta(minutes=15)).timestamp() * 1000),
endTime=int(datetime.utcnow().timestamp() * 1000),
queryString=query
)
query_id = start_query_response['queryId']
time.sleep(2)
response = logs.get_query_results(queryId=query_id)
if not response['results']:
print("No suspicious events found. Exiting.")
return {"status": "no_action"}
result = response['results'][0]
event_dict = {item['field']: item['value'] for item in result}
# Use requestParameters.userName (which contains the actual username)
target_user = event_dict.get('requestParameters.userName')
source_ip = event_dict.get('sourceIPAddress', 'Unknown')
if target_user in EXEMPT_USERS:
print(f"User {target_user} is EXEMPT. Skipping.")
return {"status": "skipped"}
print(f"🚨 LOCKING DOWN user: {target_user}")
# Attach Deny-All inline policy
deny_policy = {
"Version": "2012-10-17",
"Statement": [{
"Effect": "Deny",
"Action": "*",
"Resource": "*"
}]
}
iam.put_user_policy(
UserName=target_user,
PolicyName='AUTO_LOCKDOWN_BY_SOC',
PolicyDocument=json.dumps(deny_policy)
)
# Send SNS alert
sns.publish(
TopicArn='arn:aws:sns:us-east-1:459653582331:SOC-PrivilegeEscalation-Alerts',
Subject=f"🚨 Auto-Contained: {target_user}",
Message=f"User: {target_user}\nIP: {source_ip}\nTime: {datetime.utcnow()}"
)
return {"status": "contained", "user": target_user}
except Exception as e:
print(f"Error: {e}")
return {"status": "error", "message": str(e)}
```
🧹 清理
bash
# 移除 Deny-All 策略
aws iam delete-user-policy --user-name ScapegoatUser --policy-name AUTO_LOCKDOWN_BY_SOC
# 移除 AdministratorAccess
aws iam detach-user-policy --user-name ScapegoatUser --policy-arn arn:aws:iam::aws:policy/AdministratorAccess
# 删除测试用户（可选）
aws iam delete-user --user-name ScapegoatUser
# 删除 Lambda function
aws lambda delete-function --function-name SOCResponder
# 删除 EventBridge rule
aws events delete-rule --name Trigger-SOC-Responder
# 删除 SNS topic
aws sns delete-topic --topic-arn arn:aws:sns:us-east-1:459653582331:SOC-PrivilegeEscalation-Alerts
📚 我的收获
云安全基础
使用 CloudTrail 设置安全审计日志
IAM 事件跟踪和用户活动监控
为自动化角色设计最小权限 IAM
监控与可观测性
为安全模式构建高效的 metric filter
创建可操作的 CloudWatch Alarm
使用 CloudWatch Logs Insights 进行实时查询
事件驱动架构
设计无服务器事件 pipeline
配置 Lambda 进行自动事件响应
处理跨服务 IAM 权限
真实场景故障排除
IAM 权限问题（拒绝与允许）
CloudTrail 日志传输延迟和时序
CloudWatch Logs 查询中的空字段错误
Lambda 超时管理
CloudWatch Logs 字段映射（userIdentity 与 requestParameters）
🚀 未来增强
集成 Slack/Teams webhook 以进行实时聊天告警
基于 IP 的白名单（仅阻止可疑 IP）
基于时间的规则（仅在非工作时间阻止）
多账户 AWS Organizations 支持
SIEM 集成（Splunk、Datadog、Sentinel）
使用 CloudWatch Anomaly Detection 进行异常检测
使用 Grafana 或 QuickSight 构建 Dashboard
自动创建事件工单（Jira、ServiceNow📄 许可证
本项目是开源的，并基于 MIT License 提供。
🔗 与我联系
LinkedIn: https://www.linkedin.com/in/blessing-m-8361351a3/)
Email:radxzi@gmail.com
⭐ 显示您的支持
如果您觉得这个项目有用，请在 GitHub 上给它点个 ⭐！
[IAM 内联策略 AUTO_LOCKDOWN_BY_SOC]
[处于 ALARM 状态的 CloudWatch Alarm]
[SNS 邮件通知]
1. IAM ScapegoatUser 权限
显示已附加 AUTO_LOCKDOWN_BY_SOC 内联策略
text
Permissions policies (4)
├── AdministratorAccess (Managed policy)
├── AUTO_LOCKDOWN_BY_SOC (Inline policy - Deny-All) ✅
├── IAMUserChangePassword (Managed policy)
└── ReadOnlyAccess (Managed policy)
2. Lambda 执行日志
text
🚨 LOCKING DOWN user: ScapegoatUser
✅ Deny-All policy attached to ScapegoatUser
Duration: 2459.82 ms
3. CloudWatch Alarm 状态
Privilege-Escalation-Alarm → ALARM 状态
IAM-Security-Detection-Alarm → ALARM 状态
4. SNS 邮件告警
text
Subject: Alarm "IAM-Security-Detection-Alarm" in ALARM state
Body: Threshold crossed: 1 datapoint >= threshold
Time: Saturday 11 July, 2026 05:07:14 UTC
🚀 设置说明
前置条件
具有管理员访问权限的 AWS 账户
已配置 AWS CLI
以下服务的 IAM 权限：CloudTrail、CloudWatch、IAM、Lambda、EventBridge、SNS
第 1 步：创建测试用户
bash
aws iam create-user --user-name ScapegoatUser
aws iam attach-user-policy --user-name ScapegoatUser --policy-arn arn:aws:iam::aws:policy/ReadOnlyAccess
第 2 步：创建 CloudTrail Trail
启用具有多区域支持的 CloudTrail
将日志转发到 CloudWatch Logs 组：cloudtrail-log-group
第 3 步：创建 Metric Filter
Filter pattern: { $.eventName = "AttachUserPolicy" }
Metric namespace: Security/SOC
Metric name: PrivilegeEscalationCount
第 4 步：创建 CloudWatch Alarm
指标：Security/SOC/PrivilegeEscalationCount
统计：Sum，周期：1 分钟
阈值：1 个数据点中 > 0
第 5 步：创建 EventBridge Rule
Event pattern: CloudWatch Alarm State Change
目标：Lambda 函数 SOCResponder
第 6 步：部署 Lambda 函数
运行环境：Python 3.12
超时：30 秒
具有以下权限的 IAM 角色：Logs、IAM、SNS
第 7 步：创建 SNS Topic
Topic 名称：SOC-PrivilegeEscalation-Alerts
订阅电子邮件 endpoint
📁 仓库结构
text
cloud-security-pipeline/
├── README.md # This file
├── architecture/
│ └── diagram.png # Architecture diagram
├── cloudformation/
│ └── pipeline.yaml # CloudFormation template
├── lambda/
│ └── responder.py # Lambda function code
├── cloudtrail/
│ └── metric-filter.json # Metric filter configuration
├── alarms/
│ └── alarm-config.json # CloudWatch Alarm config
├── eventbridge/
│ └── rule.json # EventBridge rule
└── screenshots/
├── lambda-logs.png
├── inline-policy.png
├── alarm-state.png
└── sns-email.png
💻 Lambda 函数代码
python
import boto3
import os
import json
import time
from datetime import datetime, timedelta
logs = boto3.client('logs')
iam = boto3.client('iam')
sns = boto3.client('sns')
EXEMPT_USERS = os.environ.get('EXEMPT_USERS', 'your-admin-username').split(',')
LOG_GROUP = os.environ.get('LOG_GROUP', 'cloudtrail-log-group')
def lambda_handler(event, context):
print(f"Event Triggered: {json.dumps(event)}")
```
query = f"""
fields @timestamp, userIdentity.userName, sourceIPAddress, eventName, requestParameters.userName
| filter eventName = "AttachUserPolicy"
| sort @timestamp desc
| limit 1
"""
try:
start_query_response = logs.start_query(
logGroupName=LOG_GROUP,
startTime=int((datetime.utcnow() - timedelta(minutes=15)).timestamp() * 1000),
endTime=int(datetime.utcnow().timestamp() * 1000),
queryString=query
)
query_id = start_query_response['queryId']
time.sleep(2)
response = logs.get_query_results(queryId=query_id)
if not response['results']:
print("No suspicious events found. Exiting.")
return {"status": "no_action"}
result = response['results'][0]
event_dict = {item['field']: item['value'] for item in result}
# Use requestParameters.userName (which contains the actual username)
target_user = event_dict.get('requestParameters.userName')
source_ip = event_dict.get('sourceIPAddress', 'Unknown')
if target_user in EXEMPT_USERS:
print(f"User {target_user} is EXEMPT. Skipping.")
return {"status": "skipped"}
print(f"🚨 LOCKING DOWN user: {target_user}")
# Attach Deny-All inline policy
deny_policy = {
"Version": "2012-10-17",
"Statement": [{
"Effect": "Deny",
"Action": "*",
"Resource": "*"
}]
}
iam.put_user_policy(
UserName=target_user,
PolicyName='AUTO_LOCKDOWN_BY_SOC',
PolicyDocument=json.dumps(deny_policy)
)
# Send SNS alert
sns.publish(
TopicArn='arn:aws:sns:us-east-1:459653582331:SOC-PrivilegeEscalation-Alerts',
Subject=f"🚨 Auto-Contained: {target_user}",
Message=f"User: {target_user}\nIP: {source_ip}\nTime: {datetime.utcnow()}"
)
return {"status": "contained", "user": target_user}
except Exception as e:
print(f"Error: {e}")
return {"status": "error", "message": str(e)}
```
🧹 清理
bash
# 移除 Deny-All 策略
aws iam delete-user-policy --user-name ScapegoatUser --policy-name AUTO_LOCKDOWN_BY_SOC
# 移除 AdministratorAccess
aws iam detach-user-policy --user-name ScapegoatUser --policy-arn arn:aws:iam::aws:policy/AdministratorAccess
# 删除测试用户（可选）
aws iam delete-user --user-name ScapegoatUser
# 删除 Lambda function
aws lambda delete-function --function-name SOCResponder
# 删除 EventBridge rule
aws events delete-rule --name Trigger-SOC-Responder
# 删除 SNS topic
aws sns delete-topic --topic-arn arn:aws:sns:us-east-1:459653582331:SOC-PrivilegeEscalation-Alerts
📚 我的收获
云安全基础
使用 CloudTrail 设置安全审计日志
IAM 事件跟踪和用户活动监控
为自动化角色设计最小权限 IAM
监控与可观测性
为安全模式构建高效的 metric filter
创建可操作的 CloudWatch Alarm
使用 CloudWatch Logs Insights 进行实时查询
事件驱动架构
设计无服务器事件 pipeline
配置 Lambda 进行自动事件响应
处理跨服务 IAM 权限
真实场景故障排除
IAM 权限问题（拒绝与允许）
CloudTrail 日志传输延迟和时序
CloudWatch Logs 查询中的空字段错误
Lambda 超时管理
CloudWatch Logs 字段映射（userIdentity 与 requestParameters）
🚀 未来增强
集成 Slack/Teams webhook 以进行实时聊天告警
基于 IP 的白名单（仅阻止可疑 IP）
基于时间的规则（仅在非工作时间阻止）
多账户 AWS Organizations 支持
SIEM 集成（Splunk、Datadog、Sentinel）
使用 CloudWatch Anomaly Detection 进行异常检测
使用 Grafana 或 QuickSight 构建 Dashboard
自动创建事件工单（Jira、ServiceNow📄 许可证
本项目是开源的，并基于 MIT License 提供。
🔗 与我联系
LinkedIn: https://www.linkedin.com/in/blessing-m-8361351a3/)
Email:radxzi@gmail.com
⭐ 显示您的支持
如果您觉得这个项目有用，请在 GitHub 上给它点个 ⭐！标签：AWS, DPI, IAM, 自动化响应, 逆向工具