实时邮件威胁检测面对的不是一个漂亮的 demo，而是持续涌入、格式复杂、内容高度不可信的消息流。Abnormal AI 分享的实践重点，是把 Amazon Bedrock AgentCore Code Interpreter 当作 Agent 的临时计算草稿纸：需要时创建，处理完成后销毁，让 Agent 能在隔离环境中完成解析、计算和证据整理，同时避免把不可信邮件内容直接带进核心服务。
这个思路的价值不只在于“让模型会执行代码”。真正困难的是如何设计临时沙箱的生命周期、权限边界、输入输出协议和故障处理，使它能够服务于实时检测以及大规模消息处理。
为什么邮件安全 Agent 需要一个临时计算环境
邮件威胁分析经常需要处理模型不擅长直接完成的任务，例如：
- 解码 MIME、HTML、压缩附件或混合格式正文；
- 提取 URL、域名、附件元数据和邮件头；
- 对时间、地址、跳转链路等结构化数据做计算；
- 运行轻量级规则、统计或特征提取；
- 将多个中间结果整理成可供 Agent 推理的证据。
如果把这些工作全部放在主应用进程中，邮件正文或附件就可能影响宿主服务。即使 Agent 只生成了“看起来安全”的代码，也不能假设输入数据可信。Code Interpreter 的定位更像一个一次性工作区：Agent 可以在其中处理数据，但工作区不应成为长期状态，也不应拥有超出任务需要的网络、文件和凭证权限。
在实时检测场景中，沙箱还需要满足几个工程约束：启动延迟要可控；失败不能阻塞整条消息流水线；输出必须结构化；资源上限必须明确；同一个会话不能意外泄露上一个邮件任务的数据。
沙箱设计的四个关键决策
1. 用任务级生命周期，而不是长期共享会话
对于邮件安全，最容易接受的隔离模型是“一封邮件或一个分析任务对应一个临时计算环境”。任务结束后清理文件和会话，避免以下问题：
- 上一封邮件的附件残留在下一次分析中；
- Agent 把旧任务的上下文误认为当前证据；
- 长期运行环境积累临时文件和未知状态；
- 调试信息或敏感内容在多个租户之间串联。
这会牺牲一部分复用效率，但换来更简单的安全边界。若后续需要优化成本或延迟，可以复用经过严格清理和重置的 worker；在没有证明清理可靠之前，不应默认复用。
2. 只把必要数据放入沙箱
主服务不应把完整邮件上下文、所有租户信息和长期凭证一股脑传进去。更稳妥的输入协议可以包含：
{
"task_id": "msg-20250308-001",
"tenant_id": "tenant-a",
"artifacts": [
{
"name": "body.html",
"sha256": "...",
"content_ref": "object://temporary-input/body.html"
}
],
"limits": {
"timeout_seconds": 8,
"max_output_bytes": 65536
},
"requested_operations": [
"extract_urls",
"inspect_headers",
"summarize_redirects"
]
}
这里的 content_ref 只是示意。生产实现应使用短期授权、一次性对象路径或由平台托管的输入通道，而不是把永久访问密钥写入 Agent 的环境变量。输入也应经过大小限制和类型校验，避免压缩炸弹、超大 HTML 或异常嵌套结构拖垮分析任务。
3. 把执行结果变成证据，而不是自由文本
Agent 不应该只接收一段“分析完成”的自然语言。更可靠的输出包括：
- 任务状态和执行耗时；
- 提取出的 URL、域名、附件哈希和邮件头字段；
- 每个结果的来源文件和行号、字节偏移或解析路径；
- 规则命中项及其置信度；
- 可重试错误和不可重试错误；
- 明确的“未知”状态。
结构化结果能让上游策略引擎进行确定性判断，也方便审计和离线回放。模型可以解释证据，但不应该替代证据本身。
4. 默认禁止无关网络访问
邮件安全分析通常不需要让代码自由访问互联网。开放网络会带来数据外传、恶意域名访问、供应链下载和不可重复结果等风险。可以把外部查询拆成受控工具：由主服务或专门的信誉查询服务访问允许的 API，再把有限结果传回沙箱。
如果业务确实需要检查 URL 跳转，建议使用隔离的抓取服务、域名白名单、请求超时、响应大小限制和审计日志，而不是给 Code Interpreter 一个通用公网出口。
一个可运行的本地安全原型
下面的 Python 示例不是 Amazon Bedrock AgentCore API 调用，而是一个可以直接运行的沙箱边界原型。它演示了几个生产上值得保留的原则：临时目录、输入大小限制、超时、固定输出格式和任务结束后的清理。将它迁移到 AgentCore Code Interpreter 时，可以把 run_task 内部替换为平台的会话创建、代码执行和会话销毁调用。
运行前无需安装第三方依赖：
from __future__ import annotations
import json
import re
import signal
import tempfile
from pathlib import Path
MAX_INPUT_BYTES = 256 * 1024
MAX_OUTPUT_ITEMS = 100
class Timeout(Exception):
pass
def timeout_handler(signum, frame):
raise Timeout("analysis timed out")
def extract_urls(text: str) -> list[str]:
urls = re.findall(r"https?://[^\\s<>\"']+", text, flags=re.IGNORECASE)
return sorted(set(urls))[:MAX_OUTPUT_ITEMS]
def run_task(task_id: str, raw_email: bytes, timeout_seconds: int = 2) -> dict:
if len(raw_email) > MAX_INPUT_BYTES:
return {"task_id": task_id, "status": "rejected", "reason": "input_too_large"}
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(timeout_seconds)
try:
with tempfile.TemporaryDirectory(prefix="email-task-") as workdir:
input_path = Path(workdir) / "message.eml"
input_path.write_bytes(raw_email)
text = raw_email.decode("utf-8", errors="replace")
result = {
"task_id": task_id,
"status": "completed",
"bytes": len(raw_email),
"urls": extract_urls(text),
"evidence": ["message.eml"],
}
return result
except Timeout:
return {"task_id": task_id, "status": "timeout"}
finally:
signal.alarm(0)
if __name__ == "__main__":
sample = b"Subject: test\\n\\nVisit https://example.com/login"
print(json.dumps(run_task("demo-001", sample), ensure_ascii=False, indent=2))
这个原型还不等于生产级隔离：Python 进程本身不应被当成对抗恶意代码的完整安全边界。生产环境应依赖 AgentCore Code Interpreter 或其他具备资源隔离能力的沙箱，并额外配置 CPU、内存、磁盘、执行时长、网络出口和凭证访问策略。
面向大规模实时检测的运行方式
在十亿级消息规模下，关键不是让每封邮件都调用更多模型，而是把“是否需要计算”做成分层决策：
- 先在廉价、确定性的流水线上完成 MIME 解析、大小校验、哈希计算和基础特征提取。
- 只有遇到复杂附件、可疑重定向或需要解释的消息时，才创建 Code Interpreter 任务。
- 为每个任务设置明确的截止时间；超时后返回降级结果，而不是无限等待。
- 将沙箱执行与主检测流水线解耦，通过队列、并发上限和背压保护下游服务。
- 对任务创建、代码执行、文件访问和销毁分别记录指标与审计事件。
可以用下面的配置作为一个工程起点，再结合实际 AgentCore 接口映射字段。它是部署设计示例，不是特定 AWS API 的固定配置：
agent_task:
name: email-artifact-inspection
lifecycle: ephemeral
timeout_seconds: 8
max_concurrency: 500
input:
max_bytes: 262144
allowed_types: [text/plain, text/html, message/rfc822]
filesystem:
writable_paths: [/tmp/task]
persist_after_task: false
network:
mode: deny_by_default
allowed_services:
- reputation-query.internal
credentials:
inject: false
output:
format: json
max_bytes: 65536
include_evidence_locations: true
failure_policy:
timeout: return_degraded_result
sandbox_error: retry_once_then_fallback
要特别监控四类指标：沙箱创建延迟、任务执行延迟、资源耗尽比例和每个任务的输出大小。它们比单纯的模型响应时间更能说明系统是否接近容量边界。
生产落地前的检查清单
- 隔离：不同任务、租户和消息之间是否有明确的文件与会话边界？
- 最小权限：沙箱是否没有长期凭证，网络是否默认拒绝？
- 输入防护：是否限制文件大小、类型、压缩层级和解析深度？
- 输出协议：结果是否可被机器校验，是否携带证据位置和错误类型？
- 生命周期：成功、失败、超时和取消时是否都会清理临时资源？
- 弹性：高峰期是否有并发上限、队列和降级路径？
- 审计：是否能回放某次 Agent 决策使用了哪些输入和计算结果？
- 成本：是否只为真正需要代码执行的邮件创建沙箱？
Abnormal AI 的实践给出的核心启发是：Agent 的计算能力不应直接等同于主系统的权限。把 Code Interpreter 设计成短生命周期、最小权限、结构化输出的计算草稿纸，才能在实时邮件安全这类高吞吐场景中兼顾灵活性、隔离性和可运营性。