typehint x509.base (pyca#5899)

mathiasertl · web-flow · commit 8d41a94bba76 · 2021-03-15T18:44:02.000-04:00
* start typing x509.base * statically type x509.base * typehint X509Backend interface * typehint at least the X509Backend interface * make _CertificateRevocationList/_CertificateSigningRequest actual subclasses of the interface (as done before for Certificate in f16bff2) * tell mypy to ignore lines with deliberately wrong types * signature_hash_algorithm always returns a hash algorithm (it's not optional) * Revert "signature_hash_algorithm always returns a hash algorithm (it's not optional)" This reverts commit f6a5b17. * hash algorithm is actually optional * fix import style * typehint parsed_version to int, which it de facto always is * minimize changes * break import cycle with conditional imports * ignore access to private members of openssl implementation * reformat code with Black * test check for missing public key
diff --git a/.coveragerc b/.coveragerc
@@ -15,3 +15,4 @@ exclude_lines =
     @abc.abstractmethod
     @abc.abstractproperty
     @typing.overload
+    if typing.TYPE_CHECKING
diff --git a/src/cryptography/hazmat/backends/__init__.py b/src/cryptography/hazmat/backends/__init__.py
@@ -9,7 +9,7 @@
 _default_backend: typing.Optional[Backend] = None
 
 
-def default_backend():
+def default_backend() -> Backend:
     global _default_backend
 
     if _default_backend is None:
diff --git a/src/cryptography/hazmat/backends/interfaces.py b/src/cryptography/hazmat/backends/interfaces.py
@@ -4,6 +4,23 @@
 
 
 import abc
+import typing
+
+
+if typing.TYPE_CHECKING:
+    from cryptography.hazmat._types import _PRIVATE_KEY_TYPES
+    from cryptography.hazmat.primitives import hashes
+    from cryptography.x509.base import (
+        Certificate,
+        CertificateBuilder,
+        CertificateRevocationList,
+        CertificateRevocationListBuilder,
+        CertificateSigningRequest,
+        CertificateSigningRequestBuilder,
+        RevokedCertificate,
+        RevokedCertificateBuilder,
+    )
+    from cryptography.x509.name import Name
 
 
 class CipherBackend(metaclass=abc.ABCMeta):
@@ -262,69 +279,86 @@ def load_der_parameters(self, data):
 
 class X509Backend(metaclass=abc.ABCMeta):
     @abc.abstractmethod
-    def load_pem_x509_certificate(self, data):
+    def load_pem_x509_certificate(self, data: bytes) -> "Certificate":
         """
         Load an X.509 certificate from PEM encoded data.
         """
 
     @abc.abstractmethod
-    def load_der_x509_certificate(self, data):
+    def load_der_x509_certificate(self, data: bytes) -> "Certificate":
         """
         Load an X.509 certificate from DER encoded data.
         """
 
     @abc.abstractmethod
-    def load_der_x509_csr(self, data):
+    def load_der_x509_csr(self, data: bytes) -> "CertificateSigningRequest":
         """
         Load an X.509 CSR from DER encoded data.
         """
 
     @abc.abstractmethod
-    def load_pem_x509_csr(self, data):
+    def load_pem_x509_csr(self, data: bytes) -> "CertificateSigningRequest":
         """
         Load an X.509 CSR from PEM encoded data.
         """
 
     @abc.abstractmethod
-    def create_x509_csr(self, builder, private_key, algorithm):
+    def create_x509_csr(
+        self,
+        builder: "CertificateSigningRequestBuilder",
+        private_key: "_PRIVATE_KEY_TYPES",
+        algorithm: typing.Optional["hashes.HashAlgorithm"],
+    ) -> "CertificateSigningRequest":
         """
         Create and sign an X.509 CSR from a CSR builder object.
         """
 
     @abc.abstractmethod
-    def create_x509_certificate(self, builder, private_key, algorithm):
+    def create_x509_certificate(
+        self,
+        builder: "CertificateBuilder",
+        private_key: "_PRIVATE_KEY_TYPES",
+        algorithm: typing.Optional["hashes.HashAlgorithm"],
+    ) -> "Certificate":
         """
         Create and sign an X.509 certificate from a CertificateBuilder object.
         """
 
     @abc.abstractmethod
-    def create_x509_crl(self, builder, private_key, algorithm):
+    def create_x509_crl(
+        self,
+        builder: "CertificateRevocationListBuilder",
+        private_key: "_PRIVATE_KEY_TYPES",
+        algorithm: typing.Optional["hashes.HashAlgorithm"],
+    ) -> "CertificateRevocationList":
         """
         Create and sign an X.509 CertificateRevocationList from a
         CertificateRevocationListBuilder object.
         """
 
     @abc.abstractmethod
-    def create_x509_revoked_certificate(self, builder):
+    def create_x509_revoked_certificate(
+        self, builder: "RevokedCertificateBuilder"
+    ) -> "RevokedCertificate":
         """
         Create a RevokedCertificate object from a RevokedCertificateBuilder
         object.
         """
 
     @abc.abstractmethod
-    def x509_name_bytes(self, name):
+    def x509_name_bytes(self, name: "Name") -> bytes:
         """
         Compute the DER encoded bytes of an X509 Name object.
         """
 
     @abc.abstractmethod
-    def load_pem_x509_crl(self, data):
+    def load_pem_x509_crl(self, data: bytes) -> "CertificateRevocationList":
         """
         Load an X.509 CRL from PEM encoded data.
         """
 
     @abc.abstractmethod
-    def load_der_x509_crl(self, data):
+    def load_der_x509_crl(self, data: bytes) -> "CertificateRevocationList":
         """
         Load an X.509 CRL from DER encoded data.
         """
diff --git a/src/cryptography/hazmat/backends/openssl/backend.py b/src/cryptography/hazmat/backends/openssl/backend.py
@@ -6,6 +6,7 @@
 import collections
 import contextlib
 import itertools
+import typing
 import warnings
 from contextlib import contextmanager
 
@@ -18,6 +19,7 @@
     encode_der,
     encode_der_integer,
 )
