Passed
Push — master ( 0b26fe...78dc7d )
by Olivier
04:21
created

InternalServer._address_space_fixes()   A

Complexity

Conditions 1

Size

Total Lines 13

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 8
CRAP Score 1

Importance

Changes 1
Bugs 1 Features 0
Metric Value
cc 1
c 1
b 1
f 0
dl 0
loc 13
ccs 8
cts 8
cp 1
crap 1
rs 9.4285
1
"""
2
Internal server implementing opcu-ua interface.
3
Can be used on server side or to implement binary/https opc-ua servers
4
"""
5
6 1
from datetime import datetime
7 1
from copy import copy, deepcopy
8 1
from datetime import timedelta
9 1
from os import path
10 1
import logging
11 1
from threading import Lock
12 1
from enum import Enum
13 1
try:
14 1
    from urllib.parse import urlparse
15 1
except ImportError:
16 1
    from urlparse import urlparse
17
18
19 1
from opcua import ua
20 1
from opcua.common import utils
21 1
from opcua.common.callback import (CallbackType, ServerItemCallback,
22
                                   CallbackDispatcher)
23 1
from opcua.common.node import Node
24 1
from opcua.server.history import HistoryManager
25 1
from opcua.server.address_space import AddressSpace
26 1
from opcua.server.address_space import AttributeService
27 1
from opcua.server.address_space import ViewService
28 1
from opcua.server.address_space import NodeManagementService
29 1
from opcua.server.address_space import MethodService
30 1
from opcua.server.subscription_service import SubscriptionService
31 1
from opcua.server.standard_address_space import standard_address_space
32 1
from opcua.server.users import User
33 1
from opcua.common import xmlimporter
34
35
36 1
class SessionState(Enum):
37 1
    Created = 0
38 1
    Activated = 1
39 1
    Closed = 2
40
41
42 1
class ServerDesc(object):
43 1
    def __init__(self, serv, cap=None):
44 1
        self.Server = serv
45 1
        self.Capabilities = cap
46
47
48 1
class InternalServer(object):
49
50 1
    def __init__(self, shelffile=None):
51 1
        self.logger = logging.getLogger(__name__)
52
53 1
        self.server_callback_dispatcher = CallbackDispatcher()
54
55 1
        self.endpoints = []
56 1
        self._channel_id_counter = 5
57 1
        self.allow_remote_admin = True
58 1
        self.disabled_clock = False  # for debugging we may want to disable clock that writes too much in log
59 1
        self._known_servers = {}  # used if we are a discovery server
60
61 1
        self.aspace = AddressSpace()
62 1
        self.attribute_service = AttributeService(self.aspace)
63 1
        self.view_service = ViewService(self.aspace)
64 1
        self.method_service = MethodService(self.aspace)
65 1
        self.node_mgt_service = NodeManagementService(self.aspace)
66
67 1
        self.load_standard_address_space(shelffile)
68
69 1
        self.loop = utils.ThreadLoop()
70 1
        self.asyncio_transports = []
71 1
        self.subscription_service = SubscriptionService(self.loop, self.aspace)
72
73 1
        self.history_manager = HistoryManager(self)
74
75
        # create a session to use on server side
76 1
        self.isession = InternalSession(self, self.aspace, self.subscription_service, "Internal", user=User.Admin)
