From: Kefu Chai Date: Sat, 13 Jun 2026 02:23:39 +0000 (+0800) Subject: python-common/cryptotools: reimplement on top of cryptography X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=d814088ce7d0297e134bb0f3894a11c53d7a554a;p=ceph.git python-common/cryptotools: reimplement on top of cryptography the last change dropped X509Req, but InternalCryptoCaller still uses pyOpenSSL, which keeps deprecating and removing its OpenSSL.crypto and OpenSSL.SSL APIs. pyOpenSSL's own README [1] says: If you are using pyOpenSSL for anything other than making a TLS connection you should move to cryptography and drop your pyOpenSSL dependency. none of this module makes a TLS connection, and we already depend on cryptography and use it in cephadm/ssl_cert_utils.py. so reimplement the caller with cryptography: - create_private_key(): rsa.generate_private_key() dumped as PKCS8 PEM, the same format pyOpenSSL produced. - create_self_signed_cert(): x509.CertificateBuilder, with the dname fields mapped to their NameOIDs. - _load_cert() and get_cert_issuer_info(): load_pem_x509_certificate() and cert.issuer. - certificate_days_to_expire(): cert.not_valid_after. - verify_tls(): compare the cert's and the key's public keys instead of loading them into an SSL.Context. password_hash() and verify_password() already used bcrypt, so they are left as is. the output formats and error behavior are unchanged, hence the existing tls tests pass without changes. [1] https://github.com/pyca/pyopenssl/blob/main/README.rst Signed-off-by: Kefu Chai --- diff --git a/src/mypy.ini b/src/mypy.ini index 9fe2f82a535..0be1f430484 100755 --- a/src/mypy.ini +++ b/src/mypy.ini @@ -34,6 +34,9 @@ ignore_errors = True [mypy-OpenSSL] ignore_missing_imports = True +[mypy-cryptography,cryptography.*] +ignore_missing_imports = True + [mypy-prettytable] ignore_missing_imports = True diff --git a/src/python-common/ceph/cryptotools/internal.py b/src/python-common/ceph/cryptotools/internal.py index db3d6a5c048..b5410f1177e 100644 --- a/src/python-common/ceph/cryptotools/internal.py +++ b/src/python-common/ceph/cryptotools/internal.py @@ -1,25 +1,43 @@ """Internal execution of cryptographic functions for the ceph mgr """ -from typing import Dict, Any, Tuple, Union +from typing import Any, Dict, NoReturn, Tuple, Union from uuid import uuid4 import datetime import warnings -from OpenSSL import crypto, SSL import bcrypt +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives.serialization import load_pem_private_key +from cryptography.x509.oid import NameOID from .caller import CryptoCaller, CryptoCallError +# the relative distinguished names accepted by create_self_signed_cert(), +# mapped to their x509 object identifiers. mgr_util validates the dname +# fields against the same set before we get here. +_RDN_OIDS = { + 'C': NameOID.COUNTRY_NAME, + 'ST': NameOID.STATE_OR_PROVINCE_NAME, + 'L': NameOID.LOCALITY_NAME, + 'O': NameOID.ORGANIZATION_NAME, + 'OU': NameOID.ORGANIZATIONAL_UNIT_NAME, + 'CN': NameOID.COMMON_NAME, + 'emailAddress': NameOID.EMAIL_ADDRESS, +} + + class InternalError(CryptoCallError): pass class InternalCryptoCaller(CryptoCaller): - def fail(self, msg: str) -> None: + def fail(self, msg: str) -> NoReturn: raise InternalError(msg) def password_hash(self, password: str, salt_password: str) -> str: @@ -36,93 +54,100 @@ class InternalCryptoCaller(CryptoCaller): return ok def create_private_key(self) -> str: - pkey = crypto.PKey() - pkey.generate_key(crypto.TYPE_RSA, 2048) - return crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey).decode() + pkey = rsa.generate_private_key(public_exponent=65537, key_size=2048) + # PKCS8 PEM, the format pyOpenSSL's dump_privatekey() produced + return pkey.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode() def create_self_signed_cert( self, dname: Dict[str, str], pkey: str ) -> str: - _pkey = crypto.load_privatekey(crypto.FILETYPE_PEM, pkey) + _pkey: Any = load_pem_private_key(pkey.encode(), password=None) - # create a self-signed cert and populate its subject with the dname - # settings - cert = crypto.X509() - subj = cert.get_subject() + # build the subject, which is also the issuer as the cert is + # self-signed, from the dname settings + attrs = [] for k, v in dname.items(): - setattr(subj, k, v) - cert.set_subject(subj) - cert.set_serial_number(int(uuid4())) - cert.gmtime_adj_notBefore(0) - cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60) # 10 years - cert.set_issuer(cert.get_subject()) - cert.set_pubkey(_pkey) - cert.sign(_pkey, 'sha512') - return crypto.dump_certificate(crypto.FILETYPE_PEM, cert).decode() - - def _load_cert(self, crt: Union[str, bytes]) -> Any: + oid = _RDN_OIDS.get(k) + if oid is None: + self.fail('unsupported certificate subject field: %s' % k) + attrs.append(x509.NameAttribute(oid, v)) + name = x509.Name(attrs) + + now = datetime.datetime.now(datetime.timezone.utc) + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(_pkey.public_key()) + .serial_number(int(uuid4())) + .not_valid_before(now) + .not_valid_after(now + datetime.timedelta(days=10 * 365)) + .sign(_pkey, hashes.SHA512()) + ) + return cert.public_bytes(serialization.Encoding.PEM).decode() + + def _load_cert(self, crt: Union[str, bytes]) -> x509.Certificate: crt_buffer = crt.encode() if isinstance(crt, str) else crt try: - cert = crypto.load_certificate(crypto.FILETYPE_PEM, crt_buffer) - except (ValueError, crypto.Error) as e: + cert = x509.load_pem_x509_certificate(crt_buffer) + except ValueError as e: self.fail('Invalid certificate: %s' % str(e)) return cert - def _issuer_info(self, cert: Any) -> Tuple[str, str]: - components = cert.get_issuer().get_components() - org_name = cn = '' - for c in components: - if c[0].decode() == 'O': # org comp - org_name = c[1].decode() - elif c[0].decode() == 'CN': # common name comp - cn = c[1].decode() + def _name_value(self, name: x509.Name, oid: x509.ObjectIdentifier) -> str: + attrs = name.get_attributes_for_oid(oid) + if not attrs: + return '' + # cryptography types NameAttribute.value as str in some releases and + # str | bytes in others; keep it Any so the bytes path type-checks + # either way (like the other cryptography objects in this file). + value: Any = attrs[0].value + return value if isinstance(value, str) else value.decode('utf-8') + + def _issuer_info(self, cert: x509.Certificate) -> Tuple[str, str]: + org_name = self._name_value(cert.issuer, NameOID.ORGANIZATION_NAME) + cn = self._name_value(cert.issuer, NameOID.COMMON_NAME) return (org_name, cn) def certificate_days_to_expire(self, crt: str) -> int: - x509 = self._load_cert(crt) - no_after = x509.get_notAfter() - if not no_after: - self.fail("Certificate does not have an expiration date.") - - end_date = datetime.datetime.strptime( - no_after.decode(), '%Y%m%d%H%M%SZ' - ) + cert = self._load_cert(crt) + # not_valid_after is naive UTC; not_valid_after_utc only exists since + # cryptography 42, so stick with the portable accessor and silence + # its deprecation warning, same as for utcnow(). + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + not_after = cert.not_valid_after + now = datetime.datetime.utcnow() - if x509.has_expired(): - org, cn = self._issuer_info(x509) - msg = 'Certificate issued by "%s/%s" expired on %s' % ( - org, - cn, - end_date, + if now > not_after: + org, cn = self._issuer_info(cert) + self.fail( + 'Certificate issued by "%s/%s" expired on %s' + % (org, cn, not_after) ) - self.fail(msg) - # Certificate still valid, calculate and return days until expiration - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - days_until_exp = (end_date - datetime.datetime.utcnow()).days - return int(days_until_exp) + return int((not_after - now).days) def get_cert_issuer_info(self, crt: str) -> Tuple[str, str]: return self._issuer_info(self._load_cert(crt)) def verify_tls(self, crt: str, key: str) -> None: try: - _key = crypto.load_privatekey(crypto.FILETYPE_PEM, key) - _key.check() - except (ValueError, crypto.Error) as e: + _key: Any = load_pem_private_key(key.encode(), password=None) + except (ValueError, TypeError) as e: self.fail('Invalid private key: %s' % str(e)) - _crt = self._load_cert(crt) - try: - context = SSL.Context(SSL.TLSv1_METHOD) - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - context.use_certificate(_crt) - context.use_privatekey(_key) - context.check_privatekey() - except crypto.Error as e: + _crt: Any = self._load_cert(crt) + # the cert and key match iff they share the same public key + pem = serialization.Encoding.PEM + spki = serialization.PublicFormat.SubjectPublicKeyInfo + key_pub = _key.public_key().public_bytes(pem, spki) + crt_pub = _crt.public_key().public_bytes(pem, spki) + if key_pub != crt_pub: self.fail( - 'Private key and certificate do not match up: %s' % str(e) + 'Invalid cert/key pair: ' + 'private key and certificate do not match up' ) - except SSL.Error as e: - self.fail(f'Invalid cert/key pair: {e}')