自定义脚本安全护栏

自定义脚本安全护栏在网关内部运行由你编写的检查逻辑。当你要对接的服务不属于内置服务提供方时使用它——自研分类器、AISIX 尚未集成的商业产品，或者需要组合多项检查的策略。

其他每一种安全护栏类型都只对接一家服务提供方的协议。内容检查类 API 并没有行业标准，因此要对接一个使用自有协议的服务，通常意味着在它前面构建并运维一个转换服务。自定义脚本把这层转换放进网关内部，于是它是十几行配置，而不是一次部署。

本指南将编写一个针对自有服务检查提示词的脚本，把它只绑定到单个模型以便安全试用，并验证 AISIX 会在调用上游模型前阻断命中的请求。

准备工作

开始前，请准备以下内容：

阅读安全护栏行为，了解钩子点、执行模式和失败策略。
一个网关可以访问的检查服务。脚本在你自己的网关内运行，因此它可以是你私有网络中的服务。
以下配置路径之一：
- AISIX Cloud，其中包含环境、已接入的网关和具有写权限范围的 Admin Token。对于 On-Premises，请按照 AISIX Cloud 快速入门操作。如需申请混合云访问权限，请联系 API7。
- 加载声明式 resources.yaml 文件的开源 AISIX 网关。
可用的模型别名和能发送 Chat Completions 请求的调用方 API Key。
curl。AISIX Cloud 路径还会用到 jq。

编写脚本

脚本是一个 ES 模块，导出 checkInput、checkOutput 或两者。未导出的钩子会被跳过，因此一个脚本可以只覆盖一个方向。

export async function checkInput(ctx) {
  const resp = await fetch("https://screening.internal/scan", {
    method: "POST",
    headers: {
      "content-type": "application/json",
      "x-api-key": ctx.secrets.SCAN_KEY,
    },
    body: JSON.stringify({ text: ctx.text }),
  });
  if (!resp.ok) {
    throw new Error("screening service returned " + resp.status);
  }
  const body = await resp.json();
  return body.outcome.deny
    ? { action: "block", reason_code: body.outcome.rule }
    : { action: "none" };
}

像这个脚本在非 2xx 响应时那样抛出异常，会把裁决交给安全护栏的失败策略，而不是自行猜测。按默认策略这会阻断请求——检查类控制在无法完成检查时正应如此。

钩子收到什么

字段	说明
`ctx.hook`	`"input"` 或 `"output"`。
`ctx.text`	所有文本片段拼接后的结果，适用于把请求作为整体检查的脚本。
`ctx.segments`	以数组形式给出的文本片段。做改写的脚本需为每个片段返回一个替换值。
`ctx.messages`	带角色的片段，形如 `{role, text}`。
`ctx.model`	请求指向的模型。仅输入钩子有。
`ctx.secrets`	该安全护栏上配置的凭据，按名称索引。

钩子返回什么

返回值	结果
`{action: "none"}`	放行。
`{action: "block", reason_code, reason}`	拒绝。两个字段都只进入网关日志，都不会到达调用方。
`{action: "mask", segments, counts}`	放行，但内容已被替换。`segments` 的条目数必须与 `ctx.segments` 一致。

返回其他任何内容——包括什么都不返回——都会被当作失败，而不是放行。一个什么都没判定的脚本，不应被读成一个判定为允许的脚本。

脚本可以调用什么

API	用途
`fetch(url, init)`	调用你的服务。返回 `{status, ok, headers, text(), json()}`。
`console.log/info/warn/error/debug`	写入网关日志。
`crypto.hmac(alg, key, data, outEncoding, keyEncoding)`	HMAC-SHA1 或 HMAC-SHA256，用于需要签名请求的服务。
`crypto.hash(alg, data, outEncoding)`	SHA-1 或 SHA-256。
`crypto.base64Encode/base64Decode(data)`	Base64。
`crypto.randomUUID()`	随机 UUID，可用作 nonce。
`aisix.embed(model, texts)`	使用同一环境中的 `embedding` 模型做向量化，用于按语义检查。

crypto.hmac 接受并返回 "hex" 或 "base64"，读取密钥时支持 "utf8"、"hex" 或 "base64"。之所以支持以 hex 读取密钥，是为了能表达派生链——每一步的原始输出作为下一步的密钥：

let k = crypto.hmac("sha256", "AWS4" + ctx.secrets.SECRET, date, "hex");
k = crypto.hmac("sha256", k, region, "hex", "hex");

没有文件系统、没有环境变量访问、没有定时器。fetch 失败或算法不受支持都会抛出异常，因此你的脚本可以自行决定如何处理。

创建自定义脚本安全护栏

请先把它绑定到单个模型，而不是整个环境。脚本是你刚写的代码，把它限定在你自己控制的模型上，意味着出错只影响你正在测试的流量。参见下文调试脚本。

AISIX Cloud

创建安全护栏，并带上脚本读取的凭据：

curl -X POST "$ADMIN_ENDPOINT/api/environments/$ENV_ID/guardrails" \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "in-house-screening",
    "enabled": true,
    "kind": "custom",
    "hook_point": "input",
    "fail_open": false,
    "config": {
      "script": "export async function checkInput(ctx) { const r = await fetch(\"https://screening.internal/scan\", { method: \"POST\", headers: { \"content-type\": \"application/json\", \"x-api-key\": ctx.secrets.SCAN_KEY }, body: JSON.stringify({ text: ctx.text }) }); if (!r.ok) { throw new Error(\"screening returned \" + r.status); } const b = await r.json(); return b.outcome.deny ? { action: \"block\", reason_code: b.outcome.rule } : { action: \"none\" }; }",
      "secrets": { "SCAN_KEY": "your-screening-service-key" },
      "timeout_ms": 5000
    }
  }' | jq

