Passed
Pull Request — master (#209)
by
unknown
03:30 queued 01:11
created

asyncua.server.uaprocessor.UaProcessor.process()   C

Complexity

Conditions 9

Size

Total Lines 28
Code Lines 27

Duplication

Lines 0
Ratio 0 %

Importance

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