From 809be53dc8b46a60d99401e40b9625871662b12c Mon Sep 17 00:00:00 2001 From: Simon Pichugin Date: Jun 06 2018 13:07:05 +0000 Subject: Issue 49646 - Improve TLS cert processing in lib389 CLI Description: Add a new option to setup.inf file - self_sign_cert_valid_months which accepts int value. dscreate should renew Self-signed CA if it is expired or it will expire less than in 2 months. Also, we need to import it to other existing instances. Remove the NSS DB in the test topology during a teardown. Fix small issues in nss_tls_test. Add format_cmd_list function to utils.py. It returns a nicely formatted quoted representation of the command list we put into subprocess call. Add more logging to nss_tls.py. Change the default validity period to 24 months. https://pagure.io/389-ds-base/issue/49646 Reviewed by: mreynolds, tbordaz, mhonek (Thanks!) --- diff --git a/src/lib389/lib389/instance/options.py b/src/lib389/lib389/instance/options.py index 91fcec6..1dee6f5 100644 --- a/src/lib389/lib389/instance/options.py +++ b/src/lib389/lib389/instance/options.py @@ -192,6 +192,11 @@ class Slapd2Base(Options2): self._helptext['self_sign_cert'] = "Issue a self signed certificate during the setup process. This is not suitable for production TLS, but aids simplifying setup of TLS (you only need to replace a certificate instead)" self._example_comment['self_sign_cert'] = True + self._options['self_sign_cert_valid_months'] = 24 + self._type['self_sign_cert_valid_months'] = int + self._helptext['self_sign_cert_valid_months'] = "Set a number of months for which the self signed cert will be issued." + self._example_comment['self_sign_cert_valid_months'] = True + # In the future, make bin and sbin /usr/[s]bin, but we may need autotools assistance from Ds self._options['bin_dir'] = ds_paths.bin_dir self._type['bin_dir'] = str diff --git a/src/lib389/lib389/instance/setup.py b/src/lib389/lib389/instance/setup.py index 4b64b66..d3f56cb 100644 --- a/src/lib389/lib389/instance/setup.py +++ b/src/lib389/lib389/instance/setup.py @@ -430,19 +430,28 @@ class SetupDs(object): # Does this work? assert_c(ds_instance.exists(), "Instance failed to install, does not exist when expected") - # Create a certificate database. tlsdb = NssSsl(dbpath=slapd['cert_dir']) if not tlsdb._db_exists(): tlsdb.reinit() if slapd['self_sign_cert']: - # If it doesn't exist, create a cadb. - ssca_path = os.path.join(slapd['sysconf_dir'], 'dirsrv/ssca/') + etc_dirsrv_path = os.path.join(slapd['sysconf_dir'], 'dirsrv/') + ssca_path = os.path.join(etc_dirsrv_path, 'ssca/') ssca = NssSsl(dbpath=ssca_path) + # If it doesn't exist, create a CA DB if not ssca._db_exists(): ssca.reinit() - ssca.create_rsa_ca() + ssca.create_rsa_ca(months=slapd['self_sign_cert_valid_months']) + # If CA is expired or will expire soon, + # Reissue it and resign the existing certs that were signed by the cert previously + elif ssca.rsa_ca_needs_renew(): + ca = ssca.renew_rsa_ca(months=slapd['self_sign_cert_valid_months']) + # Import CA to the existing instances except the one we install now (we import it later) + for dir in os.listdir(etc_dirsrv_path): + if dir.startswith("slapd-") and dir != slapd['cert_dir']: + tlsdb_inst = NssSsl(dbpath=os.path.join(etc_dirsrv_path, dir)) + tlsdb_inst.import_rsa_crt(ca) csr = tlsdb.create_rsa_key_and_csr() (ca, crt) = ssca.rsa_ca_sign_csr(csr) diff --git a/src/lib389/lib389/nss_ssl.py b/src/lib389/lib389/nss_ssl.py index 26c8f70..bd1f81c 100644 --- a/src/lib389/lib389/nss_ssl.py +++ b/src/lib389/lib389/nss_ssl.py @@ -20,10 +20,11 @@ import shutil import logging # from nss import nss import subprocess +from datetime import datetime, timedelta from subprocess import check_output from lib389.passwd import password_generate -from lib389.utils import ensure_str, ensure_bytes +from lib389.utils import ensure_str, ensure_bytes, format_cmd_list import uuid KEYBITS = 4096 @@ -36,11 +37,13 @@ CERT_SUFFIX = 'O=testing,L=389ds,ST=Queensland,C=AU' ISSUER = 'CN=ssca.389ds.example.com,%s' % CERT_SUFFIX SELF_ISSUER = 'CN={HOSTNAME},givenName={GIVENNAME},%s' % CERT_SUFFIX USER_ISSUER = 'CN={HOSTNAME},%s' % CERT_SUFFIX -VALID = 2 +VALID = 24 +VALID_MIN = 61 # Days # My logger log = logging.getLogger(__name__) + class NssSsl(object): def __init__(self, dirsrv=None, dbpassword=None, dbpath=None): self.dirsrv = dirsrv @@ -55,6 +58,10 @@ class NssSsl(object): else: self.dbpassword = dbpassword + self.db_files = {"dbm_backend": ["%s/%s" % (self._certdb, f) for f in ("key3.db", "cert8.db", "secmod.db")], + "sql_backend": ["%s/%s" % (self._certdb, f) for f in ("key4.db", "cert9.db", "pkcs11.txt")], + "support": ["%s/%s" % (self._certdb, f) for f in ("noise.txt", "pin.txt", "pwdfile.txt")]} + def detect_alt_names(self, alt_names=[]): """Attempt to determine appropriate subject alternate names for a host. Returns the list of names we derive. @@ -100,18 +107,12 @@ class NssSsl(object): with open(fpath, 'w') as f: f.write(noise) - def reinit(self): """ Re-init (create) the nss db. """ # 48886: The DB that DS ships with is .... well, broken. Purge it! - for f in ('key3.db', 'cert8.db', 'key4.db', 'cert9.db', 'secmod.db', 'pkcs11.txt'): - try: - # Perhaps we should be backing these up instead ... - os.remove("%s/%s" % (self._certdb, f )) - except: - pass + assert self.remove_db() try: os.makedirs(self._certdb) @@ -132,27 +133,39 @@ class NssSsl(object): # 48886; This needs to be sql format ... cmd = ['/usr/bin/certutil', '-N', '-d', self._certdb, '-f', '%s/%s' % (self._certdb, PWD_TXT)] self._generate_noise('%s/noise.txt' % self._certdb) - self.log.debug("nss cmd: %s" % cmd) + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) result = ensure_str(check_output(cmd, stderr=subprocess.STDOUT)) self.log.debug("nss output: %s" % result) return True def _db_exists(self): - """ - Check that a nss db exists at the certpath - """ - key3 = os.path.exists("%s/key3.db" % (self._certdb)) - cert8 = os.path.exists("%s/cert8.db" % (self._certdb)) - key4 = os.path.exists("%s/key4.db" % (self._certdb)) - cert9 = os.path.exists("%s/cert9.db" % (self._certdb)) - secmod = os.path.exists("%s/secmod.db" % (self._certdb)) - pkcs11 = os.path.exists("%s/pkcs11.txt" % (self._certdb)) - - if ((key3 and cert8 and secmod) or (key4 and cert9 and pkcs11)): + """Check that a nss db exists at the certpath""" + + if all(map(os.path.exists, self.db_files["dbm_backend"])) or \ + all(map(os.path.exists, self.db_files["sql_backend"])): return True return False - def create_rsa_ca(self): + def remove_db(self): + """Remove nss db files at the certpath""" + + files = self.db_files["dbm_backend"] + \ + self.db_files["sql_backend"] + \ + self.db_files["support"] + + for file in files: + try: + os.remove(file) + except FileNotFoundError: + pass + + if os.path.isdir(self._certdb) and not os.listdir(self._certdb): + os.removedirs(self._certdb) + + assert not self._db_exists() + return True + + def create_rsa_ca(self, months=VALID): """ Create a self signed CA. """ @@ -175,7 +188,7 @@ class NssSsl(object): '-t', 'CT,,', '-v', - '%s' % VALID, + '%s' % months, '--keyUsage', 'certSigning', '-d', @@ -185,7 +198,7 @@ class NssSsl(object): '-f', '%s/%s' % (self._certdb, PWD_TXT), ] - self.log.debug("nss cmd: %s" % cmd) + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) result = ensure_str(check_output(cmd, stderr=subprocess.STDOUT)) self.log.debug("nss output: %s" % result) # Now extract the CAcert to a well know place. @@ -199,13 +212,112 @@ class NssSsl(object): self._certdb, '-a', ] - self.log.debug("nss cmd: %s" % cmd) + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) certdetails = check_output(cmd, stderr=subprocess.STDOUT) with open('%s/ca.crt' % self._certdb, 'w') as f: f.write(ensure_str(certdetails)) - check_output(['/usr/bin/c_rehash', self._certdb], stderr=subprocess.STDOUT) + cmd = ['/usr/bin/c_rehash', self._certdb] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) return True + def rsa_ca_needs_renew(self): + """Check is our self signed CA is expired or + will expire less than a minimum period of time (VALID_MIN) + """ + + cmd = [ + '/usr/bin/certutil', + '-L', + '-n', + CA_NAME, + '-d', + self._certdb, + ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + certdetails = check_output(cmd, stderr=subprocess.STDOUT, encoding='utf-8') + end_date_str = certdetails.split("Not After : ")[1].split("\n")[0] + date_format = '%a %b %d %H:%M:%S %Y' + end_date = datetime.strptime(end_date_str, date_format) + + if end_date - datetime.now() < timedelta(days=VALID_MIN): + return True + else: + return False + + def renew_rsa_ca(self, months=VALID): + """Renew the self signed CA.""" + + csr_path = os.path.join(self._certdb, 'CA_renew.csr') + crt_path = '%s/ca.crt' % self._certdb + + # Create noise. + self._generate_noise('%s/noise.txt' % self._certdb) + + # Generate a CSR for a new CA cert + cmd = [ + '/usr/bin/certutil', + '-R', + '-s', + ISSUER, + '-g', + '%s' % KEYBITS, + '-k', + 'NSS Certificate DB:%s' % CA_NAME, + '-d', + self._certdb, + '-z', + '%s/noise.txt' % self._certdb, + '-f', + '%s/%s' % (self._certdb, PWD_TXT), + '-a', + '-o', csr_path, + ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) + + # Sign the CSR with our old CA + cmd = [ + '/usr/bin/certutil', + '-C', + '-d', + self._certdb, + '-f', + '%s/%s' % (self._certdb, PWD_TXT), + '-a', + '-i', csr_path, + '-o', crt_path, + '-c', CA_NAME, + '--keyUsage', + 'certSigning', + '-t', + 'CT,,', + '-v', + '%s' % months, + ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) + + cmd = ['/usr/bin/c_rehash', self._certdb] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) + + # Import the new CA to our DB instead of the old CA + cmd = [ + '/usr/bin/certutil', + '-A', + '-n', CA_NAME, + '-t', "CT,,", + '-a', + '-i', crt_path, + '-d', self._certdb, + '-f', '%s/%s' % (self._certdb, PWD_TXT), + ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) + + return crt_path + def _rsa_cert_list(self): cmd = [ '/usr/bin/certutil', @@ -242,6 +354,7 @@ class NssSsl(object): '-f', '%s/%s' % (self._certdb, PWD_TXT), ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) result = ensure_str(check_output(cmd, stderr=subprocess.STDOUT)) lines = result.split('\n')[1:-1] @@ -304,8 +417,7 @@ class NssSsl(object): have_user = True return have_user - - def create_rsa_key_and_cert(self, alt_names=[]): + def create_rsa_key_and_cert(self, alt_names=[], months=VALID): """ Create a key and a cert that is signed by the self signed ca @@ -336,7 +448,7 @@ class NssSsl(object): '-t', ',,', '-v', - '%s' % VALID, + '%s' % months, '-d', self._certdb, '-z', @@ -344,8 +456,7 @@ class NssSsl(object): '-f', '%s/%s' % (self._certdb, PWD_TXT), ] - - self.log.debug("nss cmd: %s" % cmd) + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) result = ensure_str(check_output(cmd, stderr=subprocess.STDOUT)) self.log.debug("nss output: %s" % result) return True @@ -381,8 +492,6 @@ class NssSsl(object): '-8', ','.join(alt_names), '-g', '%s' % KEYBITS, - '-v', - '%s' % VALID, '-d', self._certdb, '-z', @@ -392,71 +501,90 @@ class NssSsl(object): '-a', '-o', csr_path, ] - - self.log.debug("nss cmd: %s" % cmd) + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) check_output(cmd, stderr=subprocess.STDOUT) return csr_path - def rsa_ca_sign_csr(self, csr_path): + def rsa_ca_sign_csr(self, csr_path, months=VALID): """ Given a CSR, sign it with our CA certificate (if present). This emits a signed certificate which can be imported with import_rsa_crt. """ crt_path = 'crt'.join(csr_path.rsplit('csr', 1)) ca_path = '%s/ca.crt' % self._certdb - check_output([ + cmd = [ '/usr/bin/certutil', '-C', '-d', self._certdb, '-f', '%s/%s' % (self._certdb, PWD_TXT), + '-v', + '%s' % months, '-a', '-i', csr_path, '-o', crt_path, '-c', CA_NAME, - ], stderr=subprocess.STDOUT) + ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) return (ca_path, crt_path) - def import_rsa_crt(self, ca, crt): + def import_rsa_crt(self, ca=None, crt=None): """Given a signed certificate from a ca, import the CA and certificate to our database. + + """ - shutil.copyfile(ca, '%s/ca.crt' % self._certdb) - check_output(['/usr/bin/c_rehash', self._certdb], stderr=subprocess.STDOUT) - check_output([ - '/usr/bin/certutil', - '-A', - '-n', CA_NAME, - '-t', "CT,,", - '-a', - '-i', '%s/ca.crt' % self._certdb, - '-d', self._certdb, - '-f', - '%s/%s' % (self._certdb, PWD_TXT), - ], stderr=subprocess.STDOUT) - check_output([ - '/usr/bin/certutil', - '-A', - '-n', CERT_NAME, - '-t', ",,", - '-a', - '-i', crt, - '-d', self._certdb, - '-f', - '%s/%s' % (self._certdb, PWD_TXT), - ], stderr=subprocess.STDOUT) - check_output([ - '/usr/bin/certutil', - '-V', - '-d', self._certdb, - '-n', CERT_NAME, - '-u', 'YCV' - ], stderr=subprocess.STDOUT) - def create_rsa_user(self, name): + assert ca is not None or crt is not None, "At least one parameter should be specified (ca or crt)" + + if ca is not None: + shutil.copyfile(ca, '%s/ca.crt' % self._certdb) + cmd = ['/usr/bin/c_rehash', self._certdb] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) + cmd = [ + '/usr/bin/certutil', + '-A', + '-n', CA_NAME, + '-t', "CT,,", + '-a', + '-i', '%s/ca.crt' % self._certdb, + '-d', self._certdb, + '-f', + '%s/%s' % (self._certdb, PWD_TXT), + ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) + + if crt is not None: + cmd = [ + '/usr/bin/certutil', + '-A', + '-n', CERT_NAME, + '-t', ",,", + '-a', + '-i', crt, + '-d', self._certdb, + '-f', + '%s/%s' % (self._certdb, PWD_TXT), + ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) + cmd = [ + '/usr/bin/certutil', + '-V', + '-d', self._certdb, + '-n', CERT_NAME, + '-u', 'YCV' + ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) + + def create_rsa_user(self, name, months=VALID): """ Create a key and cert for a user to authenticate to the directory. @@ -488,7 +616,7 @@ class NssSsl(object): '-t', ',,', '-v', - '%s' % VALID, + '%s' % months, '-d', self._certdb, '-z', @@ -496,22 +624,25 @@ class NssSsl(object): '-f', '%s/%s' % (self._certdb, PWD_TXT), ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) result = ensure_str(check_output(cmd, stderr=subprocess.STDOUT)) self.log.debug("nss output: %s" % result) # Now extract this into PEM files that we can use. # pk12util -o user-william.p12 -d . -k pwdfile.txt -n user-william -W '' - check_output([ + cmd = [ 'pk12util', '-d', self._certdb, '-o', '%s/%s%s.p12' % (self._certdb, USER_PREFIX, name), '-k', '%s/%s' % (self._certdb, PWD_TXT), '-n', '%s%s' % (USER_PREFIX, name), '-W', '""' - ], stderr=subprocess.STDOUT) + ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) # openssl pkcs12 -in user-william.p12 -passin pass:'' -out file.pem -nocerts -nodes # Extract the key - check_output([ + cmd = [ 'openssl', 'pkcs12', '-in', '%s/%s%s.p12' % (self._certdb, USER_PREFIX, name), @@ -519,9 +650,11 @@ class NssSsl(object): '-out', '%s/%s%s.key' % (self._certdb, USER_PREFIX, name), '-nocerts', '-nodes' - ], stderr=subprocess.STDOUT) + ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) # Extract the cert - check_output([ + cmd = [ 'openssl', 'pkcs12', '-in', '%s/%s%s.p12' % (self._certdb, USER_PREFIX, name), @@ -530,16 +663,20 @@ class NssSsl(object): '-nokeys', '-clcerts', '-nodes' - ], stderr=subprocess.STDOUT) + ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) # Convert the cert for userCertificate attr - check_output([ + cmd = [ 'openssl', 'x509', '-inform', 'PEM', '-outform', 'DER', '-in', '%s/%s%s.crt' % (self._certdb, USER_PREFIX, name), '-out', '%s/%s%s.der' % (self._certdb, USER_PREFIX, name), - ], stderr=subprocess.STDOUT) + ] + self.log.debug("nss cmd: %s" % format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) return subject diff --git a/src/lib389/lib389/tests/nss_ssl_test.py b/src/lib389/lib389/tests/nss_ssl_test.py index 4741da5..0fbbfd3 100644 --- a/src/lib389/lib389/tests/nss_ssl_test.py +++ b/src/lib389/lib389/tests/nss_ssl_test.py @@ -25,8 +25,9 @@ else: log = logging.getLogger(__name__) + def test_external_ca(): - """ Test the behaviour of our system ca database. + """Test the behaviour of our system ca database. :id: 321c85c1-9cf3-413f-9e26-99a8b327509c @@ -54,9 +55,9 @@ def test_external_ca(): (ca, crt) = ssca.rsa_ca_sign_csr(csr) tlsdb.import_rsa_crt(ca, crt) + def test_nss_ssca_users(topo): - """ - Validate that we can submit user certs to the ds ca for signing. + """Validate that we can submit user certs to the ds ca for signing. :id: a47e47ed-2056-440b-8797-d13fa89098f6 :steps: @@ -68,11 +69,13 @@ def test_nss_ssca_users(topo): 2. It works. 3. It works. """ + ssca = NssSsl(dbpath=topo.standalone.get_ssca_dir()) - if not ssca._rsa_ca_exists(): + if not ssca._db_exists(): ssca.reinit() - ssca.create_rsa_ca() + if not ssca._rsa_ca_exists(): + ssca.create_rsa_ca() # It better exist now! assert(ssca._rsa_ca_exists() is True) @@ -80,9 +83,10 @@ def test_nss_ssca_users(topo): # Check making users certs. They should never conflict for user in ('william', 'noriko', 'mark'): # Create the user cert - assert(ssca.create_rsa_user(user) is True) + assert(ssca.create_rsa_user(user) is not None) # Assert it exists now assert(ssca._rsa_user_exists(user) is True) + assert(ssca._rsa_user_exists('non_existen') is False) if __name__ == "__main__": diff --git a/src/lib389/lib389/topologies.py b/src/lib389/lib389/topologies.py index 372d26e..e550643 100644 --- a/src/lib389/lib389/topologies.py +++ b/src/lib389/lib389/topologies.py @@ -19,6 +19,7 @@ from lib389.utils import generate_ds_params from lib389.mit_krb5 import MitKrb5 from lib389.saslmap import SaslMappings from lib389.replica import ReplicationManager, Replicas +from lib389.nss_ssl import NssSsl from lib389._constants import * DEBUGGING = os.getenv('DEBUGGING', default=False) @@ -29,6 +30,14 @@ else: log = logging.getLogger(__name__) +def _remove_ssca_db(topology): + ssca = NssSsl(dbpath=topology[0].get_ssca_dir()) + if ssca._db_exists(): + return ssca.remove_db() + else: + return True + + def _create_instances(topo_dict, suffix): """Create requested instances without replication or any other modifications @@ -235,6 +244,7 @@ def topology_st(request): if DEBUGGING: topology.standalone.stop() else: + assert _remove_ssca_db(topology) if topology.standalone.exists(): topology.standalone.delete() request.addfinalizer(fin) @@ -301,6 +311,7 @@ def topology_st_gssapi(request): if DEBUGGING: topology.standalone.stop() else: + assert _remove_ssca_db(topology) if topology.standalone.exists(): topology.standalone.delete() krb.destroy_realm() @@ -320,6 +331,7 @@ def topology_i2(request): if DEBUGGING: [inst.stop() for inst in topology] else: + assert _remove_ssca_db(topology) [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) @@ -336,6 +348,7 @@ def topology_i3(request): if DEBUGGING: [inst.stop() for inst in topology] else: + assert _remove_ssca_db(topology) [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) @@ -351,6 +364,7 @@ def topology_m1(request): if DEBUGGING: [inst.stop() for inst in topology] else: + assert _remove_ssca_db(topology) [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) @@ -367,6 +381,7 @@ def topology_m1c1(request): if DEBUGGING: [inst.stop() for inst in topology] else: + assert _remove_ssca_db(topology) [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) @@ -383,6 +398,7 @@ def topology_m2(request): if DEBUGGING: [inst.stop() for inst in topology] else: + assert _remove_ssca_db(topology) [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) @@ -399,6 +415,7 @@ def topology_m3(request): if DEBUGGING: [inst.stop() for inst in topology] else: + assert _remove_ssca_db(topology) [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) @@ -415,6 +432,7 @@ def topology_m4(request): if DEBUGGING: [inst.stop() for inst in topology] else: + assert _remove_ssca_db(topology) [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) @@ -432,6 +450,7 @@ def topology_m2c2(request): if DEBUGGING: [inst.stop() for inst in topology] else: + assert _remove_ssca_db(topology) [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) @@ -467,6 +486,7 @@ def topology_m1h1c1(request): if DEBUGGING: [inst.stop() for inst in topology] else: + assert _remove_ssca_db(topology) [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) diff --git a/src/lib389/lib389/utils.py b/src/lib389/lib389/utils.py index d2f5f44..9bccee8 100644 --- a/src/lib389/lib389/utils.py +++ b/src/lib389/lib389/utils.py @@ -33,6 +33,7 @@ import time import sys import filecmp import six +import shlex from socket import getfqdn from ldapurl import LDAPUrl from contextlib import closing @@ -987,6 +988,7 @@ def pseudolocalize(string): pseudo_string += char return pseudo_string + def assert_c(condition, msg="Assertion Failed"): """This is the same as assert, but assert is compiled out when optimisation is enabled. This prevents compiling out. @@ -994,3 +996,8 @@ def assert_c(condition, msg="Assertion Failed"): if not condition: raise AssertionError(msg) + +def format_cmd_list(cmd): + """Format the subprocess command list to the quoted shell string""" + + return " ".join(map(shlex.quote, cmd))