+from cryptography.hazmat._types import _PRIVATE_KEY_TYPES
 from cryptography.hazmat.backends.interfaces import Backend as BackendInterface
 from cryptography.hazmat.backends.openssl import aead
 from cryptography.hazmat.backends.openssl.ciphers import _CipherContext
@@ -137,6 +139,7 @@
 from cryptography.hazmat.primitives.kdf import scrypt
 from cryptography.hazmat.primitives.serialization import pkcs7, ssh
 from cryptography.x509 import ocsp
+from cryptography.x509.name import Name
 
 
 _MemoryBIO = collections.namedtuple("_MemoryBIO", ["bio", "char_ptr"])
@@ -895,7 +898,12 @@ def _x509_check_signature_params(self, private_key, algorithm):
                 "MD5 hash algorithm is only supported with RSA keys"
             )
 
-    def create_x509_csr(self, builder, private_key, algorithm):
+    def create_x509_csr(
+        self,
+        builder: x509.CertificateSigningRequestBuilder,
+        private_key: _PRIVATE_KEY_TYPES,
+        algorithm: typing.Optional[hashes.HashAlgorithm],
+    ) -> _CertificateSigningRequest:
         if not isinstance(builder, x509.CertificateSigningRequestBuilder):
             raise TypeError("Builder type mismatch.")
         self._x509_check_signature_params(private_key, algorithm)
@@ -920,7 +928,9 @@ def create_x509_csr(self, builder, private_key, algorithm):
 
         # Set subject public key.
         public_key = private_key.public_key()
-        res = self._lib.X509_REQ_set_pubkey(x509_req, public_key._evp_pkey)
+        res = self._lib.X509_REQ_set_pubkey(
+            x509_req, public_key._evp_pkey  # type: ignore[union-attr]
+        )
         self.openssl_assert(res == 1)
 
         # Add extensions.
@@ -960,16 +970,25 @@ def create_x509_csr(self, builder, private_key, algorithm):
             self.openssl_assert(res == 1)
 
         # Sign the request using the requester's private key.
-        res = self._lib.X509_REQ_sign(x509_req, private_key._evp_pkey, evp_md)
+        res = self._lib.X509_REQ_sign(
+            x509_req, private_key._evp_pkey, evp_md  # type: ignore[union-attr]
+        )
         if res == 0:
             errors = self._consume_errors_with_text()
             raise ValueError("Signing failed", errors)
 
         return _CertificateSigningRequest(self, x509_req)
 
-    def create_x509_certificate(self, builder, private_key, algorithm):
+    def create_x509_certificate(
+        self,
+        builder: x509.CertificateBuilder,
+        private_key: _PRIVATE_KEY_TYPES,
+        algorithm: typing.Optional[hashes.HashAlgorithm],
+    ) -> _Certificate:
         if not isinstance(builder, x509.CertificateBuilder):
             raise TypeError("Builder type mismatch.")
+        if builder._public_key is None:
+            raise TypeError("Builder has no public key.")
         self._x509_check_signature_params(private_key, algorithm)
 
         # Resolve the signature algorithm.
