Add support for server username/password auth

opcua/server/internal_server.py

@@ -3,13 +3,15 @@ Internal server implementing opcu-ua interface.
 Can be used on server side or to implement binary/https opc-ua servers
 """
 
+from datetime import datetime, timedelta
+from copy import copy
+from struct import unpack_from, unpack
 import os
 import asyncio
 import logging
 from enum import Enum
 from copy import copy, deepcopy
 from urllib.parse import urlparse
-from datetime import datetime, timedelta
 from typing import Coroutine
 
 from opcua import ua
@@ -22,6 +24,13 @@ from .users import User
 
 __all__ = ["InternalServer"]
 
+use_crypto = True
+try:
+    from opcua.crypto import uacrypto
+except ImportError:
+    logging.getLogger(__name__).warning("cryptography is not installed, use of crypto disabled")
+    use_crypto = False
+
 
 class SessionState(Enum):
     Created = 0
@@ -45,6 +54,8 @@ class InternalServer:
         self.allow_remote_admin = True
         self.disabled_clock = False  # for debugging we may want to disable clock that writes too much in log
         self._known_servers = {}  # used if we are a discovery server
+        self.certificate = None
+        self.private_key = None
         self.aspace = AddressSpace()
         self.attribute_service = AttributeService(self.aspace)
         self.view_service = ViewService(self.aspace)
@@ -54,6 +65,9 @@ class InternalServer:
         self.asyncio_transports = []
         self.subscription_service: SubscriptionService = SubscriptionService(self.loop, self.aspace)
         self.history_manager = HistoryManager(self)
+
+        self.user_manager = default_user_manager # defined at the end of this file
+
         # create a session to use on server side
         self.isession = InternalSession(self, self.aspace, self.subscription_service, "Internal", user=User.Admin)
         self.current_time_node = Node(self.isession, ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
@@ -249,7 +263,41 @@ class InternalServer:
         """
         self.aspace.set_attribute_value(nodeid, ua.AttributeIds.Value, datavalue)
 
+    def set_user_manager(self, user_manager):
+        """
+        set up a function which that will check for authorize users. Input function takes username
+        and password as paramters and returns True of user is allowed access, False otherwise.
+        """
+        self.user_manager = user_manager
 
+    def check_user_token(self, isession, token):
+        """
+        unpack the username and password for the benefit of the user defined user manager
+        """
+        userName = token.UserName
+        passwd = token.Password
+
+        # decrypt password is we can
+        if str(token.EncryptionAlgorithm) != "None":
+            if use_crypto == False:
+                return False;
+            try:
+                if token.EncryptionAlgorithm == "http://www.w3.org/2001/04/xmlenc#rsa-1_5":
+                    raw_pw = uacrypto.decrypt_rsa15(self.private_key, passwd)
+                elif token.EncryptionAlgorithm == "http://www.w3.org/2001/04/xmlenc#rsa-oaep":
+                    raw_pw = uacrypto.decrypt_rsa_oaep(self.private_key, passwd)
+                else:
+                    self.logger.warning("Unknown password encoding '{0}'".format(token.EncryptionAlgorithm))
+                    return False
+                length = unpack_from('<I', raw_pw)[0] - len(isession.nonce)
+                passwd = raw_pw[4:4 + length]
+                passwd = passwd.decode('utf-8')
+            except Exception as exp:
+                self.logger.warning("Unable to decrypt password")
+                return False
+
+        # call user_manager
+        return self.user_manager(self, isession, userName, passwd)
 
 
 class InternalSession:
@@ -311,9 +359,9 @@ class InternalSession:
         self.state = SessionState.Activated
         id_token = params.UserIdentityToken
         if isinstance(id_token, ua.UserNameIdentityToken):
-            if self.iserver.allow_remote_admin and id_token.UserName in ('admin', 'Admin'):
-                self.user = User.Admin
-        self.logger.info('Activated internal session %s for user %s', self.name, self.user)
+            if self.iserver.check_user_token(self, id_token) == False:
+                raise utils.ServiceError(ua.StatusCodes.BadUserAccessDenied)
+        self.logger.info("Activated internal session %s for user %s", self.name, self.user)
         return result
 
     async def read(self, params):
@@ -388,3 +436,12 @@ class InternalSession:
         if acks is None:
             acks = []
         return self.subscription_service.publish(acks)
+
+
+def default_user_manager(iserver, isession, userName, password):
+    """
+    Default user_manager, does nothing much but check for admin
+    """
+    if iserver.allow_remote_admin and userName in ("admin", "Admin"):
+        isession.user = User.Admin
+    return True

opcua/server/server.py

@@ -75,8 +75,6 @@ class Server:
         self.bserver: BinaryServer = None
         self._discovery_clients = {}
         self._discovery_period = 60
-        self.certificate = None
-        self.private_key = None
         self._policies = []
         self.nodes = Shortcuts(self.iserver.isession)
         # enable all endpoints by default
@@ -117,10 +115,10 @@ class Server:
         """
         load server certificate from file, either pem or der
         """
-        self.certificate = await uacrypto.load_certificate(path)
+        self.iserver.certificate = await uacrypto.load_certificate(path)
 
-    async def load_private_key(self, path: str):
-        self.private_key = await uacrypto.load_private_key(path)
+    async def load_private_key(self, path):
+        self.iserver.private_key = await uacrypto.load_private_key(path)
 
     def disable_clock(self, val: bool=True):
         """
@@ -244,7 +242,7 @@ class Server:
             self._policies = [ua.SecurityPolicyFactory()]
 
         if self._security_policy != [ua.SecurityPolicyType.NoSecurity]:
-            if not (self.certificate and self.private_key):
+            if not (self.iserver.certificate and self.iserver.private_key):
                 self.logger.warning("Endpoints other than open requested but private key and certificate are not set.")
                 return
 
@@ -256,16 +254,16 @@ class Server:
                                     ua.MessageSecurityMode.SignAndEncrypt)
                 self._policies.append(ua.SecurityPolicyFactory(security_policies.SecurityPolicyBasic256Sha256,
                                                                ua.MessageSecurityMode.SignAndEncrypt,
-                                                               self.certificate,
-                                                               self.private_key)
+                                                               self.iserver.certificate,
+                                                               self.iserver.private_key)
                                      )
             if ua.SecurityPolicyType.Basic256Sha256_Sign in self._security_policy:
                 self._set_endpoints(security_policies.SecurityPolicyBasic256Sha256,
                                     ua.MessageSecurityMode.Sign)
                 self._policies.append(ua.SecurityPolicyFactory(security_policies.SecurityPolicyBasic256Sha256,
                                                                ua.MessageSecurityMode.Sign,
-                                                               self.certificate,
-                                                               self.private_key)
+                                                               self.iserver.certificate,
+                                                               self.iserver.private_key)
                                      )
 
     def _set_endpoints(self, policy=ua.SecurityPolicy, mode=ua.MessageSecurityMode.None_):
@@ -298,8 +296,8 @@ class Server:
         edp = ua.EndpointDescription()
         edp.EndpointUrl = self.endpoint.geturl()
         edp.Server = appdesc
-        if self.certificate:
-            edp.ServerCertificate = uacrypto.der_from_x509(self.certificate)
+        if self.iserver.certificate:
+            edp.ServerCertificate = uacrypto.der_from_x509(self.iserver.certificate)
         edp.SecurityMode = mode
         edp.SecurityPolicyUri = policy.URI
         edp.UserIdentityTokens = idtokens