状态：✅ 已完成（含 mybiz 动手实验实录） 前置阅读：02-Mac本地部署实操、03-技术架构与业务模块分析、04-DAG编排引擎深入、05-FreeTaxi异步检测模式 日期：2026-09-14 素材来源：safety-api / safety-basic / safety-admin 源码逐行核实 + 本地环境真实实验 读者定位：要把自己的业务接进 JoySafety 的读者；想搞懂"鉴权、限流、配置同步"在代码里怎么落地的读者

〇、先建立全局认识：接入一个业务 = 打通两条链路

业务接入不是"调一个接口"那么简单，它是一条配置链路和一条请求链路同时打通：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

【配置链路】（只跑一次，之后变更才重跑）
  运维脚本 ──► safety-admin:8006 /config/defense/manage/business/new + online
            ──► MySQL business_info 表（版本化状态机）
  safety-api:8007 每 60s HTTP 轮询 admin /config/defense/api/conf
            ──► 内存 volatile Map 全量替换 ──► 监听器重建该业务 DAG

【请求链路】（每次审核都跑）
  业务方 ──HTTP+HMAC-SHA1──► POST /llmsec/api/defense/v2/{accessKey}
    → checkParams 参数校验 → rateLimited 限流 → verify 验签 → preProcess 预处理
    → 按 responseMode 分派（sync 直接跑 DAG / free_taxi·http·mq 异步入队）
    → DAG 逐节点调 skill 服务（keywords/bert/knowledge）→ 返回 riskCode

三个最关键的工程认知（后面全部细节围绕它们）：

1. api 不直连 MySQL。业务配置（含 secretKey）由 admin 从 MySQL 读出，api 通过 60s 轮询拉到内存——所以新建业务后最长要等约 60 秒才能调通，这是正常现象不是故障。
2. 鉴权三段式：限流（按 accessKey 的 Guava RateLimiter）→ 时效（timestamp 超 expireSeconds 拒绝）→ 签名（HmacSHA1+Base64）。accessKey 以 test 开头整体跳过第三段。
3. 多业务隔离靠"按业务名索引"：配置 Map、DAG root Map、限流器 Map、敏感词场景过滤，全部以业务名为 key，互不串扰；只有 all 词库是全局的。

一、业务配置详解（admin 侧）

1.1 创建接口与字段

接口文档：safety/safety-admin/docs/api/business_api.md；入口 BusinessManageController.java:41（/config/defense/manage/business/new）。必填字段：

建表 SQL 见 safety/safety-admin/docs/tables.sql:33（business_info），另有两列提纲里没提但很重要：access_targets（授权接口集合，默认 ["defenseV2","defenseResultV2"]，见 BusinessInfoVO.java:17）和版本化三件套 version/status/editor*。

checkConf 的超时去哪了：SessionContext.timeout()（safety-basic/.../core/session/SessionContext.java:36）按 fromRole 取对应 checkConf 的 timeoutMilliseconds，DAG 主循环每跳一个节点前检查一次（FunctionExecutorImpl.java:297），超时抛 exec_timeout(50002)。这就是笔记-03 说的"三层超时"里最外层的业务级超时。

1.2 版本化：上线才算数

业务配置同样走笔记-02 附录 C 的通用流水线：new(status=edit) → online；改配置走 newVersion → update → upgrade（upgrade 把旧版本置 offline、新版本置 online 原子切换）。只有 online 状态的记录会被 api 轮询到——只 new 不 online 的业务，api 永远看不见。

1.3 官方示例

example/safety-skills/safety-knowledge/business_init.py（注意实际路径在 safety-knowledge 子目录下）建了 test 演示业务：secretKey=123456、qpsLimit=20、两侧 checkConf 都是 checkNum=1 / timeoutMilliseconds=5000。

二、配置怎么流转：admin → MySQL → api 轮询 → 内存缓存 → DAG 重建

这是"新建业务等 60 秒"的完整答案，核心在 safety-api/.../service/DynamicConfigService.java（注意：DefenseServiceImpl、RateLimitServiceImpl 都在 service/ 包下，不是 service/impl/）：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

@PostConstruct
private void init() throws IOException { loadConf(); }          // 启动时先全量加载一次

