AI-Infra-Guard 数据自动同步 API：不重启服务、不重建镜像热更新 AI 安全规则

AI-Infra-Guard 数据自动同步 API：不重启服务、不重建镜像热更新 AI 安全规则

【免费下载链接】AI-Infra-GuardA full-stack AI Red Teaming platform securing AI ecosystems via Agent Scan, Skills Scan, MCP scan, AI Infra scan and LLM jailbreak evaluation.项目地址: https://gitcode.com/GitHub_Trending/ai/AI-Infra-Guard

A.I.G（AI-Infra-Guard）依赖data/目录下的规则文件完成 AI 基础设施扫描：指纹识别、CVE 漏洞匹配、MCP 安全插件与越狱评估数据集都存放于此。Data Auto-Sync API（接口文档）让你通过两个 HTTP 端点从官方 GitHub 仓库拉取最新规则，直接覆盖到服务器工作目录，整个过程无需重启进程或重建 Docker 镜像。读完本文，你将掌握同步接口的完整调用方式、响应字段含义、异步轮询的最佳实践，以及该机制在 common/websocket/update_api.go 中的实现原理与并发/安全设计。

接口总览

两个操作共用同一路径/api/v1/system/update-data，以 HTTP 方法区分：

• POST：触发同步（异步执行，立即返回）；
• GET：查询同步状态/进度。

路由在 common/websocket/server.go 中注册于v1.Group("/system")组下：

system := v1.Group("/system") system.Use(setupIdentityMiddleware()) { system.POST("/update-data", HandleTriggerDataUpdate) system.GET("/update-data", HandleGetUpdateStatus) }

同步会更新哪些数据目录

从 update_api.go 的常量定义可以看到，同步固定拉取以下 6 个data/子目录：

dataDirsDefault = "fingerprints,vuln,vuln_en,mcp,eval,agents"

这些目录中的规则文件会被扫描器加载执行（例如指纹规则由 common/fingerprints/preload/preload.go 中的Runner驱动，通过 DSL 表达式匹配 HTTP 响应）。由于更新是直接写盘的，新规则在下次扫描时即生效。

端点一：触发数据同步POST /api/v1/system/update-data

请求不带任何参数。从源码可以看到，ref与目录列表均为包级常量（main分支、全部 6 个目录），调用方无法指定其他分支或目录——这是刻意的设计：

// common/websocket/update_api.go const ref = defaultGitHubBranch // 恒为 "main" const dirs = dataDirsDefault // 恒为全部 data 子目录

响应字段（data对象）

所有响应均包裹在项目统一的{status, message, data}信封中。按项目约定，status = 0表示正常（空闲/运行中/成功），status = 1表示最近一次同步失败（见 HandleGetUpdateStatus 中的状态码判定逻辑）。

cURL 示例

curl -X POST http://localhost:8088/api/v1/system/update-data

示例响应（同步已启动）

{ "status": 0, "message": "sync started", "data": { "running": true, "started_at": "2026-04-20T10:00:00Z", "message": "cloning repository…", "files_updated": 0, "ref": "main" } }

端点二：查询同步状态GET /api/v1/system/update-data

响应信封与字段定义同触发端点。轮询该端点即可观察同步的实时进度——data.message会随执行阶段变化（克隆中 → 复制中 → 完成/失败）。

cURL 示例

curl http://localhost:8088/api/v1/system/update-data

示例响应（同步进行中）

{ "status": 0, "message": "copying data directories…", "data": { "running": true, "started_at": "2026-04-20T10:00:00Z", "message": "copying data directories…", "files_updated": 0, "ref": "main" } }

示例响应（同步完成）

{ "status": 0, "message": "sync complete — 312 file(s) updated from ref \"main\"", "data": { "running": false, "success": true, "started_at": "2026-04-20T10:00:00Z", "finished_at": "2026-04-20T10:00:45Z", "message": "sync complete — 312 file(s) updated from ref \"main\"", "files_updated": 312, "ref": "main" } }

示例响应（同步失败）

{ "status": 1, "message": "git clone failed: exit status 128\nfatal: unable to access 'https://github.com/...'", "data": { "running": false, "success": false, "started_at": "2026-04-20T10:00:00Z", "finished_at": "2026-04-20T10:00:05Z", "message": "git clone failed: exit status 128\nfatal: unable to access 'https://github.com/...'", "files_updated": 0, "ref": "main" } }

失败场景的典型原因包括：服务器PATH中缺少git可执行文件，或无法访问github.com:443（内网环境常见）。此时message会携带 git 的完整错误输出，便于排查。

典型工作流与 Python 完整示例

