UaProcessor._process_message() - Code Metrics - Inspection of "Merge pull request #613 from brubbel/modify_subscr..." - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 1a9d89...1da59b )

by Olivier

created 2018-06-05 19:32 UTC

UaProcessor._process_message() F

↳ Parent: UaProcessor

Complexity

Conditions

Size

Total Lines

358

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	186
CRAP Score	50.1784

Importance

Changes	3
Bugs	1	Features	1

Metric	Value
cc	35
c	3
b	1
f	1
dl	0
loc	358
ccs	186
cts	242
cp	0.7685
crap	50.1784
rs	2

How to fix Long Method Complexity


import logging
from threading import RLock, Lock
import time

from opcua import ua
from opcua.ua.ua_binary import nodeid_from_binary, struct_from_binary
from opcua.ua.ua_binary import struct_to_binary, uatcp_to_binary
from opcua.common import utils
from opcua.common.connection import SecureConnection


class PublishRequestData(object):

    def __init__(self):
        self.requesthdr = None
        self.algohdr = None
        self.seqhdr = None
        self.timestamp = time.time()


class UaProcessor(object):

    def __init__(self, internal_server, socket):
        self.logger = logging.getLogger(__name__)
        self.iserver = internal_server
        self.name = socket.get_extra_info('peername')
        self.sockname = socket.get_extra_info('sockname')
        self.session = None
        self.socket = socket
        self._socketlock = Lock()
        self._datalock = RLock()
        self._publishdata_queue = []
        self._publish_result_queue = []  # used when we need to wait for PublishRequest
        self._connection = SecureConnection(ua.SecurityPolicy())

    def set_policies(self, policies):
        self._connection.set_policy_factories(policies)

    def send_response(self, requesthandle, algohdr, seqhdr, response, msgtype=ua.MessageType.SecureMessage):
        with self._socketlock:
            response.ResponseHeader.RequestHandle = requesthandle
            data = self._connection.message_to_binary(
                struct_to_binary(response), message_type=msgtype, request_id=seqhdr.RequestId, algohdr=algohdr)

            self.socket.write(data)

    def open_secure_channel(self, algohdr, seqhdr, body):
        request = struct_from_binary(ua.OpenSecureChannelRequest, body)

        self._connection.select_policy(
            algohdr.SecurityPolicyURI, algohdr.SenderCertificate, request.Parameters.SecurityMode)

        channel = self._connection.open(request.Parameters, self.iserver)
        # send response
        response = ua.OpenSecureChannelResponse()
        response.Parameters = channel
        self.send_response(request.RequestHeader.RequestHandle, None, seqhdr, response, ua.MessageType.SecureOpen)

    def forward_publish_response(self, result):
        self.logger.info("forward publish response %s", result)
        with self._datalock:
            while True:
                if len(self._publishdata_queue) == 0:
                    self._publish_result_queue.append(result)
                    self.logger.info("Server wants to send publish answer but no publish request is available,"
                                     "enqueing notification, length of result queue is %s",
                                     len(self._publish_result_queue))
                    return
                requestdata = self._publishdata_queue.pop(0)
                if requestdata.requesthdr.TimeoutHint == 0 or requestdata.requesthdr.TimeoutHint != 0 and time.time() - requestdata.timestamp < requestdata.requesthdr.TimeoutHint / 1000:
                    break

        response = ua.PublishResponse()
        response.Parameters = result

        self.send_response(requestdata.requesthdr.RequestHandle, requestdata.algohdr, requestdata.seqhdr, response)

    def process(self, header, body):
        msg = self._connection.receive_from_header_and_body(header, body)
        if isinstance(msg, ua.Message):
            if header.MessageType == ua.MessageType.SecureOpen:
                self.open_secure_channel(msg.SecurityHeader(), msg.SequenceHeader(), msg.body())

            elif header.MessageType == ua.MessageType.SecureClose:
                self._connection.close()
                return False

            elif header.MessageType == ua.MessageType.SecureMessage:
                return self.process_message(msg.SecurityHeader(), msg.SequenceHeader(), msg.body())
        elif isinstance(msg, ua.Hello):
            ack = ua.Acknowledge()
            ack.ReceiveBufferSize = msg.ReceiveBufferSize
            ack.SendBufferSize = msg.SendBufferSize
            data = uatcp_to_binary(ua.MessageType.Acknowledge, ack)
            self.socket.write(data)
        elif isinstance(msg, ua.ErrorMessage):
            self.logger.warning("Received an error message type")
        elif msg is None:
            pass  # msg is a ChunkType.Intermediate of an ua.MessageType.SecureMessage
        else:
            self.logger.warning("Unsupported message type: %s", header.MessageType)
            raise utils.ServiceError(ua.StatusCodes.BadTcpMessageTypeInvalid)
        return True

    def process_message(self, algohdr, seqhdr, body):
        typeid = nodeid_from_binary(body)
        requesthdr = struct_from_binary(ua.RequestHeader, body)
        try:
            return self._process_message(typeid, requesthdr, algohdr, seqhdr, body)
        except utils.ServiceError as e:
            status = ua.StatusCode(e.code)
            response = ua.ServiceFault()
            response.ResponseHeader.ServiceResult = status
            self.logger.info("sending service fault response: %s (%s)", status.doc, status.name)
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
            return True

    def _process_message(self, typeid, requesthdr, algohdr, seqhdr, body):
        if typeid == ua.NodeId(ua.ObjectIds.CreateSessionRequest_Encoding_DefaultBinary):
            self.logger.info("Create session request")
            params = struct_from_binary(ua.CreateSessionParameters, body)

            # create the session on server
            self.session = self.iserver.create_session(self.name, external=True)
            # get a session creation result to send back
            sessiondata = self.session.create_session(params, sockname=self.sockname)

            response = ua.CreateSessionResponse()
            response.Parameters = sessiondata
            response.Parameters.ServerCertificate = self._connection.security_policy.client_certificate
            if self._connection.security_policy.server_certificate is None:
                data = params.ClientNonce
            else:
                data = self._connection.security_policy.server_certificate + params.ClientNonce
            response.Parameters.ServerSignature.Signature = \
                self._connection.security_policy.asymmetric_cryptography.signature(data)

            response.Parameters.ServerSignature.Algorithm = "http://www.w3.org/2000/09/xmldsig#rsa-sha1"

            self.logger.info("sending create session response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.CloseSessionRequest_Encoding_DefaultBinary):
            self.logger.info("Close session request")
            deletesubs = ua.ua_binary.Primitives.Boolean.unpack(body)

            self.session.close_session(deletesubs)

            response = ua.CloseSessionResponse()
            self.logger.info("sending close session response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.ActivateSessionRequest_Encoding_DefaultBinary):
            self.logger.info("Activate session request")
            params = struct_from_binary(ua.ActivateSessionParameters, body)

            if not self.session:
                self.logger.info("request to activate non-existing session")
                raise utils.ServiceError(ua.StatusCodes.BadSessionIdInvalid)

            if self._connection.security_policy.client_certificate is None:
                data = self.session.nonce
            else:
                data = self._connection.security_policy.client_certificate + self.session.nonce
            self._connection.security_policy.asymmetric_cryptography.verify(data, params.ClientSignature.Signature)

            result = self.session.activate_session(params)

            response = ua.ActivateSessionResponse()
            response.Parameters = result

            self.logger.info("sending read response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.ReadRequest_Encoding_DefaultBinary):
            self.logger.info("Read request")
            params = struct_from_binary(ua.ReadParameters, body)

            results = self.session.read(params)

            response = ua.ReadResponse()
            response.Results = results

            self.logger.info("sending read response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.WriteRequest_Encoding_DefaultBinary):
            self.logger.info("Write request")
            params = struct_from_binary(ua.WriteParameters, body)

            results = self.session.write(params)

            response = ua.WriteResponse()
            response.Results = results

            self.logger.info("sending write response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.BrowseRequest_Encoding_DefaultBinary):
            self.logger.info("Browse request")
            params = struct_from_binary(ua.BrowseParameters, body)

            results = self.session.browse(params)

            response = ua.BrowseResponse()
            response.Results = results

            self.logger.info("sending browse response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.GetEndpointsRequest_Encoding_DefaultBinary):
            self.logger.info("get endpoints request")
            params = struct_from_binary(ua.GetEndpointsParameters, body)

            endpoints = self.iserver.get_endpoints(params, sockname=self.sockname)

            response = ua.GetEndpointsResponse()
            response.Endpoints = endpoints

            self.logger.info("sending get endpoints response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.FindServersRequest_Encoding_DefaultBinary):
            self.logger.info("find servers request")
            params = struct_from_binary(ua.FindServersParameters, body)

            servers = self.iserver.find_servers(params)

            response = ua.FindServersResponse()
            response.Servers = servers

            self.logger.info("sending find servers response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.RegisterServerRequest_Encoding_DefaultBinary):
            self.logger.info("register server request")
            serv = struct_from_binary(ua.RegisteredServer, body)

            self.iserver.register_server(serv)

            response = ua.RegisterServerResponse()

            self.logger.info("sending register server response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.RegisterServer2Request_Encoding_DefaultBinary):
            self.logger.info("register server 2 request")
            params = struct_from_binary(ua.RegisterServer2Parameters, body)

            results = self.iserver.register_server2(params)

            response = ua.RegisterServer2Response()
            response.ConfigurationResults = results

            self.logger.info("sending register server 2 response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.TranslateBrowsePathsToNodeIdsRequest_Encoding_DefaultBinary):
            self.logger.info("translate browsepaths to nodeids request")
            params = struct_from_binary(ua.TranslateBrowsePathsToNodeIdsParameters, body)

            paths = self.session.translate_browsepaths_to_nodeids(params.BrowsePaths)

            response = ua.TranslateBrowsePathsToNodeIdsResponse()
            response.Results = paths

            self.logger.info("sending translate browsepaths to nodeids response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.AddNodesRequest_Encoding_DefaultBinary):
            self.logger.info("add nodes request")
            params = struct_from_binary(ua.AddNodesParameters, body)

            results = self.session.add_nodes(params.NodesToAdd)

            response = ua.AddNodesResponse()
            response.Results = results

            self.logger.info("sending add node response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.DeleteNodesRequest_Encoding_DefaultBinary):
            self.logger.info("delete nodes request")
            params = struct_from_binary(ua.DeleteNodesParameters, body)

            results = self.session.delete_nodes(params)

            response = ua.DeleteNodesResponse()
            response.Results = results

            self.logger.info("sending delete node response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.AddReferencesRequest_Encoding_DefaultBinary):
            self.logger.info("add references request")
            params = struct_from_binary(ua.AddReferencesParameters, body)

            results = self.session.add_references(params.ReferencesToAdd)

            response = ua.AddReferencesResponse()
            response.Results = results

            self.logger.info("sending add references response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.DeleteReferencesRequest_Encoding_DefaultBinary):
            self.logger.info("delete references request")
            params = struct_from_binary(ua.DeleteReferencesParameters, body)

            results = self.session.delete_references(params.ReferencesToDelete)

            response = ua.DeleteReferencesResponse()
            response.Parameters.Results = results

            self.logger.info("sending delete references response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)


        elif typeid == ua.NodeId(ua.ObjectIds.CreateSubscriptionRequest_Encoding_DefaultBinary):
            self.logger.info("create subscription request")
            params = struct_from_binary(ua.CreateSubscriptionParameters, body)

            result = self.session.create_subscription(params, self.forward_publish_response)

            response = ua.CreateSubscriptionResponse()
            response.Parameters = result

            self.logger.info("sending create subscription response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.ModifySubscriptionRequest_Encoding_DefaultBinary):
            self.logger.info("modify subscription request")
            params = struct_from_binary(ua.ModifySubscriptionParameters, body)

            result = self.session.modify_subscription(params, self.forward_publish_response)

            response = ua.ModifySubscriptionResponse()
            response.Parameters = result

            self.logger.info("sending modify subscription response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.DeleteSubscriptionsRequest_Encoding_DefaultBinary):
            self.logger.info("delete subscriptions request")
            params = struct_from_binary(ua.DeleteSubscriptionsParameters, body)

            results = self.session.delete_subscriptions(params.SubscriptionIds)

            response = ua.DeleteSubscriptionsResponse()
            response.Results = results

            self.logger.info("sending delte subscription response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.CreateMonitoredItemsRequest_Encoding_DefaultBinary):
            self.logger.info("create monitored items request")
            params = struct_from_binary(ua.CreateMonitoredItemsParameters, body)
            results = self.session.create_monitored_items(params)

            response = ua.CreateMonitoredItemsResponse()
            response.Results = results

            self.logger.info("sending create monitored items response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.ModifyMonitoredItemsRequest_Encoding_DefaultBinary):
            self.logger.info("modify monitored items request")
            params = struct_from_binary(ua.ModifyMonitoredItemsParameters, body)
            results = self.session.modify_monitored_items(params)

            response = ua.ModifyMonitoredItemsResponse()
            response.Results = results

            self.logger.info("sending modify monitored items response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.DeleteMonitoredItemsRequest_Encoding_DefaultBinary):
            self.logger.info("delete monitored items request")
            params = struct_from_binary(ua.DeleteMonitoredItemsParameters, body)

            results = self.session.delete_monitored_items(params)

            response = ua.DeleteMonitoredItemsResponse()
            response.Results = results

            self.logger.info("sending delete monitored items response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.HistoryReadRequest_Encoding_DefaultBinary):
            self.logger.info("history read request")
            params = struct_from_binary(ua.HistoryReadParameters, body)

            results = self.session.history_read(params)

            response = ua.HistoryReadResponse()
            response.Results = results

            self.logger.info("sending history read response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.RegisterNodesRequest_Encoding_DefaultBinary):
            self.logger.info("register nodes request")
            params = struct_from_binary(ua.RegisterNodesParameters, body)
            self.logger.info("Node registration not implemented")

            response = ua.RegisterNodesResponse()
            response.Parameters.RegisteredNodeIds = params.NodesToRegister

            self.logger.info("sending register nodes response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.UnregisterNodesRequest_Encoding_DefaultBinary):
            self.logger.info("unregister nodes request")
            params = struct_from_binary(ua.UnregisterNodesParameters, body)

            response = ua.UnregisterNodesResponse()

            self.logger.info("sending unregister nodes response")
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.PublishRequest_Encoding_DefaultBinary):
            self.logger.info("publish request")

            if not self.session:
                return False

            params = struct_from_binary(ua.PublishParameters, body)

            data = PublishRequestData()
            data.requesthdr = requesthdr
            data.seqhdr = seqhdr
            data.algohdr = algohdr
            with self._datalock:
                self._publishdata_queue.append(data)  # will be used to send publish answers from server
                if self._publish_result_queue:
                    result = self._publish_result_queue.pop(0)
                    self.forward_publish_response(result)
            self.session.publish(params.SubscriptionAcknowledgements)
            self.logger.info("publish forward to server")

        elif typeid == ua.NodeId(ua.ObjectIds.RepublishRequest_Encoding_DefaultBinary):
            self.logger.info("re-publish request")

            params = struct_from_binary(ua.RepublishParameters, body)
            msg = self.session.republish(params)

            response = ua.RepublishResponse()
            response.NotificationMessage = msg

            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        elif typeid == ua.NodeId(ua.ObjectIds.CloseSecureChannelRequest_Encoding_DefaultBinary):
            self.logger.info("close secure channel request")
            self._connection.close()
            response = ua.CloseSecureChannelResponse()
            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
            return False

        elif typeid == ua.NodeId(ua.ObjectIds.CallRequest_Encoding_DefaultBinary):
            self.logger.info("call request")

            params = struct_from_binary(ua.CallParameters, body)

            results = self.session.call(params.MethodsToCall)

            response = ua.CallResponse()
            response.Results = results

            self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)

        else:
            self.logger.warning("Unknown message received %s", typeid)
            raise utils.ServiceError(ua.StatusCodes.BadNotImplemented)

        return True

    def close(self):
        """
        to be called when client has disconnected to ensure we really close
        everything we should
        """
        self.logger.info("Cleanup client connection: %s", self.name)
        if self.session:
            self.session.close_session(True)


1
2	1	import logging
3	1	from threading import RLock, Lock
4	1	import time
5
6	1	from opcua import ua
7	1	from opcua.ua.ua_binary import nodeid_from_binary, struct_from_binary
8	1	from opcua.ua.ua_binary import struct_to_binary, uatcp_to_binary
9	1	from opcua.common import utils
10	1	from opcua.common.connection import SecureConnection
11
12
13	1	class PublishRequestData(object):
14
15	1	def __init__(self):
16	1	self.requesthdr = None
17	1	self.algohdr = None
18	1	self.seqhdr = None
19	1	self.timestamp = time.time()
20
21
22	1	class UaProcessor(object):
23
24	1	def __init__(self, internal_server, socket):
25	1	self.logger = logging.getLogger(__name__)
26	1	self.iserver = internal_server
27	1	self.name = socket.get_extra_info('peername')
28	1	self.sockname = socket.get_extra_info('sockname')
29	1	self.session = None
30	1	self.socket = socket
31	1	self._socketlock = Lock()
32	1	self._datalock = RLock()
33	1	self._publishdata_queue = []
34	1	self._publish_result_queue = [] # used when we need to wait for PublishRequest
35	1	self._connection = SecureConnection(ua.SecurityPolicy())
36
37	1	def set_policies(self, policies):
38	1	self._connection.set_policy_factories(policies)
39
40	1	def send_response(self, requesthandle, algohdr, seqhdr, response, msgtype=ua.MessageType.SecureMessage):
41	1	with self._socketlock:
42	1	response.ResponseHeader.RequestHandle = requesthandle
43	1	data = self._connection.message_to_binary(
44		struct_to_binary(response), message_type=msgtype, request_id=seqhdr.RequestId, algohdr=algohdr)
45
46	1	self.socket.write(data)
47
48	1	def open_secure_channel(self, algohdr, seqhdr, body):
49	1	request = struct_from_binary(ua.OpenSecureChannelRequest, body)
50
51	1	self._connection.select_policy(
52		algohdr.SecurityPolicyURI, algohdr.SenderCertificate, request.Parameters.SecurityMode)
53
54	1	channel = self._connection.open(request.Parameters, self.iserver)
55		# send response
56	1	response = ua.OpenSecureChannelResponse()
57	1	response.Parameters = channel
58	1	self.send_response(request.RequestHeader.RequestHandle, None, seqhdr, response, ua.MessageType.SecureOpen)
59
60	1	def forward_publish_response(self, result):
61	1	self.logger.info("forward publish response %s", result)
62	1	with self._datalock:
63	1	while True:
64	1	if len(self._publishdata_queue) == 0:
65		self._publish_result_queue.append(result)
66		self.logger.info("Server wants to send publish answer but no publish request is available,"
67		"enqueing notification, length of result queue is %s",
68		len(self._publish_result_queue))
69		return
70	1	requestdata = self._publishdata_queue.pop(0)
71	1	if requestdata.requesthdr.TimeoutHint == 0 or requestdata.requesthdr.TimeoutHint != 0 and time.time() - requestdata.timestamp < requestdata.requesthdr.TimeoutHint / 1000:
72	1	break
73
74	1	response = ua.PublishResponse()
75	1	response.Parameters = result
76
77	1	self.send_response(requestdata.requesthdr.RequestHandle, requestdata.algohdr, requestdata.seqhdr, response)
78
79	1	def process(self, header, body):
80	1	msg = self._connection.receive_from_header_and_body(header, body)
81	1	if isinstance(msg, ua.Message):
82	1	if header.MessageType == ua.MessageType.SecureOpen:
83	1	self.open_secure_channel(msg.SecurityHeader(), msg.SequenceHeader(), msg.body())
84
85	1	elif header.MessageType == ua.MessageType.SecureClose:
86	1	self._connection.close()
87	1	return False
88
89	1	elif header.MessageType == ua.MessageType.SecureMessage:
90	1	return self.process_message(msg.SecurityHeader(), msg.SequenceHeader(), msg.body())
91	1	elif isinstance(msg, ua.Hello):
92	1	ack = ua.Acknowledge()
93	1	ack.ReceiveBufferSize = msg.ReceiveBufferSize
94	1	ack.SendBufferSize = msg.SendBufferSize
95	1	data = uatcp_to_binary(ua.MessageType.Acknowledge, ack)
96	1	self.socket.write(data)
97		elif isinstance(msg, ua.ErrorMessage):
98		self.logger.warning("Received an error message type")
99		elif msg is None:
100		pass # msg is a ChunkType.Intermediate of an ua.MessageType.SecureMessage
101		else:
102		self.logger.warning("Unsupported message type: %s", header.MessageType)
103		raise utils.ServiceError(ua.StatusCodes.BadTcpMessageTypeInvalid)
104	1	return True
105
106	1	def process_message(self, algohdr, seqhdr, body):
107	1	typeid = nodeid_from_binary(body)
108	1	requesthdr = struct_from_binary(ua.RequestHeader, body)
109	1	try:
110	1	return self._process_message(typeid, requesthdr, algohdr, seqhdr, body)
111	1	except utils.ServiceError as e:
112	1	status = ua.StatusCode(e.code)
113	1	response = ua.ServiceFault()
114	1	response.ResponseHeader.ServiceResult = status
115	1	self.logger.info("sending service fault response: %s (%s)", status.doc, status.name)
116	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
117	1	return True
118
119	1	def _process_message(self, typeid, requesthdr, algohdr, seqhdr, body):
120	1	if typeid == ua.NodeId(ua.ObjectIds.CreateSessionRequest_Encoding_DefaultBinary):
121	1	self.logger.info("Create session request")
122	1	params = struct_from_binary(ua.CreateSessionParameters, body)
123
124		# create the session on server
125	1	self.session = self.iserver.create_session(self.name, external=True)
126		# get a session creation result to send back
127	1	sessiondata = self.session.create_session(params, sockname=self.sockname)
128
129	1	response = ua.CreateSessionResponse()
130	1	response.Parameters = sessiondata
131	1	response.Parameters.ServerCertificate = self._connection.security_policy.client_certificate
132	1	if self._connection.security_policy.server_certificate is None:
133	1	data = params.ClientNonce
134		else:
135	1	data = self._connection.security_policy.server_certificate + params.ClientNonce
136	1	response.Parameters.ServerSignature.Signature = \
137		self._connection.security_policy.asymmetric_cryptography.signature(data)
138
139	1	response.Parameters.ServerSignature.Algorithm = "http://www.w3.org/2000/09/xmldsig#rsa-sha1"
140
141	1	self.logger.info("sending create session response")
142	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
143
144	1	elif typeid == ua.NodeId(ua.ObjectIds.CloseSessionRequest_Encoding_DefaultBinary):
145	1	self.logger.info("Close session request")
146	1	deletesubs = ua.ua_binary.Primitives.Boolean.unpack(body)
147
148	1	self.session.close_session(deletesubs)
149
150	1	response = ua.CloseSessionResponse()
151	1	self.logger.info("sending close session response")
152	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
153
154	1	elif typeid == ua.NodeId(ua.ObjectIds.ActivateSessionRequest_Encoding_DefaultBinary):
155	1	self.logger.info("Activate session request")
156	1	params = struct_from_binary(ua.ActivateSessionParameters, body)
157
158	1	if not self.session:
159		self.logger.info("request to activate non-existing session")
160		raise utils.ServiceError(ua.StatusCodes.BadSessionIdInvalid)
161
162	1	if self._connection.security_policy.client_certificate is None:
163	1	data = self.session.nonce
164		else:
165	1	data = self._connection.security_policy.client_certificate + self.session.nonce
166	1	self._connection.security_policy.asymmetric_cryptography.verify(data, params.ClientSignature.Signature)
167
168	1	result = self.session.activate_session(params)
169
170	1	response = ua.ActivateSessionResponse()
171	1	response.Parameters = result
172
173	1	self.logger.info("sending read response")
174	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
175
176	1	elif typeid == ua.NodeId(ua.ObjectIds.ReadRequest_Encoding_DefaultBinary):
177	1	self.logger.info("Read request")
178	1	params = struct_from_binary(ua.ReadParameters, body)
179
180	1	results = self.session.read(params)
181
182	1	response = ua.ReadResponse()
183	1	response.Results = results
184
185	1	self.logger.info("sending read response")
186	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
187
188	1	elif typeid == ua.NodeId(ua.ObjectIds.WriteRequest_Encoding_DefaultBinary):
189	1	self.logger.info("Write request")
190	1	params = struct_from_binary(ua.WriteParameters, body)
191
192	1	results = self.session.write(params)
193
194	1	response = ua.WriteResponse()
195	1	response.Results = results
196
197	1	self.logger.info("sending write response")
198	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
199
200	1	elif typeid == ua.NodeId(ua.ObjectIds.BrowseRequest_Encoding_DefaultBinary):
201	1	self.logger.info("Browse request")
202	1	params = struct_from_binary(ua.BrowseParameters, body)
203
204	1	results = self.session.browse(params)
205
206	1	response = ua.BrowseResponse()
207	1	response.Results = results
208
209	1	self.logger.info("sending browse response")
210	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
211
212	1	elif typeid == ua.NodeId(ua.ObjectIds.GetEndpointsRequest_Encoding_DefaultBinary):
213	1	self.logger.info("get endpoints request")
214	1	params = struct_from_binary(ua.GetEndpointsParameters, body)
215
216	1	endpoints = self.iserver.get_endpoints(params, sockname=self.sockname)
217
218	1	response = ua.GetEndpointsResponse()
219	1	response.Endpoints = endpoints
220
221	1	self.logger.info("sending get endpoints response")
222	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
223
224	1	elif typeid == ua.NodeId(ua.ObjectIds.FindServersRequest_Encoding_DefaultBinary):
225	1	self.logger.info("find servers request")
226	1	params = struct_from_binary(ua.FindServersParameters, body)
227
228	1	servers = self.iserver.find_servers(params)
229
230	1	response = ua.FindServersResponse()
231	1	response.Servers = servers
232
233	1	self.logger.info("sending find servers response")
234	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
235
236	1	elif typeid == ua.NodeId(ua.ObjectIds.RegisterServerRequest_Encoding_DefaultBinary):
237	1	self.logger.info("register server request")
238	1	serv = struct_from_binary(ua.RegisteredServer, body)
239
240	1	self.iserver.register_server(serv)
241
242	1	response = ua.RegisterServerResponse()
243
244	1	self.logger.info("sending register server response")
245	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
246
247	1	elif typeid == ua.NodeId(ua.ObjectIds.RegisterServer2Request_Encoding_DefaultBinary):
248		self.logger.info("register server 2 request")
249		params = struct_from_binary(ua.RegisterServer2Parameters, body)
250
251		results = self.iserver.register_server2(params)
252
253		response = ua.RegisterServer2Response()
254		response.ConfigurationResults = results
255
256		self.logger.info("sending register server 2 response")
257		self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
258
259	1	elif typeid == ua.NodeId(ua.ObjectIds.TranslateBrowsePathsToNodeIdsRequest_Encoding_DefaultBinary):
260	1	self.logger.info("translate browsepaths to nodeids request")
261	1	params = struct_from_binary(ua.TranslateBrowsePathsToNodeIdsParameters, body)
262
263	1	paths = self.session.translate_browsepaths_to_nodeids(params.BrowsePaths)
264
265	1	response = ua.TranslateBrowsePathsToNodeIdsResponse()
266	1	response.Results = paths
267
268	1	self.logger.info("sending translate browsepaths to nodeids response")
269	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
270
271	1	elif typeid == ua.NodeId(ua.ObjectIds.AddNodesRequest_Encoding_DefaultBinary):
272	1	self.logger.info("add nodes request")
273	1	params = struct_from_binary(ua.AddNodesParameters, body)
274
275	1	results = self.session.add_nodes(params.NodesToAdd)
276
277	1	response = ua.AddNodesResponse()
278	1	response.Results = results
279
280	1	self.logger.info("sending add node response")
281	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
282
283	1	elif typeid == ua.NodeId(ua.ObjectIds.DeleteNodesRequest_Encoding_DefaultBinary):
284	1	self.logger.info("delete nodes request")
285	1	params = struct_from_binary(ua.DeleteNodesParameters, body)
286
287	1	results = self.session.delete_nodes(params)
288
289	1	response = ua.DeleteNodesResponse()
290	1	response.Results = results
291
292	1	self.logger.info("sending delete node response")
293	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
294
295	1	elif typeid == ua.NodeId(ua.ObjectIds.AddReferencesRequest_Encoding_DefaultBinary):
296	1	self.logger.info("add references request")
297	1	params = struct_from_binary(ua.AddReferencesParameters, body)
298
299	1	results = self.session.add_references(params.ReferencesToAdd)
300
301	1	response = ua.AddReferencesResponse()
302	1	response.Results = results
303
304	1	self.logger.info("sending add references response")
305	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
306
307	1	elif typeid == ua.NodeId(ua.ObjectIds.DeleteReferencesRequest_Encoding_DefaultBinary):
308	1	self.logger.info("delete references request")
309	1	params = struct_from_binary(ua.DeleteReferencesParameters, body)
310
311	1	results = self.session.delete_references(params.ReferencesToDelete)
312
313	1	response = ua.DeleteReferencesResponse()
314	1	response.Parameters.Results = results
315
316	1	self.logger.info("sending delete references response")
317	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
318
319
320	1	elif typeid == ua.NodeId(ua.ObjectIds.CreateSubscriptionRequest_Encoding_DefaultBinary):
321	1	self.logger.info("create subscription request")
322	1	params = struct_from_binary(ua.CreateSubscriptionParameters, body)
323
324	1	result = self.session.create_subscription(params, self.forward_publish_response)
325
326	1	response = ua.CreateSubscriptionResponse()
327	1	response.Parameters = result
328
329	1	self.logger.info("sending create subscription response")
330	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
331
332	1	elif typeid == ua.NodeId(ua.ObjectIds.ModifySubscriptionRequest_Encoding_DefaultBinary):
333		self.logger.info("modify subscription request")
334		params = struct_from_binary(ua.ModifySubscriptionParameters, body)
335
336		result = self.session.modify_subscription(params, self.forward_publish_response)
337
338		response = ua.ModifySubscriptionResponse()
339		response.Parameters = result
340
341		self.logger.info("sending modify subscription response")
342		self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
343
344	1	elif typeid == ua.NodeId(ua.ObjectIds.DeleteSubscriptionsRequest_Encoding_DefaultBinary):
345	1	self.logger.info("delete subscriptions request")
346	1	params = struct_from_binary(ua.DeleteSubscriptionsParameters, body)
347
348	1	results = self.session.delete_subscriptions(params.SubscriptionIds)
349
350	1	response = ua.DeleteSubscriptionsResponse()
351	1	response.Results = results
352
353	1	self.logger.info("sending delte subscription response")
354	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
355
356	1	elif typeid == ua.NodeId(ua.ObjectIds.CreateMonitoredItemsRequest_Encoding_DefaultBinary):
357	1	self.logger.info("create monitored items request")
358	1	params = struct_from_binary(ua.CreateMonitoredItemsParameters, body)
359	1	results = self.session.create_monitored_items(params)
360
361	1	response = ua.CreateMonitoredItemsResponse()
362	1	response.Results = results
363
364	1	self.logger.info("sending create monitored items response")
365	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
366
367	1	elif typeid == ua.NodeId(ua.ObjectIds.ModifyMonitoredItemsRequest_Encoding_DefaultBinary):
368		self.logger.info("modify monitored items request")
369		params = struct_from_binary(ua.ModifyMonitoredItemsParameters, body)
370		results = self.session.modify_monitored_items(params)
371
372		response = ua.ModifyMonitoredItemsResponse()
373		response.Results = results
374
375		self.logger.info("sending modify monitored items response")
376		self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
377
378	1	elif typeid == ua.NodeId(ua.ObjectIds.DeleteMonitoredItemsRequest_Encoding_DefaultBinary):
379	1	self.logger.info("delete monitored items request")
380	1	params = struct_from_binary(ua.DeleteMonitoredItemsParameters, body)
381
382	1	results = self.session.delete_monitored_items(params)
383
384	1	response = ua.DeleteMonitoredItemsResponse()
385	1	response.Results = results
386
387	1	self.logger.info("sending delete monitored items response")
388	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
389
390	1	elif typeid == ua.NodeId(ua.ObjectIds.HistoryReadRequest_Encoding_DefaultBinary):
391		self.logger.info("history read request")
392		params = struct_from_binary(ua.HistoryReadParameters, body)
393
394		results = self.session.history_read(params)
395
396		response = ua.HistoryReadResponse()
397		response.Results = results
398
399		self.logger.info("sending history read response")
400		self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
401
402	1	elif typeid == ua.NodeId(ua.ObjectIds.RegisterNodesRequest_Encoding_DefaultBinary):
403		self.logger.info("register nodes request")
404		params = struct_from_binary(ua.RegisterNodesParameters, body)
405		self.logger.info("Node registration not implemented")
406
407		response = ua.RegisterNodesResponse()
408		response.Parameters.RegisteredNodeIds = params.NodesToRegister
409
410		self.logger.info("sending register nodes response")
411		self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
412
413	1	elif typeid == ua.NodeId(ua.ObjectIds.UnregisterNodesRequest_Encoding_DefaultBinary):
414		self.logger.info("unregister nodes request")
415		params = struct_from_binary(ua.UnregisterNodesParameters, body)
416
417		response = ua.UnregisterNodesResponse()
418
419		self.logger.info("sending unregister nodes response")
420		self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
421
422	1	elif typeid == ua.NodeId(ua.ObjectIds.PublishRequest_Encoding_DefaultBinary):
423	1	self.logger.info("publish request")
424
425	1	if not self.session:
426		return False
427
428	1	params = struct_from_binary(ua.PublishParameters, body)
429
430	1	data = PublishRequestData()
431	1	data.requesthdr = requesthdr
432	1	data.seqhdr = seqhdr
433	1	data.algohdr = algohdr
434	1	with self._datalock:
435	1	self._publishdata_queue.append(data) # will be used to send publish answers from server
436	1	if self._publish_result_queue:
437		result = self._publish_result_queue.pop(0)
438		self.forward_publish_response(result)
439	1	self.session.publish(params.SubscriptionAcknowledgements)
440	1	self.logger.info("publish forward to server")
441
442	1	elif typeid == ua.NodeId(ua.ObjectIds.RepublishRequest_Encoding_DefaultBinary):
443		self.logger.info("re-publish request")
444
445		params = struct_from_binary(ua.RepublishParameters, body)
446		msg = self.session.republish(params)
447
448		response = ua.RepublishResponse()
449		response.NotificationMessage = msg
450
451		self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
452
453	1	elif typeid == ua.NodeId(ua.ObjectIds.CloseSecureChannelRequest_Encoding_DefaultBinary):
454		self.logger.info("close secure channel request")
455		self._connection.close()
456		response = ua.CloseSecureChannelResponse()
457		self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
458		return False
459
460	1	elif typeid == ua.NodeId(ua.ObjectIds.CallRequest_Encoding_DefaultBinary):
461	1	self.logger.info("call request")
462
463	1	params = struct_from_binary(ua.CallParameters, body)
464
465	1	results = self.session.call(params.MethodsToCall)
466
467	1	response = ua.CallResponse()
468	1	response.Results = results
469
470	1	self.send_response(requesthdr.RequestHandle, algohdr, seqhdr, response)
471
472		else:
473	1	self.logger.warning("Unknown message received %s", typeid)
474	1	raise utils.ServiceError(ua.StatusCodes.BadNotImplemented)
475
476	1	return True
477
478	1	def close(self):
479		"""
480		to be called when client has disconnected to ensure we really close
481		everything we should
482		"""
483	1	self.logger.info("Cleanup client connection: %s", self.name)
484	1	if self.session:
485		self.session.close_session(True)
486

FreeOpcUa / python-opcua

Push — master ( 1a9d89...1da59b )

UaProcessor._process_message() F

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like