shanto12/langgraph-cyber-ir-lab

# LangGraph 网络安全事件响应实验室 一个可部署的、**完全合成的**网络安全事件响应培训平台。它提供了一个完整的、生产级的底层架构——包括身份验证、公共控制平面、持久化任务队列、有状态的模拟 SOC、Postgres + pgvector schema、种子场景和知识语料库——让你可以专注于自己构建以下三个学习领域： - **`agent_runtime/graph/`** — LangGraph 多 Agent 调查 - **`agent_runtime/rag/`** — RAG 摄入 + 检索 pipeline - **`agent_runtime/tools/`** — 基于 mock SOC 的 LangChain tool 包装器 这三个包之外的所有内容都已经构建完成，并且现在可以针对确定性的**训练脚手架**（`apps/agent_worker/stub_runner.py`）运行。你只需在强类型的接口背后实现真正的业务逻辑，然后切换功能开关——无需修改任何平台代码。 ## 边界：已构建的 vs. 你要构建的 | 领域 | 状态 | 位置 | | --- | --- | --- | | 身份验证、会话、CSRF、安全标头、速率限制 | 已构建 | `apps/control_api/` | | 公共 REST API + HTMX UI + SSE 实时时间线 | 已构建 | `apps/control_api/` | | 持久化 Postgres 任务队列 + 事件发件箱 | 已构建 | `packages/database/`、migrations | | 场景铸造器（DeepSeek 草稿 → 验证 → 确定性回退） | 已构建 | `apps/agent_worker/foundry.py` | | 确定性调查**脚手架** + 评估器 | 已构建 | `apps/agent_worker/stub_runner.py`、`evaluator.py` | | 有状态的模拟 SIEM / EDR / 身份 / 邮件 / TI / 防火墙 / 工单 / 资产 | 已构建 | `apps/mock_soc_api/` | | 种子场景、知识语料库、RAG 评估集 | 已构建 | `scenarios/`、`knowledge/` | | 平台与你的代码之间的类型化衔接 | 已构建 | `agent_runtime/interfaces.py` | | **LangGraph 调查图** | **由你构建** | `agent_runtime/graph/` | | **RAG 摄入与检索** | **由你构建** | `agent_runtime/rag/` | | **LangChain tool 包装器** | **由你构建** | `agent_runtime/tools/` | 这三个学习包以带有详细文档的存根形式提供——每个包都公开了一个单一的 `build_*` 工厂，该工厂会抛出 `NotImplementedError` 并列出具体的 TODO 列表。请阅读 `agent_runtime/graph/__init__.py`、`rag/__init__.py` 和 `tools/__init__.py` 中的模块文档字符串，以确切了解需要实现什么。 ## 架构 三个服务加上一个数据库。**`control-api` 是唯一的公共服务。** Worker 和 mock SOC 没有公共域名；它们通过私有网络以及（针对 mock SOC）共享的 `INTERNAL_SERVICE_TOKEN` 进行通信。 ``` flowchart TD User([Analyst / browser]) -->|HTTPS| CA["control-api (PUBLIC)
FastAPI + HTMX + Jinja
Auth, REST /api/v1, SSE"] CA -->|enqueue job| DB[("PostgreSQL + PGVector
durable queue + event outbox
23 tables")] CA -->|read events / stream SSE| DB W["agent-worker (PRIVATE)
claim / heartbeat / complete
scenario foundry + runner + evaluator"] -->|claim job, write events| DB W -->|tools + containment| SOC["mock-soc-api (PRIVATE)
in-memory fake SOC
INTERNAL_SERVICE_TOKEN"] CA -.->|approval decision| DB W -.->|blocks on approval| DB ``` **`control-api`**（公共） — FastAPI + HTMX + Jinja。包含 Argon2id 密码哈希、Postgres 中的不透明服务端会话、双重提交 CSRF、登录速率限制（通过审计行）、安全标头。在 `/api/v1` 下提供 REST API、HTMX UI 以及 SSE 实时时间线。在启动时运行 `alembic upgrade head` 并以幂等方式进行播种。将任务排入 Postgres 队列。 **`agent-worker`**（私有，无域名） — 一个持久化任务循环（领取 / 心跳 / 完成，包含租约、重试和陈旧租约恢复）。托管**场景铸造器**（DeepSeek 草稿 → 严格验证 → 确定性回退）、调查运行器和确定性评估器。`INVESTIGATION_RUNNER` 标志用于选择运行器（`stub` 目前已提供并运行；`graph` 运行你的实现）。一个小型的 FastAPI 应用暴露 `/health` 和 `/ready` 以供探针检测。 **`mock-soc-api`**（私有，无域名，无数据库，内存运行） — 有状态的虚假 SIEM、EDR、身份、邮件、threat-intel、防火墙、工单和资产 API。所有业务域路由都需要 `INTERNAL_SERVICE_TOKEN`；`/health` 和 `/ready` 是开放的。一个合成安全边界会拒绝真实的 IP/域名/电子邮件（HTTP 422）。遏制操作是幂等且经过审计的。 **PostgreSQL + PGVector**（私有，基于卷的存储） — 23 张表：持久化队列（`FOR UPDATE SKIP LOCKED`、租约、重试、陈旧恢复）、事务性事件发件箱以及知识表。`knowledge_chunks.embedding` 是一个 **nullable `vector(384)`**（BAAI/bge-small-en-v1.5），保持为空以供你实现 RAG。隐藏的真实答案位于 `scenario_ground_truth` 中，并且**绝不会**由分析师/Agent 路由返回。 ### 调查流程（脚手架） 冻结的场景会被转换为 `RUN_INVESTIGATION` 任务。运行器会发出实时事件流——`run_created`、`phase_changed`、`evidence_collected`、`hypothesis_proposed`、`tool_called`、`tool_result`、`finding_created`、`approval_requested`、`approval_decided`、`containment_executed`、`report_generated`、`evaluation_completed`——并且**会阻塞并等待恰好一次人工审批**。获得批准后，它会通过 mock SOC 执行幂等的遏制操作，然后生成确定性报告和评估。`control-api` 启动运行（`HTTP 202` + `run_id`）并通过 SSE 流式传输事件。 ## 快速开始（本地开发） **前置条件：** Python 3.13、[uv](https://docs.astral.sh/uv/)，以及一个带有 `pgvector` 扩展的 PostgreSQL 16 实例（例如 CI 中使用的 `pgvector/pgvector:pg16` Docker 镜像）。 ``` # 1. 将依赖项安装到 .venv 中（base + dev group，以及所有 service extras） uv sync --extra control-api --extra agent-worker --extra mock-soc # 2. 配置环境（复制文档中记录的变量名；填入本地值） cp .env.example .env # 本地运行至少需要： # DATABASE_URL=postgresql+asyncpg://lab:lab@localhost:5432/lab # INTERNAL_SERVICE_TOKEN=$(openssl rand -hex 32) # ADMIN_BOOTSTRAP_PASSWORD=... # 在 admin seed 时进行 Argon2id 哈希处理 # MODEL_API_KEY=... # 仅当您使用 scenario foundry 时需要 # 3. 应用 migrations uv run alembic upgrade head # 4. 运行 services（每个都绑定 $PORT，默认为 8080）。共享代码通过 # PYTHONPATH 导入 — 请按照 Dockerfiles 中的方式进行设置。 export PYTHONPATH=.:packages uv run uvicorn apps.control_api.main:app --port 8080 # public UI + API; seeds on startup uv run uvicorn apps.agent_worker.main:app --port 8081 # background worker + health app uv run uvicorn apps.mock_soc_api.main:app --port 8082 # mock SOC ``` 打开 访问 HTMX UI，或打开 查看 OpenAPI schema。默认运行器是确定性脚手架，因此你可以在编写任何自己的代码之前，驱动一次端到端的完整调查流程。 ### 运行测试 ``` export PYTHONPATH=.:packages # Fast 测试，无数据库 uv run pytest tests/unit tests/contract -q # 验证 frozen seed 场景 uv run python -m scenarios.validate # Integration 测试（需要 DATABASE_URL + 正在运行的 Postgres/pgvector） uv run alembic upgrade head uv run pytest tests/integration -q ``` 使用指定的工具进行 lint 和类型检查： ``` uv run ruff check . uv run ruff format --check . uv run mypy packages apps # advisory; agent_runtime/{graph,rag,tools} are excluded ``` ## 功能开关与配置 配置通过环境变量进行（名称记录在 `.env.example` 中；**仓库中没有真正的密钥——它们只存在于 Railway 中**）。 | 标志 | 值 | 含义 | | --- | --- | --- | | `INVESTIGATION_RUNNER` | `stub`（默认） \| `graph` | `stub` 运行提供的脚手架；`graph` 运行 `agent_runtime/graph`。 | | `RETRIEVAL_BACKEND` | `stub`（默认） \| `pgvector` | `stub` 不返回任何分块；`pgvector` 运行 `agent_runtime/rag`。 | 模型配置（非机密）：`MODEL_PROVIDER=deepseek`、`MODEL_BASE_URL=https://api.deepseek.com`、`MODEL_NAME=deepseek-v4-flash`、`MODEL_THINKING=disabled`。密钥（仅在 Railway 中）：`MODEL_API_KEY`、`INTERNAL_SERVICE_TOKEN`、`ADMIN_BOOTSTRAP_PASSWORD`。 ## 仓库结构 ``` langgraph-cyber-ir-lab/ ├── agent_runtime/ # ── YOUR learning area ── │ ├── interfaces.py # typed Protocols (the seams): EventPublisher, ApprovalService, │ │ # ToolRegistry, RetrievalService, CheckpointService, InvestigationRunner │ ├── graph/ # TODO: LangGraph multi-agent investigation (build_runner) │ ├── rag/ # TODO: RAG ingestion + retrieval (build_retrieval_service) │ └── tools/ # TODO: LangChain tool wrappers (build_tool_registry) ├── apps/ │ ├── control_api/ # public FastAPI: auth, REST /api/v1, HTMX UI, SSE, seeding │ ├── agent_worker/ # durable job loop, scenario foundry, stub runner, evaluator │ └── mock_soc_api/ # stateful in-memory mock SOC + safety boundary + audit ├── packages/ │ ├── contracts/ # shared Pydantic models + enums │ ├── database/ # SQLAlchemy async models + session/engine helpers │ ├── mock_soc_client/ # typed async client + SocToolRegistry (used by the stub runner) │ └── observability/ # structlog configuration ├── scenarios/ # 5 frozen seeds + loader/validator + schemas + private ground truth ├── knowledge/ # 18 docs (7 policies, 6 playbooks, 5 reference) + manifest + 40 RAG eval Qs ├── infrastructure/ │ ├── migrations/ # Alembic (0001_initial → 23 tables) │ └── requirements/ # pinned per-service requirement files (source of truth for Docker) ├── tests/ # unit, contract, integration, e2e ├── docs/ # DEPENDENCY_BASELINE.md (+ api/architecture/deployment/learning) ├── pyproject.toml # local dev (uv) + tooling config └── .env.example # every environment variable, names only ``` ### 内容 - **5 个冻结的种子场景** — `phishing_credential_compromise`、`malware_execution`、`business_email_compromise`、`identity_session_theft`、`data_exfiltration`，每个都包含诱饵和评分标准。 - **18 份知识文档**（7 份策略、6 份手册、5 份参考资料）+ 一个清单，以及 **40 个 RAG 评估问题**（25 个可回答、7 个跨文档、5 个不可回答、3 个模棱两可）。 ## 你的起点 你要构建对接的衔接点是 **`agent_runtime/interfaces.py`** 中的 `Protocol`。提供的脚手架基于它们实现了 `InvestigationRunner`；你的任务是在三个学习包中实现相同的接口并切换标志。 1. **Tools**（`agent_runtime/tools/build_tool_registry`） — 将每个 `mock_soc_client.MockSocClient` 方法包装为带有严格 Pydantic 参数 schema 的 LangChain `@tool`；将遏制工具标记为需要批准；组装一个 `ToolRegistry`。*不要重新实现 mock SOC——通过 `MockSocClient` 调用它。* 2. **RAG**（`agent_runtime/rag/build_retrieval_service`） — 加载并对 `knowledge/` 语料库进行分块，生成 `bge-small-en-v1.5`（384 维）embedding 存入 `knowledge_chunks`，注册索引版本，并实现带有引用的混合 pgvector + 关键字搜索。针对 `knowledge/evaluation/rag_questions.json` 进行评估。然后设置 `RETRIEVAL_BACKEND=pgvector`。 3. **Graph**（`agent_runtime/graph/build_runner`） — 构建一个 LangGraph `StateGraph`，其中包含路由到专家 Agent 的 supervisor、并行证据扇出、在遏制操作之前连接到 `ctx.approvals` 的 `interrupt()`，以及连接到 `ctx.checkpoints` 的 Postgres checkpointer。然后设置 `INVESTIGATION_RUNNER=graph`。 每个存根模块的文档字符串都列出了完整的 TODO 清单。`RunContext`（在 `interfaces.py` 中）为你的运行器提供了它所需的一切：`publisher`、`approvals`、`tools`、`retrieval`、`checkpoints`、`soc`、`session_factory` 和 `settings`。 ## 部署 每个服务都有自己的 `Dockerfile`（构建上下文 = 仓库根目录），并根据其指定的 `infrastructure/requirements/*.txt` 进行安装。镜像在 **GitHub Actions 中构建并推送到 GHCR**，或者直接由 **Railway** 从连接的仓库的 Dockerfile 中构建。 - `control-api` 运行 `alembic upgrade head`，然后提供服务；播种操作在启动时以幂等方式运行。它是唯一具有公共域名的服务。 - `agent-worker` 和 `mock-soc-api` 是私有的（无域名）；mock SOC 还额外受到 `INTERNAL_SERVICE_TOKEN` 的访问控制。 **CI**（`.github/workflows/ci.yml`）：ruff lint + format、mypy（建议性）、单元/契约测试 + 种子验证、Postgres/pgvector 集成测试、Alembic up → down → up 迁移检查、gitleaks、pip-audit（建议性）以及 Docker 构建（在 `main` 分支上推送）。 ## 文档 - [`docs/DEPENDENCY_BASELINE.md`](docs/DEPENDENCY_BASELINE.md) — 指定的依赖项集合及理由。 - `docs/architecture/`、`docs/api/`、`docs/deployment/`、`docs/learning/` — 为更深入的参考资料保留；`docs/learning/USER_IMPLEMENTATION_ROADMAP.md` 是旨在指导构建三个学习包的指南。 核心依赖项：Python 3.13、FastAPI 0.139、Starlette 1.3.1、Pydantic 2.13.4、SQLAlchemy 2.0.51（异步）+ asyncpg、Alembic 1.18.5、pgvector 0.5.0、argon2-cffi、httpx、structlog、openai 2.45。

标签：AV绕过, FastAPI, LLM智能体, RAG, 事件响应培训, 人工智能, 安全运营, 扫描框架, 测试用例, 用户模式Hook绕过, 请求拦截, 逆向工具