removing duplicate ssl related code from ssl_cert_utils

Signed-off-by: Redouane Kachach <rkachach@redhat.com>

diff --git a/src/pybind/mgr/cephadm/ssl_cert_utils.py b/src/pybind/mgr/cephadm/ssl_cert_utils.py
index 8063cd64c4d2e2fb6bf88a6b1a04fba86d99e8da..fcc6f00eab9d3aed38fbf4e3cf49e79b26d356b9 100644 (file)
--- a/src/pybind/mgr/cephadm/ssl_cert_utils.py
+++ b/src/pybind/mgr/cephadm/ssl_cert_utils.py
@@ -2,7 +2,6 @@
 from typing import Any, Tuple, IO
 import ipaddress
 import tempfile
-import os
 import logging
 
 from datetime import datetime, timedelta
@@ -11,7 +10,7 @@ from cryptography.x509.oid import NameOID
 from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.backends import default_backend
-from OpenSSL import crypto, SSL
+from mgr_util import verify_tls_files
 
 from orchestrator import OrchestratorError
 
@@ -155,95 +154,3 @@ class SSLCerts:
         self.root_cert = given_cert
         self.root_key = serialization.load_pem_private_key(
             data=priv_key.encode('utf-8'), backend=default_backend(), password=None)
-
-
-def verify_tls(crt, key):
-    # type: (str, str) -> None
-    verify_cacrt_content(crt)
-
-    try:
-        _key = crypto.load_privatekey(crypto.FILETYPE_PEM, key)
-        _key.check()
-    except (ValueError, crypto.Error) as e:
-        raise SSLConfigException('Invalid private key: {}'.format(str(e)))
-    try:
-        _crt = crypto.load_certificate(crypto.FILETYPE_PEM, crt)
-    except ValueError as e:
-        raise SSLConfigException('Invalid certificate key: {}'.format(str(e))
-                                 )
-
-    try:
-        context = SSL.Context(SSL.TLSv1_METHOD)
-        context.use_certificate(_crt)
-        context.use_privatekey(_key)
-        context.check_privatekey()
-    except crypto.Error as e:
-        logger.warning(f'Private key and certificate do not match up: {e}')
-
-
-def verify_tls_files(cert_fname, pkey_fname):
-    # type: (str, str) -> None
-    """Basic checks for TLS certificate and key files
-
-    Do some validations to the private key and certificate:
-    - Check the type and format
-    - Check the certificate expiration date
-    - Check the consistency of the private key
-    - Check that the private key and certificate match up
-
-    :param cert_fname: Name of the certificate file
-    :param pkey_fname: name of the certificate public key file
-
-    :raises SSLConfigException: An error with a message
-
-    """
-    if not cert_fname or not pkey_fname:
-        raise SSLConfigException('no certificate configured')
-
-    verify_cacrt(cert_fname)
-
-    if not os.path.isfile(pkey_fname):
-        raise SSLConfigException('private key %s does not exist' % pkey_fname)
-
-    try:
-        with open(pkey_fname) as f:
-            pkey = crypto.load_privatekey(crypto.FILETYPE_PEM, f.read())
-            pkey.check()
-    except (ValueError, crypto.Error) as e:
-        raise SSLConfigException(
-            'Invalid private key {}: {}'.format(pkey_fname, str(e)))
-    try:
-        context = SSL.Context(SSL.TLSv1_METHOD)
-        context.use_certificate_file(cert_fname, crypto.FILETYPE_PEM)
-        context.use_privatekey_file(pkey_fname, crypto.FILETYPE_PEM)
-        context.check_privatekey()
-    except crypto.Error as e:
-        logger.warning(
-            f'Private key {pkey_fname} and certificate {cert_fname} do not match up: {e}')
-
-
-def verify_cacrt(cert_fname):
-    # type: (str) -> None
-    """Basic validation of a ca cert"""
-
-    if not cert_fname:
-        raise SSLConfigException("CA cert not configured")
-    if not os.path.isfile(cert_fname):
-        raise SSLConfigException("Certificate {} does not exist".format(cert_fname))
-
-    try:
-        with open(cert_fname) as f:
-            verify_cacrt_content(f.read())
-    except ValueError as e:
-        raise SSLConfigException(
-            'Invalid certificate {}: {}'.format(cert_fname, str(e)))
-
-
-def verify_cacrt_content(crt):
-    # type: (str) -> None
-    try:
-        x509 = crypto.load_certificate(crypto.FILETYPE_PEM, crt)
-        if x509.has_expired():
-            logger.warning(f'Certificate has expired: {crt}')
-    except (ValueError, crypto.Error) as e:
-        raise SSLConfigException(f'Invalid certificate: {e}')

diff --git a/src/pybind/mgr/mgr_util.py b/src/pybind/mgr/mgr_util.py
index ba938401175aab17c3220df0a19ee822a89dd82e..a499b49c51e2e11d410f2d90c35a8e89c36980d2 100644 (file)
--- a/src/pybind/mgr/mgr_util.py
+++ b/src/pybind/mgr/mgr_util.py
@@ -417,6 +417,7 @@ def test_port_allocation(addr: str, port: int) -> None:
     """Checks if the port is available
     :raises PortAlreadyInUse: in case port is already in use
     :raises Exception: any generic error other than port already in use
+    If no exception is raised, the port can be assumed available
     """
     try:
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)