GuardDuty 调查代理公开预览:把威胁发现、90 天日志与资源拓扑串成调查报告

2026-07-28 26 预计阅读时间: 1 分钟
来源: infoq.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:10 分钟

AWS 推出了 Amazon GuardDuty investigation agent 的公开预览版。它不只是重复展示一条 GuardDuty finding,而是把安全发现、最近 90 天的活动日志和云资源拓扑关联起来,生成带有风险评级、置信度以及 MITRE ATT&CK 分类的结构化报告。

这项能力瞄准的是安全运营中最耗时的一段工作:告警出现后,分析人员需要确认涉及哪些资源、前后发生了什么、行为是否符合已知攻击技术,以及事件是否值得立即升级。调查代理可以缩短信息收集和初步归因的时间,但在公开预览阶段,每个账户每天最多只能运行 10 次调查,因此调用策略和人工复核仍然很重要。

从单条 finding 走向上下文调查

传统告警处理经常从一条 finding 开始。值班人员随后切换到日志查询、资源清单、身份关系和网络配置等不同界面,手工拼出时间线。问题不只是慢,还容易因为信息分散而漏掉关键关联。

GuardDuty 调查代理将三类信息放进同一调查流程:

  • GuardDuty findings:提供触发调查的可疑行为和受影响对象。
  • 最近 90 天的活动日志:帮助还原事件前后的操作轨迹,而不是只看告警瞬间。
  • 资源拓扑:补充实例、身份、存储、网络等资源之间的关系。

输出也不只是自然语言摘要,而是结构化报告,其中包含风险评级、置信度和 MITRE ATT&CK 分类。这些字段适合接入现有分级流程:风险评级可用于安排响应顺序,置信度可以提示分析人员需要投入多少验证成本,MITRE ATT&CK 分类则有助于把事件映射到统一的攻击技术语言。

不过,置信度不能被当成“是否入侵”的最终结论。日志缺失、保留周期、资源权限以及跨账户可见性都可能影响调查上下文。更稳妥的做法是让代理完成证据汇总和初步分类,由安全分析人员对高影响动作做最终判断。

MCP 接入改变了调查入口

调查代理可以通过 AWS MCP Server 访问,这意味着调查不一定只能从控制台发起。支持 MCP 的智能体工具可以把安全调查嵌入聊天式工作台、事件响应机器人或内部 SOC 流程。

一个典型工作流可以是:

  1. GuardDuty finding 进入告警队列。
  2. 自动化规则先做去重、资产重要性判断和基础过滤。
  3. 智能体通过 AWS MCP Server 发起调查。
  4. 调查报告写入工单,附带风险、置信度和 MITRE ATT&CK 分类。
  5. 高风险事件由人员确认,并执行隔离、凭证轮换或进一步取证。

MCP 在这里解决的是工具连接问题,而不是替代安全决策。尤其要避免让调查智能体直接拥有不受约束的资源修改权限。调查权限与处置权限最好分离:前者读取 findings、日志和拓扑,后者通过单独审批的自动化执行。

可以这样实践:在调用 MCP 前做每日配额控制

公开预览每个账户每天最多进行 10 次调查。如果每条 finding 都直接触发代理,很容易在一天开始时就耗尽配额。下面这个可运行的 Python 脚本实现了一个本地调查队列:它按严重度排序、按 finding ID 去重,并在生成待提交任务前执行每日 10 次的客户端配额限制。

这里有一个明确假设:脚本只负责挑选调查任务和生成提示词,实际的 MCP 工具名称与调用参数需要按照公开预览期间的 AWS MCP Server 配置进行替换。

先保存为 schedule_investigations.py

#!/usr/bin/env python3
import argparse
import json
from datetime import datetime, timezone
from pathlib import Path

DAILY_LIMIT = 10


