otgate — OT policy gateway

otgate

一个具备 OT 感知能力的授权网关，位于 LLM Agent 与工业 OPC UA 服务之间，为 Agent 发起的每次调用应用具备工艺感知能力的访问策略——包括数值范围、速率限制和联锁。

Русская версия →

[MCP client / agent] ---MCP---> [otgate] ---OPC UA---> [OPC UA server]
                                    |
                     [policy engine + audit + shadow mode]

相同决策的终端输出（python examples/demo.py）

otgate 本身就是一个 MCP 服务。Agent 连接它时，就像连接一个普通的 OPC UA MCP；otgate 在内部连接真实的 OPC UA 服务。对 Agent 来说，每次调用都是透明的；otgate 会拦截每次调用，应用策略并记录审计日志。

为什么需要 otgate

• 现有的 OPC UA MCP 服务可以在没有授权层的情况下写入标签——Agent 拥有完整且不受限制的访问权限。
• 现有的 LLM 防护措施（LlamaFirewall、NeMo Guardrails）针对的是文本/提示词，无法理解工艺语义——它们不了解数值范围、变化率或联锁。
• otgate 正好填补了这一空白：OT 语义 + 准入策略 + 审计。

策略不是“你可以写入这个标签”，而是“你可以在范围 X 内写入，速度不得超过区间 Z 内的 Y，并且只有在紧急标签 W 未激活时才可以写入”。

适用场景：OWASP ASI02

在 OWASP Agentic 应用十大风险中，otgate 针对的风险是 ASI02 — 工具误用与滥用：Agent 虽然没有超出获授权限，却不安全地使用合法工具——例如参数污染、过度执行，或在当前系统状态下才具有危险性的调用。

身份和访问层（ASI03）回答的是此 Agent 是否可以调用此工具。但它们不会检查 Agent 写入的内容、写入的速度或写入的条件。otgate 正是负责这一层：

ASI02 子机制	otgate
参数污染	value_range — 检查参数本身
过度执行 / 规模过大	rate_interval 内的 max_rate，重启后仍然有效
在当前系统状态下不安全	interlocks — 允许操作前读取实时保护标签
不安全的组合操作（切香肠攻击）	cumulative_range — 限制一个时间窗口内的累计漂移，因此无法将被禁止的操作拆分为多个合法步骤
递归 / 调用风暴	calls_interval 内的 max_calls
安全条件无法验证	默认拒绝 — 拒绝操作而不是猜测

单步限制评估单次写入；累计限制评估一连串写入。没有后者时，Agent 可以通过连续执行多个幅度较小的合法步骤，绕过 max_rate：

step 1: 50.0 -> 52.5  (drift  +2.5)  ALLOW
step 2: 52.5 -> 55.0  (drift    +5)  ALLOW
step 3: 55.0 -> 57.5  (drift  +7.5)  ALLOW
step 4: 57.5 -> 60.0  (drift   +10)  ALLOW
step 5: 60.0 -> 62.5  (drift +12.5)  DENY  cumulative drift too large:
                                           62.5 is +12.5 from 50, allowed [-10, 10]

docs/threat-model.md 中介绍了针对 OWASP 全部十项 Agentic 风险的完整映射，包括 otgate 明确不覆盖的风险（工具链操纵、目标劫持、记忆投毒、恶意 Agent 检测）。

安装

pip install -e .           # core: FakeBackend, engine, MCP server
pip install -e ".[dev]"    # + pytest for the test suite
pip install -e ".[opcua]"  # + asyncua for a real OPC UA backend

Python 3.11+。

策略示例

策略是一个 YAML 列表，包含各标签的规则。默认拒绝未列出的所有标签。

# examples/reactor_policy.yaml
- tag: "ns=2;s=Reactor.TIC101.PV"     # temperature, read-only
  access: read

- tag: "ns=2;s=Reactor.TIC101.SP"     # temperature setpoint
  access: write_with_approval          # writes need human approval (ASK)
  value_range: [40, 80]                # only within 40..80 degC
  cumulative_range: [-10, 10]          # ... and no more than 10 degC total drift
  cumulative_interval: 3600            #     from where it stood an hour ago
  max_rate: 5                          # no faster than 5 degC ...
  rate_interval: 60                    # ... per 60 seconds
  interlocks:
    - tag: "ns=2;s=Reactor.ESD"        # blocked while emergency shutdown ...
      condition: "== true"             # ... is active
      action: deny

