Add support for server username/password auth

Author: igfarm

opcua/server/internal_server.py

@@ -2,9 +2,9 @@
 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 logging
 from threading import Lock
@@ -31,6 +31,13 @@
 from opcua.server.users import User
 #from opcua.common import xmlimporter
 
+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
@@ -57,6 +64,9 @@ def __init__(self, shelffile=None):
         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)
@@ -71,6 +81,8 @@ def __init__(self, shelffile=None):
 
         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)
 
@@ -269,7 +281,41 @@ def set_attribute_value(self, nodeid, datavalue, attr=ua.AttributeIds.Value):
         """
         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(object):
@@ -332,8 +378,8 @@ def activate_session(self, params):
         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
+            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
 
@@ -412,3 +458,12 @@ def publish(self, acks=None):
         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

@@ -90,8 +90,6 @@ def __init__(self, shelffile=None, iserver=None):
         self.bserver = None
         self._discovery_clients = {}
         self._discovery_period = 60
-        self.certificate = None
-        self.private_key = None
         self._policies = []
         self.nodes = Shortcuts(self.iserver.isession)
 
@@ -132,10 +130,10 @@ def load_certificate(self, path):
         """
         load server certificate from file, either pem or der
         """
-        self.certificate = uacrypto.load_certificate(path)
+        self.iserver.certificate = uacrypto.load_certificate(path)
 
     def load_private_key(self, path):
-        self.private_key = uacrypto.load_private_key(path)
+        self.iserver.private_key = uacrypto.load_private_key(path)
 
     def disable_clock(self, val=True):
         """
@@ -265,7 +263,7 @@ def _setup_server_nodes(self):
             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
 
@@ -277,16 +275,16 @@ def _setup_server_nodes(self):
                                     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_):
@@ -319,8 +317,8 @@ def _set_endpoints(self, policy=ua.SecurityPolicy, mode=ua.MessageSecurityMode.N
         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