reject not supported identity tokens

prevents unsupported identity tokens to create a session.
For example reject anonymous connection if not set via set_security_IDs, currently this was possible and was a security issue.

Author: schroeder-

opcua/server/internal_server.py

@@ -73,6 +73,7 @@ def __init__(self, shelffile=None, user_manager=None, session_cls=None):
         self.current_time_node = Node(self.isession, ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
         self._address_space_fixes()
         self.setup_nodes()
+        self.supported_tokens = []
 
     @property
     def thread_loop(self):
@@ -346,6 +347,10 @@ def activate_session(self, params):
             result.Results.append(ua.StatusCode())
         self.state = SessionState.Activated
         id_token = params.UserIdentityToken
+        # Check if security policy is supported
+        if not isinstance(id_token, self.iserver.supported_tokens):
+            self.logger.error('Rejected active session UserIdentityToken not supported')
+            raise utils.ServiceError(ua.StatusCodes.BadIdentityTokenRejected)
         if isinstance(id_token, ua.UserNameIdentityToken):
             if self.user_manager.check_user_token(self, id_token) == False:
                 raise utils.ServiceError(ua.StatusCodes.BadUserAccessDenied)

opcua/server/server.py

@@ -263,23 +263,27 @@ def _setup_server_nodes(self):
 
     def _set_endpoints(self, policy=ua.SecurityPolicy, mode=ua.MessageSecurityMode.None_):
         idtokens = []
+        supported_token_classes = []
         if "Anonymous" in self._policyIDs:
             idtoken = ua.UserTokenPolicy()
             idtoken.PolicyId = 'anonymous'
             idtoken.TokenType = ua.UserTokenType.Anonymous
             idtokens.append(idtoken)
+            supported_token_classes.append(ua.AnonymousIdentityToken)
 
         if "Basic256Sha256" in self._policyIDs:
             idtoken = ua.UserTokenPolicy()
             idtoken.PolicyId = 'certificate_basic256sha256'
             idtoken.TokenType = ua.UserTokenType.Certificate
             idtokens.append(idtoken)
+            supported_token_classes.append(ua.X509IdentityToken)
 
         if "Username" in self._policyIDs:
             idtoken = ua.UserTokenPolicy()
             idtoken.PolicyId = 'username'
             idtoken.TokenType = ua.UserTokenType.UserName
             idtokens.append(idtoken)
+            supported_token_classes.append(ua.UserNameIdentityToken)
 
         appdesc = ua.ApplicationDescription()
         appdesc.ApplicationName = ua.LocalizedText(self.name)
@@ -299,6 +303,7 @@ def _set_endpoints(self, policy=ua.SecurityPolicy, mode=ua.MessageSecurityMode.N
         edp.TransportProfileUri = 'http://opcfoundation.org/UA-Profile/Transport/uatcp-uasc-uabinary'
         edp.SecurityLevel = 0
         self.iserver.add_endpoint(edp)
+        self.iserver.supported_tokens = tuple(supported_token_classes)
 
     def set_server_name(self, name):
         self.name = name

tests/tests_crypto_connect.py

@@ -17,6 +17,7 @@
 port_num1 = 48515
 port_num2 = 48512
 port_num3 = 48510
+port_num4 = 48533
 
 @unittest.skipIf(disable_crypto_tests, "crypto not available")
 class TestCryptoConnect(unittest.TestCase):
@@ -51,6 +52,14 @@ def setUpClass(cls):
         cls.srv_crypto2.load_private_key("examples/private-key-3072-example.pem")
         cls.srv_crypto2.start()
 
+        cls.srv_crypto_no_anoymous = Server()
+        cls.uri_crypto_no_anoymous = 'opc.tcp://127.0.0.1:{0:d}'.format(port_num4)
+        cls.srv_crypto_no_anoymous.set_endpoint(cls.uri_crypto_no_anoymous)
+        cls.srv_crypto_no_anoymous.load_certificate("examples/certificate-3072-example.der")
+        cls.srv_crypto_no_anoymous.load_private_key("examples/private-key-3072-example.pem")
+        cls.srv_crypto_no_anoymous.set_security_IDs(["Username", "Basic256Sha256"])
+        cls.srv_crypto_no_anoymous.start()
+
     @classmethod
     def tearDownClass(cls):
         # stop the server 
@@ -131,3 +140,11 @@ def test_basic256sha56_encrypt_fail(self):
                              None,
                              ua.MessageSecurityMode.None_
                              )
+
+    def test_anonymous_rejection(self):
+        # FIXME: how to make it fail???
+        clt = Client(self.uri_crypto_no_anoymous)
+        with self.assertRaises(ua.UaError) as exc_info:
+            clt.connect()
+            clt.disconnect()
+        assert ua.StatusCodes.BadIdentityTokenRejected == exc_info.type.code