@Scheduled(fixedRate = 60000, initialDelay = 60000)             // 之后每 60s 轮询
private void loadConf() throws IOException {
    ConfigResponse dynamicConf = loadDynamic(...);              // HTTP POST admin /config/defense/api/conf
    boolean shouldUpdate = shouldUpdate(preDynamicConf, dynamicConf);
    if (!shouldUpdate) { logger.info("配置无变更"); return; }     // 按 lastChangeTimestamp 比对
    ...
    this.nodeConfMap = nodeConfMap;                             // 全量替换 volatile Map（换对象不换内容，无锁）
    this.name2BusinessConf = name2BusinessConf;
    for (Runnable run : this.listeners) { pool.submit(run); }   // 回调监听器 → 重建所有业务 DAG
}

逐环节批注：

1. 轮询目标：conf.dynamic.url=http://safety-admin:8006/config/defense/api/conf（safety-api/docs/application_example.properties:5），admin 侧由 DefenseConfigApiController.java:42 提供。
2. 变更判定：admin 组装配置时取 业务表/DAG 表/关联原子能力表 三者 update_time 的最大值作为 lastChangeTimestamp（DefenseConfigServiceImpl.java:274-291）；api 侧逐业务比对，没变就直接返回（DynamicConfigService.java:113-151）。还有一个兜底：拉到的配置为空时拒绝更新，防止 admin 故障把 api 的配置清空。
3. DAG 重建：FunctionChainBuilderImpl.java:50 把自己注册成监听器，配置一变就 loadConf() 重新构建每个业务的执行链（反射实例化 Function、编译 groovy Router、重建 OkHttp client），root 节点按业务名存进 rootMap（:64）。
4. 限流器同步刷新：RateLimitServiceImpl.java:37 同样 @Scheduled(fixedRate = 60000) 重建全部 RateLimiter。

实验结论：新建业务 + 新 DAG，不用重启 api，60s 轮询即可生效（本章实验实测）。笔记-02 坑 5 里"改原子能力超时要重启"是另一个问题（部分资源重建不彻底），不要混淆。

三、请求怎么流转：一次请求的完整鉴权链路

入口 safety-api/.../contoller/DefenseApiV2Controller.java（contoller 是原仓库的拼写笔误）。两个端点：

• :37 POST /llmsec/api/defense/v2/{accessKey} —— 审核
• :159 POST /llmsec/api/defenseResult/v2/{accessKey} —— 主动拉异步结果（free_taxi/http 模式用）

3.1 主线代码逐段批注

1
2
3
4
5
6

defenseService.checkParams(request);                                  // :46  参数校验
request.setAccessTarget(ConstantVals.TARGET_DEFENSE_V2);              // :57  强制写死 "defenseV2"
if (defenseService.rateLimited(request)) { ... "触发限流" ... }        // :59  限流（在验签之前！）
if (!accessKey.startsWith("test")) { defenseService.verify(request); } // :74 test 开头免签名特赦
defenseService.preProcess(request);                                   // :78  OpenAI 格式解析/7000 字截断
switch (mode) { case sync: ... case free_taxi: ... }                  // :97  按 responseMode 分派

注意两个设计细节：

• 限流排在验签前面——恶意请求连验签的 CPU 都消耗不到，先被限流挡掉；但这也意味着业务不存在（40002）的错误会在限流阶段就抛出（DefenseServiceImpl.java:84-87：拿不到 BusinessConf 直接抛 token_error），轮询未同步时调新业务见到的就是它。
• accessTarget 由服务端写死为 defenseV2（ConstantVals.java:12），客户端传不传都一样——签名串里的 accessTarget 也必须用这个值，否则验签必挂。

3.2 checkParams：硬性规则（DefenseServiceImpl.java:61）

3.3 限流：Guava RateLimiter 内存级（RateLimitServiceImpl.java）

1
2
3
4
5
6
7

public boolean hasLimit(String accessKey, String accessTarget) {     // :77
    RateLimiter rateLimiter = rateLimiterMap.get(key(accessKey, accessTarget)); // 先查子维度 accessKey$accessTarget
    rateLimiter = rateLimiter != null ? rateLimiter : rateLimiterMap.get(accessKey); // 没有就用业务总限流
    if (rateLimiter == null) { return false; }
    if (rateLimiter.tryAcquire()) { return false; }                  // 快速路径：拿到令牌直接放行
    return !rateLimiter.tryAcquire(1, waitMillisecondsMap.get(accessKey), TimeUnit.MILLISECONDS); // :86 再等一个窗口
}

