x509_certificate: check existing certificate's signature for selfsigned and ownca provider (#407)

* Verify whether signature matches.

* Add changelog fragment.

* Forgot imports.

* Fix wrong name.

* Check whether the CA private key fits to the CA certificate. Use correct key in tests.

* Refactor code.

plugins/module_utils/crypto/cryptography_support.py

@@ -32,13 +32,36 @@ from ansible_collections.community.crypto.plugins.module_utils.version import Lo
 try:
     import cryptography
     from cryptography import x509
+    from cryptography.exceptions import InvalidSignature
     from cryptography.hazmat.backends import default_backend
     from cryptography.hazmat.primitives import serialization
+    from cryptography.hazmat.primitives.asymmetric import padding
     import ipaddress
 except ImportError:
     # Error handled in the calling module.
     pass
 
+try:
+    import cryptography.hazmat.primitives.asymmetric.rsa
+except ImportError:
+    pass
+try:
+    import cryptography.hazmat.primitives.asymmetric.ec
+except ImportError:
+    pass
+try:
+    import cryptography.hazmat.primitives.asymmetric.dsa
+except ImportError:
+    pass
+try:
+    import cryptography.hazmat.primitives.asymmetric.ed25519
+except ImportError:
+    pass
+try:
+    import cryptography.hazmat.primitives.asymmetric.ed448
+except ImportError:
+    pass
+
 try:
     # This is a separate try/except since this is only present in cryptography 2.5 or newer
     from cryptography.hazmat.primitives.serialization.pkcs12 import (
@@ -58,8 +81,13 @@ except ImportError:
     _load_pkcs12 = None
 
 from .basic import (
+    CRYPTOGRAPHY_HAS_DSA_SIGN,
+    CRYPTOGRAPHY_HAS_EC_SIGN,
     CRYPTOGRAPHY_HAS_ED25519,
+    CRYPTOGRAPHY_HAS_ED25519_SIGN,
     CRYPTOGRAPHY_HAS_ED448,
+    CRYPTOGRAPHY_HAS_ED448_SIGN,
+    CRYPTOGRAPHY_HAS_RSA_SIGN,
     CRYPTOGRAPHY_HAS_X25519,
     CRYPTOGRAPHY_HAS_X25519_FULL,
     CRYPTOGRAPHY_HAS_X448,
@@ -664,3 +692,40 @@ def _parse_pkcs12_legacy(pkcs12_bytes, passphrase=None):
         if maybe_name != backend._ffi.NULL:
             friendly_name = backend._ffi.string(maybe_name)
     return private_key, certificate, additional_certificates, friendly_name
+
+
+def cryptography_verify_signature(signature, data, hash_algorithm, signer_public_key):
+    '''
+    Check whether the given signature of the given data was signed by the given public key object.
+    '''
+    try:
+        if CRYPTOGRAPHY_HAS_RSA_SIGN and isinstance(signer_public_key, cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey):
+            signer_public_key.verify(signature, data, padding.PKCS1v15(), hash_algorithm)
+            return True
+        if CRYPTOGRAPHY_HAS_EC_SIGN and isinstance(signer_public_key, cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePublicKey):
+            signer_public_key.verify(signature, data, cryptography.hazmat.primitives.asymmetric.ec.ECDSA(hash_algorithm))
+            return True
+        if CRYPTOGRAPHY_HAS_DSA_SIGN and isinstance(signer_public_key, cryptography.hazmat.primitives.asymmetric.dsa.DSAPublicKey):
+            signer_public_key.verify(signature, data, hash_algorithm)
+            return True
+        if CRYPTOGRAPHY_HAS_ED25519_SIGN and isinstance(signer_public_key, cryptography.hazmat.primitives.asymmetric.ed25519.Ed25519PublicKey):
+            signer_public_key.verify(signature, data)
+            return True
+        if CRYPTOGRAPHY_HAS_ED448_SIGN and isinstance(signer_public_key, cryptography.hazmat.primitives.asymmetric.ed448.Ed448PublicKey):
+            signer_public_key.verify(signature, data)
+            return True
+        raise OpenSSLObjectError(u'Unsupported public key type {0}'.format(type(signer_public_key)))
+    except InvalidSignature:
+        return False
+
+
+def cryptography_verify_certificate_signature(certificate, signer_public_key):
+    '''
+    Check whether the given X509 certificate object was signed by the given public key object.
+    '''
+    return cryptography_verify_signature(
+        certificate.signature,
+        certificate.tbs_certificate_bytes,
+        certificate.signature_hash_algorithm,
+        signer_public_key
+    )

plugins/module_utils/crypto/module_backends/certificate_ownca.py

@@ -28,8 +28,10 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.support im
 )
 
 from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
+    cryptography_compare_public_keys,
     cryptography_key_needs_digest_for_signing,
     cryptography_serial_number_of_cert,
+    cryptography_verify_certificate_signature,
 )
 
 from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import (
@@ -107,6 +109,9 @@ class OwnCACertificateBackendCryptography(CertificateBackend):
         except OpenSSLBadPassphraseError as exc:
             module.fail_json(msg=str(exc))
 
+        if not cryptography_compare_public_keys(self.ca_cert.public_key(), self.ca_private_key.public_key()):
+            raise CertificateError('The CA private key does not belong to the CA certificate')
+
         if cryptography_key_needs_digest_for_signing(self.ca_private_key):
             if self.digest is None:
                 raise CertificateError(
@@ -175,6 +180,10 @@ class OwnCACertificateBackendCryptography(CertificateBackend):
 
         self._ensure_existing_certificate_loaded()
 
+        # Check whether certificate is signed by CA certificate
+        if not cryptography_verify_certificate_signature(self.existing_certificate, self.ca_cert.public_key()):
+            return True
+
         # Check subject
         if self.ca_cert.subject != self.existing_certificate.issuer:
             return True

plugins/module_utils/crypto/module_backends/certificate_selfsigned.py

@@ -22,6 +22,7 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.support im
 from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
     cryptography_key_needs_digest_for_signing,
     cryptography_serial_number_of_cert,
+    cryptography_verify_certificate_signature,
 )
 
 from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import (
@@ -135,8 +136,16 @@ class SelfSignedCertificateBackendCryptography(CertificateBackend):
         return self.cert.public_bytes(Encoding.PEM)
 
     def needs_regeneration(self):
-        return super(SelfSignedCertificateBackendCryptography, self).needs_regeneration(
-            not_before=self.notBefore, not_after=self.notAfter)
+        if super(SelfSignedCertificateBackendCryptography, self).needs_regeneration(not_before=self.notBefore, not_after=self.notAfter):
+            return True
+
+        self._ensure_existing_certificate_loaded()
+
+        # Check whether certificate is signed by private key
+        if not cryptography_verify_certificate_signature(self.existing_certificate, self.privatekey.public_key()):
+            return True
+
+        return False
 
     def dump(self, include_certificate):
         result = super(SelfSignedCertificateBackendCryptography, self).dump(include_certificate)