python opcua模块加密认证通讯配置教程

python opcua 模块签名加密通讯教程

第一章 Python 通过opcua模块进行通讯搭建

文章目录

搭建流程

通过SIEMES博途配置OPCUA服务器，并启动PLC仿真完成OPCUA服务器的搭建
通过Python pip安装OPCUA模块，搭建OPCUA客户端进行通讯
教程目标：
1、通过OPCUA加密及签名认证建立OPCUA通讯
2、获取服务端Node节点，并导出为json文件
3、获取节点数值
4、更改节点数值

action: 你需要了解SIEMENS plc博途plc项目创建流程及组态

一、搭建SIEMENS plc OPCUA服务器

1、在博途软件中创建SIEMENS 1500系列plc项目，并激活OPCUA服务器

2、配置SIEMENS OPCUA 服务器

3、配置OPCUA通讯安全策略

## 4、配置用户认证

5、配置客户端证书

6、启动仿真，启用opcua服务器

启动仿真前提是博途软件已经成功安装PLCSIM Advance

7、配置仿真环境

设置plc实例名称、ip地址、子网掩码及plc系列型号，点击start;
实例启动完成后，需要在博途中点击在线，搜索plc,并将项目下载到plc实例中，启动plc完成OPCUA服务器搭建工作

8、仿真环境搭建完成验证

可以通过UAExpert调试工具对OPCUA服务器通讯进行验证

二、调试python代码进行客户端搭建

本文以OPCUA模块搭建OPCUA通讯基本搭建，最新的OPCUA模块建议选择opcua-asnicio；
如果你对python异步陌生，那OPCUA通讯仍可适用

1.pip安装opcua模块

通过pip install opcua来为python 环境安装模块
建议通过虚拟环境安装模块

2.编写python代码

代码如下（示例）：

import time
from opcua import Client, ua
import json

# opcua client class


class OpcuaClient:
    def __init__(self, url: str,
                 user_name: str = None,
                 user_password: str = None,
                 basic256: bool = False,
                 sha256: bool = False,
                 cert_path: str = None,
                 key_path: str = None,
                 application_uri=None
                 ) -> None:
        self._url = url
        self._client = Client(url)

        if user_name and user_password:
            self._client.set_user(user_name)
            self._client.set_password(user_password)
        if basic256 and not sha256:
            self._client.set_security_string("Basic256,SignAndEncrypt,None,None")
        if sha256:
            self._client.set_security_string(f"Basic256Sha256,SignAndEncrypt,{cert_path},{key_path}")

        self._client.application_uri = "urn:example.org:FreeOpcUa:python-opcua" if application_uri is None else application_uri

        # print(f'client application uri: {self._client.application_uri}')

        self._client.connect()

    @property
    def root_node(self):
        return self._client.get_root_node()

    def get_all_childeren(self, node) -> dict:
        _node_dict = {}
        for _nc in node.get_children():
            if _nc:
                print(type(_nc), _nc)
                if _nc.get_children():
                    _sub_nc = self.get_all_childeren(_nc)
                    _sub_nc['value'] = str(_nc)
                    _node_dict[f'{_nc.get_browse_name()}'] = _sub_nc
                else:
                    _node_dict[f'{_nc.get_browse_name()}'] = str(_nc)
        return _node_dict

    def json_all_childeren(self, file_path: str) -> str:
        _time = time.perf_counter()
        _node_dict = self.get_all_childeren(self.root_node)
        with open(file_path, 'w') as f:
            json.dump(_node_dict, f, indent=4)
        print(f'generate json file time: {time.perf_counter() - _time}')
        return file_path

    def read_node(self, node):
        return self._client.get_node(node).get_value()

    def write_node(self, node, value):
        self._client.get_node(node).set_value(value)

    def read_nodes(self, nodes):
        return {node: self.read_node(node) for node in nodes}

    def write_nodes(self, nodes):
        for node, value in nodes.items():
            self.write_node(node, value)

    def close(self):
        self._client.disconnect()


