otgate — OT policy gateway

otgate

LLM エージェントと産業用 OPC UA サーバーの間に位置する、OT を考慮した認可ゲートウェイです。値の範囲、レート制限、インターロックなど、プロセスを考慮したアクセス ポリシーをエージェントのすべての呼び出しに適用します。

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

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

同じ判定をターミナル出力で表示（python examples/demo.py）

otgate 自体が MCP サーバーです。エージェントは通常の OPC UA MCP と同様に接続し、otgate は内部で実際の OPC UA サーバーへのクライアントを保持します。エージェントから見ればすべての呼び出しは透過的ですが、otgate は各呼び出しを遮断してポリシーを適用し、監査ログを記録します。

なぜ必要か

• 既存の OPC UA MCP サーバーは認可レイヤーなしでタグに書き込むため、エージェントは制限のない完全なアクセス権を持ちます。
• 既存の LLM ガードレール（LlamaFirewall、NeMo Guardrails）はテキスト／プロンプトを対象としており、プロセスのセマンティクスを理解しません。範囲、変化率、インターロックについて何も把握していません。
• otgate はまさにそのギャップを埋めます。OT のセマンティクス＋アクセス許可ポリシー＋監査を実現します。

これは「このタグに書き込んでよい」ではなく、「範囲 X 内で、間隔 Z において Y より速くない場合に限り、かつ緊急タグ W がアクティブでない間に限り書き込んでよい」というものです。

適用領域：OWASP ASI02

OWASP Top 10 for Agentic Applicationsにおいて、otgate が対象とするリスクはASI02 — Tool Misuse & Exploitationです。これは、付与された権限の範囲内にとどまりながら、正規のツールを安全でない方法で使用するエージェントを指します。たとえば、パラメーター汚染、過剰な実行、または現在のシステム状態によってのみ危険となる呼び出しです。

ID とアクセスのレイヤー（ASI03）が答えるのは、「このエージェントはこのツールを呼び出してよいか」です。書き込みの内容、速度、条件は確認しません。otgate はその確認を担います。

ASI02 のサブメカニズム	otgate
パラメーター汚染	value_range — 引数そのものをチェックします
過剰な実行／規模	rate_interval における max_rate。再起動後も状態を保持します
システム状態によって安全でない操作	interlocks — 許可する前にガードタグの現在値を読み取ります
危険な組み合わせ（サラミ攻撃）	cumulative_range — 期間内の累積ドリフトを制限するため、禁止された操作を小さな許可済みステップに分割しても実行できません
再帰／呼び出しの集中	calls_interval における max_calls
検証不能な安全条件	フェイルクローズ — 推測せずに拒否します

ステップごとの制限は単一の書き込みを判定し、累積制限は一連の書き込みを判定します。後者がなければ、エージェントは小さな許可済みステップを多数実行して 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]

otgate が明示的に対象外とするもの（ツールチェーン操作、目標の乗っ取り、メモリ汚染、不正エージェントの検出など）を含む、OWASP のエージェント関連リスク全 10 項目との対応については、**docs/threat-model.md**をご覧ください。

インストール

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 レジストリには io.github.alikup-ai/otgate-opcua として、PyPI には otgate として公開されています。 レジストリ名に opcua が含まれているのは、検索がサーバー名のみを対象とし、説明文をインデックス化しないためです。そのため、単独の 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	(オフ)	シャドウモードを有効にするには 1／true
OTGATE_RATE_HISTORY	(メモリ内)	変化率の履歴を再起動後も保持するための JSONL ファイルのパス
OTGATE_OPCUA_ENDPOINT	—	バックエンドが asyncua の場合に必須
OTGATE_AGENT_TOKEN	—	単一エージェント HTTP チャネル用のベアラートークン
OTGATE_AGENTS	—	マルチエージェントモード用 agents.yaml のパス（単一エージェント設定を上書き）
OTGATE_OPERATOR_TOKEN	—	オペレーター HTTP チャネル用のベアラートークン（すべてのエージェントトークンと異なる必要があります）
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 は、この呼び出し元はこのノードにアクセスしてよいかに応えます。デフォルトは読み取り専用で、制御プロファイル、許可リスト対象、証明書のピン留め、監査付き制御呼び出しを提供します。

