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

# 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())