77
78 1
        self.current_time_node = Node(self.isession, ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
79 1
        self._address_space_fixes()
80 1
        self.setup_nodes()
81
82 1
    def setup_nodes(self):
83
        """
84
        Set up some nodes as defined by spec
85
        """
86 1
        uries = ["http://opcfoundation.org/UA/"]
87 1
        ns_node = Node(self.isession, ua.NodeId(ua.ObjectIds.Server_NamespaceArray))
88 1
        ns_node.set_value(uries)
89
90 1
    def load_standard_address_space(self, shelffile=None):
91
        # check for a python shelf file, in windows the file extension is also needed for the check
92 1
        shelffile_win = shelffile
93 1
        if shelffile_win:
94
            shelffile_win += ".dat"
95
96 1
        if shelffile and (path.isfile(shelffile) or path.isfile(shelffile_win)):
97
            # import address space from shelf
98
            self.aspace.load_aspace_shelf(shelffile)
99
        else:
100
            # import address space from code generated from xml
101 1
            standard_address_space.fill_address_space(self.node_mgt_service)
102
            # import address space directly from xml, this has performance impact so disabled
103
            # importer = xmlimporter.XmlImporter(self.node_mgt_service)
104
            # importer.import_xml("/path/to/python-opcua/schemas/Opc.Ua.NodeSet2.xml", self)
105
106
            # if a cache file was supplied a shelve of the standard address space can now be built for next start up
107 1
            if shelffile:
108
                self.aspace.make_aspace_shelf(shelffile)
109
110 1
    def _address_space_fixes(self):
111
        """
112
        Looks like the xml definition of address space has some error. This is a good place to fix them
113
        """
114
115 1
        it = ua.AddReferencesItem()
116 1
        it.SourceNodeId = ua.NodeId(ua.ObjectIds.BaseObjectType)
117 1
        it.ReferenceTypeId = ua.NodeId(ua.ObjectIds.Organizes)
118 1
        it.IsForward = False
119 1
        it.TargetNodeId = ua.NodeId(ua.ObjectIds.ObjectTypesFolder)
120 1
        it.TargetNodeClass = ua.NodeClass.Object
121
122 1
        results = self.isession.add_references([it])
123
 
124 1
    def load_address_space(self, path):
125
        """
126
        Load address space from path
127
        """
128
        self.aspace.load(path)
129
130 1
    def dump_address_space(self, path):
131
        """
132
        Dump current address space to path
133
        """
134
        self.aspace.dump(path)
135
136 1
    def start(self):
137 1
        self.logger.info("starting internal server")
138 1
        for edp in self.endpoints:
139 1
            self._known_servers[edp.Server.ApplicationUri] = ServerDesc(edp.Server)
140 1
        self.loop.start()
141 1
        Node(self.isession, ua.NodeId(ua.ObjectIds.Server_ServerStatus_State)).set_value(0, ua.VariantType.Int32)
142 1
        Node(self.isession, ua.NodeId(ua.ObjectIds.Server_ServerStatus_StartTime)).set_value(datetime.utcnow())
143 1
        if not self.disabled_clock:
144 1
            self._set_current_time()
145
146 1
    def stop(self):
147 1
        self.logger.info("stopping internal server")
148 1
        self.isession.close_session()
149 1
        self.loop.stop()
150 1
        self.history_manager.stop()
151
152 1
    def _set_current_time(self):
153 1
        self.current_time_node.set_value(datetime.utcnow())
154 1
        self.loop.call_later(1, self._set_current_time)
155
156 1
    def get_new_channel_id(self):
157 1
        self._channel_id_counter += 1
158 1
        return self._channel_id_counter
159
160 1
    def add_endpoint(self, endpoint):
161 1
        self.endpoints.append(endpoint)
162
163 1
    def get_endpoints(self, params=None, sockname=None):
164 1
        self.logger.info("get endpoint")
165 1
        if sockname:
166
            # return to client the ip address it has access to
167 1
            edps = []
168 1
            for edp in self.endpoints:
169 1
                edp1 = copy(edp)
170 1
                url = urlparse(edp1.EndpointUrl)
171 1
                url = url._replace(netloc=sockname[0] + ":" + str(sockname[1]))
172 1
                edp1.EndpointUrl = url.geturl()
173 1
                edps.append(edp1)
174 1
            return edps
175 1
        return self.endpoints[:]
176
177 1
    def find_servers(self, params):
178 1
        if not params.ServerUris:
179 1
            return [desc.Server for desc in self._known_servers.values()]
180 1
        servers = []
181 1
        for serv in self._known_servers.values():
182 1
            serv_uri = serv.Server.ApplicationUri.split(":")
183 1
            for uri in params.ServerUris:
184 1
                uri = uri.split(":")
185 1
                if serv_uri[:len(uri)] == uri:
186 1
                    servers.append(serv.Server)
187 1
                    break
188 1
        return servers
189
190 1
    def register_server(self, server, conf=None):
191 1
        appdesc = ua.ApplicationDescription()
192 1
        appdesc.ApplicationUri = server.ServerUri
193 1
        appdesc.ProductUri = server.ProductUri
194
        # FIXME: select name from client locale
195 1
        appdesc.ApplicationName = server.ServerNames[0]
196 1
        appdesc.ApplicationType = server.ServerType
197 1
        appdesc.DiscoveryUrls = server.DiscoveryUrls
198
        # FIXME: select discovery uri using reachability from client network
199 1
        appdesc.GatewayServerUri = server.GatewayServerUri
200 1
        self._known_servers[server.ServerUri] = ServerDesc(appdesc, conf)
201
202 1
    def register_server2(self, params):
203
        return self.register_server(params.Server, params.DiscoveryConfiguration)
204
205 1
    def create_session(self, name, user=User.Anonymous, external=False):
206 1
        return InternalSession(self, self.aspace, self.subscription_service, name, user=user, external=external)
207
208 1
    def enable_history_data_change(self, node, period=timedelta(days=7), count=0):
209
        """
210
        Set attribute Historizing of node to True and start storing data for history
211
        """
212 1
        node.set_attribute(ua.AttributeIds.Historizing, ua.DataValue(True))
213 1
        node.set_attr_bit(ua.AttributeIds.AccessLevel, ua.AccessLevel.HistoryRead)
214 1
        node.set_attr_bit(ua.AttributeIds.UserAccessLevel, ua.AccessLevel.HistoryRead)
215 1
        self.history_manager.historize_data_change(node, period, count)
216
217 1
    def disable_history_data_change(self, node):
218
        """
219
        Set attribute Historizing of node to False and stop storing data for history
220
        """
221 1
        node.set_attribute(ua.AttributeIds.Historizing, ua.DataValue(False))
222 1
        node.unset_attr_bit(ua.AttributeIds.AccessLevel, ua.AccessLevel.HistoryRead)
223 1
        node.unset_attr_bit(ua.AttributeIds.UserAccessLevel, ua.AccessLevel.HistoryRead)
224 1
        self.history_manager.dehistorize(node)
225
226 1
    def enable_history_event(self, source, period=timedelta(days=7), count=0):
227
        """
228
        Set attribute History Read of object events to True and start storing data for history
229
        """
230 1
        event_notifier = source.get_event_notifier()
231 1
        if ua.EventNotifier.SubscribeToEvents not in event_notifier:
232
            raise ua.UaError("Node does not generate events", event_notifier)
233
234 1
        if ua.EventNotifier.HistoryRead not in event_notifier:
235 1
            event_notifier.add(ua.EventNotifier.HistoryRead)
236 1
            source.set_event_notifier(event_notifier)
237
238 1
        self.history_manager.historize_event(source, period, count)
239
240 1
    def disable_history_event(self, source):
241
        """
242
        Set attribute History Read of node to False and stop storing data for history
243
        """
244 1
        source.unset_attr_bit(ua.AttributeIds.EventNotifier, ua.EventNotifier.HistoryRead)
245 1
        self.history_manager.dehistorize(source)
246
247 1
    def subscribe_server_callback(self, event, handle):
248
        """
249
        Create a subscription from event to handle
250
        """
251
        self.server_callback_dispatcher.addListener(event, handle)
252
253 1
    def unsubscribe_server_callback(self, event, handle):
254
        """
255
        Remove a subscription from event to handle
256
        """
257
        self.server_callback_dispatcher.removeListener(event, handle)
258
259
260 1
class InternalSession(object):
261 1
    _counter = 10
262 1
    _auth_counter = 1000
263
264 1
    def __init__(self, internal_server, aspace, submgr, name, user=User.Anonymous, external=False):
265 1
        self.logger = logging.getLogger(__name__)
266 1
        self.iserver = internal_server
267 1
        self.external = external  # define if session is external, we need to copy some objects if it is internal
268 1
        self.aspace = aspace
269 1
        self.subscription_service = submgr
270 1
        self.name = name
271 1
        self.user = user
272 1
        self.nonce = None
273 1
        self.state = SessionState.Created
274 1
        self.session_id = ua.NodeId(self._counter)
275 1
        InternalSession._counter += 1
276 1
        self.authentication_token = ua.NodeId(self._auth_counter)
277 1
        InternalSession._auth_counter += 1
278 1
        self.subscriptions = []
279 1
        self.logger.info("Created internal session %s", self.name)
280 1
        self._lock = Lock()
281
282 1
    def __str__(self):
283
        return "InternalSession(name:{0}, user:{1}, id:{2}, auth_token:{3})".format(
284
            self.name, self.user, self.session_id, self.authentication_token)
285
286 1
    def get_endpoints(self, params=None, sockname=None):
287 1
        return self.iserver.get_endpoints(params, sockname)
288
289 1
    def create_session(self, params, sockname=None):
290 1
        self.logger.info("Create session request")
291
292 1
        result = ua.CreateSessionResult()
293 1
        result.SessionId = self.session_id
294 1
        result.AuthenticationToken = self.authentication_token
295 1
        result.RevisedSessionTimeout = params.RequestedSessionTimeout
296 1
        result.MaxRequestMessageSize = 65536
297 1
        self.nonce = utils.create_nonce(32)
298 1
        result.ServerNonce = self.nonce
299 1
        result.ServerEndpoints = self.get_endpoints(sockname=sockname)
300
301 1
        return result
302
303 1
    def close_session(self, delete_subs=True):
304 1
        self.logger.info("close session %s with subscriptions %s", self, self.subscriptions)
305 1
        self.state = SessionState.Closed
306 1
        self.delete_subscriptions(self.subscriptions[:])
307
308 1
    def activate_session(self, params):
309 1
        self.logger.info("activate session")
310 1
        result = ua.ActivateSessionResult()
311 1
        if self.state != SessionState.Created:
312
            raise utils.ServiceError(ua.StatusCodes.BadSessionIdInvalid)
313 1
        self.nonce = utils.create_nonce(32)
314 1
        result.ServerNonce = self.nonce
315 1
        for _ in params.ClientSoftwareCertificates:
316
            result.Results.append(ua.StatusCode())
317 1
        self.state = SessionState.Activated
318 1
        id_token = params.UserIdentityToken
319 1
        if isinstance(id_token, ua.UserNameIdentityToken):
320 1
            if self.iserver.allow_remote_admin and id_token.UserName in ("admin", "Admin"):
321 1
                self.user = User.Admin
322 1
        self.logger.info("Activated internal session %s for user %s", self.name, self.user)
323 1
        return result
324
325 1
    def read(self, params):
326 1
        results = self.iserver.attribute_service.read(params)
327 1
        if self.external:
328 1
            return results
329 1
        return [deepcopy(dv) for dv in results]
330
331 1
    def history_read(self, params):
332 1
        return self.iserver.history_manager.read_history(params)
333
334 1
    def write(self, params):
335 1
        if not self.external:
336
            # If session is internal we need to store a copy og object, not a reference,
337
            # otherwise users may change it and we will not generate expected events
338 1
            params.NodesToWrite = [deepcopy(ntw) for ntw in params.NodesToWrite]
339 1
        return self.iserver.attribute_service.write(params, self.user)
340
341 1
    def browse(self, params):
342 1
        return self.iserver.view_service.browse(params)
343
344 1
    def translate_browsepaths_to_nodeids(self, params):
345 1
        return self.iserver.view_service.translate_browsepaths_to_nodeids(params)
346
347 1
    def add_nodes(self, params):
348 1
        return self.iserver.node_mgt_service.add_nodes(params, self.user)
349
350 1
    def delete_nodes(self, params):
351 1
        return self.iserver.node_mgt_service.delete_nodes(params, self.user)
352
353 1
    def add_references(self, params):
354 1
        return self.iserver.node_mgt_service.add_references(params, self.user)
355
356 1
    def delete_references(self, params):
357
        return self.iserver.node_mgt_service.delete_references(params, self.user)
358
359 1
    def add_method_callback(self, methodid, callback):
360 1
        return self.aspace.add_method_callback(methodid, callback)
361
362 1
    def call(self, params):
363 1
        return self.iserver.method_service.call(params)
364
365 1
    def create_subscription(self, params, callback):
366 1
        result = self.subscription_service.create_subscription(params, callback)
367 1
        with self._lock:
368 1
            self.subscriptions.append(result.SubscriptionId)
369 1
        return result
370
371 1
    def create_monitored_items(self, params):
372 1
        subscription_result = self.subscription_service.create_monitored_items(params)
373 1
        self.iserver.server_callback_dispatcher.dispatch(
374
            CallbackType.ItemSubscriptionCreated, ServerItemCallback(params, subscription_result))
375 1
        return subscription_result
376
377 1
    def modify_monitored_items(self, params):
378
        subscription_result = self.subscription_service.modify_monitored_items(params)
379
        self.iserver.server_callback_dispatcher.dispatch(
380
            CallbackType.ItemSubscriptionModified, ServerItemCallback(params, subscription_result))
381
        return subscription_result
382
383 1
    def republish(self, params):
384
        return self.subscription_service.republish(params)
385
386 1
    def delete_subscriptions(self, ids):
387 1
        for i in ids:
388 1
            with self._lock:
389 1
                if i in self.subscriptions:
390 1
                    self.subscriptions.remove(i)
391 1
        return self.subscription_service.delete_subscriptions(ids)
392
393 1
    def delete_monitored_items(self, params):
394 1
        subscription_result = self.subscription_service.delete_monitored_items(params)
395 1
        self.iserver.server_callback_dispatcher.dispatch(
396
            CallbackType.ItemSubscriptionDeleted, ServerItemCallback(params, subscription_result))
397 1
        return subscription_result
398
399 1
    def publish(self, acks=None):
400 1
        if acks is None:
401 1
            acks = []
402
        return self.subscription_service.publish(acks)
403