为基于Asyncua的OPC UA Python服务器添加用户名密码安全认证

要为你的asyncua服务器添加用户名密码认证，需完成以下核心步骤，使用对应函数即可实现需求：

1. 定义用户验证回调函数 #

创建异步函数校验客户端传入的用户名和密码，返回验证结果及用户角色：

async def user_validator(username, password):
    # 管理员用户：拥有全部权限
    if username == "admin" and password == "admin_pass":
        return True, ua.UserRole.Admin
    # 无权限用户：仅允许基础访问（或完全拒绝）
    elif username == "guest" and password == "guest_pass":
        return True, ua.UserRole.User
    # 无效凭证
    else:
        return False, ua.UserRole.Anonymous

2. 绑定用户验证函数到服务器 #

通过服务器的user_manager注册验证逻辑：

server.user_manager.set_user_validator(user_validator)

3. 配置服务器安全策略与令牌类型 #

指定支持的安全策略（包含无加密模式，适合调试），并启用用户名密码令牌认证：

# 设置支持的安全策略：无加密、签名、签名加密三种模式
server.set_security_policy([
    ua.SecurityPolicyType.NoSecurity,
    ua.SecurityPolicyType.Basic256Sha256_Sign,
    ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt
])
# 允许使用用户名密码令牌进行身份认证
server.set_security_IDs([ua.UserTokenType.UserName])

4. 为节点设置角色权限 #

针对不同用户角色配置节点访问权限，比如让管理员拥有读写权限，无权限用户无法访问：

# 获取系统角色ID
admin_role = await server.get_role_id(ua.UserRole.Admin)
user_role = await server.get_role_id(ua.UserRole.User)

# 给MyVariable设置权限：管理员可读写，普通用户无权限
await myvar.set_role_permissions([
    ua.RolePermissionType(
        RoleId=admin_role,
        Permissions=ua.PermissionType.Read | ua.PermissionType.Write | ua.PermissionType.Browse
    ),
    ua.RolePermissionType(
        RoleId=user_role,
        Permissions=ua.PermissionType.None_
    )
])

# 对MyObject和自定义结构体变量执行相同权限配置
await myobj.set_role_permissions([
    ua.RolePermissionType(
        RoleId=admin_role,
        Permissions=ua.PermissionType.Read | ua.PermissionType.Write | ua.PermissionType.Browse
    ),
    ua.RolePermissionType(
        RoleId=user_role,
        Permissions=ua.PermissionType.None_
    )
])

await mystruct.set_role_permissions([
    ua.RolePermissionType(
        RoleId=admin_role,
        Permissions=ua.PermissionType.Read | ua.PermissionType.Write | ua.PermissionType.Browse
    ),
    ua.RolePermissionType(
        RoleId=user_role,
        Permissions=ua.PermissionType.None_
    )
])

修改后的完整代码 #

将上述内容整合到原始代码中，最终实现如下：

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 user_validator(username, password):
    # 管理员用户
    if username == "admin" and password == "admin_pass":
        return True, ua.UserRole.Admin
    # 无权限用户
    elif username == "guest" and password == "guest_pass":
        return True, ua.UserRole.User
    else:
        return False, ua.UserRole.Anonymous

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

    # 配置用户认证逻辑
    server.user_manager.set_user_validator(user_validator)
    server.set_security_policy([
        ua.SecurityPolicyType.NoSecurity,
        ua.SecurityPolicyType.Basic256Sha256_Sign,
        ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt
    ])
    server.set_security_IDs([ua.UserTokenType.UserName])

    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()

    # 设置节点角色权限
    admin_role = await server.get_role_id(ua.UserRole.Admin)
    user_role = await server.get_role_id(ua.UserRole.User)

    # 配置MyVariable权限
    await myvar.set_role_permissions([
        ua.RolePermissionType(
            RoleId=admin_role,
            Permissions=ua.PermissionType.Read | ua.PermissionType.Write | ua.PermissionType.Browse
        ),
        ua.RolePermissionType(
            RoleId=user_role,
            Permissions=ua.PermissionType.None_
        )
    ])

    # 配置MyObject权限
    await myobj.set_role_permissions([
        ua.RolePermissionType(
            RoleId=admin_role,
            Permissions=ua.PermissionType.Read | ua.PermissionType.Write | ua.PermissionType.Browse
        ),
        ua.RolePermissionType(
            RoleId=user_role,
            Permissions=ua.PermissionType.None_
        )
    ])

    # 配置自定义结构体变量权限
    await mystruct.set_role_permissions([
        ua.RolePermissionType(
            RoleId=admin_role,
            Permissions=ua.PermissionType.Read | ua.PermissionType.Write | ua.PermissionType.Browse
        ),
        ua.RolePermissionType(
            RoleId=user_role,
            Permissions=ua.PermissionType.None_
        )
    ])
    
    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(f"{var}\n{n_struct.FirstValue}\n{n_struct.SecondValue}\n{n_struct.ThirdValue}\n{n_struct.FourthValue}\n{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)

说明 #

• 可根据需求修改用户名、密码及权限配置。
• 若需加密通信，需确保客户端配置对应安全策略（如Basic256Sha256_SignAndEncrypt）。
• 无权限用户（guest）尝试访问节点时会被拒绝，管理员用户拥有完整读写和浏览权限。

内容的提问来源于stack exchange，提问作者Hrathen23