Adds a certificate validation hook to the client and server

With the hook you can register callback method with the following type `CertificateValidatorMethod`:
```
Callable[[x509.Certificate, ApplicationDescription], Awaitable[None]]
```

The method should raise an `ServiceError` exception when the certificate is invalid.
The makes it possible to provide your own logic for validation.

An default implementation of such certificate validation hook is also provided by `ucrypto.validator.CertificateValidator`.

`CertificateValidator` supports checking:
- Uri
- Timerange
- Key Usage
- Extended Key Usage
- Trust
- Revoked

For the last two checks a instance of a `TrustStore` needs to be provided to `CertificateValidator`.

Which checks must be performed can be configured with `CertificateValidatorOptions`. Those are provide to the constructor of `CertificateValidator` and can be changed later with `CertificateValidator.set_validate_options`.

Examples of client and server with encryption are changed to demonstrate the use of the validator.

Basic testing of the validator is present.

asyncua/client/client.py

@@ -18,9 +18,11 @@
 from ..common.shortcuts import Shortcuts
 from ..common.structures import load_type_definitions, load_enums
 from ..common.structures104 import load_data_type_definitions
-from ..common.utils import create_nonce
+from ..common.utils import create_nonce, ServiceError
 from ..common.ua_utils import value_to_datavalue, copy_dataclass_attr
 from ..crypto import uacrypto, security_policies
+from ..crypto.validator import CertificateValidatorMethod
+from ..crypto.uacrypto import x509
 
 _logger = logging.getLogger(__name__)
 
@@ -86,6 +88,8 @@ def __init__(self, url: str, timeout: float = 4, watchdog_intervall: float = 1.0
         self._locale = ["en"]
         self._watchdog_intervall = watchdog_intervall
         self._closing: bool = False
+        self.certificate_validator: Optional[CertificateValidatorMethod]= None
+        """hook to validate a certificate, raises a ServiceError when not valid"""
 
     async def __aenter__(self):
         await self.connect()
@@ -489,6 +493,15 @@ async def create_session(self) -> ua.CreateSessionResult:
             raise ua.UaError("Server certificate mismatch")
         # remember PolicyId's: we will use them in activate_session()
         ep = Client.find_endpoint(response.ServerEndpoints, self.security_policy.Mode, self.security_policy.URI)
+
+        if self.certificate_validator and server_certificate:
+            try:
+                await self.certificate_validator(x509.load_der_x509_certificate(server_certificate), ep.Server)
+            except ServiceError as exp:
+                status = ua.StatusCode(exp.code)
+                _logger.error("create_session fault response: %s (%s)", status.doc, status.name)
+                raise ua.UaStatusCodeError(exp.code) from exp
+
         self._policy_ids = ep.UserIdentityTokens
         #  Actual maximum number of milliseconds that a Session shall remain open without activity
         if self.session_timeout != response.RevisedSessionTimeout:

asyncua/crypto/validator.py

@@ -0,0 +1,133 @@
+from typing import Callable, Awaitable, Optional
+import logging
+from datetime import datetime
+from enum import Flag, auto
+from cryptography import x509
+from cryptography.x509.oid import ExtendedKeyUsageOID
+from asyncua import ua
+from asyncua.common.utils import ServiceError
+from asyncua.ua import ApplicationDescription
+from asyncua.crypto.truststore import TrustStore
+
+_logger = logging.getLogger(__name__)
+
+# Use for storing method that can validate a certificate on a create_session
+CertificateValidatorMethod = Callable[[x509.Certificate, ApplicationDescription], Awaitable[None]]
+
+class CertificateValidatorOptions(Flag):
+    """
+    Flags for which certificate validation should be performed
+
+    Three default sets of flags are provided:
+    - BASIC_VALIDATION
+    - EXT_VALIDATION
+    - TRUSTED_VALIDATION
+    """
+    TIME_RANGE = auto()
+    URI = auto()
+    KEY_USAGE = auto()
+    EXT_KEY_USAGE = auto()
+    TRUSTED = auto()
+    REVOKED = auto()
+
+    PEER_CLIENT = auto()
+    """Expect role of the peer is client (mutal exclusive with PEER_SERVER)"""
+    PEER_SERVER = auto()
+    """Expect role of the peer is server (mutal exclusive with PEER_CLIENT)"""
+
+    BASIC_VALIDATION = TIME_RANGE | URI
+    """Option set with: Only check time range and uri"""
+    EXT_VALIDATION = TIME_RANGE | URI | KEY_USAGE | EXT_KEY_USAGE
+    """Option set with: Check time, uri, key usage and extended key usage"""
+    TRUSTED_VALIDATION = TIME_RANGE | URI | KEY_USAGE | EXT_KEY_USAGE | TRUSTED | REVOKED
+    """Option set with: Check time, uri, key usage, extended key usage, is trusted (direct or by CA) and not revoked (CRL)"""
+
+
+class CertificateValidator:
+    """
+    CertificateValidator contains a basic certificate validator including trusted store with revocation list support.
+    The CertificateValidator can be used as a CertificateValidatorMethod.
+
+    Default CertificateValidatorOptions.BASIC_VALIDATION is used.
+    """
+
+    def __init__(self, options: CertificateValidatorOptions = CertificateValidatorOptions.BASIC_VALIDATION | CertificateValidatorOptions.PEER_CLIENT, trust_store: Optional[TrustStore] = None):
+        self._options = options
+        self._trust_store: Optional[TrustStore] = trust_store
+
+    def set_validate_options(self, options: CertificateValidatorOptions):
+        """ Change the use validation options at runtime"""
+
+        self._options = options
+
+    async def validate(self, cert: x509.Certificate, app_description: ua.ApplicationDescription):
+        """ Validate if a certificate is valid based on the validation options.
+        When not valid is raises a ServiceError with an UA Result Code.
+
+        Args:
+            cert (x509.Certificate): certificate to check
+            app_description (ua.ApplicationDescription): application descriptor of the client/server
+
+        Raises:
+            BadCertificateTimeInvalid: When current time is not in the time range of the certificate
+            BadCertificateUriInvalid: Uri from certificate doesn't match application descriptor uri
+            BadCertificateUseNotAllowed: KeyUsage or ExtendedKeyUsage fields mismatch
+            BadCertificateInvalid: General when part of certifcate fields can't be found
+            BadCertificateUntrusted: Not trusted by TrustStore
+            ApplicationDescription: Certifacate in CRL of the TrustStore
+        """
+
+        if CertificateValidatorOptions.TIME_RANGE in self._options:
+            now = datetime.utcnow()
+            if cert.not_valid_after < now:
+                raise ServiceError(ua.StatusCodes.BadCertificateTimeInvalid)
+            elif cert.not_valid_before > now:
+                raise ServiceError(ua.StatusCodes.BadCertificateTimeInvalid)
+        try:
+            san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)
+            if CertificateValidatorOptions.URI in self._options:
+                san_uri = san.value.get_values_for_type(x509.UniformResourceIdentifier)
+                if app_description.ApplicationUri not in san_uri:
+                    raise ServiceError(ua.StatusCodes.BadCertificateUriInvalid)
+            if CertificateValidatorOptions.KEY_USAGE in self._options:
+
+                key_usage = cert.extensions.get_extension_for_class(x509.KeyUsage).value
+                if key_usage.data_encipherment is False or \
+                key_usage.digital_signature is False or \
+                key_usage.content_commitment is False or \
+                key_usage.key_encipherment is False:
+                    raise ServiceError(ua.StatusCodes.BadCertificateUseNotAllowed)
+            if CertificateValidatorOptions.EXT_KEY_USAGE in self._options:
+                oid = ExtendedKeyUsageOID.SERVER_AUTH if CertificateValidatorOptions.PEER_SERVER in self._options else ExtendedKeyUsageOID.CLIENT_AUTH
+
+                if oid not in cert.extensions.get_extension_for_class(x509.ExtendedKeyUsage).value:
+                    raise ServiceError(ua.StatusCodes.BadCertificateUseNotAllowed)
+
+                if CertificateValidatorOptions.PEER_SERVER in self._options and \
+                    app_description.ApplicationType not in [ua.ApplicationType.Server, ua.ApplicationType.ClientAndServer]:
+                    _logger.warning('mismatch between application type and certificate ExtendedKeyUsage')
+                    raise ServiceError(ua.StatusCodes.BadCertificateUseNotAllowed)
+                elif CertificateValidatorOptions.PEER_CLIENT in self._options and \
+                    app_description.ApplicationType not in [ua.ApplicationType.Client, ua.ApplicationType.ClientAndServer]:
+                    _logger.warning('mismatch between application type and certificate ExtendedKeyUsage')
+                    raise ServiceError(ua.StatusCodes.BadCertificateUseNotAllowed)
+
+
+            # if hostname is not None:
+            #     san_dns_names = san.value.get_values_for_type(x509.DNSName)
+            #     if hostname not in san_dns_names:
+            #         raise ServiceError(ua.StatusCodes.BadCertificateHostNameInvalid) from exc
+        except x509.ExtensionNotFound as exc:
+            raise ServiceError(ua.StatusCodes.BadCertificateInvalid) from exc
+
+        if CertificateValidatorOptions.TRUSTED in self._options or CertificateValidatorOptions.REVOKED in self._options:
+
+            if CertificateValidatorOptions.TRUSTED in self._options:
+                if self._trust_store and not self._trust_store.is_trusted(cert):
+                    raise ServiceError(ua.StatusCodes.BadCertificateUntrusted)
+            if CertificateValidatorOptions.REVOKED in self._options:
+                if self._trust_store and self._trust_store.is_revoked(cert):
+                    raise ServiceError(ua.StatusCodes.BadCertificateRevoked)
+
+    async def __call__(self, cert: x509.Certificate, app_description: ua.ApplicationDescription):
+        return await self.validate(cert, app_description)

