janebjobtech-cloud/Security-Incident-Response-Automation

GitHub: janebjobtech-cloud/Security-Incident-Response-Automation

一个基于AWS Lambda的自动化安全事件响应系统,用于快速检测和处理云环境中的安全威胁。

Stars: 0 | Forks: 0

# 安全事件响应自动化 一个使用 Python Lambda 函数的自动化、事件驱动型 AWS 安全事件响应系统——可在 20 分钟内检测、调查、遏制并通报未经授权的访问、异常 API 活动以及 S3 配置错误。 # 🛡️ 安全事件响应自动化 ### CLCS 基于 AI 的云自动化与脚本 | 实验室 ## 📋 目录 - [概述](#overview) - [架构](#architecture) - [系统组件](#system-components) - [检测逻辑与阈值](#detection-logic--thresholds) - [项目结构](#project-structure) - [Lambda 函数](#lambda-functions) - [通用工具模块](#a1-common-utilities) - [检测:未经授权的访问](#a2-detection--unauthorized-access) - [检测:异常 API 活动](#a3-detection--unusual-api-activity) - [检测:S3 配置错误](#a4-detection--s3-misconfiguration) - [调查工作流](#a5-investigation-workflow) - [遏制工作流](#a6-containment-workflow) - [通知处理器](#a7-notification-handler) - [事件响应剧本](#incident-response-playbooks) - [部署概述](#deployment-overview) - [测试与验证](#testing--validation) - [成果](#results) - [未来增强](#future-enhancements) - [参考资料](#references) ## 概述 本实验室为一家模拟的金融服务公司实现了一个完全自动化的云安全事件响应系统。该系统用一个无服务器、事件驱动的流水线取代了缓慢的手动流程,用于检测、调查、遏制并通报三类高优先级事件: | 事件类型 | 自动化之前 | 自动化之后 | |---|---|---| | 未经授权的访问 | 4–6 小时 | < 20 分钟 | | 异常 API 活动 | 4–6 小时 | < 20 分钟 | | S3 配置错误 | 4–6 小时 | < 20 分钟 | **使用的关键技术:** `Python 3.11` · `AWS Lambda` · `CloudTrail` · `AWS Config` · `Amazon SNS` · `EventBridge` · `IAM` · `S3` · `EC2` · `CloudWatch Logs` ## 架构 该系统遵循**模块化、事件驱动架构**,包含从检测到通知的四个逻辑层: ``` CloudTrail Logs │ ▼ ┌─────────────────────┐ │ Detection Layer │ ← Lambda + CloudWatch Logs + EventBridge + AWS Config │ (3 Lambda Functions)│ └────────┬────────────┘ │ Incident Event ▼ ┌─────────────────────┐ │ Investigation Layer │ ← Lambda: collects user context, logs, metadata │ (1 Lambda Function) │ └────────┬────────────┘ │ Investigation Result ▼ ┌─────────────────────┐ │ Containment Layer │ ← Lambda: disables users, quarantines instances, blocks access │ (1 Lambda Function) │ └────────┬────────────┘ │ Actions Taken ▼ ┌─────────────────────┐ │ Notification Layer │ ← SNS: structured alert to security team │ (SNS + Lambda) │ └─────────────────────┘ ``` **部署区域:** `us-east-1` **运行时:** 所有 Lambda 函数均使用 `Python 3.11` **IAM 模型:** 共享执行角色,按服务分配最小权限 ## 系统组件 ### 检测层 使用三种触发机制识别安全事件: | 触发条件 | 机制 | 目标函数 | |---|---|---| | 控制台登录失败 | CloudWatch Logs 订阅筛选器 | `detect_unauthorized_access.py` | | 敏感 API 调用 | EventBridge 定时规则(每 5 分钟) | `detect_unusual_api_activity.py` | | S3 合规性违规 | 基于 AWS Config 变更的 EventBridge 规则 | `detect_public_s3_buckets.py` | ### 调查层 当事件触发时,自动收集上下文证据: - 用户身份和 IAM 权限快照 - 过去 30 分钟的 CloudTrail 活动 - 资源元数据(EC2、S3 等) - 网络流量和事件时间线 - 将 CloudTrail 日志副本复制到取证 S3 存储桶 ### 遏制层 根据事件类型执行有针对性的修复: - **未经授权的访问** → 禁用 IAM 用户,强制密码重置 - **异常 API 活动** → 限制 IAM 角色,隔离 EC2 实例,阻止可疑 IP - **配置错误** → 应用 S3 阻止公共访问,移除公共 ACL ### 通知层 Amazon SNS 发送结构化告警,包括: - 事件类型和严重级别 - 调查结果 - 已采取的遏制措施 - 建议的后续步骤 高严重性事件路由至值班工程师。低严重性事件排队发送给安全运维团队。 ## 检测逻辑与阈值 | 事件类型 | 检测逻辑 | 阈值 | 自动响应 | |---|---|---|---| | 未经授权的访问 | 登录失败 + 地理位置异常 | 10 分钟内 ≥ 5 次失败 | 触发调查 → 禁用用户 | | 异常 API 活动 | 来自新 IP 的敏感 API 调用或大量删除操作 | 5 分钟内 > 20 次删除调用 | 标记潜在入侵 → 限制角色 | | S3 配置错误 | 公共存储桶 ACL 或存储桶策略 | 检测到任何违规 | 自动恢复为安全配置 | ## 项目结构 ``` security-incident-response/ │ ├── common_utils.py # Shared helpers: SNS, snapshots, CloudTrail queries │ ├── detection/ │ ├── detect_unauthorized_access.py # CloudWatch Logs trigger — failed logins │ ├── detect_unusual_api_activity.py# EventBridge schedule — sensitive API calls │ └── detect_public_s3_buckets.py # AWS Config trigger — public bucket detection │ ├── workflows/ │ ├── investigation_workflow.py # Collects evidence, user context, logs │ └── containment_workflow.py # Executes remediation actions │ ├── notification/ │ └── notification_handler.py # Generic SNS notification dispatcher │ └── playbooks/ ├── playbook_unauthorized_access.md ├── playbook_unusual_api_activity.md └── playbook_s3_misconfiguration.md ``` ## Lambda 函数 ### A.1 通用工具模块 **文件:** `common_utils.py` 所有 Lambda 函数导入的共享模块。初始化 boto3 客户端并提供可复用的辅助函数。 **关键函数:** | 函数 | 用途 | |---|---| | `publish_notification(message)` | 将结构化事件数据发布到 SNS | | `snapshot_instance(instance_id)` | 创建取证 EC2 EBS 快照 | | `get_root_volume_id(instance_id)` | 解析用于快照的根卷 | | `copy_cloudtrail_logs(start, end)` | 将日志范围复制到取证 S3 存储桶 | | `get_recent_user_activity(user, min)` | 查询 CloudTrail 获取用户近期事件 | | `get_user_policies(username)` | 返回附加和内联的 IAM 策略 | **运行时所需的环境变量:** ``` SNS_TOPIC_ARN = arn:aws:sns:us-east-1::SecurityIncidents QUARANTINE_SG_ID = sg-0 FORENSIC_BUCKET = ``` ### A.2 检测 — 未经授权的访问 **文件:** `detect_unauthorized_access.py` **触发器:** CloudWatch Logs 对 CloudTrail `ConsoleLogin` 事件的订阅筛选器 **逻辑:** 1. 从 CloudWatch 接收 `ConsoleLogin` 记录 2. 筛选 `Failure` 响应 3. 调用 `is_threshold_exceeded()` — 查询 CloudTrail 获取 10 分钟内 ≥ 5 次登录失败 4. 如果满足阈值 → 构建事件载荷 → 发布到 SNS **可配置常量:** ``` FAILED_THRESHOLD = 5 # Number of failures before alerting WINDOW_MINUTES = 10 # Lookback window in minutes ``` ### A.3 检测 — 异常 API 活动 **文件:** `detect_unusual_api_activity.py` **触发器:** EventBridge 定时规则 — 每 5 分钟运行一次 **逻辑:** 1. 查询 CloudTrail 获取过去 5 分钟的事件 2. 标记来自新的/未知 IP 地址的敏感 API 调用 3. 统计所有删除类型操作;如果计数超过阈值则告警 4. 为敏感 API 滥用和大规模删除分别发布事件 **监控的 API:** ``` SENSITIVE_APIS = [ "DeleteBucket", "DeleteObject", "TerminateInstances", "PutUserPolicy", "AttachUserPolicy" ] ``` **可配置常量:** ``` MASS_DELETE_THRESHOLD = 20 # Delete operations before alerting WINDOW_MINUTES = 5 # Lookback window in minutes ``` ### A.4 检测 — S3 配置错误 **文件:** `detect_public_s3_buckets.py` **触发器:** AWS Config 合规性变更规则(EventBridge)或定期扫描 **逻辑:** 1. 列出账户中的所有 S3 存储桶 2. 检查每个存储桶的 ACL 是否有 `AllUsers` 或 `AuthenticatedUsers` 授权 3. 为任何可公开访问的存储桶构建并发布事件 ### A.5 调查工作流 **文件:** `investigation_workflow.py` **触发器:** 由检测函数通过 EventBridge 或直接 Lambda 调用 **收集:** - 用户 IAM 策略(附加和内联) - 用户过去 30 分钟的 CloudTrail 活动 - 资源元数据(EC2 实例状态、启动时间、标签) - CloudTrail 日志取证副本在 S3 中的路径 **输出:** 向 SNS 发布结构化的调查结果通知,为安全分析师提供完整上下文。 ### A.6 遏制工作流 **文件:** `containment_workflow.py` **触发器:** 调查完成后调用 **按事件类型采取的行动:** | 事件类型 | 遏制行动 | |---|---| | 未经授权的访问 | `disable_user()` → 通过 IAM 强制密码重置 | | 异常 API 活动 | `quarantine_instance()` → 将 EC2 重新分配到隔离安全组 + 创建取证快照 | | S3 配置错误 | `remove_public_access()` → 对所有四项控件应用 S3 阻止公共访问 | **发布:** 向 SNS 发送遏制阶段通知,列出已采取的每一项行动。 ### A.7 通知处理器 **文件:** `notification_handler.py` **触发器:** 所有事件阶段的 EventBridge 路由规则 一个通用调度器,格式化并转发任何阶段事件(检测、调查、遏制)作为结构化的 SNS 告警。当使用 EventBridge 在所有 Lambda 阶段进行路由时,作为集中式通知中继。 ## 事件响应剧本 ### 剧本 1 — 未经授权的访问尝试 | 步骤 | 行动 | |---|---| | 1. 检测触发 | 10 分钟内 ≥ 5 次失败登录 或 来自异常 IP 的登录 | | 2. 调查 | 收集用户身份、IAM 权限、源 IP、MFA 状态、30 分钟 CloudTrail 历史 | | 3. 遏制 | 禁用 IAM 登录,撤销会话,强制密码重置,应用临时拒绝所有策略 | | 4. 证据 | 将 CloudTrail 日志复制到取证存储桶;记录时间戳、IP、事件 ID | | 5. 通知 | SNS 告警:用户、源 IP、遏制状态、建议的修复措施 | | 6. 升级 | 重复尝试 → 升级到二级安全分析师 | ### 剧本 2 — 异常 API 活动 | 步骤 | 行动 | |---|---| | 1. 检测触发 | 来自新 IP 的敏感 API 调用 或 5 分钟内 ≥ 20 次删除操作 | | 2. 调查 | 识别用户/角色,检索 CloudTrail 事件,检查 IP 历史,审查 IAM 信任策略 | | 3. 遏制 | 限制 IAM 角色权限,通过 NACL/SG 阻止可疑 IP,暂停破坏性操作 | | 4. 证据 | 复制 CloudTrail 日志,捕获 IAM 角色策略状态,保存 API 事件载荷 | | 5. 通知 | SNS 告警:API 调用、用户/角色、IP 地址、遏制行动 | | 6. 升级 | 持续破坏性活动 → 完全禁用 IAM 角色 | ### 剧本 3 — 公共 S3 存储桶配置错误 | 步骤 | 行动 | |---|---| | 1. 检测触发 | AWS Config 规则违规 或 Lambda 检测到公共 ACL/存储桶策略 | | 2. 调查 | 检索存储桶 ACL 和策略,识别最后修改者,审查 CloudTrail 中的策略变更 | | 3. 遏制 | 应用 S3 阻止公共访问(所有 4 项控件),移除公共 ACL,恢复到安全基线 | | 4. 证据 | 将先前的存储桶策略保存到取证 S3 存储桶;记录负责用户 | | 5. 通知 | SNS 告警:存储桶名称、配置错误详情、遏制状态 | | 6. 升级 | 如果敏感数据已暴露 → 启动数据泄露协议 | ## 部署概述 所有六个 Lambda 函数都作为独立的 Python 3.11 运行时容器部署在 `us-east-1`,使用**共享的 IAM 执行角色**,其权限范围限定为最小权限,覆盖: `CloudTrail` · `IAM` · `EC2` · `S3` · `SNS` · `CloudWatch Logs` **敏感标识符在运行时作为环境变量注入** — 它们从未硬编码在源代码中: ``` SNS_TOPIC_ARN → SNS topic for all incident alerts FORENSIC_BUCKET → S3 bucket for evidence storage QUARANTINE_SG_ID → Security group for EC2 isolation ``` 这种模型意味着: - **零基础设施管理** — 完全无服务器 - **成本高效** — 仅在执行时计费 - **自动扩展** — 原生随 AWS 扩展 - **无需打补丁** ## 测试与验证 每种事件类型都通过模拟真实世界攻击场景进行了测试: | 测试场景 | 方法 | |---|---| | 未经授权的访问 | 对测试 IAM 用户重复进行控制台登录失败尝试 | | 异常 API 活动 | 从新 IP 地址脚本化调用敏感 API | | S3 配置错误 | 对测试存储桶故意应用公共 ACL | **捕获的验证产物(见附录):** - Lambda 执行日志(CloudWatch) - CloudTrail 事件匹配 - S3 存储桶策略变更前后对比 - SNS 通知送达回执 ## 成果 | 阶段 | 测量时间 | |---|---| | 检测 | 2–5 分钟 | | 调查 | ≤ 10 分钟 | | 遏制 | ≤ 5 分钟 | | **总响应时间** | **< 20 分钟** | **相比手动流程的改进:** 响应时间减少约 93%(从 4–6 小时缩短到 20 分钟以下)。 ## 未来增强 - [ ] **基于 ML 的异常检测** — 用基于用户/角色行为基线的模型取代静态阈值 - [ ] **使用 DynamoDB 跟踪已知 IP** — 使 `is_new_ip_for_user()` 达到生产就绪状态 - [ ] **多云集成** — 将检测和遏制扩展到 Azure 和 GCP - [ ] **采用 SOAR 平台** — 与 Splunk SOAR 或 Palo Alto XSOAR 集成以实现完全编排 - [ ] **自动执行运行手册** — 通过与工单系统(如 Jira、ServiceNow)集成来闭环升级流程 ## 参考资料 - Gartner. (2023). *安全编排、自动化和响应解决方案市场指南.* https://www.gartner.com/en/documents/security-orchestration-automation-response - IBM Security. (2023). *2023 年数据泄露成本报告.* https://www.ibm.com/reports/data-breach - NIST. (2021). *SP 800-61 Rev. 2: 计算机安全事件处理指南.* https://doi.org/10.6028/NIST.SP.800-61r2 - Gogolin, G. (2021). 数字取证工具包. In *数字取证解析* (pp. 39–54). CRC Press. https://doi.org/10.1201/9781003049357-3 - Kharb, L., & Chahal, D. (2023). 云访问安全代理:强化云安全. *国际研究出版与评论杂志, 4*(8), 642–644. https://doi.org/10.55248/gengpi.4.823.50412 - Mujahid, B. (2023). *云取证:调查云环境中的安全事件.* OSF 预印本. https://doi.org/10.31219/osf.io/4ty6u
**CLCS 660 — 基于 AI 的云自动化与脚本** 马里兰大学全球校区 · 第 7 单元实验作业 学生:Jane Buro · 提交日期:2026 年 4 月 28 日
标签:Amazon SNS, AWS, AWS Config, CloudTrail, DPI, EventBridge, FTP漏洞扫描, IAM, Lambda, Python, S3配置错误检测, URL发现, 事件驱动, 云安全事件响应, 安全包含, 安全检测, 安全编排, 安全调查, 安全通知, 异常API活动检测, 无后门, 无服务器安全, 未授权访问检测, 漏洞探索, 结构化查询, 网络安全, 自动化事件响应, 自动化安全, 逆向工具, 金融安全, 隐私保护