From 83d41432e14dd6fdbaf3bc173e8b9dfcc6e363fe Mon Sep 17 00:00:00 2001
From: William Brown <william@blackhats.net.au>
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 <william@blackhats.net.au>

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 <path> -n <nickname>')
+    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)
+
+