访问级别：read、write、write_with_approval、deny。

写入检查顺序（以首个未通过的检查为准）：策略中的标签 → 访问权限 → 数值范围 → 联锁 → 变化率 → 审批 → 允许。

演示

python examples/demo.py

otgate demo — agent interacting with a reactor via the policy gateway
---------------------------------------------------------------------
  read temperature (PV)                            ALLOW read allowed -> 55.0
  write setpoint = 60 degC (in range)              ASK   write requires human approval
  write setpoint = 200 degC (out of range)         DENY  value 200.0 is outside allowed range [40.0, 80.0]
  write setpoint = 60 degC while ESD active        DENY  interlock active: ns=2;s=Reactor.ESD == True (current ns=2;s=Reactor.ESD = True)
  write setpoint = 78 degC one second later        DENY  rate of change too high: |78.0 - 50.0| = 28 in 1s exceeds 5 per 60s

audit log
---------
{"timestamp": "...", "action": "read",  "node_id": "ns=2;s=Reactor.TIC101.PV", "value": null,  "decision": "ALLOW", "reason": "read allowed", "shadow": false, "executed": true}
{"timestamp": "...", "action": "write", "node_id": "ns=2;s=Reactor.TIC101.SP", "value": 60.0,  "decision": "ASK",   "reason": "write requires human approval", "shadow": false, "executed": false}
{"timestamp": "...", "action": "write", "node_id": "ns=2;s=Reactor.TIC101.SP", "value": 200.0, "decision": "DENY",  "reason": "value 200.0 is outside allowed range [40.0, 80.0]", "shadow": false, "executed": false}
{"timestamp": "...", "action": "write", "node_id": "ns=2;s=Reactor.TIC101.SP", "value": 60.0,  "decision": "DENY",  "reason": "interlock active: ns=2;s=Reactor.ESD == True (current ns=2;s=Reactor.ESD = True)", "shadow": false, "executed": false}
{"timestamp": "...", "action": "write", "node_id": "ns=2;s=Reactor.TIC101.SP", "value": 78.0,  "decision": "DENY",  "reason": "rate of change too high: |78.0 - 50.0| = 28 in 1s exceeds 5 per 60s", "shadow": false, "executed": false}

作为 MCP 服务运行

已发布到官方 MCP Registry，名称为 io.github.alikup-ai/otgate-opcua，并发布到 PyPI，名称为 otgate。 注册表名称中带有 opcua，因为注册表的搜索只会匹配服务名称，不会索引 描述，因此只有已经知道 otgate 这个词的人才能搜到仅名为 otgate 的条目。

otgate 提供以下工具：read_tag、write_tag、browse、get_audit_log，以及 操作员审批工具 list_pending、approve、deny。

otgate                      # console script (stdio transport)
python -m otgate.server     # equivalent

通过环境变量进行配置（因此切换后端无需修改代码）：

变量	默认值	含义
OTGATE_BACKEND	fake	fake 或 asyncua
OTGATE_POLICY	examples/reactor_policy.yaml	策略 YAML 路径
OTGATE_AUDIT	audit.jsonl	审计日志路径
OTGATE_SHADOW	(off)	设置为 1/true 以启用影子模式
OTGATE_RATE_HISTORY	(in-memory)	JSONL 文件路径，用于让变化率历史记录在重启后保留
OTGATE_OPCUA_ENDPOINT	—	后端为 asyncua 时必需
OTGATE_AGENT_TOKEN	—	单 Agent HTTP 通道的 bearer token
OTGATE_AGENTS	—	多 Agent 模式下 agents.yaml 的路径（覆盖单 Agent 模式）
OTGATE_OPERATOR_TOKEN	—	操作员 HTTP 通道的 bearer token（必须与所有 Agent token 不同）
OTGATE_HOST / OTGATE_AGENT_PORT / OTGATE_OPERATOR_PORT	127.0.0.1 / 8770 / 8771	HTTP 绑定主机和端口