凭据加密存储且不会回显。之后读取时会显示它们的名称，以便你确认脚本可以读到哪些凭据，但不会显示值。

将它绑定到单个模型：

curl -X POST "$ADMIN_ENDPOINT/api/environments/$ENV_ID/guardrail_attachments" \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"guardrail_id": "'"$GUARDRAIL_ID"'", "scope_type": "model", "scope_id": "'"$MODEL_ID"'", "enabled": true}' | jq

你也可以在 AISIX Cloud 控制台中创建和编辑该安全护栏，控制台提供脚本编辑器和凭据的名称/值编辑器。

开源 AISIX 网关

在 resources.yaml 文件中添加该安全护栏。使用块标量可以让脚本保持可读：

guardrails:
  - name: in-house-screening
    enabled: true
    kind: custom
    hook_point: input
    fail_open: false
    script: |
      export async function checkInput(ctx) {
        const resp = await fetch("https://screening.internal/scan", {
          method: "POST",
          headers: {
            "content-type": "application/json",
            "x-api-key": ctx.secrets.SCAN_KEY,
          },
          body: JSON.stringify({ text: ctx.text }),
        });
        if (!resp.ok) {
          throw new Error("screening service returned " + resp.status);
        }
        const body = await resp.json();
        return body.outcome.deny
          ? { action: "block", reason_code: body.outcome.rule }
          : { action: "none" };
      }
    secrets:
      SCAN_KEY: your-screening-service-key
    timeout_ms: 5000

guardrail_attachments:
  - guardrail_name: in-house-screening
    scope_type: model
    scope_id: your-model-alias
    enabled: true

用 SIGHUP 重新加载网关。

验证安全护栏

发送一个你的检查服务会拒绝的请求：

curl -i "$GATEWAY_ENDPOINT/v1/chat/completions" \
  -H "Authorization: Bearer $CALLER_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model": "your-model-alias", "messages": [{"role": "user", "content": "something your service denies"}]}'

AISIX 返回 422 和 content_filter 错误，并且不会调用上游模型。无关的请求照常返回模型的回答。

调试脚本

没有单独的控制台来试跑脚本。请按调试路由的方式来调试它：在一个只有你自己在用的模型上验证。

创建一个用完即弃的模型。 指向与正式模型相同的服务提供方，别名取 screening-test 之类。
只把安全护栏绑定到该模型，即上文的 scope_type: model。环境中的其他内容不受影响。
向该别名发送请求，并查看网关日志。脚本中的 console.log 会输出到那里，因此你可以打印服务返回了什么、脚本做了什么判定。
行为符合预期后再放大范围。 把安全护栏改绑到环境或你真正想检查的模型上，然后删除测试模型。

脚本行为不符合预期时，先检查两件事：

语法错误在保存时就会被拒绝，并给出行号和列号。如果安全护栏保存成功，说明脚本能够解析。
运行期失败的脚本会出现在网关日志和该请求的用量事件中，原因字段会区分失败种类：custom_timeout 表示脚本耗尽了预算，custom_script_error 表示脚本抛出了异常，custom_bad_verdict 表示脚本返回的不是一个裁决。

检查模型响应

导出 checkOutput，并把 hook_point 设为 output 或 both。响应钩子以相同的结构接收模型的回复。

对于流式响应，AISIX 按窗口释放内容：每个窗口先被检查，通过后再释放。这样流式客户端仍能保持响应，而不必等待整段回答。如果你更希望缓冲整段响应后一次性检查，可将 stream_processing_mode 设为 buffer_full。

改写内容

返回 {action: "mask", segments}，为 ctx.segments 中的每个条目提供一个替换值：

export function checkInput(ctx) {
  const masked = ctx.segments.map((s) => s.replace(/\d{3}-\d{2}-\d{4}/g, "<SSN>"));
  const changed = masked.filter((s, i) => s !== ctx.segments[i]).length;
  if (changed === 0) return { action: "none" };
  return { action: "mask", segments: masked, counts: { US_SSN: changed } };
}

counts 是可选的，用于按名称记录你检出了什么，写入该请求的用量事件。只记录名称——记录匹配到的内容会把它带进遥测数据。

在 AISIX 无法把文本替换回去的调用位置上，改写请求会阻断而不是放行原文，因此策略不会只被执行一半。

限制与失败行为

配置项	默认值	说明
`timeout_ms`	`5000`	单次钩子调用的预算，涵盖脚本本身及它发起的所有调用。单次调用的超时请在脚本内自行设置。
`max_memory_bytes`	`16777216`	脚本的内存上限。单个响应体最多可占用其四分之一，因为响应体需要在同一块内存中解析。
`fail_open`	`false`	无法完成检查时如何处理请求。
`output_fail_open`	`false`	同上，用于响应。

该预算分别覆盖两类失控：陷入循环的脚本会被中断，等待一个永不返回的调用的脚本会被放弃。无论哪种情况，都由失败策略决定最终裁决。

每次调用都在全新的沙箱中运行，因此脚本存下的任何内容都不会留到下一个请求。需要持久保存的检查状态应放在你自己的服务里。

后续步骤

安全护栏行为——钩子点、执行模式和失败策略的详细说明。
语义筛查——如果内置的按语义检查已经够用。
安全护栏概述——内置服务提供方的完整列表。

准备工作​

编写脚本​

钩子收到什么​

钩子返回什么​

脚本可以调用什么​

创建自定义脚本安全护栏​

AISIX Cloud​

开源 AISIX 网关​

验证安全护栏​

调试脚本​

检查模型响应​

改写内容​

限制与失败行为​

后续步骤​