-
Notifications
You must be signed in to change notification settings - Fork 669
Description
I am trying to write a client program in Python to connect to the OPC UA server, which is started using the software KepServer. I want to write data to the nodes on the server.
I successfully connected to the server and was able to read data, but when I tried to write data, an error occurred: 'User does not have permission to perform the requested operation.' (BadUserAccessDenied).
The security policy has been set up on the server side: Basic256Sha256, SignAndEncrypt, but I have configured certificates and trusted them on both the server and client sides, but the client's certificate is self generated.
And strangely enough, I used software such as UA Expert as a client to connect to the server. The connection was successful and I was able to write, but I don't know where the problem lies. Can someone help me? thank you.
from opcua import Client, ua
import logging
import sys
import os
import time
# Set up logging
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
logger = logging.getLogger(__name__)
# Enable verbose debugging (get more authentication-related information)
logging.getLogger("opcua").setLevel(logging.DEBUG)
def test_connection_with_server_cert(server_url="opc.tcp://10.10.181.4:48032", username="DELL", password="admin"):
"""Test connection using server certificate and user credentials"""
logger.info(f"Testing connection to {server_url} using server certificate and user authentication...")
# Client certificate and private key file paths
client_cert_path = "client_cert.der"
client_key_path = "client_private_key.pem"
# Server certificate path
server_cert_path = "server_cer.cer"
# Check if certificate files exist
if not os.path.exists(client_cert_path) or not os.path.exists(client_key_path):
logger.error("Client certificate files do not exist, please run the certificate generation script first")
return False
if not os.path.exists(server_cert_path):
logger.error(f"Server certificate file ({server_cert_path}) does not exist, please confirm the path is correct")
return False
try:
# Create client instance
client = Client(server_url)
# Set application URI (to match certificate)
client.application_uri = "urn:example:client:fixed"
client.name = "Python OPC UA Client"
# Create trusted directory
client_cert_dir = os.path.dirname(client_cert_path) or "."
trusted_folder = os.path.join(client_cert_dir, "trusted")
if not os.path.exists(trusted_folder):
os.makedirs(trusted_folder)
# Copy server certificate to trusted directory
target_path = os.path.join(trusted_folder, os.path.basename(server_cert_path))
if not os.path.exists(target_path):
import shutil
shutil.copy(server_cert_path, target_path)
logger.info(f"Server certificate added to trusted list: {target_path}")
# Set security policy and certificate
logger.info("Setting security policy: Basic256Sha256,SignAndEncrypt")
client.set_security_string(f"Basic256Sha256,SignAndEncrypt,{client_cert_path},{client_key_path}")
# Create username/password token
logger.info(f"Setting user authentication information: username={username}")
identity_token = ua.UserNameIdentityToken()
identity_token.UserName = username
identity_token.Password = password.encode('utf-8')
identity_token.EncryptionAlgorithm = "" # Do not encrypt password
# Set user identity authentication method
client.user_identity_token = identity_token
# Connect to server
logger.info("Connecting...")
client.connect()
logger.info("✓ Connection successful!")
# Perform basic tests
logger.info("Performing basic tests...")
# Get root node
root = client.get_root_node()
logger.info(f"Root node: {root}")
# Get objects node
objects = client.get_objects_node()
logger.info(f"Objects node: {objects}")
# Browse child nodes
children = objects.get_children()
logger.info(f"Found {len(children)} object child nodes:")
for i, child in enumerate(children[:5]): # Only display first 5
try:
browse_name = child.get_browse_name()
node_id = child.nodeid
logger.info(f" Node {i+1}: {browse_name} (ID: {node_id})")
except Exception as e:
logger.warning(f" Node {i+1}: Unable to get information - {e}")
# Try to read server time
try:
server_time_node = client.get_node("ns=0;i=2258") # Server.ServerStatus.CurrentTime
server_time = server_time_node.get_value()
logger.info(f"Server time: {server_time}")
except Exception as e:
logger.warning(f"Unable to read server time: {e}")
# Disconnect
client.disconnect()
logger.info("Disconnected")
return True
except Exception as e:
error_msg = str(e)
logger.error(f"Connection failed: {error_msg}")
try:
client.disconnect()
except:
pass
return False
def test_anonymous_connection(server_url="opc.tcp://10.10.181.4:48032"):
"""Try anonymous connection (without username/password)"""
logger.info(f"Trying anonymous connection to {server_url}...")
try:
# Create client instance
client = Client(server_url)
# Connect to server (without setting username/password)
client.connect()
logger.info("✓ Anonymous connection successful!")
# Get root node
root = client.get_root_node()
logger.info(f"Root node: {root}")
# Disconnect
client.disconnect()
logger.info("Disconnected")
return True
except Exception as e:
logger.error(f"Anonymous connection failed: {e}")
try:
client.disconnect()
except:
pass
return False
if __name__ == "__main__":
# Get server URL from command line arguments, or use default value
server_url = sys.argv[1] if len(sys.argv) > 1 else "opc.tcp://10.10.181.4:48032"
# Username and password can be specified through command line arguments
username = sys.argv[2] if len(sys.argv) > 2 else "DELL"
password = sys.argv[3] if len(sys.argv) > 3 else "admin"
# First try username/password connection
if not test_connection_with_server_cert(server_url, username, password):
logger.info("----------------------------------------")
logger.info("Username/password authentication failed, trying anonymous connection...")
test_anonymous_connection(server_url)