批注：

• key 结构：业务总限流用 accessKey；AuthControlConf.qpsLimits（safety-basic/.../core/auth/AuthControlConf.java:39）可按 accessTarget 配子维度，key 为 accessKey$accessTarget（:72-73）。
• qpsLimit 不传或 ≤0 时默认 10 QPS（:46）。
• 它不是"超限即拒"：tryAcquire() 失败后还会再等 1000/qpsLimit + 10 毫秒（:48），等到令牌照样放行。实验证实：qpsLimit=2 时串行压测根本打不出限流（请求被拖慢到 ~470ms 自动排队），必须并发突发才能看到 50001（见第七章）。这是"节流"而非"熔断"，对同步调用方很友好，但做容量评估时别误以为超 QPS 会直接报错。
• 纯内存实现、60s 全量重建——多实例部署时每个实例独立计数，实际全局限流 ≈ qpsLimit × 实例数。

3.4 验签：verify → BusinessConf → basicCheck（AuthRequest.java:26）

DefenseServiceImpl.verify()（:92）很短：plainText 为空就回填 content → 按 accessKey 查 BusinessConf → request.basicCheck(businessConf)。basicCheck 的实际执行顺序（与常见直觉不同，以代码为准）：

1
2
3
4
5
6
7

if (controlConf == null) throw "未授权";                              // :27 业务不存在/未上线
// :35-39 ① accessTarget 授权范围：配置了 accessTargets 且请求带 accessTarget 时，必须在集合内
// :41-48 ② 必填校验：timestamp>0、requestId/accessKey/plainText/signature 非空
if (System.currentTimeMillis() - timestamp > controlConf.getExpireSeconds() * 1000)
    throw sign_expire;                                                // :50 ③ 防重放：超 expireSeconds(默认300s) 拒收
boolean verify = SignUtil.verify(controlConf, this);                  // :54 ④ HMAC 验签
if (!verify) throw sign_not_matched;

即：授权范围 → 必填 → 时效 → 验签。两点值得记：

• accessTarget 校验是有条件的（集合非空且请求非空才查），而 controller 已强制写入 defenseV2，所以 defense 接口实际总会校验；默认授权集合 ["defenseV2","defenseResultV2"] 覆盖了拉结果接口。
• 时效检查只看"比服务端晚多少"，不管未来时间（timestamp 大于当前时间永远合法），客户端时钟快一点没事、慢 5 分钟必挂。

3.5 签名算法本体（safety-basic/.../core/util/SignUtil.java）

1
2
3
4
5
6
7
8
9

public static String sign(AuthControlConf authConf, AuthRequest authRequest) {   // :36
    StringBuilder plainBuilder = new StringBuilder();
    plainBuilder.append("accessKey=").append(authConf.getAccessKey())
            .append("&").append("accessTarget=").append(authRequest.getAccessTarget())
            .append("&").append("requestId=").append(authRequest.getRequestId())
            .append("&").append("timestamp=").append(authRequest.getTimestamp())
            .append("&").append("plainText=").append(authRequest.getPlainText());
    return hmacSHA1Encrypt(plainBuilder.toString(), authConf.getSecretKey());
}

签名串是带 key 的 accessKey=..&accessTarget=..&requestId=..&timestamp=..&plainText=..（顺序固定），用业务 secretKey 做 HmacSHA1，结果 Base64（:62-73）。verify()（:57）就是"服务端重算一遍再字符串相等比较"——所以 secretKey 永不上网，网传的是签名。

3.6 test 特赦实测

DefenseApiV2Controller.java:74 的 startsWith("test") 让整个 verify 被跳过。实测：timestamp=1（1970 年）、不带 signature 调 /v2/test 依然返回 code=0。生产业务千万不要用 test 开头的 accessKey，且这个硬编码前缀意味着你无法通过配置关掉它。

3.7 preProcess（DefenseServiceImpl.java:107）

