Use cryptography.x509.verification for certification verifier.

Author: dajiaji

cwt/algs/asymmetric.py

@@ -1,6 +1,8 @@
 from typing import Any, Dict, List
 
-from certvalidator import CertificateValidator, ValidationContext
+from cryptography.x509 import Certificate, DNSName, load_der_x509_certificate
+from cryptography.x509.oid import NameOID
+from cryptography.x509.verification import PolicyBuilder, Store
 
 from ..cose_key_interface import COSEKeyInterface
 from ..exceptions import VerifyError
@@ -11,28 +13,32 @@ def __init__(self, params: Dict[int, Any]):
         super().__init__(params)
 
         self._key: Any = b""
-        self._cert = b""
-        self._intermediates = []
+        self._cert: Certificate = None
+        self._intermediates: List[Certificate] = []
 
         if 33 in params:
             if not isinstance(params[33], (bytes, list)):
                 raise ValueError("x5c(33) should be bytes(bstr) or list.")
             certs = [params[33]] if isinstance(params[33], bytes) else params[33]
-            self._cert = certs[0]
+            self._cert = load_der_x509_certificate(certs[0])
             if len(certs) > 1:
-                self._intermediates = certs[1:]
+                for c in certs[1:]:
+                    self._intermediates.append(load_der_x509_certificate(c))
             return
 
-    def validate_certificate(self, ca_certs: List[bytes]) -> bool:
+    def validate_certificate(self, ca_certs: List[Certificate]) -> bool:
         if not ca_certs:
             raise ValueError("ca_certs should be set.")
         if not self._cert:
             return False
 
-        ctx = ValidationContext(trust_roots=ca_certs)
+        store = Store(ca_certs)
+        builder = PolicyBuilder().store(store)
+        verifier = builder.build_server_verifier(
+            DNSName(self._cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value)
+        )
         try:
-            validator = CertificateValidator(self._cert, self._intermediates, validation_context=ctx)
-            validator.validate_usage(set(["digital_signature"]), extended_optional=True)
+            verifier.verify(self._cert, self._intermediates)
         except Exception as err:
             raise VerifyError("Failed to validate the certificate bound to the key.") from err
         return True

cwt/cose.py

@@ -1,7 +1,7 @@
 from typing import Any, Dict, List, Optional, Tuple, Union
 
-from asn1crypto import pem
 from cbor2 import CBORTag
+from cryptography.x509 import load_pem_x509_certificates
 
 from .cbor_processor import CBORProcessor
 from .const import (
@@ -54,11 +54,8 @@ def __init__(
         if ca_certs:
             if not isinstance(ca_certs, str):
                 raise ValueError("ca_certs should be str.")
-            self._trust_roots: List[bytes] = []
             with open(ca_certs, "rb") as f:
-                for _, _, der_bytes in pem.unarmor(f.read(), multiple=True):
-                    self._ca_certs.append(der_bytes)
-
+                self._ca_certs = load_pem_x509_certificates(f.read())
         if not isinstance(deterministic_header, bool):
             raise ValueError("deterministic_header should be bool.")
         self._deterministic_header = deterministic_header

poetry.lock

@@ -26,9 +26,7 @@ exclude = [
 
 [tool.poetry.dependencies]
 python = "^3.9.2,<4.0"
-asn1crypto = "^1.4.0"
 cbor2 = "^5.4.2"
-certvalidator = "^0.11.1"
 cryptography = ">=42.0.1,<45"
 Sphinx = {version = ">=7.1,<8", optional = true, extras = ["docs"]}
 sphinx-autodoc-typehints = {version = ">=1.25.2,<3.0.0", optional = true, extras = ["docs"]}

pyproject.toml

@@ -175,7 +175,7 @@ def test_sample_readme_signed_cwt_es256(self):
         decoded = cwt.decode(token, public_key)
         assert 1 in decoded and decoded[1] == "coaps://as.example"
 
-    def test_sample_readme_signed_cwt_es256_with_cert(self):
+    def test_sample_readme_signed_cwt_es256_with_cert_missing_required_extension(self):
         # with open(key_path("cacert.pem")) as f:
         #     k1 = x509.load_pem_x509_certificate(f.read().encode("utf-8"))
 
@@ -196,10 +196,14 @@ def test_sample_readme_signed_cwt_es256_with_cert(self):
         token = cwt.encode({"iss": "coaps://as.example", "sub": "dajiaji", "cti": "123"}, private_key)
 
         decoder = CWT.new(ca_certs=key_path("cacert.pem"))
-        decoded = decoder.decode(token, public_key)
-        assert 1 in decoded and decoded[1] == "coaps://as.example"
+        with pytest.raises(VerifyError) as err:
+            decoder.decode(token, public_key)
+            pytest.fail("decode() should fail.")
+        assert "Failed to validate the certificate bound to the key." in str(err.value)
+        # decoded = decoder.decode(token, public_key)
+        # assert 1 in decoded and decoded[1] == "coaps://as.example"
 
-    def test_sample_readme_signed_cwt_es256_with_cert_without_intermediates(self):
+    def test_sample_readme_signed_cwt_es256_with_cert_missing_required_extension_without_intermediates(self):
         with open(key_path("private_key_cert_es256.pem")) as f:
             private_key = COSEKey.from_pem(f.read(), kid="P-256-01")
 
@@ -209,8 +213,12 @@ def test_sample_readme_signed_cwt_es256_with_cert_without_intermediates(self):
         token = cwt.encode({"iss": "coaps://as.example", "sub": "dajiaji", "cti": "123"}, private_key)
 
         decoder = CWT.new(ca_certs=key_path("cacert.pem"))
-        decoded = decoder.decode(token, public_key)
-        assert 1 in decoded and decoded[1] == "coaps://as.example"
+        with pytest.raises(VerifyError) as err:
+            decoder.decode(token, public_key)
+            pytest.fail("decode() should fail.")
+        assert "Failed to validate the certificate bound to the key." in str(err.value)
+        # decoded = decoder.decode(token, public_key)
+        # assert 1 in decoded and decoded[1] == "coaps://as.example"
 
     def test_sample_readme_signed_cwt_es256_with_another_ca_cert(self):
         with open(key_path("private_key_cert_es256.pem")) as f: