Fix SimpleRoleRuleset

asyncua/crypto/permission_rules.py

@@ -1,23 +1,21 @@
 from asyncua import ua
 from asyncua.server.users import UserRole
 
-WRITE_TYPES = [
- ua.ObjectIds.WriteRequest_Encoding_DefaultBinary,
+ADMIN_TYPES = [
  ua.ObjectIds.RegisterServerRequest_Encoding_DefaultBinary,
  ua.ObjectIds.RegisterServer2Request_Encoding_DefaultBinary,
  ua.ObjectIds.AddNodesRequest_Encoding_DefaultBinary,
  ua.ObjectIds.DeleteNodesRequest_Encoding_DefaultBinary,
  ua.ObjectIds.AddReferencesRequest_Encoding_DefaultBinary,
  ua.ObjectIds.DeleteReferencesRequest_Encoding_DefaultBinary,
- ua.ObjectIds.RegisterNodesRequest_Encoding_DefaultBinary,
- ua.ObjectIds.UnregisterNodesRequest_Encoding_DefaultBinary
 ]
 
-READ_TYPES = [
+USER_TYPES = [
  ua.ObjectIds.CreateSessionRequest_Encoding_DefaultBinary,
  ua.ObjectIds.CloseSessionRequest_Encoding_DefaultBinary,
  ua.ObjectIds.ActivateSessionRequest_Encoding_DefaultBinary,
  ua.ObjectIds.ReadRequest_Encoding_DefaultBinary,
+ ua.ObjectIds.WriteRequest_Encoding_DefaultBinary,
  ua.ObjectIds.BrowseRequest_Encoding_DefaultBinary,
  ua.ObjectIds.GetEndpointsRequest_Encoding_DefaultBinary,
  ua.ObjectIds.FindServersRequest_Encoding_DefaultBinary,
@@ -33,7 +31,9 @@
  ua.ObjectIds.CloseSecureChannelRequest_Encoding_DefaultBinary,
  ua.ObjectIds.CallRequest_Encoding_DefaultBinary,
  ua.ObjectIds.SetMonitoringModeRequest_Encoding_DefaultBinary,
- ua.ObjectIds.SetPublishingModeRequest_Encoding_DefaultBinary
+ ua.ObjectIds.SetPublishingModeRequest_Encoding_DefaultBinary,
+ ua.ObjectIds.RegisterNodesRequest_Encoding_DefaultBinary,
+ ua.ObjectIds.UnregisterNodesRequest_Encoding_DefaultBinary,
 ]
 
 
@@ -49,15 +49,15 @@ def check_validity(self, user, action_type, body):
 class SimpleRoleRuleset(PermissionRuleset):
  """
  Standard simple role-based ruleset.
- Admins alone can write, admins and users can read, and anonymous users can't do anything.
+ Admins alone can change address space, admins and users can read/write, and anonymous users can't do anything.
  """
 
  def __init__(self):
- write_ids = list(map(ua.NodeId, WRITE_TYPES))
- read_ids = list(map(ua.NodeId, READ_TYPES))
+ admin_ids = list(map(ua.NodeId, ADMIN_TYPES))
+ user_ids = list(map(ua.NodeId, USER_TYPES))
  self._permission_dict = {
- UserRole.Admin: set().union(write_ids, read_ids),
- UserRole.User: set().union(read_ids),
+ UserRole.Admin: set().union(admin_ids, user_ids),
+ UserRole.User: set().union(user_ids),
  UserRole.Anonymous: set()
  }
 

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)
@@ -90,9 +88,9 @@ async def test_permissions_admin(srv_crypto_one_cert):
  assert await clt.get_objects_node().get_children()
  objects = clt.nodes.objects
  child = await objects.get_child(['0:MyObject', '0:MyVariable'])
- await child.read_value()
  await child.set_value(42.0)
-
+ assert await child.read_value() == 42.0
+ await child.add_property(0, "MyProperty1", 3)
 
 async def test_permissions_user(srv_crypto_one_cert):
  clt = Client(uri_crypto_cert)
@@ -108,9 +106,10 @@ async def test_permissions_user(srv_crypto_one_cert):
  assert await clt.get_objects_node().get_children()
  objects = clt.nodes.objects
  child = await objects.get_child(['0:MyObject', '0:MyVariable'])
- await child.read_value()
+ await child.set_value(44.0)
+ assert await child.read_value() == 44.0
  with pytest.raises(ua.uaerrors.BadUserAccessDenied):
- await child.set_value(42)
+ await child.add_property(0, "MyProperty2", 3)
 
 
 async def test_permissions_anonymous(srv_crypto_one_cert):
@@ -123,6 +122,7 @@ async def test_permissions_anonymous(srv_crypto_one_cert):
  server_certificate=srv_crypto_params[0][1],
  mode=ua.MessageSecurityMode.SignAndEncrypt
  )
- await clt.connect()
- await clt.get_endpoints()
- await clt.disconnect()
+ async with clt:
+ await clt.get_endpoints()
+ with pytest.raises(ua.uaerrors.BadUserAccessDenied):
+ await clt.nodes.objects.get_children()