验签后执行：配置了 openai.stream.business 的业务的 robot 消息按 OpenAI 流式格式解析；openai.request.business 的 user 消息解析成 openaiMessages；content 超 7000 字截断（GlobalConf.MAX_CHECK_LEN，GlobalConf.java:49）、plainText 超 14000 截断。

四、签名算法与三语言复刻

三份 demo 算法完全一致，只是语言壳不同。以 quickstart/api_test.py 为准逐行批注：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

def do_sign(body, secret_key):
    timestamp = int(time.time() * 1000)                    # 毫秒时间戳（不是秒！）
    body["timestamp"] = timestamp
    body["accessTarget"] = "defenseV2"                     # 必须与服务端写死值一致
    if 'plainText' not in body or not body['plainText']:
        body['plainText'] = body['content']                # plainText 默认取 content
    text = "&".join([k + "=" + str(body[k]) for k in
        ["accessKey", "accessTarget", "requestId", "timestamp", "plainText"]])  # 顺序固定
    sig = gen_sign(text, secret_key)                       # hmac-sha1 → base64
    body['signature'] = sig

三语言对照（全部在 safety/safety-demo/ 下）：

跨语言坑：为什么签名用 plainText 而不是整个 JSON body？safety-api/docs/api.md:36 写得很坦白——不同语言对 JSON 默认值/序列化的处理不一致，对整个 body 签名跨语言必翻车，所以只签一个纯文本字段。自己写客户端时：plainText 传什么，服务端就签什么，它和 content 不一致时以 plainText 为准。

五、四种 responseMode 怎么选

经验法则：默认 sync；只在"审核不能阻塞主流程"（流式生成）时上 free_taxi；http/mq 是给不要求实时处置的场景。选 free_taxi/http 别忘了 checkParams 要求 sessionId 必填。

六、多业务隔离机制

1
2
3
4
5

name2BusinessConf  (业务名 → BusinessConf：secretKey/qpsLimit/checkConf)
nodeConfMap        (业务名 → DAG 节点配置)
rootMap            (业务名 → DAG 执行链 root)
rateLimiterMap     (accessKey / accessKey$accessTarget → RateLimiter)
敏感词             (business_scene 列：精确匹配业务名，或 "all"/空 = 全局生效)

1. 配置按业务名索引：上面四张 Map 各管一段，业务之间互不串扰（DynamicConfigService.java:53-54、FunctionChainBuilderImpl.java:46）。鉴权时按 URL 里的 accessKey 查出本业务的 secretKey——A 业务的签名过不了 B 业务的验签，天然隔离。
2. 限流隔离：每个业务独立 RateLimiter，mybiz 被打爆不影响 test。
3. 敏感词的全局例外：keywords 服务按请求里的 businessName 做场景过滤，business_scene="all"（或空）的词对所有业务生效（safety-keywords/.../SensitiveWordsManageServiceImpl.java:180-191 的 isSceneMatch）；业务名通过 DAG 节点调用时透传（KeywordFunction.java:71-76）。所以"全司通用黑词"灌到 all，业务专属词灌到自己名下。
4. 子业务：messageInfo.ext.subBizId 存在时按 accessKey$subBizId 先查配置，查不到回退主业务（ConfigService.java:33-43、DynamicConfigService.java:223-227）——一套 secretKey 下挂多个子场景的玩法，开源版没有配套管理接口，需要手工造配置，了解即可。

七、动手实验：mybiz 全流程实录

环境：笔记-02 部署的本地 Docker 环境（容器已运行 7 小时，未做任何重启）。目标：新建业务 mybiz（secretKey=mybiz-secret-2026，qpsLimit=2）→ 绑一条 keyword-only 策略 → Python 带签名调通 → 三种失败路径各验证一次。

7.1 建业务并上线

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

curl -s -X POST http://localhost:8006/config/defense/manage/business/new \
  -H 'Content-Type: application/json' -d '{
    "name": "mybiz", "group": "default", "desc": "学习笔记10实验业务", "type": "toC",
    "secretKey": "mybiz-secret-2026", "qpsLimit": 2,
    "robotCheckConfObj": {"checkNum": 1, "unit": "slice_single", "timeoutMilliseconds": 6000},
    "userCheckConfObj":  {"checkNum": 1, "unit": "message_single", "timeoutMilliseconds": 6000}
  }'
