Include support for x509 certificate chains in DER format (#1822)

* Fix certificate chain handling in x509_from_der function

This change enhances the x509_from_der function to properly handle certificate chains
by extracting and using the first certificate when a chain is detected.

When a certificate chain is received, the cryptography library raises a ValueError
with 'extradata' in the error message. This fix catches that specific error and
implements DER parsing to extract the first certificate from the chain.

This allows the library to work with certificate chains that some OPC UA servers
might provide, improving compatibility.

Fixes #1148 and #1245

* Update CHANGELOG.md with certificate chain fix

* Improve exception handling comment in x509_from_der function

* Add tests for certificate chain handling

* Fix code formatting issues

* Apply Ruff formatting changes

* Remove unused os import from test file

Author: ryanmickler

CHANGELOG.md

@@ -19,6 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   [#1277](https://github.com/FreeOpcUa/opcua-asyncio/pull/1277)
 - Fixed incorrect function signature in Node and it's Sync wrapper
   [#1690](https://github.com/FreeOpcUa/opcua-asyncio/pull/1690)
+- Fix certificate chain handling in x509_from_der function
+  Fixes [#1148](https://github.com/FreeOpcUa/opcua-asyncio/issues/1148) and [#1245](https://github.com/FreeOpcUa/opcua-asyncio/issues/1245)
 
 ## [1.0.2] - 2022-04-05
 

asyncua/crypto/uacrypto.py

@@ -57,9 +57,56 @@ async def load_certificate(path_or_content: Union[bytes, str, Path], extension:
 
 
 def x509_from_der(data):
+    """Load X.509 certificate from DER data.
+
+    This function handles both single certificates and certificate chains.
+    When a certificate chain is detected, it extracts and returns the first
+    certificate in the chain.
+
+    Args:
+        data (bytes): DER-encoded certificate data
+
+    Returns:
+        x509.Certificate or None: Loaded certificate object or None if data is empty
+    """
     if not data:
         return None
-    return x509.load_der_x509_certificate(data, default_backend())
+    try:
+        # First try to load as a single certificate
+        return x509.load_der_x509_certificate(data, default_backend())
+    except ValueError as e:
+        # The specific error 'ParseError { kind: ExtraData }' indicates we might have a certificate chain
+        # This happens when the cryptography library encounters additional data after a valid certificate
+        if "extradata" not in str(e).lower():
+            raise
+
+        # Try parsing as certificate chain by manually extracting the first certificate
+        # using DER encoding rules: https://en.wikipedia.org/wiki/X.690#DER_encoding
+        offset = 0
+        while offset < len(data):
+            if data[offset] != 0x30:  # 0x30 is the ASN.1 SEQUENCE tag that starts a certificate
+                break
+
+            # Parse the length field according to DER rules
+            length_byte = data[offset + 1]
+            if length_byte < 128:  # Short form length
+                cert_len = length_byte
+                header_len = 2  # Tag (1 byte) + length (1 byte)
+            else:  # Long form length
+                # First byte indicates how many bytes are used to represent the length
+                num_len_bytes = length_byte & 0x7F
+                # Read the actual length from the following bytes
+                cert_len = int.from_bytes(data[offset + 2 : offset + 2 + num_len_bytes], "big")
+                header_len = 2 + num_len_bytes  # Tag (1) + length indicator (1) + length bytes
+
+            # Extract just the first certificate from the chain
+            total_len = header_len + cert_len
+            cert_data = data[offset : offset + total_len]
+            # Return only the first certificate in the chain
+            return x509.load_der_x509_certificate(cert_data, default_backend())
+
+        # If we get here, we have a certificate chain but no valid certificates
+        raise ValueError("No valid certificates found in the certificate chain")
 
 
 async def load_private_key(

tests/test_cert_chain.py

@@ -0,0 +1,111 @@
+"""Tests for certificate chain handling in the uacrypto module.
+
+This module tests the enhanced x509_from_der function which now supports:
+1. Loading single certificates (original functionality)
+2. Loading the first certificate from a certificate chain
+3. Proper error handling for invalid certificate data
+
+These tests ensure that the function correctly handles certificate chains
+that some OPC UA servers might provide, improving compatibility.
+"""
+
+import pytest
+from pathlib import Path
+
+from asyncua.crypto import uacrypto
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+
+pytestmark = pytest.mark.asyncio
+
+BASE_DIR = Path(__file__).parent.parent
+EXAMPLE_CERT_PATH = BASE_DIR / "examples" / "certificate-example.der"
+EXAMPLE_CERT_3072_PATH = BASE_DIR / "examples" / "certificate-3072-example.der"
+PEER_CERT_PATH = BASE_DIR / "examples" / "certificates" / "peer-certificate-example-1.der"
+
+
+def create_cert_chain():
+    """Create a certificate chain for testing.
+
+    This function creates a simulated certificate chain by concatenating
+    two different certificates. In a real-world scenario, certificate chains
+    would contain a server certificate followed by intermediate CA certificates.
+
+    Returns:
+        tuple: (first_cert_data, cert_chain_data)
+            - first_cert_data: The DER-encoded data of the first certificate
+            - cert_chain_data: The DER-encoded data of the entire certificate chain
+    """
+    # Load the example certificates
+    with open(EXAMPLE_CERT_PATH, "rb") as f:
+        cert1_data = f.read()
+
+    with open(PEER_CERT_PATH, "rb") as f:
+        cert2_data = f.read()
+
+    # Create a certificate chain by concatenating two different certificates
+    cert_chain = cert1_data + cert2_data
+
+    return cert1_data, cert_chain
+
+
+def test_x509_from_der_single_cert():
+    """Test that x509_from_der works with a single certificate."""
+    # Test with the standard example certificate
+    with open(EXAMPLE_CERT_PATH, "rb") as f:
+        cert_data = f.read()
+
+    cert = uacrypto.x509_from_der(cert_data)
+    assert cert is not None
+    assert isinstance(cert, x509.Certificate)
+
+    # Test with the 3072-bit example certificate
+    with open(EXAMPLE_CERT_3072_PATH, "rb") as f:
+        cert_data_3072 = f.read()
+
+    cert_3072 = uacrypto.x509_from_der(cert_data_3072)
+    assert cert_3072 is not None
+    assert isinstance(cert_3072, x509.Certificate)
+
+    # Test with the peer certificate
+    with open(PEER_CERT_PATH, "rb") as f:
+        peer_cert_data = f.read()
+
+    peer_cert = uacrypto.x509_from_der(peer_cert_data)
+    assert peer_cert is not None
+    assert isinstance(peer_cert, x509.Certificate)
+
+
+def test_x509_from_der_cert_chain():
+    """Test that x509_from_der works with a certificate chain."""
+    first_cert_data, cert_chain = create_cert_chain()
+
+    # Load the certificate chain using x509_from_der
+    cert = uacrypto.x509_from_der(cert_chain)
+
+    # Verify that the certificate was loaded correctly
+    assert cert is not None
+    assert isinstance(cert, x509.Certificate)
+
+    # Verify that the loaded certificate is the first one in the chain
+    # by comparing it with the original certificate
+    original_cert = x509.load_der_x509_certificate(first_cert_data, default_backend())
+    assert cert.public_bytes(serialization.Encoding.DER) == original_cert.public_bytes(serialization.Encoding.DER)
+
+
+def test_x509_from_der_invalid_data():
+    """Test that x509_from_der handles invalid data correctly."""
+    # Test with None
+    assert uacrypto.x509_from_der(None) is None
+
+    # Test with empty bytes
+    assert uacrypto.x509_from_der(b"") is None
+
+    # Test with invalid data that doesn't start with a SEQUENCE tag
+    with pytest.raises(ValueError):
+        uacrypto.x509_from_der(b"invalid data")
+
+    # Test with data that starts with a SEQUENCE tag but is otherwise invalid
+    with pytest.raises(ValueError):
+        uacrypto.x509_from_der(b"\x30\x03\x01\x02\x03")