1. 触发同步—— 调用POST /api/v1/system/update-data，立即返回；
2. 轮询完成—— 周期性调用GET /api/v1/system/update-data，直到data.running为false；
3. 检查结果—— 检查data.success与data.message；
4. 无需重启—— 更新后的规则在下一次扫描时生效。

import requests import time BASE_URL = "http://localhost:8088" # 触发同步 resp = requests.post(f"{BASE_URL}/api/v1/system/update-data") print(resp.json()) # 轮询直到完成 while True: status = requests.get(f"{BASE_URL}/api/v1/system/update-data").json() data = status["data"] print(f"[{data['message']}] files_updated={data['files_updated']}") if not data["running"]: break time.sleep(3) if data.get("success"): print(f"Sync complete — {data['files_updated']} file(s) updated") else: print(f"Sync failed: {data['message']}")

源码级实现解析

异步执行与并发控制

HandleTriggerDataUpdate（update_api.go）的关键行为：

1. 加锁检查updateStatus.Running——同一时刻只允许一个同步任务；若已有任务在跑，直接返回sync already running及当前状态快照，不会启动第二个任务；
2. 置Running = true、记录StartedAt，随后go runDataUpdate(ref, dirs)异步执行，HTTP 请求立即返回；
3. 状态读取（GET）通过sync.Mutex保护并在锁内做值拷贝快照，避免读到撕裂状态。

同步执行流程（runDataUpdate）

核心函数 runDataUpdate 分为三个阶段：

os.MkdirTemp("aig-data-sync-*") # 1. 系统临时目录，defer RemoveAll 自动清理 ↓ validateRef(ref) # 2. 防御性校验 ref（恒为常量 "main"） ↓ git clone --depth 1 --branch main <repo> <tmpDir> Env: GIT_TERMINAL_PROMPT=0 # 3. 浅克隆；禁止交互式凭据提示，避免挂起 ↓ copyDataDirs(tmpDir, dirs) # 4. 递归复制 6 个 data/ 子目录到工作目录 ↓ finish(true, "sync complete — N file(s) updated from ref \"main\"")

细节要点：

• 浅克隆（--depth 1）使下载量最小化，只取main分支最新一个提交；
• GIT_TERMINAL_PROMPT=0保证即便凭据缺失也会快速失败，而不是阻塞 goroutine；
• 克隆失败时，message拼接 git 的CombinedOutput()输出（对应文档中exit status 128的失败示例）；
• 复制阶段若某子目录在当前 ref 中不存在（如上游删除了data/agents/），会被静默跳过而非报错——copyDataDirs 中对os.Stat不存在的目录直接continue。

安全设计：白名单 + 路径穿越防护

虽然当前 API 不接收任何目录参数，源码仍做了完整的纵深防御（对安全工具自身而言是恰当的示范）：

• allowedDataDirs白名单（6 个目录），白名单之外的目录名一律拒绝（update_api.go）；
• 复制前用filepath.Rel二次验证源路径不会逃逸出srcRoot/data/；
• copyDir用os.DirFS只以裸文件名读取文件，并对每个目标路径做“必须留在目标根内”的约束检查，防止符号链接逃逸（update_api.go）；
• validateRef使用正则^[a-zA-Z0-9._\-/]+$限制 git 参数字符集，防止参数注入。

测试佐证

common/websocket/update_api_test.go 提供了对复制逻辑的完整测试覆盖，用伪克隆仓库（fake clone）验证行为：

• TestCopyDataDirs_selectiveDirs：指定fingerprints,vuln时只复制这两个目录（mcp不被复制、仓库根部的README.md不会被带入）；
• TestCopyDataDirs_allDirs：使用dataDirsDefault全量复制，文件计数正确；
• TestCopyDataDirs_missingSubdir：源中缺失vuln_en时静默跳过、不返回错误——印证了上面“缺失子目录不报错”的实现事实；
• TestSplitDirs：逗号分隔目录列表解析的边界情况（空格、空串）。

运行前提与限制

1. git二进制必须在服务器PATH中——同步完全依赖 shell 外的exec.Command("git", ...)，Docker 镜像需包含 git；
2. 服务器需能访问github.com:443——失败时可通过 GET 状态接口的message字段看到 git 原始错误；
3. 只能拉取main分支的全部 6 个目录，不提供分支/目录级别的定制参数；
4. 单并发——重复 POST 不会排队，直接返回当前运行状态；
5. 覆盖式写入——复制过程直接覆盖同名文件（权限0o644），若你在本地手工修改过data/下的规则文件，同步后会被上游版本覆盖。

参考路径

【免费下载链接】AI-Infra-GuardA full-stack AI Red Teaming platform securing AI ecosystems via Agent Scan, Skills Scan, MCP scan, AI Infra scan and LLM jailbreak evaluation.项目地址: https://gitcode.com/GitHub_Trending/ai/AI-Infra-Guard