MCP 客户端配置示例（FakeBackend）：

{
  "mcpServers": {
    "otgate": {
      "command": "otgate",
      "env": { "OTGATE_BACKEND": "fake" }
    }
  }
}

相关项目：底层访问层

otgate 管理的是进程当前允许执行的操作——范围、相对于实时值的变化率、联锁、漂移和审批。它并不打算充当底层访问层，而这方面已有优秀方案： IndustriAgents/OPCUA-MCP 用于回答 此调用方是否有权访问此节点——默认只读，支持控制配置、允许列表中的目标、固定证书以及经过审计的控制调用。

这两个问题确实不同。向允许列表中的目标写入 40 °C，满足所有访问规则；但如果标签读数为 22 °C，且升温速率违反热限制，这仍然是错误的。没有进程限制的访问控制会允许这样的写入；没有访问控制的进程限制则无从依托。

审计记录中有一个明显差异，体现了这种区分：他们的验收标准要求记录包含配置、工具、决策和目标 ID，但不包含值，这对于访问日志来说是合理的。otgate 会记录值，因为“拒绝 200.0，超出 [40, 80]”这句话不包含具体值就无法写出。

连接真实的 OPC UA 服务

无需修改代码——安装额外依赖，并将配置指向你的服务：

pip install -e ".[opcua]"
export OTGATE_BACKEND=asyncua
export OTGATE_OPCUA_ENDPOINT=opc.tcp://localhost:4840
otgate

后端通过抽象接口（otgate/backends/base.py）接入，提供两种实现——FakeBackend（无依赖的内存模拟器）和 AsyncuaBackend（真实 OPC UA）。asyncua 采用延迟导入，因此假后端、测试和演示都可以在不安装它的情况下运行。

影子模式

设置 OTGATE_SHADOW=1 后，引擎允许的写入不会在后端执行；审计记录会改为写入一条“WOULD execute”记录，并标记 executed: false。这样，你可以运行 Agent，观察它会对进程做什么，而不会产生任何实际影响。

人工审批工作流

write_with_approval 标签会返回 ASK。otgate 不会静默丢弃写入，而是将其暂存为带有 TTL 的待处理请求，并向 Agent 提供一个审批 ID。随后，人工通过操作员工具进行处理：

• list_pending — 显示暂存的写入请求。
• approve <id> — 批准并执行，执行前会根据当前进程状态重新检查策略。如果请求等待期间联锁触发，或值漂移到允许范围之外，审批会被记录，但写入会被阻止——审批表示意图，绝不是盲目执行。
• deny <id> — 拒绝；写入不会执行。

如果在 TTL 内无人响应，请求会过期（故障安全：沉默表示否）。每一步——ASK、批准/拒绝以及最终执行——都会记录在审计日志中。

stdio 服务会在一个进程中公开所有工具（适合本地使用）。对于实际部署，请运行下方的两个经过身份验证的 HTTP 通道，这样 Agent 在物理上就无法访问 approve/deny。

部署与安全

除本地试验外，请通过 HTTP 运行 otgate，并使用两个独立通道，每个通道各自使用一个 Bearer 令牌：

• Agent 通道 — read_tag、write_tag、browse
• 操作员通道 — get_audit_log、list_pending、approve、deny

Agent 只持有 Agent 令牌，因此甚至看不到审批工具——隔离由通道实现，而非依赖约定。

export OTGATE_AGENT_TOKEN=$(openssl rand -hex 32)
export OTGATE_OPERATOR_TOKEN=$(openssl rand -hex 32)   # must differ from agent
export OTGATE_BACKEND=asyncua
export OTGATE_OPCUA_ENDPOINT=opc.tcp://127.0.0.1:4840
otgate-http     # agent on :8770/mcp, operator on :8771/mcp

每个请求都必须携带 Authorization: Bearer <token>；其他情况都会返回 401。令牌比较采用恒定时间算法。如果任一令牌未设置，或者两个令牌相同，otgate 将拒绝启动。

网络隔离不可或缺。 otgate 是一个软策略层：如果可以直接访问 OPC UA 服务，Agent 就能完全绕过网关，策略也就毫无意义。因此，安全部署必须满足以下条件：

