Trust on First Use (TOFU) - Beckhoff

[Seversonic]

Hi All,

Trying to use this library to write a small script that will setup Beckhoff's OPCUA server. Beckhoff uses something called trust on first use, where after an install you can connect without authentication to create a password.

I am able get some of a response using client.set_security_string, and leaving the cert and key blank, but it does throw a file not found error when looking for the key.

I think if I understand this correctly, that the implementation WONT support a secure connection without a certificate and key? so me trying to automate this TOFU will not work?

/home/ryan/method.py:65: DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()
INFO:asyncua.client.ua_client.UaClient:opening connection
INFO:asyncua.uaprotocol:updating client limits to: TransportLimits(max_recv_buffer=65536, max_send_buffer=65536, max_chunk_count=64, max_message_size=4194304)
INFO:asyncua.client.ua_client.UASocketProtocol:open_secure_channel
INFO:asyncua.client.ua_client.UASocketProtocol:close_secure_channel
INFO:asyncua.client.ua_client.UASocketProtocol:Request to close socket received
INFO:asyncua.client.client:find_endpoint [EndpointDescription(EndpointUrl='opc.tcp://XXXXXX:4840', Server=ApplicationDescription(ApplicationUri='urn:CX-71CE4D:BeckhoffAutomation:TcOpcUaServer:1', ProductUri='urn:BeckhoffAutomation:TcOpcUaServer', ApplicationName=LocalizedText(Locale='en', Text='TcOpcUaServer@CX-XXXXXX'), ApplicationType_=<ApplicationType.Server: 0>, GatewayServerUri=None, DiscoveryProfileUri=None, DiscoveryUrls=['opc.tcp://XXXXXX:4840']), ServerCertificate=b


', SecurityMode=<MessageSecurityMode.SignAndEncrypt: 3>, SecurityPolicyUri='http://opcfoundation.org/UA/SecurityPolicy#Basic256Sha256', UserIdentityTokens=[UserTokenPolicy(PolicyId='Anonymous-[0]-SignAndEncrypt-Basic256Sha256', TokenType=<UserTokenType.Anonymous: 0>, IssuedTokenType=None, IssuerEndpointUrl=None, SecurityPolicyUri=None)], TransportProfileUri='http://opcfoundation.org/UA-Profile/Transport/uatcp-uasc-uabinary', SecurityLevel=115), EndpointDescription(EndpointUrl='opc.tcp://XXXXXX:4840', Server=ApplicationDescription(ApplicationUri='urn:CX-XXXXXX:BeckhoffAutomation:TcOpcUaServer:1', ProductUri='urn:BeckhoffAutomation:TcOpcUaServer', ApplicationName=LocalizedText(Locale='en', Text='TcOpcUaServer@XXXXXX'), ApplicationType_=<ApplicationType.Server: 0>, GatewayServerUri=None, DiscoveryProfileUri=None, DiscoveryUrls=['opc.tcp://XXXXXX:4840']), ServerCertificate=

Certificate here

', SecurityMode=<MessageSecurityMode.SignAndEncrypt: 3>, SecurityPolicyUri='http://opcfoundation.org/UA/SecurityPolicy#Aes256_Sha256_RsaPss', UserIdentityTokens=[UserTokenPolicy(PolicyId='Anonymous-[0]-SignAndEncrypt-Aes256_Sha256_RsaPss', TokenType=<UserTokenType.Anonymous: 0>, IssuedTokenType=None, IssuerEndpointUrl=None, SecurityPolicyUri=None)], TransportProfileUri='http://opcfoundation.org/UA-Profile/Transport/uatcp-uasc-uabinary', SecurityLevel=125), EndpointDescription(EndpointUrl='opc.tcp://XXXXXX:4840', Server=ApplicationDescription(ApplicationUri='urn:CX-XXXXXX:BeckhoffAutomation:TcOpcUaServer:1', ProductUri='urn:BeckhoffAutomation:TcOpcUaServer', ApplicationName=LocalizedText(Locale='en', Text='TcOpcUaServer@XXXXXX'), ApplicationType_=<ApplicationType.Server: 0>, GatewayServerUri=None, DiscoveryProfileUri=None, DiscoveryUrls=['opc.tcp://XXXXXX:4840']), ServerCertificate=

', SecurityMode=<MessageSecurityMode.SignAndEncrypt: 3>, SecurityPolicyUri='http://opcfoundation.org/UA/SecurityPolicy#Aes128_Sha256_RsaOaep', UserIdentityTokens=[UserTokenPolicy(PolicyId='Anonymous-[0]-SignAndEncrypt-Aes128_Sha256_RsaOaep', TokenType=<UserTokenType.Anonymous: 0>, IssuedTokenType=None, IssuerEndpointUrl=None, SecurityPolicyUri=None)], TransportProfileUri='http://opcfoundation.org/UA-Profile/Transport/uatcp-uasc-uabinary', SecurityLevel=120)] <MessageSecurityMode.SignAndEncrypt: 3> 'http://opcfoundation.org/UA/SecurityPolicy#Basic256Sha256'
INFO:asyncua.client.ua_client.UASocketProtocol:Socket has closed connection
ERROR:asyncua:error
Traceback (most recent call last):
  File "/home/ryan/method.py", line 45, in task
    await client.set_security_string("Basic256Sha256,SignAndEncrypt,,")
  File "/home/ryan/.local/lib/python3.10/site-packages/asyncua/client/client.py", line 180, in set_security_string
    return await self.set_security(policy_class, parts[2], parts[3], client_key_password, parts[4] if len(parts) >= 5 else None, mode)
  File "/home/ryan/.local/lib/python3.10/site-packages/asyncua/client/client.py", line 216, in set_security
    return await self._set_security(policy, certificate, private_key, server_certificate, mode)
  File "/home/ryan/.local/lib/python3.10/site-packages/asyncua/client/client.py", line 229, in _set_security
    cert = await uacrypto.load_certificate(certificate.path_or_content, certificate.extension)
  File "/home/ryan/.local/lib/python3.10/site-packages/asyncua/crypto/uacrypto.py", line 51, in load_certificate
    content = await get_content(path_or_content)
  File "/home/ryan/.local/lib/python3.10/site-packages/asyncua/crypto/uacrypto.py", line 39, in get_content
    async with aiofiles.open(path_or_content, mode='rb') as f:
  File "/home/ryan/.local/lib/python3.10/site-packages/aiofiles/base.py", line 98, in __aenter__
    self._obj = await self._coro
  File "/home/ryan/.local/lib/python3.10/site-packages/aiofiles/threadpool/__init__.py", line 94, in _open
    f = yield from loop.run_in_executor(executor, cb)
  File "/usr/local/lib/python3.10/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
FileNotFoundError: [Errno 2] No such file or directory: ''
INFO:asyncua.client.client:disconnect
INFO:asyncua.client.ua_client.UaClient:close_session
WARNING:asyncua.client.ua_client.UaClient:close_session but connection wasn't established
WARNING:asyncua.client.ua_client.UaClient:close_secure_channel was called but connection is closed

[Seversonic]

code here

import asyncio
import logging
from asyncua import Client, Node, ua

logging.basicConfig(level=logging.INFO)
_logger = logging.getLogger('asyncua')


async def browse_nodes(node: Node):
    """
    Build a nested node tree dict by recursion (filtered by OPC UA objects and variables).
    """
    node_class = await node.read_node_class()
    children = []
    for child in await node.get_children():
        if await child.read_node_class() in [ua.NodeClass.Object, ua.NodeClass.Variable]:
            children.append(
                await browse_nodes(child)
            )
    if node_class != ua.NodeClass.Variable:
        var_type = None
    else:
        try:
            var_type = (await node.read_data_type_as_variant_type()).value
        except ua.UaError:
            _logger.warning('Node Variable Type could not be determined for %r', node)
            var_type = None
    return {
        'id': node.nodeid.to_string(),
        'name': (await node.read_display_name()).Text,
        'cls': node_class.value,
        'children': children,
        'type': var_type,
    }


async def task(loop):
    url = "opc.tcp://192.168.178.111:4840"
    # url = "opc.tcp://localhost:4840/freeopcua/server/"
    try:
        client = Client(url=url)
        client.set_user('test')
        client.set_password('test')
        # client.set_security_string()
        #await client.set_security_string("Basic256Sha256,SignAndEncrypt,,") #this works, but throws errors with no certs
        await client.set_security(policy="Basic256Sha256",mode="SignAndEncrypt", certificate="",private_key="" )
        await client.connect()
        ###await client.find_servers()
        ####await client.find_servers()
        # Client has a few methods to get proxy to UA nodes that should always be in address space such as Root or Objects
        #root = client.nodes.root
        #_logger.info("Objects node is: %r", root)

        # Node objects have methods to read and write node attributes as well as browse or populate address space
        #_logger.info("Children of root are: %r", await root.get_children())

        #tree = await browse_nodes(client.nodes.objects)
        #_logger.info('Node tree: %r', tree)
    except Exception:
        _logger.exception('error')
    finally:
        await client.disconnect()


def main():
    loop = asyncio.get_event_loop()
    loop.set_debug(True)
    loop.run_until_complete(task(loop))
    loop.close()


if __name__ == "__main__":
    main()

[Seversonic]

I changed to using the client with auth example and i've had a lot more success. However, I'm getting and untrusted response now

/home/ryan/test.py:101: DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()
ERROR:asyncua.crypto.uacrypto:certificate does not contain the hostname in DNSNames ########. Some applications will check this.
TRYING
self-signed certificate
ERROR:asyncua.client.client:create_session fault response: The certificate is not trusted. (BadCertificateUntrusted)
ERROR:__main__:The certificate is not trusted.(BadCertificateUntrusted)

[schroeder-]

If you can connect without encryption (don't call set_security), then you don't need a certificate. If you need SecurityPolicyBasic256Sha256 or any other security policy, then you need a certificate.

You can create a certificate with the example:
https://github.com/FreeOpcUa/opcua-asyncio/blob/master/examples/client-with-encryption.py

Here is a minimalstic example (not complete):

from pathlib import Path
from cryptography.x509.oid import ExtendedKeyUsageOID
from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256
from asyncua.crypto.cert_gen import setup_self_signed_certificate
from asyncua.crypto.validator import CertificateValidator, CertificateValidatorOptions
from asyncua.crypto.truststore import TrustStore
import socket
....

cert_base = Path(__file__).parent
cert = Path(cert_base / "client_cert.der")
private_key = Path(cert_base / "client_cert.pem")

....
async def task(....):
     try:
        host_name = socket.gethostname()
        client_app_uri = f"urn:{host_name}:foobar:myselfsignedclient"
        url = "opc.tcp://127.0.0.1:4840/freeopcua/server/"
    
        await setup_self_signed_certificate(private_key,
                                            cert,
                                            client_app_uri,
                                            host_name,
                                            [ExtendedKeyUsageOID.CLIENT_AUTH],
                                            {
                                                'countryName': 'CN',
                                                'stateOrProvinceName': 'AState',
                                                'localityName': 'Foo',
                                                'organizationName': "Bar Ltd",
                                            })
         client = Client(url=url)
        client.application_uri = client_app_uri
        await client.set_security(
            SecurityPolicyBasic256Sha256,
            certificate=str(cert),
            private_key=str(private_key)
        )
        ......

Anonymous connection is the default, if you don't call anything.
If you want to use password and username you have to add this:

client.set_user('test')
client.set_password('test')

[Seversonic]

I believe I do need to set security for this first use (this is the only way ive been able to connect using UAExpert).

I need to initialize this server before I can enable the none/none endpoint with a username and password (ultimate goal).

Which I am also getting when I use UAExpert, but I dont know how to get around this in this library

Update:
Switched to BASIC_VALIDATION stopped that error from appearing, but I've had to comment out the server_certificate from the set_security method.

Still erroring out, but some progress has been made

If I print out client.nodes.objects, it returns i=85 (not sure if this is a default, but it lines up with what I see in UAExpert), but no children are being displayed.

My tree and method I need to call

[schroeder-]

Methods should normaly, have childs. But never the less you can call it without them. Because you know the needed datatypes from the documentation, at least with python.
If you want to call the method with UAexpert, you should be able to call begin a call with double click, or from the context menu.

[Seversonic]

I currently use UAExpert to do this process, but it needs to be done a lot, which is why I'm trying to automate this whole process.

I was able to get close to something working with the method call, but only if the server was remote. If I try localhost as a target, it doesnt find anything.

Connectcall fails on localhost/127.0.0.1/host_name
host name interestingly returns 127.0.1.1 in the connect call failure.

I can find the server with UAExpert using localhost or 127.0.0.1 and it resolves it to a proper HostName. I believe this is an issue with Beckhoff's implementation, as I can find other OPC servers when hosting them locally.

[oroulet]

I am a bit interested in this since we use a lot beckhoff at work.
the doc says you need to first connect wihtout authentication but they show that encryptoin is needed so you need a certificate and set up things. something like that:

async def main():
    client = Client("opc.tcp://localhost:53530/OPCUA/SimulationServer/")
    await client.set_security_string("Basic256Sha256,Sign,certificate-example.der,private-key-example.pem")
    async with client:
        node =  await client.nodes.objects.get_child("/NS:Initiilization")
        res = await node.call_method("NS:TurstOnFirstUse", user, password)
       # check res here



if __name__ == "__main__":
    logging.basicConfig(level=logging.WARN)
    asyncio.run(main())

btw use the asyncio.run method and the error over looks like path to certificate is wrong

[Seversonic]

Thank you all for the help!

Through a lot of testing, and your code snippets, I was able to get the Initialization process to work with the following code:

The only issue right now is that I cant get it working on localhost, it only works remotely.

import asyncio
import logging
import sys
import socket
from pathlib import Path
from cryptography.x509.oid import ExtendedKeyUsageOID
sys.path.insert(0, "..")
from asyncua import Client, Node
from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256
from asyncua.crypto.cert_gen import setup_self_signed_certificate
from asyncua.crypto.validator import CertificateValidator, CertificateValidatorOptions
from asyncua.crypto.truststore import TrustStore
from asyncua import ua

#Set what to display during output, INFO, ERROR, DEBUG
logging.basicConfig(level=logging.ERROR)
_logger = logging.getLogger(__name__)

cert_idx = 4
cert_base = Path(__file__).parent
cert = Path(cert_base / f"certificates/peer-certificate-example-{cert_idx}.der")
private_key = Path(cert_base / f"certificates/peer-private-key-example-{cert_idx}.pem")


host_name = socket.gethostname()
print(host_name)
client_app_uri = f"urn:{host_name}:foobar:myselfsignedclient"
url = "opc.tcp://192.168.178.111:4840"


async def task(loop):
    
    
    #create dummy certificates for connecting to the server
    await setup_self_signed_certificate(private_key,
                                        cert,
                                        client_app_uri,
                                        host_name,
                                        [ExtendedKeyUsageOID.CLIENT_AUTH],
                                        {
                                            'countryName': 'Country',
                                            'stateOrProvinceName': 'Prov',
                                            'localityName': 'City',
                                            'organizationName': "Company",
                                        })
    client = Client(url=url)
    client.application_uri = client_app_uri
    
    #set the security types for connecting to the server
    await client.set_security(
        SecurityPolicyBasic256Sha256,
        certificate=str(cert),
        private_key=str(private_key)
        #server_certificate="certificate-example.der"
    )    
    
    try:
        async with client:
            #TODO turn these into parameters to pass into a script eventually
            namespace = "http://Beckhoff.com/TwinCAT/TF6100/Server/Initialization"
            user = "username"
            password = "password"
            
            # calling a method on server
            obj = await client.nodes.root.get_child("Objects/2:Initialization")
            res = await obj.call_method("2:TrustOnFirstUse", user, password) #this is called by the OPC BrowseName (which is namespace and text name of method)
            #print(obj)
            _logger.info("method result is: %r", res)
            #while True:
                #await asyncio.sleep(1)

           
    except ua.UaError as exp:
        _logger.error(exp)


def main():
    loop = asyncio.get_event_loop()
    loop.set_debug(False)
    loop.run_until_complete(task(loop))
    loop.close()


if __name__ == "__main__":
    main()