From 83d41432e14dd6fdbaf3bc173e8b9dfcc6e363fe Mon Sep 17 00:00:00 2001 From: William Brown Date: Nov 07 2019 00:44:33 +0000 Subject: Ticket 50007, 50648 - improve x509 handling. Bug Description: certutil and nssdb have fallen from favour and are really not easy to use. Most admins are used to PEM files and much easier TLS management. Fix Description: Add some basic and simple wrappers to support a set of common operations such as listing TLS certs, making a CSR and importing PEM files - including from services like let's encrypt https://pagure.io/389-ds-base/issue/50007 https://pagure.io/389-ds-base/pull-request/50648 Author: William Brown Review by: spichugi, mreynolds (Thanks!) --- diff --git a/src/lib389/cli/dsctl b/src/lib389/cli/dsctl index 3be9540..7dd4b8d 100755 --- a/src/lib389/cli/dsctl +++ b/src/lib389/cli/dsctl @@ -16,12 +16,13 @@ import sys import signal import os from lib389.utils import get_instance_list -from lib389.cli_base import _get_arg +from lib389.cli_base import _get_arg, setup_script_logger, disconnect_instance from lib389 import DirSrv from lib389.cli_ctl import instance as cli_instance from lib389.cli_ctl import dbtasks as cli_dbtasks from lib389.cli_base import disconnect_instance, setup_script_logger from lib389.cli_base import format_error_to_dict +from lib389.cli_ctl import tls as cli_tls from lib389.cli_ctl.instance import instance_remove_all from lib389._constants import DSRC_CONTAINER @@ -52,6 +53,7 @@ subparsers = parser.add_subparsers(help="action") if not os.path.exists(DSRC_CONTAINER): cli_instance.create_parser(subparsers) cli_dbtasks.create_parser(subparsers) +cli_tls.create_parser(subparsers) argcomplete.autocomplete(parser) diff --git a/src/lib389/lib389/cli_ctl/tls.py b/src/lib389/lib389/cli_ctl/tls.py new file mode 100644 index 0000000..ffa5625 --- /dev/null +++ b/src/lib389/lib389/cli_ctl/tls.py @@ -0,0 +1,144 @@ + +from lib389.nss_ssl import NssSsl, CERT_NAME, CA_NAME +from lib389.cli_base import _warn + +def show_servercert(inst, log, args): + tls = NssSsl(dirsrv=inst) + log.info(tls.display_cert_details(CERT_NAME)) + +def list_client_cas(inst, log, args): + tls = NssSsl(dirsrv=inst) + # This turns an array of [('CA', 'C,,')] + for c in tls.list_client_ca_certs(): + log.info(c[0]) + +def list_cas(inst, log, args): + tls = NssSsl(dirsrv=inst) + # This turns an array of [('CA', 'C,,')] + for c in tls.list_ca_certs(): + log.info(c[0]) + +def show_cert(inst, log, args): + tls = NssSsl(dirsrv=inst) + nickname = args.nickname + log.info(tls.display_cert_details(nickname)) + +def import_client_ca(inst, log, args): + tls = NssSsl(dirsrv=inst) + cert_path = args.cert_path + nickname = args.nickname + if nickname.lower() == CERT_NAME.lower() or nickname.lower() == CA_NAME.lower(): + log.error("You may not import a CA with the nickname %s or %s" % (CERT_NAME, CA_NAME)) + return + tls.add_cert(nickname=nickname, input_file=cert_path) + tls.edit_cert_trust(nickname, "T,,") + +def import_ca(inst, log, args): + tls = NssSsl(dirsrv=inst) + cert_path = args.cert_path + nickname = args.nickname + if nickname.lower() == CERT_NAME.lower() or nickname.lower() == CA_NAME.lower(): + log.error("You may not import a CA with the nickname %s or %s" % (CERT_NAME, CA_NAME)) + return + tls.add_cert(nickname=nickname, input_file=cert_path) + tls.edit_cert_trust(nickname, "C,,") + +def import_key_cert_pair(inst, log, args): + tls = NssSsl(dirsrv=inst) + key_path = args.key_path + cert_path = args.cert_path + tls.add_server_key_and_cert(key_path, cert_path) + +def generate_key_csr(inst, log, args): + tls = NssSsl(dirsrv=inst) + alt_names = args.alt_names + subject = args.subject + out_path = tls.create_rsa_key_and_csr(alt_names, subject) + log.info(out_path) + +def import_server_cert(inst, log, args): + tls = NssSsl(dirsrv=inst) + cert_path = args.cert_path + tls.import_rsa_crt(crt=cert_path) + +def remove_cert(inst, log, args, warn=True): + tls = NssSsl(dirsrv=inst) + nickname = args.nickname + if warn: + _warn(nickname, msg="Deleting certificate %s" % nickname) + tls.del_cert(nickname) + + +def create_parser(subparsers): + tls_parser = subparsers.add_parser('tls', help="Manage TLS certificates") + + subcommands = tls_parser.add_subparsers(help='action') + + list_ca_parser = subcommands.add_parser('list-ca', help='list server certificate authorities including intermediates') + list_ca_parser.set_defaults(func=list_cas) + + list_client_ca_parser = subcommands.add_parser('list-client-ca', help='list client certificate authorities including intermediates') + list_client_ca_parser.set_defaults(func=list_client_cas) + + show_servercert_parser = subcommands.add_parser('show-server-cert', help='Show the active server certificate that clients will see and verify') + show_servercert_parser.set_defaults(func=show_servercert) + + show_cert_parser = subcommands.add_parser('show-cert', help='Show a certificate\'s details referenced by it\'s nickname. This is analogous to certutil -L -d -n ') + show_cert_parser.add_argument('nickname', help="The nickname (friendly name) of the certificate to display") + show_cert_parser.set_defaults(func=show_cert) + + generate_server_cert_csr_parser = subcommands.add_parser( + 'generate-server-cert-csr', + help="Generate a Server-Cert certificate signing request - the csr is then submitted to a CA for verification, and when signed you import with import-ca and import-server-cert" + ) + generate_server_cert_csr_parser.add_argument('--subject', '-s', + default=None, + help="Certificate Subject field to use") + generate_server_cert_csr_parser.add_argument('alt_names', nargs='*', + help="Certificate requests subject alternative names. These are auto-detected if not provided") + generate_server_cert_csr_parser.set_defaults(func=generate_key_csr) + + import_client_ca_parser = subcommands.add_parser( + 'import-client-ca', + help="Import a CA trusted to issue user (client) certificates. This is part of how client certificate authentication functions." + ) + import_client_ca_parser.add_argument('cert_path', + help="The path to the x509 cert to import as a client trust root") + import_client_ca_parser.add_argument('nickname', help="The name of the certificate once imported") + import_client_ca_parser.set_defaults(func=import_client_ca) + + import_ca_parser = subcommands.add_parser( + 'import-ca', + help="Import a CA or intermediate CA for signing this servers certificates (aka Server-Cert). You should import all the CA's in the chain as required." + ) + import_ca_parser.add_argument('cert_path', + help="The path to the x509 cert to import as a server CA") + import_ca_parser.add_argument('nickname', help="The name of the certificate once imported") + import_ca_parser.set_defaults(func=import_ca) + + import_server_cert_parser = subcommands.add_parser( + 'import-server-cert', + help="Import a new Server-Cert after the csr has been signed from a CA." + ) + import_server_cert_parser.add_argument('cert_path', + help="The path to the x509 cert to import as Server-Cert") + import_server_cert_parser.set_defaults(func=import_server_cert) + + import_server_key_cert_parser = subcommands.add_parser( + 'import-server-key-cert', + help="Import a new key and Server-Cert after having been signed from a CA. This is used if you have an external csr tool or a service like lets encrypt that generates PEM keys externally." + ) + import_server_key_cert_parser.add_argument('cert_path', + help="The path to the x509 cert to import as Server-Cert") + import_server_key_cert_parser.add_argument('key_path', + help="The path to the x509 key to import associated to Server-Cert") + import_server_key_cert_parser.set_defaults(func=import_key_cert_pair) + + remove_cert_parser = subcommands.add_parser( + 'remove-cert', + help="Delete a certificate from this database. This will remove it from acting as a CA, a client CA or the Server-Cert role." + ) + remove_cert_parser.add_argument('nickname', help="The name of the certificate to delete") + remove_cert_parser.set_defaults(func=remove_cert) + + diff --git a/src/lib389/lib389/nss_ssl.py b/src/lib389/lib389/nss_ssl.py index 8af7132..afe9213 100644 --- a/src/lib389/lib389/nss_ssl.py +++ b/src/lib389/lib389/nss_ssl.py @@ -34,7 +34,7 @@ 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 = 24 +VALID = 24 # Months VALID_MIN = 61 # Days # My logger @@ -68,9 +68,9 @@ class NssSsl(object): :returns: list[str] """ if self.dirsrv and self.dirsrv.host not in alt_names: - alt_names.append(self.dirsrv.host) + alt_names.append(ensure_str(self.dirsrv.host)) if len(alt_names) == 0: - alt_names.append(socket.gethostname()) + alt_names.append(ensure_str(socket.gethostname())) return alt_names def generate_cert_subject(self, alt_names=[]): @@ -389,6 +389,11 @@ only. (ssl_flag, mime_flag, jar_flag) = trust_flags.split(',') return 'C' in ssl_flag + def _rsa_cert_is_caclienttrust(self, cert_tuple): + trust_flags = cert_tuple[1] + (ssl_flag, mime_flag, jar_flag) = trust_flags.split(',') + return 'T' in ssl_flag + def _rsa_cert_is_user(self, cert_tuple): """ Check an RSA cert is user trust @@ -480,15 +485,20 @@ only. self.log.debug("nss output: %s", result) return True - def create_rsa_key_and_csr(self, alt_names=[]): + def create_rsa_key_and_csr(self, alt_names=[], subject=None): """Create a new RSA key and the certificate signing request. This request can be submitted to a CA for signing. The returned certifcate can be added with import_rsa_crt. """ csr_path = os.path.join(self._certdb, '%s.csr' % CERT_NAME) - alt_names = self.detect_alt_names(alt_names) - subject = self.generate_cert_subject(alt_names) + if len(alt_names) == 0: + alt_names = self.detect_alt_names(alt_names) + if subject is None: + subject = self.generate_cert_subject(alt_names) + + self.log.debug(f"CSR subject -> {subject}") + self.log.debug(f"CSR alt_names -> {alt_names}") # Wait a second to avoid an NSS bug with serial ids based on time. time.sleep(1) @@ -755,6 +765,19 @@ only. check_output(cmd, stderr=subprocess.STDOUT) + def display_cert_details(self, nickname): + cmd = [ + '/usr/bin/certutil', + '-d', self._certdb, + '-n', nickname, + '-L', + '-f', + '%s/%s' % (self._certdb, PWD_TXT), + ] + self.log.debug("display_cert_details cmd: %s", format_cmd_list(cmd)) + return check_output(cmd, stderr=subprocess.STDOUT, encoding='utf-8') + + def get_cert_details(self, nickname): """Get the trust flags, subject DN, issuer, and expiration date @@ -769,18 +792,7 @@ only. for cert in all_certs: if cert[0] == nickname: trust_flags = cert[1] - cmd = [ - '/usr/bin/certutil', - '-d', self._certdb, - '-n', nickname, - '-L', - '-f', - '%s/%s' % (self._certdb, PWD_TXT), - ] - self.log.debug("get_cert_details cmd: %s", format_cmd_list(cmd)) - - # Expiration date - certdetails = check_output(cmd, stderr=subprocess.STDOUT, encoding='utf-8') + certdetails = self.display_cert_details(nickname) 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) @@ -817,7 +829,6 @@ only. # Did not find cert with that name raise ValueError("Certificate '{}' not found in NSS database".format(nickname)) - def list_certs(self, ca=False): all_certs = self._rsa_cert_list() certs = [] @@ -827,6 +838,20 @@ only. certs.append(self.get_cert_details(cert[0])) return certs + def list_ca_certs(self): + return [ + cert + for cert in self._rsa_cert_list() + if self._rsa_cert_is_catrust(cert) + ] + + def list_client_ca_certs(self): + return [ + cert + for cert in self._rsa_cert_list() + if self._rsa_cert_is_caclienttrust(cert) + ] + def add_cert(self, nickname, input_file, ca=False): """Add server or CA cert @@ -854,3 +879,55 @@ only. ] self.log.debug("add_cert cmd: %s", format_cmd_list(cmd)) check_output(cmd, stderr=subprocess.STDOUT) + + def add_server_key_and_cert(self, input_key, input_cert): + if not os.path.exists(input_key): + raise ValueError("The key file ({}) does not exist".format(input_key)) + if not os.path.exists(input_cert): + raise ValueError("The cert file ({}) does not exist".format(input_cert)) + + self.log.debug(f"Importing key and cert -> {input_key}, {input_cert}") + + p12_bundle = "%s/temp_server_key_cert.p12" % self._certdb + + # Remove the p12 if it exists + if os.path.exists(p12_bundle): + os.remove(p12_bundle) + + # Transform to p12 + cmd = [ + 'openssl', + 'pkcs12', + '-export', + '-in', input_cert, + '-inkey', input_key, + '-out', p12_bundle, + '-name', CERT_NAME, + '-passout', 'pass:', + '-aes128' + ] + self.log.debug("nss cmd: %s", format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) + # Remove the server-cert if it exists, because else the import name fails. + try: + self.del_cert(CERT_NAME) + except: + pass + try: + # Import it + cmd = [ + 'pk12util', + '-v', + '-i', p12_bundle, + '-d', self._certdb, + '-k', '%s/%s' % (self._certdb, PWD_TXT), + '-W', "", + ] + self.log.debug("nss cmd: %s", format_cmd_list(cmd)) + check_output(cmd, stderr=subprocess.STDOUT) + finally: + # Remove the p12 + if os.path.exists(p12_bundle): + os.remove(p12_bundle) + +