1. OPC UA 服务仅接受来自 otgate 主机/进程的连接（绑定到 loopback 或私有网段；通过防火墙阻止其他所有连接）。
2. Agent 只能访问 Agent HTTP 通道——不能访问 OPC UA 服务，也不能访问操作员通道。
3. 操作员通道仅对负责审批的人工用户/控制台开放。

换句话说：只有当 otgate 是通往服务的唯一路径时，它才有效。它可以补充硬件联锁 / SIS，但绝不能取而代之。

具有不同权限的多个 Agent

要运行多个 Agent——每个 Agent 都有自己的令牌、策略以及审计记录中的身份——请将 OTGATE_AGENTS 指向一个 agents.yaml 文件：

# examples/agents.yaml
- id: diagnostics
  token: "…secret A…"
  policy: policies/diagnostics.yaml   # read-only
- id: optimizer
  token: "…secret B…"
  policy: policies/optimizer.yaml     # may write the setpoint (with approval)

export OTGATE_AGENTS=examples/agents.yaml
export OTGATE_OPERATOR_TOKEN=$(openssl rand -hex 32)   # must differ from all agent tokens
export OTGATE_OPERATOR_PORT=8779                       # outside the agent port range
otgate-http

每个 Agent 都通过自己的令牌在独立端口上提供服务（端口从 OTGATE_AGENT_PORT 起算），因此诊断 Agent 在物理上无法使用优化器的权限，反之亦然。所有 Agent 共用一个后端、一份审计日志、一份速率历史记录和一个审批存储库——因此速率限制按标签全局生效，操作员也能看到所有 Agent 的待处理审批。每条审计记录都会包含发起调用的 agent；操作员批准暂存的写入时，归属的是发起请求的 Agent，而不是操作员。

启动时采用故障关闭策略：如果 ID/令牌重复、策略文件缺失、操作员令牌与 Agent 令牌相同，或操作员端口与 Agent 端口范围冲突，otgate 都会拒绝启动。

弹性与健康检查

后端无法访问时，otgate 会默认拒绝，绝不会把服务中断变成悄无声息的成功：

• 如果无法读取速率或漂移基线，则拒绝写入：无法验证的基线与无法验证的联锁一视同仁。

• 服务中断期间的读取或写入会返回 ERROR 决策（不是异常，也不是虚假成功），并以 executed: false 记录到审计日志中。

• 如果无法读取联锁，则拒绝写入——otgate 不会执行无法进行安全检查的写入。

• 服务中断期间获批的暂存写入不会执行；它会保持待处理状态（或在默认拒绝的重新检查中被拒绝），以便操作员重试。

真实的 OPC UA 后端会自动重新连接：连接断开后会触发有次数限制的退避重试，并在放弃前重试一次操作；最终失败时会返回 BackendError。标签缺失仍只是普通错误，绝不会触发重新连接。

每个 HTTP 通道都提供无需身份验证的 GET /health，供存活探针使用——仅返回状态，绝不返回标签数据：

curl http://127.0.0.1:8770/health
# {"status":"ok","backend":"up","pending_approvals":0}      # 200
# {"status":"degraded","backend":"down","pending_approvals":0}  # 503 when the backend is unreachable

开放 /health 不会开放其他任何内容——/mcp 仍然需要令牌。

如果 otgate 进程本身退出，Agent 只会失去与其通道的连接（没有进程，就无法访问），而物理过程会继续在 PLC 上运行——这是更安全的方向。请在监督程序下运行 otgate（例如 systemd 或容器重启策略），并让监控程序检查 /health。

性能与保真度

otgate 是策略层，而不是信号转换器——它决定调用能否通过，不会重塑数值。通过真实的 opc.tcp:// 连接，与捆绑的 asyncua 反应堆模拟器进行对比测量：