この2つの問いは、実際に異なります。許可リストに載った対象への 40 °C の書き込みは、すべてのアクセスルールを満たしますが、タグの値が 22 °C で、ランプ率が温度上限に違反するなら、やはり誤りです。プロセス制限のないアクセス制御では、その書き込みが許可されます。アクセス制御のないプロセス制限には、そもそも制御の基盤がありません。

監査ログにも違いが表れています。OPCUA-MCP の受け入れ基準では、レコードにプロファイル、ツール、判断、対象の 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）の2つの実装があります。asyncua は遅延インポートされるため、fake バックエンド、テスト、デモは asyncua なしで実行できます。

シャドウモード

OTGATE_SHADOW=1 の場合、エンジンが許可した書き込みはバックエンドで実行されません。代わりに、監査ログに executed: false を含む「WOULD execute」の行が記録されます。これにより、実際の影響を与えずにエージェントを実行し、プロセスに対してどのような操作を行うかを確認できます。

人間による承認ワークフロー

write_with_approval タグは ASK を返します。書き込みを黙って破棄するのではなく、 otgate は TTL 付きの保留リクエストとして保管し、エージェントに 承認 ID を渡します。その後、人間がオペレーター用ツールで処理します。

• list_pending — 保留中の書き込みリクエストを表示します。
• approve <id> — 現在のプロセス状態に対してポリシーを再確認したうえで承認し、 実行します。リクエストの保留中にインターロックが作動した場合や、値が範囲外に 変動した場合、承認は記録されますが、書き込みはブロックされます。承認は 実行の意思を示すものであり、無条件の実行指示ではありません。
• deny <id> — 拒否します。書き込みは実行されません。

TTL 内に誰からも応答がないリクエストは期限切れになります（フェイルセーフ： 応答がなければ拒否）。すべての段階（ASK、承認／拒否、最終的な実行）は監査記録に 残ります。

stdio サーバーはすべてのツールを 1 つのプロセスで公開します（ローカルでの利用には 問題ありません）。実際のデプロイでは、以下の認証済み HTTP チャネルを 2 つ 使用し、エージェントが物理的に approve／deny にアクセスできないようにします。

デプロイとセキュリティ

ローカルでの実験を超える用途では、otgate を HTTP 経由で2 つの独立したチャネル として実行し、それぞれに専用のベアラートークンを設定してください。

• エージェントチャネル — read_tag、write_tag、browse
• オペレーターチャネル — get_audit_log、list_pending、approve、deny

エージェントが保持するのはエージェント用トークンのみであるため、承認ツールを 見ることさえできません。分離は慣例ではなく、チャネルによって実現されます。

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 になります。トークンは一定時間で比較されます。どちらかの トークンが未設定の場合、または 2 つのトークンが同じ場合、otgate は起動を拒否します。

ネットワークの分離は必須です。 otgate はソフトウェアによるポリシーレイヤーです。 OPC UA サーバーに直接アクセスできる場合、エージェントはゲートウェイを完全に 迂回でき、ポリシーは意味をなしません。そのため、安全にデプロイするには次の条件が 必要です。

1. OPC UA サーバーは otgate のホスト／プロセスからの接続のみを受け付けます （ループバックまたはプライベートセグメントにバインドし、それ以外はすべて ファイアウォールで遮断します）。
2. エージェントがアクセスできるのはエージェント用 HTTP チャネルのみです。 OPC UA サーバーにもオペレーターチャネルにもアクセスできません。
3. オペレーターチャネルにアクセスできるのは、承認を行う人間／コンソールのみです。