def load_json(path, default):
    file = Path(path)
    if not file.exists():
        return default
    return json.loads(file.read_text(encoding="utf-8"))


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("findings", help="Path to a JSON array of GuardDuty findings")
    parser.add_argument("--state", default="investigation-state.json")
    args = parser.parse_args()

    findings = load_json(args.findings, [])
    today = datetime.now(timezone.utc).date().isoformat()
    state = load_json(args.state, {"date": today, "used": 0, "ids": []})

    if state.get("date") != today:
        state = {"date": today, "used": 0, "ids": []}

    remaining = max(0, DAILY_LIMIT - int(state.get("used", 0)))
    investigated_ids = set(state.get("ids", []))

    candidates = [
        item for item in findings
        if item.get("id") and item["id"] not in investigated_ids
    ]
    candidates.sort(key=lambda item: float(item.get("severity", 0)), reverse=True)
    selected = candidates[:remaining]

    tasks = []
    for finding in selected:
        tasks.append({
            "finding_id": finding["id"],
            "severity": finding.get("severity"),
            "resource": finding.get("resource", "unknown"),
            "agent_instruction": (
                "Investigate this GuardDuty finding using available AWS MCP tools. "
                "Correlate the finding with up to 90 days of activity logs and resource "
                "topology. Return evidence, timeline, affected resources, risk rating, "
                "confidence score, MITRE ATT&CK classification, and recommended next steps. "
                "Do not perform remediation actions."
            )
        })

    state["used"] += len(selected)
    state["ids"].extend(item["id"] for item in selected)
    Path(args.state).write_text(json.dumps(state, indent=2), encoding="utf-8")

    print(json.dumps({
        "date": today,
        "daily_limit": DAILY_LIMIT,
        "selected": len(tasks),
        "remaining_after_selection": DAILY_LIMIT - state["used"],
        "tasks": tasks
    }, indent=2))


if __name__ == "__main__":
    main()

准备一个最小输入文件 findings.json

[
  {
    "id": "finding-001",
    "severity": 8.7,
    "resource": "i-0123456789abcdef0"
  },
  {
    "id": "finding-002",
    "severity": 5.2,
    "resource": "arn:aws:iam::123456789012:user/example"
  }
]

运行:

python3 schedule_investigations.py findings.json > investigation-tasks.json
cat investigation-tasks.json

生成的 tasks 可以交给支持 AWS MCP Server 的智能体客户端。生产环境不要把“已选中”直接视为“调查成功”:应在 MCP 调用成功后再扣减配额,失败则重试或释放名额。还可以在排序中加入资产等级、是否面向公网、账户环境和重复告警数量,而不是只看 severity。

报告进入 SOC 后还要补哪些控制

结构化输出的价值在于可编排,但自动化链路需要明确边界:

  • 保留原始证据:报告中引用的 finding、日志事件和资源关系应能回溯,避免工单只剩一段总结。
  • 记录调查元数据:保存发起账户、时间、finding ID、工具版本和调查状态,便于审计与重放。
  • 验证高风险结论:高风险且低置信度的事件通常更需要人工分析,而不是自动关闭或自动隔离。
  • 限制智能体权限:优先授予只读调查权限,不要默认开放删除资源、修改策略或停用凭证等能力。
  • 规划配额:每天 10 次的预览配额更适合用于高价值告警、聚合后的事件或人工点选调查。
  • 准备预览变更:工具接口、字段和配额在正式发布前可能变化,集成层应避免硬编码过多细节。

是否值得现在接入

如果团队已经使用 GuardDuty,并且分析人员经常花时间在日志、资源关系和 ATT&CK 映射之间来回切换,这个公开预览值得在非关键流程中验证。可以先选择一个账户和少量高严重度 findings,对比代理报告与人工调查结果,重点观察证据完整度、误判类型、平均调查耗时和配额消耗。

适合的采用顺序是:先让代理生成报告,再让报告辅助工单分级,最后才考虑有限度的响应自动化。调查速度可以交给智能体提升,但隔离生产资源、撤销身份权限等高影响动作,仍应保留审批、审计和回滚机制。


相关推荐