[Python知识库]OPCUA和asyncua --- [2] 加密

使用asyncua来做加密非常容易，比open62541要简单很多，这也体现了python的优势。

一 Server

代码如下，

import asyncio
import sys

import logging
sys.path.insert(0, "..")
from asyncua import Server
from asyncua import ua
from asyncua.crypto.permission_rules import SimpleRoleRuleset
from asyncua.server.users import UserRole
from asyncua.server.user_managers import CertificateUserManager

logging.basicConfig(level=logging.INFO)


async def main():

    cert_user_manager = CertificateUserManager()
    # 添加被信任的Client证书，这里添加2个Client证书
    await cert_user_manager.add_admin("uaexpert.der", name='test_user1')
    await cert_user_manager.add_admin("peer-certificate-example-1.der", name='test_user2')

    server = Server(user_manager=cert_user_manager)

    await server.init()

    server.set_endpoint("opc.tcp://127.0.0.1:4840/freeopcua/server/")
    server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt],
                               permission_ruleset=SimpleRoleRuleset())
    # 加载Server证书和密匙
    await server.load_certificate("certificate-example.der")
    await server.load_private_key("private-key-example.pem")

    idx = 0

    # populating our address space
    myobj = await server.nodes.objects.add_object(idx, "MyObject")
    myvar = await myobj.add_variable(idx, "MyVariable", 0.0)
    await myvar.set_writable()  # Set MyVariable to be writable by clients

    # starting!

    async with server:
        while True:
            await asyncio.sleep(1)
            current_val = await myvar.get_value()
            count = current_val + 0.1
            await myvar.write_value(count)


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

Client证书有2个，其中之一是UaExpert的，Server证书和密匙是asyncua例子目录下的，需要把Server证书放到UaExpert的trusted目录下，

与open62541不同的地方是：密匙的格式是pem，不是der

打开UaExpert，可以看到加密的endpoint，与设置一样，

连接这个endpoint，最后连接成功，

二 Client

import asyncio
import logging
import sys
sys.path.insert(0, "..")
from asyncua import Client, Node, ua
from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256

# client证书和密匙
cert = f"peer-certificate-example-1.der"
private_key = f"peer-private-key-example-1.pem"


async def task(loop):
    url = "opc.tcp://127.0.0.1:4840/freeopcua/server/"
    
    client = Client(url=url)
    
    await client.set_security(
        SecurityPolicyBasic256Sha256,
        certificate=cert, # client自己的证书
        private_key=private_key, # client自己的密匙
        server_certificate="certificate-example.der" # 信任的服务器证书
    )
    
    async with client:
        objects = client.nodes.objects
        child = await objects.get_child(['0:MyObject', '0:MyVariable'])
        print('current: ', await child.get_value())
        await child.set_value(42)
        print('after modified: ', await child.get_value())


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


if __name__ == "__main__":
    main()

主要是client.set_security()函数，设置了安全策略类型，client的证书和密匙，server的证书。由于server里已经添加了client的证书，这样就可以互相信任。

运行Client可以获取当前变量的值，然后修改掉这个值,

三 总结

本文主要讲述如何使用加密通信，可以看出使用asyncua加密非常简单，也没有其它依赖，只要导入相关模块就行了。