if __name__ == '__main__':
    url = 'opc.tcp://192.168.10.1:4840'
    user_name = 'user1'
    user_password = 'Azxs1254'
    cert_path = "./opc_comm/my_cert.pem"
    key_path = "./opc_comm/my_private_key.pem"
    my_client = OpcuaClient(url, user_name, user_password, True, True, cert_path, key_path)

    print(my_client.root_node.get_browse_name())

    # print(client.get_all_childeren(client.root_node))

    # get all children nodes and save to json file
    my_client.json_all_childeren('./opc_comm/OPC_Nodes.json')

    # read node value
    print(my_client.read_node('ns=3;s="my_data"."my_bool1"'))

    my_bool = False
    try:
        while True:
            _time = time.perf_counter()
            my_bool = not my_bool
            my_client.write_node('ns=3;s=\"my_data\".\"my_bool1\"',
                                 ua.DataValue(ua.Variant(my_bool, ua.VariantType.Boolean)))
            print(f'write value: {str(my_bool):<8},write time: {time.perf_counter() - _time:10.8f}')
            time.sleep(1)
    except KeyboardInterrupt:
        raise
    finally:
        my_client.close()

该处使用的url网络请求的数据。

3.生成客户端通讯证书及私钥

1、通过generate_certificate.sh生成客户端证书及私钥

如果服务器设置了加密及认证，则客户端必须配置客户端证书及私钥，否则无法建立通讯
如果服务器安全策略设置为None，并且用户认证设置为访客则无需设置证书及用户名、密码直接通讯即可
运行.sh扩展名文件需要unix环境，如果在windows下运行可以选择安装git软件并用git bash软件执行文件

.sh文件代码如下：

: '
Generate your own x509v3 Certificate

Step 1: Change ssl.conf (subjectAltname, country, organizationName, ...)

ssl.conf:

[ req ]
default_bits = 2048
default_md = sha256
distinguished_name = subject
req_extensions = req_ext
x509_extensions = req_ext
string_mask = utf8only
prompt = no

[ req_ext ]
basicConstraints = CA:FALSE
nsCertType = client, server
keyUsage = nonRepudiation, digitalSignature, keyEncipherment, dataEncipherment, keyCertSign
extendedKeyUsage= serverAuth, clientAuth
nsComment = "OpenSSL Generated Certificat"
subjectKeyIdentifier=hash
authorityKeyIdentifier=keyid,issuer
subjectAltName = URI:urn:opcua:python:server,IP: 127.0.0.1

[ subject ]
countryName = DE
stateOrProvinceName = HE
localityName = HE
organizationName = AndreasHeine
commonName = PythonOpcUaServer

Step 2: openssl genrsa -out key.pem 2048
Step 3: openssl req -x509 -days 365 -new -out certificate.pem -key key.pem -config ssl.conf

this way is proved with Siemens OPC UA Client/Server!
'

openssl req -x509 -newkey rsa:2048 -keyout my_private_key.pem -out my_cert.pem -days 355 -nodes -addext "subjectAltName = URI:urn:example.org:FreeOpcUa:python-opcua"
openssl x509 -outform der -in my_cert.pem -out my_cert.der

2、配置python opcua客户端

配置用户名、密码、服务器url、认证文件地址、私钥文件地址，如下图

if __name__ == '__main__':
    url = 'opc.tcp://192.168.10.1:4840'
    user_name = 'user1'
    user_password = 'Azxs1254'
    cert_path = "./opc_comm/my_cert.pem"
    key_path = "./opc_comm/my_private_key.pem"
    my_client = OpcuaClient(url, user_name, user_password, True, True, cert_path, key_path)

3.通讯测试

总结

OPCUA通讯如果在测试环境可以不需要设置安全策略及验证，但数据安全性较低
如果设置加密及签名，则在通讯将需要准备客户端证书及私钥
在通讯时如果报application_uri错误，检查.sh文件中的uri是否与python client初始代码中的名称一致，如果不一致需要修改client.application_uri的字符串与.sh中的保持一致

生成的json文件树状图如下：

作者：PittDing