Metadata-Version: 2.5
Name: otgate
Version: 0.1.1
Summary: OT-aware authorization gateway between an LLM agent and an industrial OPC UA server.
Project-URL: Homepage, https://github.com/alikup-ai/otgate
Project-URL: Repository, https://github.com/alikup-ai/otgate
Project-URL: Issues, https://github.com/alikup-ai/otgate/issues
Project-URL: Changelog, https://github.com/alikup-ai/otgate/commits/main
Author: Ali K.
License: Apache-2.0
License-File: LICENSE
Keywords: guardrails,industrial,llm,mcp,opc-ua,ot-security
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Security
Requires-Python: >=3.11
Requires-Dist: mcp<2,>=1.2.0
Requires-Dist: pyyaml>=6.0
Provides-Extra: dev
Requires-Dist: httpx>=0.27; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Provides-Extra: opcua
Requires-Dist: asyncua>=1.0; extra == 'opcua'
Description-Content-Type: text/markdown

# otgate

[![CI](https://github.com/alikup-ai/otgate/actions/workflows/ci.yml/badge.svg)](https://github.com/alikup-ai/otgate/actions/workflows/ci.yml)

An OT-aware authorization gateway that sits between an LLM agent and an industrial OPC UA server, applying a process-aware access policy — value ranges, rate limits, interlocks — to every call the agent makes.

[Русская версия →](README.ru.md)

<!-- Ownership tokens for the official MCP Registry. It verifies a PyPI package
     against server.json by looking for this string in the package description,
     which is this file. Invisible when rendered. Do not remove.

     Two of them on purpose. The registry searches by server NAME only
     (substring match, per its OpenAPI spec) and does not index descriptions,
     so the published name carries "opcua" to be findable at all. The bare
     "otgate" entry is kept deprecated rather than deleted, and keeping its
     token here means its status can still be managed. -->
<!-- mcp-name: io.github.alikup-ai/otgate-opcua -->
<!-- mcp-name: io.github.alikup-ai/otgate -->

<p align="center">
  <img src="assets/demo.gif" alt="otgate intercepting an agent's writes to a reactor: the policy checks run in order — value range, interlocks, rate of change, drift, call budget, human approval — and each write is allowed, parked for approval, or denied with a reason" width="100%">
</p>

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

<details>
<summary>The same decisions as terminal output (<code>python examples/demo.py</code>)</summary>

<p align="center">
  <img src="assets/demo.svg" alt="otgate demo: an LLM agent's writes to a reactor gated by policy — ALLOW, ASK, DENY with reasons" width="100%">
</p>

</details>

otgate is itself an MCP server. The agent connects to it as if it were an ordinary OPC UA MCP; internally otgate holds a client to the real OPC UA server. Every call is transparent to the agent, and otgate intercepts each one to apply policy and record an audit line.

## Why

- Existing OPC UA MCP servers write to tags with **no authorization layer** — the agent has full, unbounded access.
- Existing LLM guardrails (LlamaFirewall, NeMo Guardrails) work on **text/prompts** and do not understand process semantics — they know nothing about ranges, rates of change, or interlocks.
- otgate closes exactly that gap: **OT semantics + admission policy + audit.**

It is not "you may write to this tag" but "you may write *within range X, no faster than Y over interval Z, and only while the emergency tag W is not active*".

## Where this fits: OWASP ASI02

In the [OWASP Top 10 for Agentic Applications](https://genai.owasp.org/), the
risk otgate targets is **ASI02 — Tool Misuse & Exploitation**: an agent that
stays inside its granted privileges but uses a legitimate tool unsafely —
parameter pollution, excessive execution, or a call that is only dangerous given
the current system state.

Identity and access layers (ASI03) answer *may this agent call this tool*. They
do not look at **what** it writes, **how fast**, or **under which conditions**.
otgate is the layer that does:

| ASI02 sub-mechanism | otgate |
| --- | --- |
| Parameter pollution | `value_range` — the argument itself is checked |
| Excessive execution / scale | `max_rate` over `rate_interval`, durable across restarts |
| Unsafe given system state | `interlocks` — reads live guard tags before allowing |
| Unsafe composition (salami attack) | `cumulative_range` — bounds total drift over a window, so slicing a forbidden move into legal steps does not work |
| Recursion / call storms | `max_calls` over `calls_interval` |
| Unverifiable safety condition | fails **closed** — denies rather than guesses |

Per-step limits judge one write; cumulative limits judge a *series*. Without the
latter an agent walks around `max_rate` with many small legal steps:

```
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]
```

A full mapping against all ten OWASP agentic risks — including what otgate
explicitly does **not** cover (tool-chain manipulation, goal hijacking, memory
poisoning, rogue-agent detection) — is in
**[docs/threat-model.md](docs/threat-model.md)**.

## Install

```bash
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+.

## Policy example

A policy is a YAML list of per-tag rules. Any tag **not** listed is denied by default.

```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
```

Access levels: `read`, `write`, `write_with_approval`, `deny`.

Write check order (first failing check wins): tag-in-policy → access → value range → interlocks → rate of change → approval → allow.

## Demo

```bash
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}
```

## Run as an MCP server

otgate exposes: `read_tag`, `write_tag`, `browse`, `get_audit_log`, and the
operator approval tools `list_pending`, `approve`, `deny`.

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

Configuration is via environment variables (so the backend switches without code changes):

| Variable | Default | Meaning |
| --- | --- | --- |
| `OTGATE_BACKEND` | `fake` | `fake` or `asyncua` |
| `OTGATE_POLICY` | `examples/reactor_policy.yaml` | policy YAML path |
| `OTGATE_AUDIT` | `audit.jsonl` | audit log path |
| `OTGATE_SHADOW` | *(off)* | `1`/`true` to enable shadow mode |
| `OTGATE_RATE_HISTORY` | *(in-memory)* | path to a JSONL file to make rate-of-change history **survive restarts** |
| `OTGATE_OPCUA_ENDPOINT` | — | required when backend is `asyncua` |
| `OTGATE_AGENT_TOKEN` | — | bearer token for the single-agent HTTP channel |
| `OTGATE_AGENTS` | — | path to `agents.yaml` for multi-agent mode (overrides single-agent) |
| `OTGATE_OPERATOR_TOKEN` | — | bearer token for the operator HTTP channel (must differ from every agent token) |
| `OTGATE_HOST` / `OTGATE_AGENT_PORT` / `OTGATE_OPERATOR_PORT` | `127.0.0.1` / `8770` / `8771` | HTTP bind host and ports |

Example MCP client config (FakeBackend):

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

### Connecting a real OPC UA server

No code changes — install the extra and point the config at your server:

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

The backend lives behind an abstract interface (`otgate/backends/base.py`) with two implementations — `FakeBackend` (in-memory simulator, no dependencies) and `AsyncuaBackend` (real OPC UA). `asyncua` is imported lazily, so the fake backend, the tests and the demo run without it.

## Shadow mode

With `OTGATE_SHADOW=1`, a write that the engine **allows** is **not** executed on the backend — instead the audit records a "WOULD execute" line with `executed: false`. This lets you run an agent and see what it *would* do to the process without any real effect.

## Human approval workflow

A `write_with_approval` tag returns `ASK`. Rather than silently drop the write,
otgate **parks** it as a pending request with a TTL and hands the agent an
approval id. A human then resolves it through the operator tools:

- `list_pending` — show parked write requests.
- `approve <id>` — approve **and** execute, *after re-checking policy* against
  the current process state. If an interlock tripped or the value drifted out of
  range while the request waited, the approval is recorded but the write is
  **blocked** — an approval is intent, never a blind execute.
- `deny <id>` — reject; the write never runs.

Requests that nobody answers within the TTL **expire** (fail-safe: silence means
no). Every step — the ASK, the approval/denial, the eventual execution — is
audited.

> The stdio server exposes all tools in one process (fine for local use). For a
> real deployment, run the **two authenticated HTTP channels** below so the agent
> physically cannot reach `approve`/`deny`.

## Deployment & security

For anything beyond local experimentation, run otgate over HTTP as **two
separate channels**, each behind its own bearer token:

- **agent channel** — `read_tag`, `write_tag`, `browse`
- **operator channel** — `get_audit_log`, `list_pending`, `approve`, `deny`

The agent holds only the agent token, so it cannot even see the approval tools —
isolation is by channel, not by convention.

```bash
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
```

Every request must carry `Authorization: Bearer <token>`; anything else gets
`401`. Tokens are compared in constant time. otgate refuses to start if either
token is unset or if the two are equal.

**Network isolation is not optional.** otgate is a soft policy layer: if the
OPC UA server is reachable directly, an agent can bypass the gateway entirely and
the policy means nothing. A safe deployment therefore requires:

1. The OPC UA server accepts connections **only** from the otgate host/process
   (bind to loopback or a private segment; firewall everything else).
2. The agent can reach **only** the agent HTTP channel — never the OPC UA server
   and never the operator channel.
3. The operator channel is reachable only by the human/console that approves.

Put another way: otgate is effective exactly when it is the *only* path to the
server. It complements — never replaces — hardware interlocks / SIS.

### Multiple agents with different rights

To run several agents — each with its own token, its own policy, and its own
identity in the audit trail — point `OTGATE_AGENTS` at an `agents.yaml`:

```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)
```

```bash
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
```

Each agent is served on its own port (from `OTGATE_AGENT_PORT` upward) behind its
own token, so the diagnostics agent physically cannot use the optimizer's rights
and vice versa. All agents share one backend, one audit log, one rate-history and
one approval store — so rate limits are global per tag and an operator sees every
agent's pending approvals. Every audit line carries the `agent` that made the
call; when an operator approves a parked write, it is attributed to the agent
that *requested* it, not the operator.

Startup is fail-closed: otgate refuses to launch on a duplicate id/token, a
missing policy file, an operator token that matches an agent token, or an
operator port that collides with the agent port range.

## Resilience & health

otgate fails **closed** when the backend is unreachable — it never turns an
outage into a silent success:

- A write whose **rate or drift baseline cannot be read** is **denied**: an
  unverifiable baseline is treated exactly like an unverifiable interlock.

- A read or write during an outage returns an **ERROR** decision (not an
  exception, not a fake success) and is written to the audit log with
  `executed: false`.
- A write whose **interlock cannot be read** is **denied** — otgate will not run
  a write it could not safety-check.
- A parked write **approved** during an outage is not executed; it stays pending
  (or is denied on the fail-closed re-check) so the operator can retry.

The real OPC UA backend **auto-reconnects**: a dropped connection triggers a
bounded retry-with-backoff and one operation retry before giving up with a
`BackendError`. A missing tag stays a plain error and never triggers reconnects.

Each HTTP channel exposes an unauthenticated **`GET /health`** for liveness
probes — status only, never tag data:

```bash
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
```

Opening `/health` does not open anything else — `/mcp` still requires the token.

If the otgate **process itself** dies, the agent simply loses its channel (no
process, no access) while the physical process keeps running on the PLC — which
is the safe direction. Run otgate under a supervisor (systemd, a container
restart policy) and point your monitor at `/health`.

## Performance & fidelity

otgate is a policy layer, not a signal transformer — it decides whether a call may
pass, it does not reshape the value. Measured against the bundled asyncua reactor
simulator over a real `opc.tcp://` connection:

- **No signal distortion.** Values written through otgate read back byte-for-byte
  identical from the server — worst absolute error `0` across values from `0.1` to
  `123456.789` and `π` to full double precision. otgate never rounds, rescales or
  coerces; the value goes to `asyncua` as-is and the type is the server's.
- **No data loss** on allowed traffic; **intentional** loss on denied traffic —
  a `DENY`/`ASK` write never reaches the server (that is the whole point), and
  every call, allowed or not, produces exactly one append-only audit line.
- **Latency overhead ≈ +0.16 ms/call** (mean) over the raw backend read — about
  **+20 %** relative on a sub-millisecond local read. The policy check itself is
  microseconds; `audit.record()` is ~0.03 ms (the file handle is kept open and
  flushed, not reopened per call). A rate-limited write costs one extra backend
  read to fetch the live baseline, and an interlocked write one read per guard
  tag — deliberate safety round-trips.

This is comfortably fast for the intended use — an LLM/agent issuing setpoints on
the scale of seconds. **otgate is not designed to sit inside a real-time control
loop** (PLC scan-cycle, millisecond determinism); deterministic control stays
below it in the PLC/DCS. Audit durability is OS-level (flushed on every record,
survives a process crash) but not `fsync`'d per line — a stricter mode is a TODO.

## Tests

```bash
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
```

The **unit** suite runs entirely on the FakeBackend with no external dependencies. Each policy rule has a test proving it works: in-range → ALLOW/ASK, out-of-range → DENY, ESD interlock → DENY, over-rate → DENY, approval tag → ASK, read-only → DENY, tag-not-in-policy → DENY — plus policy validation, audit fields, shadow mode, rate-history persistence, the approval workflow, channel auth/isolation, per-agent policies, and fail-closed/health behaviour.

The **integration** suite (`tests/test_integration_asyncua.py`) starts the asyncua reactor simulator on a fresh port and drives the same scenarios — including an approval that really lands on the server and a mid-session server stop that must fail closed — over a live `opc.tcp://` connection. It is skipped automatically when `asyncua` is not installed.

Both suites run in CI (GitHub Actions) on Python 3.11 and 3.12.

## What this does NOT do yet (v0.1)

Honest limitations — these are deliberate cuts, not oversights:

- **No agent roles.** Every caller is treated the same; there is no per-identity policy.
- **No time windows.** You cannot express "only during the day shift".
- **Scalars only.** Values are numbers or booleans; arrays/structs are not supported.
- **Single client.** No multi-user access, queueing, or locking.
- **Agent config is static.** Multiple agents, each with their own token, policy and audit identity, are supported via `agents.yaml`. But the set of agents is read at startup — there is no live add/revoke; changing it means a restart. Tokens are long-lived (no rotation/expiry).
- **`max_rate` is wall-clock, and its time base is per-process.** The *baseline* is the tag's live value, read from the backend at check time, so the limit holds even when something other than otgate moves the tag. The *time base* still comes from otgate's own record of the last write it let through: with no record there is no measurable interval, so the write is bounded to one full `rate_interval` allowance rather than being rate-checked against elapsed time. Set `OTGATE_RATE_HISTORY` to a file path to keep that time base across restarts. otgate still does not read the historian trend, so it cannot see *how* the tag got to its current value.
- **No GUI.**

## Roadmap

- Live agent management (add/revoke without restart, token rotation/expiry)
- Time-window conditions (shifts, maintenance windows)
- Historian-trend rate limits (the live-value baseline landed in v0.2; reading the actual trend is the remaining step)
- Cross-tag budgets: a group of tags sharing one cumulative allowance, so an agent cannot spread its effect across several individually-compliant tags
- More interlock actions beyond `deny`
- High availability (otgate is single-instance today; a crash drops agent access until restart)

## License

Apache-2.0. See [LICENSE](LICENSE).
