Add server-side session watchdog (#1655)

* Add server-side session watchdog

* Add test

* Remove whitespace

* Move watchdog to uaprocessor

* Restore import order

asyncua/server/internal_session.py

@@ -47,6 +47,7 @@ def __init__(self, internal_server: "InternalServer", aspace: AddressSpace, subm
         self.auth_token = ua.NodeId(self._auth_counter)
         InternalSession._auth_counter += 1
         self.logger.info('Created internal session %s', self.name)
+        self.session_timeout = None
 
     def __str__(self):
         return f'InternalSession(name:{self.name},' \
@@ -65,6 +66,7 @@ async def create_session(self, params: ua.CreateSessionParameters, sockname: Opt
         result.AuthenticationToken = self.auth_token
         result.RevisedSessionTimeout = params.RequestedSessionTimeout
         result.MaxRequestMessageSize = 65536
+        self.session_timeout = result.RevisedSessionTimeout / 1000
 
         if self.iserver.certificate_validator and params.ClientCertificate:
             await self.iserver.certificate_validator(x509.load_der_x509_certificate(params.ClientCertificate), params.ClientDescription)
@@ -75,7 +77,6 @@ async def create_session(self, params: ua.CreateSessionParameters, sockname: Opt
         ep_params = ua.GetEndpointsParameters()
         ep_params.EndpointUrl = params.EndpointUrl
         result.ServerEndpoints = await self.get_endpoints(params=ep_params, sockname=sockname)
-
         return result
 
     async def close_session(self, delete_subs=True):
@@ -85,13 +86,14 @@ async def close_session(self, delete_subs=True):
         if InternalSession._current_connections < 0:
             InternalSession._current_connections = 0
         self.state = SessionState.Closed
-        await self.delete_subscriptions(
-            [
-                id
-                for id, sub in self.subscription_service.subscriptions.items()
-                if sub.session_id == self.session_id
-            ]
-        )
+        if delete_subs:
+            await self.delete_subscriptions(
+                [
+                    id
+                    for id, sub in self.subscription_service.subscriptions.items()
+                    if sub.session_id == self.session_id
+                ]
+            )
 
     def activate_session(self, params, peer_certificate):
         self.logger.info('activate session')

asyncua/server/uaprocessor.py

@@ -1,8 +1,10 @@
+import asyncio
 import copy
 import time
 import logging
 from typing import Deque, Optional, Dict
 from collections import deque
+from datetime import datetime, timedelta, timezone
 
 from asyncua import ua
 from ..ua.ua_binary import nodeid_from_binary, struct_from_binary, struct_to_binary, uatcp_to_binary
@@ -31,6 +33,7 @@ def __init__(self, internal_server: InternalServer, transport, limits: Transport
         self.name = transport.get_extra_info('peername')
         self.sockname = transport.get_extra_info('sockname')
         self.session: Optional[InternalSession] = None
+        self.session_last_activity: Optional[datetime] = None
         self._transport = transport
         # deque for Publish Requests
         self._publish_requests: Deque[PublishRequestData] = deque()
@@ -39,6 +42,9 @@ def __init__(self, internal_server: InternalServer, transport, limits: Transport
         self._publish_results_subs: Dict[ua.IntegerId, bool] = {}
         self._limits = copy.deepcopy(limits)  # Copy limits because they get overriden
         self._connection = SecureConnection(ua.SecurityPolicy(), self._limits)
+        self._closing: bool = False
+        self._session_watchdog_task: Optional[asyncio.Task] = None
+        self._watchdog_interval: float = 1.0
 
     def set_policies(self, policies):
         self._connection.set_policy_factories(policies)
@@ -178,13 +184,17 @@ async def _process_message(self, typeid, requesthdr, seqhdr, body):
                 if self._connection.security_policy.permissions.check_validity(user, typeid, body) is False:
                     raise ua.uaerrors.BadUserAccessDenied
 
+        self.session_last_activity = datetime.now(timezone.utc)
+
         if typeid == ua.NodeId(ua.ObjectIds.CreateSessionRequest_Encoding_DefaultBinary):
             _logger.info("Create session request (%s)", user)
             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 = await self.session.create_session(params, sockname=self.sockname)
+            self._closing = False
+            self._session_watchdog_task = asyncio.create_task(self._session_watchdog_loop())
             response = ua.CreateSessionResponse()
             response.Parameters = sessiondata
             response.Parameters.ServerCertificate = self._connection.security_policy.host_certificate
@@ -202,6 +212,7 @@ async def _process_message(self, typeid, requesthdr, seqhdr, body):
             _logger.info("Close session request (%s)", user)
             if self.session:
                 deletesubs = ua.ua_binary.Primitives.Boolean.unpack(body)
+                self._closing = True
                 await self.session.close_session(deletesubs)
             else:
                 _logger.info("Request to close non-existing session (%s)", user)
@@ -507,5 +518,18 @@ async def close(self):
         everything we should
         """
         _logger.info("Cleanup client connection: %s", self.name)
+        self._closing = True
         if self.session:
             await self.session.close_session(True)
+
+    async def _session_watchdog_loop(self):
+        """
+        Checks if the session is alive
+        """
+        timeout = min(self.session.session_timeout / 2, self._watchdog_interval)
+        timeout_timedelta = timedelta(seconds=self.session.session_timeout)
+        while not self._closing:
+            await asyncio.sleep(timeout)
+            if (datetime.now(timezone.utc) - timeout_timedelta) > self.session_last_activity:
+                _logger.warning('Session timed out after %ss of inactivity', timeout_timedelta.total_seconds())
+                await self.close()

tests/test_connections.py

@@ -4,7 +4,7 @@
 import struct
 
 from asyncua import Client, Server, ua
-from asyncua.ua.uaerrors import BadMaxConnectionsReached
+from asyncua.ua.uaerrors import BadMaxConnectionsReached, BadSessionNotActivated
 from .conftest import port_num, find_free_port
 
 pytestmark = pytest.mark.asyncio
@@ -77,3 +77,20 @@ def status_change_notification(self, status: ua.StatusChangeNotification):
         # check if a exception is thrown when a normal function is called
         with pytest.raises(ConnectionError):
             await cl.get_namespace_array()
+
+
+async def test_session_watchdog():
+    port = find_free_port()
+    srv = Server()
+    await srv.init()
+    srv.set_endpoint(f"opc.tcp://127.0.0.1:{port}")
+    await srv.start()
+    client = Client(f"opc.tcp://127.0.0.1:{port}", timeout=0.5, watchdog_intervall=1)
+    client.session_timeout = 1000  # 1 second
+    await client.connect()
+    client._closing = True  # Kill the keepalive tasks
+    await asyncio.sleep(2)  # Wait for the watchdog to terminate the session due to inactivity
+    with pytest.raises(BadSessionNotActivated):
+        server_time_node = client.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
+        await server_time_node.read_value()
+    await client.disconnect()