言い換えると、otgate はサーバーへの唯一の経路である場合にのみ有効です。 otgate はハードウェアインターロック／SIS を補完するものであり、決して置き換える ものではありません。

権限の異なる複数のエージェント

それぞれに専用のトークン、ポリシー、監査記録上の ID を持つ複数のエージェントを 実行するには、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

各エージェントは専用トークンで、それぞれ独自のポート （OTGATE_AGENT_PORT から順に割り当て）上で提供されます。そのため、診断用 エージェントはオプティマイザーの権限を物理的に使えず、その逆も同様です。すべての エージェントは、1 つのバックエンド、1 つの監査ログ、1 つのレート履歴、1 つの 承認ストアを共有します。そのため、レート制限はタグごとに全体で適用され、 オペレーターはすべてのエージェントの保留中の承認を確認できます。監査ログの各行には 呼び出しを行った agent が記録されます。オペレーターが保留中の書き込みを承認した 場合、それはオペレーターではなく、リクエストしたエージェントに帰属します。

起動時はフェイルクローズです。ID／トークンの重複、ポリシーファイルの欠落、 オペレータートークンとエージェントトークンの一致、またはオペレーターのポートと エージェントのポート範囲の競合がある場合、otgate は起動を拒否します。

レジリエンスとヘルスチェック

otgate はバックエンドに到達できない場合、フェイルクローズします。障害を気付かれないままの成功にすり替えることはありません。

• レートまたはドリフトのベースラインを読み取れない書き込みは拒否されます。検証できないベースラインは、検証できないインターロックとまったく同じように扱われます。

• 障害発生中の読み取りまたは書き込みは、例外や見せかけの成功ではなく、ERROR 判定を返し、executed: false として監査ログに記録されます。

• インターロックを読み取れない書き込みは拒否されます。otgate は安全性を確認できない書き込みを実行しません。

• 障害発生中に承認された保留中の書き込みは実行されず、保留状態のままになります（またはフェイルクローズの再確認時に拒否されます）。そのため、オペレーターは再試行できます。

実際の OPC UA バックエンドは自動再接続します。接続が切れると、上限付きのバックオフ付き再試行と操作の再試行を 1 回行い、それでも失敗した場合は 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 のプロセス自体が停止した場合、エージェントは単にチャネルを失います（プロセスがなくなり、アクセスもなくなります）が、物理プロセスは PLC 上で動作し続けます。これは安全側の動作です。otgate はスーパーバイザー（systemd、コンテナーの再起動ポリシー）配下で実行し、監視対象を /health に設定してください。

パフォーマンスと忠実性

otgate はポリシーレイヤーであり、信号を変換するものではありません。呼び出しを通過させてよいかを判定しますが、値の形を変えることはありません。実際の opc.tcp:// 接続を介し、同梱の asyncua リアクターシミュレーターと比較して測定した結果は次のとおりです。

• 信号の歪みはありません。 otgate 経由で書き込んだ値は、サーバーからバイト単位でまったく同一の値として読み戻せます。0.1 から 123456.789 までの値、および倍精度の完全な精度での π において、絶対誤差の最大値は 0 です。otgate は丸め、スケーリング、型変換を行いません。値はそのまま asyncua に渡され、型はサーバー側のものになります。
• 許可されたトラフィックでデータ損失はありません。拒否されたトラフィックでは意図的な損失が発生します。DENY／ASK の書き込みはサーバーに到達しません（それが本来の目的です）。また、許可・拒否を問わず、すべての呼び出しについて追記専用の監査行が正確に 1 行生成されます。
• 生のバックエンド読み取りと比べたレイテンシーのオーバーヘッドは約 +0.16 ms／呼び出し（平均）です。1 ミリ秒未満のローカル読み取りでは、相対で約 +20 % にあたります。ポリシーチェック自体はマイクロ秒単位です。audit.record() は約 0.03 ms です（ファイルハンドルは開いたままフラッシュされ、呼び出しごとに開き直されることはありません）。レート制限付きの書き込みでは、実行時のベースラインを取得するため、バックエンド読み取りが 1 回追加されます。インターロック付きの書き込みでは、ガードタグごとに 1 回読み取ります。これらは安全性を確保するための意図的な往復処理です。