@@ -1027,7 +1046,11 @@ def create_x509_certificate(self, builder, private_key, algorithm):
         self.openssl_assert(res == 1)
 
         # Sign the certificate with the issuer's private key.
-        res = self._lib.X509_sign(x509_cert, private_key._evp_pkey, evp_md)
+        res = self._lib.X509_sign(
+            x509_cert,
+            private_key._evp_pkey,  # type: ignore[union-attr]
+            evp_md,
+        )
         if res == 0:
             errors = self._consume_errors_with_text()
             raise ValueError("Signing failed", errors)
@@ -1058,7 +1081,12 @@ def _create_asn1_time(self, time):
         self._set_asn1_time(asn1_time, time)
         return asn1_time
 
-    def create_x509_crl(self, builder, private_key, algorithm):
+    def create_x509_crl(
+        self,
+        builder: x509.CertificateRevocationListBuilder,
+        private_key: _PRIVATE_KEY_TYPES,
+        algorithm: typing.Optional[hashes.HashAlgorithm],
+    ) -> _CertificateRevocationList:
         if not isinstance(builder, x509.CertificateRevocationListBuilder):
             raise TypeError("Builder type mismatch.")
         self._x509_check_signature_params(private_key, algorithm)
@@ -1109,7 +1137,9 @@ def create_x509_crl(self, builder, private_key, algorithm):
             res = self._lib.X509_CRL_add0_revoked(x509_crl, revoked)
             self.openssl_assert(res == 1)
 
-        res = self._lib.X509_CRL_sign(x509_crl, private_key._evp_pkey, evp_md)
+        res = self._lib.X509_CRL_sign(
+            x509_crl, private_key._evp_pkey, evp_md  # type: ignore[union-attr]
+        )
         if res == 0:
             errors = self._consume_errors_with_text()
             raise ValueError("Signing failed", errors)
@@ -1170,7 +1200,9 @@ def _create_x509_extension(self, handlers, extension):
                 nid, 1 if extension.critical else 0, ext_struct
             )
 
-    def create_x509_revoked_certificate(self, builder):
+    def create_x509_revoked_certificate(
+        self, builder: x509.RevokedCertificateBuilder
+    ) -> _RevokedCertificate:
         if not isinstance(builder, x509.RevokedCertificateBuilder):
             raise TypeError("Builder type mismatch.")
 
@@ -1316,7 +1348,7 @@ def load_der_parameters(self, data):
 
         self._handle_key_loading_error()
 