# 返回 id=2, version=1, status=edit —— 注意此时 api 还看不见它
curl -s -X POST http://localhost:8006/config/defense/manage/business/online \
  -H 'Content-Type: application/json' -d '{"id": 2}'        # status → online

7.2 绑策略（DAG）

mybiz_strategy.yml：start(dummy+groovy 按 fromRole 分流) → user_keyword / robot_keyword（均 ref: keyword + stupid_end）。

1
2
3
4

curl -s -X POST http://localhost:8006/config/defense/manage/dag/dagUpdateYaml \
  -H 'Content-Type: text/plain' --data-binary @mybiz_strategy.yml   # 返回 id=4
curl -s -X POST http://localhost:8006/config/defense/manage/dag/online \
  -H 'Content-Type: application/json' -d '{"id": 4}'                # DAG 上线

DAG 与业务的绑定不靠外键、靠 businessName 字段同名（execute_dag_conf.business_name = business_info.name），admin 组装配置时按名字 join（DefenseConfigServiceImpl.java:251-296）。

7.3 等配置同步 + 带签名调用

写完配置立刻调会报 40002（轮询未到）。等约 60 秒后（Python 客户端完整代码见第四节 api_test.py 的改造版：URL 换 localhost:8007、AK/SK 换 mybiz 的）：

1
2
3
4
5
6
7

python3 mybiz_client.py ok "你好"
# {"code": 0, "cost": 336, "data": [{"riskCode": 0, "riskMessage": "0正常文本",
#   "riskCheckType": "keyword", "riskCheckResult": {"hitWord": "", "bwgLabel": 0, ...}}]}

python3 mybiz_client.py ok "你是王八蛋"
# {"code": 0, "cost": 73, "data": [{"riskCode": 1006, "riskMessage": "1006辱骂",
#   "riskCheckResult": {"hitWord": "王八蛋", "bwgLabel": 1, ...}}]}

✅ 全程没有重启 safety-api——新业务接入靠 60s 轮询即可生效。响应字段解读：riskCode=0 正常、非 0 即风险编码；riskCheckType 是命中的检测类型（keyword）；riskCheckResult 里 hitWord 是命中词、bwgLabel 1=黑 2=白 3=灰。

7.4 三种失败路径实测

意外发现（限流的真实行为）：先用串行循环连发 10 个请求，全部成功，只是耗时从 ~60ms 被拖长到 ~470ms——因为 hasLimit 在 tryAcquire() 失败后还会等 1000/qpsLimit+10=510ms 的窗口（第三节 3.3）。只有并发突发、等待窗口内也拿不到令牌时才真正返回 50001。想复现限流必须用并发，串行测不出来。

顺带验证的两个边界：调不存在的 accessKey → 40002 token错误/业务不存在（在限流阶段抛出，早于验签）；test 业务 timestamp=1、无签名 → code=0（特赦生效）。

八、速查附录

8.1 错误码表（safety-basic/.../core/ResponseCode.java）

8.2 接入一个新业务的最小清单

1. business/new + business/online（拿到 accessKey=name、自配 secretKey/qpsLimit）
2. dag/dagUpdateYaml + dag/online（businessName 与业务同名）
3. 等 ≤60s（api 轮询）→ 带签名调 POST /llmsec/api/defense/v2/{accessKey}
4. 客户端签名：accessKey=..&accessTarget=defenseV2&requestId=..&timestamp=毫秒&plainText=content，HmacSHA1(secretKey) → Base64

8.3 源码坐标速查

8.4 与提纲/直觉不符、以代码为准的事实

• basicCheck 顺序是 授权范围→必填→时效→验签，不是"必填→时效→授权→验签"（AuthRequest.java:35-57）。
• 签名串带 key= 前缀（accessKey=xx&...），不是纯值拼接（SignUtil.java:36-52）。
• DefenseServiceImpl / RateLimitServiceImpl 位于 service/ 包，没有 service/impl/ 这一层。
• 限流是"节流"不是"熔断"：超限先等 510ms（qpsLimit=2 时）再拒，串行压测打不出 50001。
• business_init.py 实际在 example/safety-skills/safety-knowledge/ 下。