Always use SimpleRoleRuleset

It hasn't been in use for SecurityPolicy None

Author: cziebuhr

asyncua/crypto/security_policies.py

@@ -508,10 +508,6 @@ def __init__(self, peer_cert, host_cert, client_pk, mode,
         self.Mode = mode
         self.peer_certificate = uacrypto.der_from_x509(peer_cert)
         self.host_certificate = uacrypto.der_from_x509(host_cert)
-        if permission_ruleset is None:
-            from asyncua.crypto.permission_rules import SimpleRoleRuleset
-            permission_ruleset = SimpleRoleRuleset()
-
         self.permissions = permission_ruleset
 
     def make_local_symmetric_key(self, secret, seed):
@@ -591,11 +587,6 @@ def __init__(self, peer_cert, host_cert, client_pk, mode, permission_ruleset=Non
         self.Mode = mode
         self.peer_certificate = uacrypto.der_from_x509(peer_cert)
         self.host_certificate = uacrypto.der_from_x509(host_cert)
-        if permission_ruleset is None:
-            from asyncua.crypto.permission_rules import SimpleRoleRuleset
-
-            permission_ruleset = SimpleRoleRuleset()
-
         self.permissions = permission_ruleset
 
     def make_local_symmetric_key(self, secret, seed):
@@ -684,10 +675,6 @@ def __init__(self, peer_cert, host_cert, client_pk, mode,
         self.Mode = mode
         self.peer_certificate = uacrypto.der_from_x509(peer_cert)
         self.host_certificate = uacrypto.der_from_x509(host_cert)
-        if permission_ruleset is None:
-            from asyncua.crypto.permission_rules import SimpleRoleRuleset
-            permission_ruleset = SimpleRoleRuleset()
-
         self.permissions = permission_ruleset
 
     def make_local_symmetric_key(self, secret, seed):
@@ -772,10 +759,6 @@ def __init__(self, peer_cert, host_cert, client_pk, mode,
         self.Mode = mode
         self.peer_certificate = uacrypto.der_from_x509(peer_cert)
         self.host_certificate = uacrypto.der_from_x509(host_cert)
-        if permission_ruleset is None:
-            from asyncua.crypto.permission_rules import SimpleRoleRuleset
-            permission_ruleset = SimpleRoleRuleset()
-
         self.permissions = permission_ruleset
 
     def make_local_symmetric_key(self, secret, seed):
@@ -859,10 +842,6 @@ def __init__(self, peer_cert, host_cert, client_pk, mode,
         self.Mode = mode
         self.peer_certificate = uacrypto.der_from_x509(peer_cert)
         self.host_certificate = uacrypto.der_from_x509(host_cert)
-        if permission_ruleset is None:
-            from asyncua.crypto.permission_rules import SimpleRoleRuleset
-            permission_ruleset = SimpleRoleRuleset()
-
         self.permissions = permission_ruleset
 
     def make_local_symmetric_key(self, secret, seed):

asyncua/server/server.py

@@ -31,6 +31,7 @@
 from ..common.connection import TransportLimits
 
 from ..crypto import security_policies, uacrypto, validator
+from ..crypto.permission_rules import SimpleRoleRuleset
 
 _logger = logging.getLogger(__name__)
 
@@ -108,7 +109,7 @@ def __init__(self, iserver: InternalServer = None, user_manager=None):
             ua.SecurityPolicyType.Aes256Sha256RsaPss_SignAndEncrypt
         ]
         # allow all certificates by default
-        self._permission_ruleset = None
+        self._permission_ruleset = SimpleRoleRuleset()
         self._policyIDs = ["Anonymous", "Basic256Sha256", "Username", "Aes128Sha256RsaOaep", "Aes256Sha256RsaPss"]
         self.certificate: Optional[x509.Certificate] = None
         # Use acceptable limits
@@ -343,7 +344,8 @@ def set_security_policy(self, security_policy, permission_ruleset=None):
 
         """
         self._security_policy = security_policy
-        self._permission_ruleset = permission_ruleset
+        if permission_ruleset is not None:
+            self._permission_ruleset = permission_ruleset
 
     def set_security_IDs(self, policy_ids):
         """

examples/server-with-encryption.py

@@ -7,7 +7,6 @@
 sys.path.insert(0, "..")
 from asyncua import Server
 from asyncua import ua
-from asyncua.crypto.permission_rules import SimpleRoleRuleset
 from asyncua.server.user_managers import CertificateUserManager
 from asyncua.crypto.cert_gen import setup_self_signed_certificate
 from asyncua.crypto.validator import CertificateValidator, CertificateValidatorOptions
@@ -38,8 +37,7 @@ async def main():
 
     await server.set_application_uri(server_app_uri)
     server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
-    server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt],
-                               permission_ruleset=SimpleRoleRuleset())
+    server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt])
 
     # Below is only required if the server should generate its own certificate,
     # It will renew also when the valid datetime range is out of range (on startup, no on runtime)

tests/test_permissions.py

@@ -4,7 +4,6 @@
 from asyncua import Client
 from asyncua import Server
 from asyncua import ua
-from asyncua.crypto.permission_rules import SimpleRoleRuleset
 from asyncua.server.users import UserRole
 from asyncua.server.user_managers import CertificateUserManager
 
@@ -58,8 +57,7 @@ async def srv_crypto_one_cert(request):
     srv = Server(user_manager=cert_user_manager)
 
     srv.set_endpoint(uri_crypto_cert)
-    srv.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt],
-                            permission_ruleset=SimpleRoleRuleset())
+    srv.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt])
     await srv.init()
     await srv.load_certificate(cert)
     await srv.load_private_key(key)