秒単位のスケールで設定値を指示する LLM／エージェントという想定用途には、十分な速度です。otgate はリアルタイム制御ループの内部で動作するようには設計されていません（PLC のスキャンサイクル、ミリ秒単位の確定性）。確定的な制御は PLC／DCS 内部で行います。監査ログの永続性は OS レベルです（各記録後にフラッシュされ、プロセスのクラッシュ後も保持されます）が、行ごとの fsync は行いません。より厳格なモードは TODO です。

テスト

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。さらに、ポリシーの検証、監査フィールド、シャドウモード、レート履歴の永続化、承認ワークフロー、チャネル認証／分離、エージェントごとのポリシー、フェイルクローズ／ヘルスチェックの動作もテストします。

統合スイート（tests/test_integration_asyncua.py）は、新しいポートで asyncua リアクターシミュレーターを起動し、稼働中の opc.tcp:// 接続を介して同じシナリオを実行します。実際にサーバーへ反映される承認や、フェイルクローズでなければならないセッション途中のサーバー停止も含まれます。asyncua がインストールされていない場合は、自動的にスキップされます。

どちらのスイートも、Python 3.11 と 3.12 で CI（GitHub Actions）上で実行されます。

まだ対応していないこと（v0.1）

制限事項を正直に記載します。これらは見落としではなく、意図的に対象外としたものです。

• エージェントのロールはありません。 すべての呼び出し元を同じように扱い、ID ごとのポリシーはありません。
• 時間帯の指定はできません。 「日勤中のみ」のような条件は設定できません。
• スカラー値のみです。 値には数値またはブール値を指定できます。配列や構造体には対応していません。
• クライアントは 1 つのみです。 マルチユーザーアクセス、キューイング、ロックには対応していません。
• エージェントの設定は静的です。 それぞれが独自のトークン、ポリシー、監査 ID を持つ複数のエージェントに、agents.yaml を使って対応できます。ただし、エージェントの一覧は起動時に読み込まれます。稼働中の追加や取り消しはできず、変更には再起動が必要です。トークンは長期間有効で、ローテーションや有効期限はありません。
• max_rate は実時間に基づき、時間の基準はプロセスごとです。 基準値には、チェック時にバックエンドから読み取ったタグの現在値を使用します。そのため、otgate 以外のものがタグの値を変更した場合でも、制限が適用されます。一方、時間の基準には、otgate が書き込みを許可した際の最終記録を使います。記録がない場合、測定可能な時間間隔がないため、経過時間に基づくレートチェックは行わず、書き込みは rate_interval 1 回分の上限値に制限されます。再起動後もこの時間の基準を保持するには、OTGATE_RATE_HISTORY にファイルパスを設定してください。otgate はヒストリアンのトレンドを読み取らないため、タグが現在値に至った経緯は確認できません。
• GUI はありません。

ロードマップ

• エージェントの稼働中管理（再起動なしでの追加や取り消し、トークンのローテーションや有効期限）
• 時間帯条件（日勤、メンテナンス時間帯）
• ヒストリアンのトレンドに基づくレート制限（現在値を基準とする機能は v0.2 で実装済みです。残るのは実際のトレンドの読み取りです）
• タグ間の予算：複数のタグで累積上限を共有し、エージェントが個別には制限内の複数タグに影響を分散できないようにします
• deny 以外のインターロックアクション
• 高可用性（現在の otgate は単一インスタンスのため、クラッシュすると再起動までエージェントはアクセスできません）

ライセンス

Apache-2.0 です。詳しくは LICENSE をご覧ください。