asyncua/server/internal_server.py

@@ -2,7 +2,7 @@
 Internal server implementing opcu-ua interface.
 Can be used on server side or to implement binary/https opc-ua servers
 """
-
+from typing import Optional
 import asyncio
 from datetime import datetime, timedelta
 from copy import copy
@@ -22,6 +22,7 @@
 from .users import User, UserRole
 from .internal_session import InternalSession
 from .event_generator import EventGenerator
+from ..crypto.validator import CertificateValidatorMethod
 
 try:
     from asyncua.crypto import uacrypto
@@ -66,6 +67,8 @@ def __init__(self, user_manager: UserManager = None):
             _logger.info("No user manager specified. Using default permissive manager instead.")
             user_manager = PermissiveUserManager()
         self.user_manager = user_manager
+        self.certificate_validator: Optional[CertificateValidatorMethod]= None
+        """hook to validate a certificate, raises a ServiceError when not valid"""
         # create a session to use on server side
         self.isession = InternalSession(
             self, self.aspace, self.subscription_service, "Internal", user=User(role=UserRole.Admin)

asyncua/server/internal_session.py

@@ -1,15 +1,19 @@
 import logging
 from enum import Enum
-from typing import Iterable, Optional, List
+from typing import Iterable, Optional, List, Tuple, TYPE_CHECKING
 
 from asyncua import ua
 from asyncua.common.session_interface import AbstractSession
 from ..common.callback import CallbackType, ServerItemCallback
 from ..common.utils import create_nonce, ServiceError
+from ..crypto.uacrypto import x509
 from .address_space import AddressSpace
 from .users import User, UserRole
 from .subscription_service import SubscriptionService
 
+if TYPE_CHECKING:
+    from .internal_server import InternalServer
+
 
 class SessionState(Enum):
     Created = 0
@@ -26,10 +30,10 @@ class InternalSession(AbstractSession):
     _counter = 10
     _auth_counter = 1000
 
-    def __init__(self, internal_server, aspace: AddressSpace, submgr: SubscriptionService, name,
+    def __init__(self, internal_server: "InternalServer", aspace: AddressSpace, submgr: SubscriptionService, name,
                  user=User(role=UserRole.Anonymous), external=False):
         self.logger = logging.getLogger(__name__)
-        self.iserver = internal_server
+        self.iserver: "InternalServer" = internal_server
         # define if session is external, we need to copy some objects if it is internal
         self.external = external
         self.aspace: AddressSpace = aspace
@@ -55,13 +59,17 @@ async def get_endpoints(self, params=None, sockname=None):
     def is_activated(self) -> bool:
         return self.state == SessionState.Activated
 
-    async def create_session(self, params, sockname=None):
+    async def create_session(self, params: ua.CreateSessionParameters, sockname: Optional[Tuple[str, int]]=None):
         self.logger.info('Create session request')
         result = ua.CreateSessionResult()
         result.SessionId = self.session_id
         result.AuthenticationToken = self.auth_token
         result.RevisedSessionTimeout = params.RequestedSessionTimeout
         result.MaxRequestMessageSize = 65536
+
+        if self.iserver.certificate_validator and params.ClientCertificate:
+            await self.iserver.certificate_validator(x509.load_der_x509_certificate(params.ClientCertificate), params.ClientDescription)
+
         self.nonce = create_nonce(32)
         result.ServerNonce = self.nonce
 

asyncua/server/server.py

@@ -29,7 +29,7 @@
 from ..common.ua_utils import get_nodes_of_namespace
 from ..common.connection import TransportLimits
 
-from ..crypto import security_policies, uacrypto
+from ..crypto import security_policies, uacrypto, validator
 
 _logger = logging.getLogger(__name__)
 
@@ -824,3 +824,17 @@ def read_attribute_value(self, nodeid, attr=ua.AttributeIds.Value):
         directly read datavalue of the Attribute
         """
         return self.iserver.read_attribute_value(nodeid, attr)
