我是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服务器上读取主代码的函数,自己做一些事情,但我只得到了错误.

谢谢.

推荐答案

您必须创建一个实现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/")

Python相关问答推荐

时间序列分解

非常奇怪:tzLocal.get_Localzone()基于python3别名的不同输出?

运行总计基于多列pandas的分组和总和

处理带有间隙(空)的duckDB上的重复副本并有效填充它们

在Wayland上使用setCellWidget时,try 编辑QTable Widget中的单元格时,PyQt 6崩溃

如何在python xsModel库中定义一个可选[December]字段,以产生受约束的SON模式

我想一列Panadas的Rashrame,这是一个URL,我保存为CSV,可以直接点击

如果满足某些条件,则用另一个数据帧列中的值填充空数据帧或数组

在www.example.com中使用`package_data`包含不包含__init__. py的非Python文件

Python Pandas获取层次路径直到顶层管理

我的字符串搜索算法的平均时间复杂度和最坏时间复杂度是多少?

OpenGL仅渲染第二个三角形,第一个三角形不可见

为什么在FastAPI中创建与数据库的连接时需要使用生成器?

Beautifulsoup:遍历一个列表,从a到z,并解析数据,以便将其存储在pdf中.

当输入是字典时,`pandas. concat`如何工作?

极柱内丢失类型信息""

如何在SQLAlchemy + Alembic中定义一个"Index()",在基表中的列上

查找数据帧的给定列中是否存在特定值

当lambda函数作为参数传递时,pyo3执行

正则表达式反向查找