向 OPC UA Python 服务器/客户端（Asyncua）添加安全性

Question

我对OPC UA和Python都是新手，但通过asyncua的例子，我创建了一个我在实际项目中需要的示例。现在我需要为服务器和客户端添加安全性，暂时使用用户名和密码就好。

这是我的功能代码，没有安全性。如果有人知道我需要使用哪些函数来创建两个用户，一个具有管理员权限，另一个没有，请告诉我。

import asyncio
from asyncua import Server, ua
from asyncua.common.methods import uamethod
from asyncua.common.structures104 import new_struct, new_struct_field

@uamethod
def func(parent, value):
    return value * 2

async def main():
    
    server = Server()
    await server.init()
    server.set_endpoint("opc.tcp://localhost:4840/freeopcua/server/")

    uri = "http://examples.freeopcua.github.io"
    idx = await server.register_namespace(uri)

    myobj = await server.nodes.objects.add_object(idx, "MyObject")
    myvar = await myobj.add_variable(idx, "MyVariable", 0.0)
    
    await server.nodes.objects.add_method(
        ua.NodeId("ServerMethod", idx),
        ua.QualifiedName("ServerMethod", idx),
        func,
        [ua.VariantType.Int64],
        [ua.VariantType.Int64],
    )
    
    struct = await new_struct(server, idx, "MyStruct", [
        new_struct_field("FirstValue", ua.VariantType.Float, 0.0),
        new_struct_field("SecondValue", ua.VariantType.Float, 0.0),
        new_struct_field("ThirdValue", ua.VariantType.Float, 0.0),
        new_struct_field("FourthValue", ua.VariantType.Float, 0.0),
        new_struct_field("FifthValue", ua.VariantType.Float, 0.0),
    ])
    
    custom_objs = await server.load_data_type_definitions()
    
    mystruct = await myobj.add_variable(idx, "my_struct", ua.Variant(ua.MyStruct(), ua.VariantType.ExtensionObject))
    await mystruct.set_writable()
    await myvar.set_writable()
    
    print("Starting server!")
   
    async with server:
   
        while True:
        
            await asyncio.sleep(0.5)
            
            n_struct = await mystruct.get_value()
            var = await myvar.read_value()

            print ("\n%f\n%f\n%f\n%f\n%f\n%f" % (var, n_struct.FirstValue,  n_struct.SecondValue, n_struct.ThirdValue, n_struct.FourthValue, n_struct.FifthValue))

try:
    loop = asyncio.get_running_loop()
    
except RuntimeError:
    loop = None
    
if loop and loop.is_running():
    print('Async event loop already running. Adding coroutine to the event loop.')
    tsk = loop.create_task(main())
    tsk.add_done_callback(
        lambda t: print(f'Task done with result={t.result()}  << return val of main()'))

else:
    print('Starting new event loop')
    result = asyncio.run(main(), debug=True)

我尝试使用 Asyncua 的加密示例，但是无法使其工作。因此，我尝试从 Asyncua 服务器上阅读主代码的功能，并自己进行操作，但是只得到了错误。

谢谢。

- Hrathen23

Answer 1

你需要创建一个实现 get_user 方法的类。

class CustomUserManager:
    def get_user(self, iserver, username=None, password=None, certificate=None):
        if username == "admin":
            if password == 'secret_admin_pw':
                return User(role=UserRole.Admin)
        elif username == "user":
            if password == 'secret_pw':
                return User(role=UserRole.User)
        return None

在你的代码中使用管理器：

user_manager = CustomUserManager()
server = Server(user_manager=cert_user_manager)
await server.init()
server.set_endpoint("opc.tcp://localhost:4840/freeopcua/server/")

- Schroeder

---

好的，这个可以工作，我可以打开服务器，但是如果我尝试连接，会出现“用户没有执行所请求操作的权限”（BadUserAccessDenied）的错误提示。但是，如果我在客户端代码中添加client.set_user('admin')和client.set_password('secret_admin_pw')，我仍然会得到同样的错误提示，而且我不知道是否需要做其他的事情。 - Hrathen23

使用 OPC Watch，我可以使用用户名和密码连接服务器。 - Hrathen23

@Hrathen23，你的服务器需要客户端提供证书吗？ - Victor Pieper

暂时不需要，首先我想只使用用户名和密码连接客户端代码，然后再尝试添加证书。理论上，我可以通过OPC Watch无需证书连接，只需要使用用户名和密码即可。 - Hrathen23

Answer 2

好的，所以出于某些原因，如果我使用：

async with Client(url=url) as client:

要作为客户端连接到URL，连接会自动启动，我无法设置用户名，所以我做的是：

client = Client(url=url)
client.set_user('admin')
client.set_password('secret_admin_pw')

await client.connect()

现在它可以工作了。

- Hrathen23