标签：asyncua 证书 await server --- client import OPCUA

使用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加密非常简单，也没有其它依赖，只要导入相关模块就行了。

标签：asyncua,证书,await,server,---,client,import,OPCUA
来源： https://blog.csdn.net/whahu1989/article/details/119298414

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。