+
+    def set_certificate_validator(self, validator: validator.CertificateValidatorMethod):
+        """
+        Assign a method to be called when certificate needs to be validated.
+
+        Function is called with certificate and application description and should raise the correct status code
+        when invalid.
+
+            async def example_validation_method(certificate: x509.Certificate, app_description: ua.ApplicationDescription):
+                ...
+                if not_valid_condition:
+                    raise ServiceError(ua.StatusCodes.BadCertificateInvalid)
+        """
+        self.iserver.certificate_validator = validator

examples/client-with-encryption.py

@@ -3,16 +3,20 @@
 import sys
 import socket
 from pathlib import Path
+from cryptography.x509.oid import ExtendedKeyUsageOID
 sys.path.insert(0, "..")
 from asyncua import Client
 from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256
 from asyncua.crypto.cert_gen import setup_self_signed_certificate
-from cryptography.x509.oid import ExtendedKeyUsageOID
+from asyncua.crypto.validator import CertificateValidator, CertificateValidatorOptions
+from asyncua.crypto.truststore import TrustStore
+from asyncua import ua
 
 
 logging.basicConfig(level=logging.INFO)
-_logger = logging.getLogger("asyncua")
+_logger = logging.getLogger(__name__)
 
+USE_TRUST_STORE = True
 
 cert_idx = 4
 cert_base = Path(__file__).parent
@@ -43,12 +47,23 @@ async def task(loop):
         private_key=str(private_key),
         server_certificate="certificate-example.der"
     )
-    async with client:
-        objects = client.nodes.objects
-        child = await objects.get_child(['0:MyObject', '0:MyVariable'])
-        print(await child.get_value())
-        await child.set_value(42)
-        print(await child.get_value())
+
+    if USE_TRUST_STORE:
+        trust_store = TrustStore([Path('examples') / 'certificates' / 'trusted' / 'certs'], [])
+        await trust_store.load()
+        validator =CertificateValidator(CertificateValidatorOptions.TRUSTED_VALIDATION|CertificateValidatorOptions.PEER_SERVER, trust_store)
+    else:
+        validator =CertificateValidator(CertificateValidatorOptions.EXT_VALIDATION|CertificateValidatorOptions.PEER_SERVER)
+    client.certificate_validator = validator
+    try:
+        async with client:
+            objects = client.nodes.objects
+            child = await objects.get_child(['0:MyObject', '0:MyVariable'])
+            print(await child.get_value())
+            await child.set_value(42)
+            print(await child.get_value())
+    except ua.UaError as exp:
+        _logger.error(exp)
 
 
 def main():

examples/server-with-encryption.py

@@ -10,12 +10,16 @@
 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
 from cryptography.x509.oid import ExtendedKeyUsageOID
+from asyncua.crypto.truststore import TrustStore
 
 
 logging.basicConfig(level=logging.INFO)
 
 
+USE_TRUST_STORE = False
+
 async def main():
     cert_base = Path(__file__).parent
     server_cert = Path(cert_base / "certificates/server-certificate-example.der")
@@ -56,6 +60,15 @@ async def main():
     await server.load_certificate(str(server_cert))
     await server.load_private_key(str(server_private_key))
 
+    if USE_TRUST_STORE:
+        trust_store = TrustStore([Path('examples') / 'certificates' / 'trusted' / 'certs'], [])
+        await trust_store.load()
+        validator = CertificateValidator(options=CertificateValidatorOptions.TRUSTED_VALIDATION | CertificateValidatorOptions.PEER_CLIENT,
+                                         trust_store = trust_store)
+    else:
+        validator = CertificateValidator(options=CertificateValidatorOptions.EXT_VALIDATION | CertificateValidatorOptions.PEER_CLIENT)
+    server.set_certificate_validator(validator)
+
     idx = 0
 
     # populating our address space