• 不会造成信号失真。 经 otgate 写入的数值，从服务器读回时逐字节完全一致——在 0.1 到 123456.789 以及完整双精度的 π 范围内，最大绝对误差为 0。otgate 从不舍入、缩放或强制转换；数值会原样传给 asyncua，类型由服务器决定。
• 允许的流量不会丢失数据；被拒绝的流量会有意丢失数据——DENY/ASK 写入绝不会到达服务器（这正是目的），而每次调用无论是否获准，都会生成且仅生成一条只追加的审计日志。
• 相比原始后端读取，延迟开销平均约为 +0.16 ms/次调用——对于亚毫秒级的本地读取，相对增幅约为 +20 %。策略检查本身只需几微秒；audit.record() 耗时约 0.03 ms（文件句柄会保持打开并刷新，不会在每次调用时重新打开）。受速率限制的写入需要额外读取一次后端以获取实时基线，而带联锁的写入则会针对每个保护标签额外读取一次——这些往返是有意为安全而设计的。

对于预期用途而言，这个速度绰绰有余——例如大模型/Agent 每隔数秒下发一次设定值。otgate 并非为实时控制回路而设计（PLC 扫描周期、毫秒级确定性）；确定性控制应由其下层的 PLC/DCS 负责。审计记录的持久性由操作系统保障（每条记录都会刷新，进程崩溃后仍可保留），但每行不会执行 fsync——更严格的模式仍待实现。

测试

pytest -m "not integration"   # unit tests — no OPC UA dependency (the default promise)
pytest -m integration         # end-to-end against a real asyncua server (needs the opcua extra)
pytest -m asi02               # just the OWASP ASI02 evidence (see docs/threat-model.md)
pytest                         # everything

单元测试套件完全在 FakeBackend 上运行，无需外部依赖。每条策略规则都有相应测试，证明其有效：范围内 → ALLOW/ASK，超出范围 → DENY，ESD 联锁 → DENY，超过速率限制 → DENY，审批标签 → ASK，只读 → DENY，标签不在策略中 → DENY；此外还会测试策略验证、审计字段、影子模式、速率历史持久化、审批工作流、通道身份验证/隔离、按 Agent 划分的策略，以及默认拒绝/健康检查行为。

集成测试套件（tests/test_integration_asyncua.py）会在新端口启动 asyncua 反应堆模拟器，并通过真实的 opc.tcp:// 连接执行相同场景——包括一项确实写入服务器的审批操作，以及会触发默认拒绝的会话中途服务器停止。当未安装 asyncua 时，该测试套件会自动跳过。

这两个测试套件都会在 Python 3.11 和 3.12 上通过 CI（GitHub Actions）运行。

目前尚未实现的功能（v0.1）

坦诚说明这些限制——都是有意取舍，并非疏忽：

• 没有 Agent 角色。 所有调用方都一视同仁；没有针对不同身份的策略。
• 没有时间窗口。 无法表达“仅在白班期间”之类的条件。
• 仅支持标量。 值只能是数字或布尔值；不支持数组或结构体。
• 仅支持单个客户端。 不支持多用户访问、排队或锁定。
• Agent 配置是静态的。 可通过 agents.yaml 支持多个 Agent，每个 Agent 都有自己的令牌、策略和审计身份。但 Agent 列表只会在启动时读取——无法实时添加或撤销；更改配置需要重启。令牌有效期很长（不会轮换或过期）。
• max_rate 使用墙上时钟，时间基准按进程计算。 基准值是检查时从后端读取的标签当前值，因此即使有 otgate 以外的程序修改标签，限制仍然有效。时间基准仍取自 otgate 自己记录的最近一次放行写入：如果没有记录，就无法测量时间间隔，因此写入的上限是一个完整的 rate_interval 允许量，而不会根据已过时间进行速率检查。将 OTGATE_RATE_HISTORY 设为文件路径，可在重启后保留此时间基准。otgate 仍不会读取历史数据趋势，因此无法得知标签如何达到当前值。
• 没有图形界面。

路线图

• 实时管理 Agent（无需重启即可添加或撤销，支持令牌轮换或过期）
• 时间窗口条件（班次、维护窗口）
• 基于历史数据趋势的速率限制（在 v0.2 中已实现实时值基准；剩下的步骤是读取实际趋势）
• 跨标签预算：一组标签共享一个累计限额，避免 Agent 将影响分散到多个各自符合限制的标签上
• 除 deny 之外的更多联锁操作
• 高可用性（目前 otgate 仅有单个实例；发生崩溃后，Agent 将无法访问，直到重启）

许可证

Apache-2.0。参见 LICENSE。