-    def load_pem_x509_certificate(self, data):
+    def load_pem_x509_certificate(self, data: bytes) -> _Certificate:
         mem_bio = self._bytes_to_bio(data)
         x509 = self._lib.PEM_read_bio_X509(
             mem_bio.bio, self._ffi.NULL, self._ffi.NULL, self._ffi.NULL
@@ -1332,7 +1364,7 @@ def load_pem_x509_certificate(self, data):
         x509 = self._ffi.gc(x509, self._lib.X509_free)
         return _Certificate(self, x509)
 
-    def load_der_x509_certificate(self, data):
+    def load_der_x509_certificate(self, data: bytes) -> _Certificate:
         mem_bio = self._bytes_to_bio(data)
         x509 = self._lib.d2i_X509_bio(mem_bio.bio, self._ffi.NULL)
         if x509 == self._ffi.NULL:
@@ -1342,7 +1374,7 @@ def load_der_x509_certificate(self, data):
         x509 = self._ffi.gc(x509, self._lib.X509_free)
         return _Certificate(self, x509)
 
-    def load_pem_x509_crl(self, data):
+    def load_pem_x509_crl(self, data: bytes) -> _CertificateRevocationList:
         mem_bio = self._bytes_to_bio(data)
         x509_crl = self._lib.PEM_read_bio_X509_CRL(
             mem_bio.bio, self._ffi.NULL, self._ffi.NULL, self._ffi.NULL
@@ -1358,7 +1390,7 @@ def load_pem_x509_crl(self, data):
         x509_crl = self._ffi.gc(x509_crl, self._lib.X509_CRL_free)
         return _CertificateRevocationList(self, x509_crl)
 
-    def load_der_x509_crl(self, data):
+    def load_der_x509_crl(self, data: bytes) -> _CertificateRevocationList:
         mem_bio = self._bytes_to_bio(data)
         x509_crl = self._lib.d2i_X509_CRL_bio(mem_bio.bio, self._ffi.NULL)
         if x509_crl == self._ffi.NULL:
@@ -1368,7 +1400,7 @@ def load_der_x509_crl(self, data):
         x509_crl = self._ffi.gc(x509_crl, self._lib.X509_CRL_free)
         return _CertificateRevocationList(self, x509_crl)
 
-    def load_pem_x509_csr(self, data):
+    def load_pem_x509_csr(self, data: bytes) -> _CertificateSigningRequest:
         mem_bio = self._bytes_to_bio(data)
         x509_req = self._lib.PEM_read_bio_X509_REQ(
             mem_bio.bio, self._ffi.NULL, self._ffi.NULL, self._ffi.NULL
@@ -1384,7 +1416,7 @@ def load_pem_x509_csr(self, data):
         x509_req = self._ffi.gc(x509_req, self._lib.X509_REQ_free)
         return _CertificateSigningRequest(self, x509_req)
 
-    def load_der_x509_csr(self, data):
+    def load_der_x509_csr(self, data: bytes) -> _CertificateSigningRequest:
         mem_bio = self._bytes_to_bio(data)
         x509_req = self._lib.d2i_X509_REQ_bio(mem_bio.bio, self._ffi.NULL)
         if x509_req == self._ffi.NULL:
@@ -2200,7 +2232,7 @@ def dh_parameters_supported(self, p, g, q=None):
     def dh_x942_serialization_supported(self):
         return self._lib.Cryptography_HAS_EVP_PKEY_DHX == 1
 
-    def x509_name_bytes(self, name):
+    def x509_name_bytes(self, name: Name) -> bytes:
         x509_name = _encode_name_gc(self, name)
         pp = self._ffi.new("unsigned char **")
         res = self._lib.i2d_X509_NAME(x509_name, pp)
diff --git a/src/cryptography/hazmat/backends/openssl/x509.py b/src/cryptography/hazmat/backends/openssl/x509.py
@@ -204,8 +204,7 @@ def extensions(self) -> x509.Extensions:
         )
 
 
-@utils.register_interface(x509.CertificateRevocationList)
-class _CertificateRevocationList(object):
+class _CertificateRevocationList(x509.CertificateRevocationList):
     def __init__(self, backend, x509_crl):
         self._backend = backend
         self._x509_crl = x509_crl
@@ -385,8 +384,7 @@ def is_signature_valid(self, public_key: _PUBLIC_KEY_TYPES) -> bool:
         return True
 
 
-@utils.register_interface(x509.CertificateSigningRequest)
-class _CertificateSigningRequest(object):
+class _CertificateSigningRequest(x509.CertificateSigningRequest):
     def __init__(self, backend, x509_req):
         self._backend = backend
         self._x509_req = x509_req
diff --git a/tests/hazmat/backends/test_openssl.py b/tests/hazmat/backends/test_openssl.py
@@ -478,7 +478,20 @@ def test_requires_certificate_builder(self):
 
         with pytest.raises(TypeError):
             backend.create_x509_certificate(
-                object(), private_key, DummyHashAlgorithm()
+                object(),  # type: ignore[arg-type]
+                private_key,
+                DummyHashAlgorithm(),
+            )
+
+    def test_builder_requires_public_key(self):
+        builder = x509.CertificateBuilder()
+        private_key = RSA_KEY_2048.private_key(backend)
+
+        with pytest.raises(TypeError):
+            backend.create_x509_certificate(
+                builder,
+                private_key,
+                DummyHashAlgorithm(),
             )
 
 
@@ -488,7 +501,9 @@ def test_requires_csr_builder(self):
 
         with pytest.raises(TypeError):
             backend.create_x509_csr(
-                object(), private_key, DummyHashAlgorithm()
+                object(),  # type: ignore[arg-type]
+                private_key,
+                DummyHashAlgorithm(),
             )
 
 
@@ -497,13 +512,19 @@ def test_invalid_builder(self):
         private_key = RSA_KEY_2048.private_key(backend)
 
         with pytest.raises(TypeError):
-            backend.create_x509_crl(object(), private_key, hashes.SHA256())
+            backend.create_x509_crl(
+                object(),  # type: ignore[arg-type]
+                private_key,
+                hashes.SHA256(),
+            )
 
 
 class TestOpenSSLCreateRevokedCertificate(object):
     def test_invalid_builder(self):
         with pytest.raises(TypeError):
-            backend.create_x509_revoked_certificate(object())
+            backend.create_x509_revoked_certificate(
+                object()  # type: ignore[arg-type]
+            )
 
 
 class TestOpenSSLSerializationWithOpenSSL(object):
