summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--base/common/python/pki/nssdb.py231
-rw-r--r--base/server/python/pki/server/cli/subsystem.py265
2 files changed, 443 insertions, 53 deletions
diff --git a/base/common/python/pki/nssdb.py b/base/common/python/pki/nssdb.py
index cad89081e..3c0ac0682 100644
--- a/base/common/python/pki/nssdb.py
+++ b/base/common/python/pki/nssdb.py
@@ -1,5 +1,6 @@
# Authors:
# Endi S. Dewata <edewata@redhat.com>
+# Dinesh Prasnath M K <dmoluguw@redhat.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the Lesser GNU General Public License as published by
@@ -294,7 +295,6 @@ class NSSDatabase(object):
exts = []
for generic_ext in generic_exts:
-
data_file = os.path.join(tmpdir, 'csr-ext-%d' % counter)
with open(data_file, 'w') as f:
f.write(generic_ext['data'])
@@ -339,95 +339,224 @@ class NSSDatabase(object):
finally:
shutil.rmtree(tmpdir)
- def create_self_signed_ca_cert(self, subject_dn, request_file, cert_file,
- serial='1', validity=240):
-
+ def create_cert(self, request_file, cert_file, serial, issuer=None,
+ key_usage_ext=None, basic_constraints_ext=None,
+ aki_ext=None, ski_ext=None, aia_ext=None, ext_key_usage_ext=None,
+ validity=None):
cmd = [
'certutil',
'-C',
- '-x',
'-d', self.directory
]
+ # Check if it's self signed
+ if issuer:
+ cmd.extend(['-c', issuer])
+ else:
+ cmd.extend('-x')
+
if self.token:
cmd.extend(['-h', self.token])
cmd.extend([
'-f', self.password_file,
- '-c', subject_dn,
'-a',
'-i', request_file,
'-o', cert_file,
- '-m', serial,
- '-v', str(validity),
- '--keyUsage', 'digitalSignature,nonRepudiation,certSigning,crlSigning,critical',
- '-2',
- '-3',
- '--extSKID',
- '--extAIA'
+ '-m', serial
])
- p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
- stderr=subprocess.STDOUT)
+ if validity:
+ cmd.extend(['-v', str(validity)])
keystroke = ''
- # Is this a CA certificate [y/N]?
- keystroke += 'y\n'
+ if aki_ext:
+ cmd.extend(['-3'])
+
+ # Enter value for the authKeyID extension [y/N]
+ if 'auth_key_id' in aki_ext:
+ keystroke += 'y\n'
+
+ # Enter value for the key identifier fields,enter to omit:
+ keystroke += aki_ext['auth_key_id']
+
+ keystroke += '\n'
+
+ # Select one of the following general name type:
+ # TODO: General Name type isn't used as of now for AKI
+ keystroke += '0\n'
+
+ if 'auth_cert_serial' in aki_ext:
+ keystroke += aki_ext['auth_cert_serial']
+
+ keystroke += '\n'
+
+ # Is this a critical extension [y/N]?
+ if 'critical' in aki_ext and aki_ext['critical']:
+ keystroke += 'y'
+
+ keystroke += '\n'
+
+ # Key Usage Constraints
+ if key_usage_ext:
+
+ cmd.extend(['--keyUsage'])
+
+ usages = []
+ for usage in key_usage_ext:
+ if key_usage_ext[usage]:
+ usages.append(usage)
+
+ cmd.extend([','.join(usages)])
- # Enter the path length constraint, enter to skip [<0 for unlimited path]:
- keystroke += '\n'
+ # Extended key usage
+ if ext_key_usage_ext:
+ cmd.extend(['--extKeyUsage'])
+ usages = []
+ for usage in ext_key_usage_ext:
+ if ext_key_usage_ext[usage]:
+ usages.append(usage)
- # Is this a critical extension [y/N]?
- keystroke += 'y\n'
+ cmd.extend([','.join(usages)])
- # Enter value for the authKeyID extension [y/N]?
- keystroke += 'y\n'
+ # Basic constraints
+ if basic_constraints_ext:
- # TODO: generate SHA1 ID (see APolicyRule.formSHA1KeyId())
- # Enter value for the key identifier fields,enter to omit:
- keystroke += '2d:7e:83:37:75:5a:fd:0e:8d:52:a3:70:16:93:36:b8:4a:d6:84:9f\n'
+ cmd.extend(['-2'])
- # Select one of the following general name type:
- keystroke += '0\n'
+ # Is this a CA certificate [y/N]?
+ if basic_constraints_ext['ca']:
+ keystroke += 'y'
- # Enter value for the authCertSerial field, enter to omit:
- keystroke += '\n'
+ keystroke += '\n'
- # Is this a critical extension [y/N]?
- keystroke += '\n'
+ # Enter the path length constraint, enter to skip [<0 for unlimited path]:
+ if basic_constraints_ext['path_length']:
+ keystroke += basic_constraints_ext['path_length']
- # TODO: generate SHA1 ID (see APolicyRule.formSHA1KeyId())
- # Adding Subject Key ID extension.
- # Enter value for the key identifier fields,enter to omit:
- keystroke += '2d:7e:83:37:75:5a:fd:0e:8d:52:a3:70:16:93:36:b8:4a:d6:84:9f\n'
+ keystroke += '\n'
- # Is this a critical extension [y/N]?
- keystroke += '\n'
+ # Is this a critical extension [y/N]?
+ if basic_constraints_ext['critical']:
+ keystroke += 'y'
- # Enter access method type for Authority Information Access extension:
- keystroke += '2\n'
+ keystroke += '\n'
- # Select one of the following general name type:
- keystroke += '7\n'
+ if ski_ext:
+ cmd.extend(['--extSKID'])
- # TODO: replace with actual hostname name and port number
- # Enter data:
- keystroke += 'http://server.example.com:8080/ca/ocsp\n'
+ # Adding Subject Key ID extension.
+ # Enter value for the key identifier fields,enter to omit:
+ if 'sk_id' in ski_ext:
+ keystroke += ski_ext['sk_id']
- # Select one of the following general name type:
- keystroke += '0\n'
+ keystroke += '\n'
- # Add another location to the Authority Information Access extension [y/N]
- keystroke += '\n'
+ # Is this a critical extension [y/N]?
+ if 'critical' in ski_ext and ski_ext['critical']:
+ keystroke += 'y'
- # Is this a critical extension [y/N]?
- keystroke += '\n'
+ keystroke += '\n'
+
+ if aia_ext:
+ cmd.extend(['--extAIA'])
+
+ # To ensure whether this is the first AIA being added
+ firstentry = True
+
+ # Enter access method type for Authority Information Access extension:
+ for s in aia_ext:
+ if not firstentry:
+ keystroke += 'y\n'
+
+ # 1. CA Issuers
+ if s == 'ca_issuers':
+ keystroke += '1'
+
+ # 2. OCSP
+ if s == 'ocsp':
+ keystroke += '2'
+ keystroke += '\n'
+ for gn in aia_ext[s]['uri']:
+ # 7. URI
+ keystroke += '7\n'
+ # Enter data
+ keystroke += gn + '\n'
+
+ # Any other number to finish
+ keystroke += '0\n'
+
+ # One entry is done.
+ firstentry = False
+
+ # Add another location to the Authority Information Access extension [y/N]
+ keystroke += '\n'
+
+ # Is this a critical extension [y/N]?
+ if 'critical' in aia_ext and aia_ext['critical']:
+ keystroke += 'y'
+
+ keystroke += '\n'
+
+ p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT)
p.communicate(keystroke)
rc = p.wait()
+ return rc
+
+ def create_self_signed_ca_cert(self, request_file, cert_file,
+ serial='1', validity=240):
+
+ # --keyUsage
+ key_usage_ext = {
+ 'digitalSignature': True,
+ 'nonRepudiation': True,
+ 'certSigning': True,
+ 'crlSigning': True,
+ 'critical': True
+ }
+
+ # -2
+ basic_constraints_ext = {
+ 'path_length': None
+ }
+
+ # FIXME: do not hard-code AKI extension
+ # -3
+ aki_ext = {
+ 'auth_key_id': '0x2d7e8337755afd0e8d52a370169336b84ad6849f'
+ }
+
+ # FIXME: do not hard-code SKI extension
+ # --extSKID
+ ski_ext = {
+ 'sk_id': '0x2d7e8337755afd0e8d52a370169336b84ad6849f'
+ }
+
+ # FIXME: do not hard-code AIA extension
+ # --extAIA
+ aia_ext = {
+ 'ocsp': {
+ 'uri': ['http://server.example.com:8080/ca/ocsp']
+ }
+
+ }
+
+ rc = self.create_cert(
+ request_file=request_file,
+ cert_file=cert_file,
+ serial=serial,
+ validity=validity,
+ key_usage_ext=key_usage_ext,
+ basic_constraints_ext=basic_constraints_ext,
+ aki_ext=aki_ext,
+ ski_ext=ski_ext,
+ aia_ext=aia_ext)
+
if rc:
raise Exception('Failed to generate self-signed CA certificate. RC: %d' % rc)
diff --git a/base/server/python/pki/server/cli/subsystem.py b/base/server/python/pki/server/cli/subsystem.py
index a9857ba5f..1381a6162 100644
--- a/base/server/python/pki/server/cli/subsystem.py
+++ b/base/server/python/pki/server/cli/subsystem.py
@@ -1,6 +1,7 @@
# Authors:
# Endi S. Dewata <edewata@redhat.com>
# Abhijeet Kasurde <akasurde@redhat.com>
+# Dinesh Prasanth M K <dmoluguw@redhat.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@@ -26,7 +27,10 @@ import getpass
import os
import subprocess
import sys
-from tempfile import mkstemp
+import random
+import tempfile
+import shutil
+import re
import pki.cli
import pki.nssdb
@@ -370,6 +374,7 @@ class SubsystemCertCLI(pki.cli.CLI):
self.add_module(SubsystemCertExportCLI())
self.add_module(SubsystemCertUpdateCLI())
self.add_module(SubsystemCertValidateCLI())
+ self.add_module(SubsystemCertRenewCLI())
@staticmethod
def print_subsystem_cert(cert, show_all=False):
@@ -1006,7 +1011,7 @@ class SubsystemCertValidateCLI(pki.cli.CLI):
# get internal token password and store in temporary file
passwd = instance.get_token_password()
- pwfile_handle, pwfile_path = mkstemp()
+ pwfile_handle, pwfile_path = tempfile.mkstemp()
os.write(pwfile_handle, passwd)
os.close(pwfile_handle)
@@ -1035,3 +1040,259 @@ class SubsystemCertValidateCLI(pki.cli.CLI):
finally:
os.unlink(pwfile_path)
+
+
+class SubsystemCertRenewCLI(pki.cli.CLI):
+ def __init__(self):
+ super(SubsystemCertRenewCLI, self).__init__(
+ 'renew', 'Renew subsystem certificate')
+
+ def usage(self):
+ print('Usage: pki-server subsystem-cert-renew [OPTIONS] <subsystem ID> <cert ID>')
+ print()
+ print(' -i, --instance <instance ID> Instance ID (default: pki-tomcat).')
+ print(' -v, --verbose Run in verbose mode.')
+ print(' --help Show help message.')
+ print(' --temp Create temporary certificate.')
+ print(' --serial <number> Provide serial number for temp certificate.')
+ print()
+
+ def execute(self, argv):
+ try:
+ opts, args = getopt.gnu_getopt(argv, 'i:v', [
+ 'instance=',
+ 'verbose',
+ 'help', 'temp', 'serial='])
+
+ except getopt.GetoptError as e:
+ print('ERROR: ' + str(e))
+ self.usage()
+ sys.exit(1)
+
+ instance_name = 'pki-tomcat'
+ is_permanent_cert = True
+ serial = None
+
+ for o, a in opts:
+ if o in ('-i', '--instance'):
+ instance_name = a
+
+ elif o in ('-v', '--verbose'):
+ self.set_verbose(True)
+
+ elif o == '--help':
+ self.usage()
+ sys.exit()
+
+ elif o == '--temp':
+ is_permanent_cert = False
+
+ elif o == '--serial':
+ serial = a
+
+ else:
+ self.print_message('ERROR: unknown option ' + o)
+ self.usage()
+ sys.exit(1)
+
+ if len(args) < 1:
+ print('ERROR: missing subsystem ID')
+ self.usage()
+ sys.exit(1)
+
+ if len(args) < 2:
+ print('ERROR: missing cert ID')
+ self.usage()
+ sys.exit(1)
+
+ subsystem_name = args[0]
+ cert_id = args[1]
+
+ instance = pki.server.PKIInstance(instance_name)
+
+ if not instance.is_valid():
+ print('ERROR: Invalid instance %s.' % instance_name)
+ sys.exit(1)
+
+ # Load the instance. Default: pki-tomcat
+ instance.load()
+
+ # Get the subsystem - Eg: ca, kra, tps, tks
+ subsystem = instance.get_subsystem(subsystem_name)
+ if not subsystem:
+ print('ERROR: No %s subsystem in instance '
+ '%s.' % (subsystem_name, instance_name))
+ sys.exit(1)
+
+ cert = subsystem.get_subsystem_cert(cert_id)
+
+ nssdb = instance.open_nssdb()
+ tmpdir = tempfile.mkdtemp()
+
+ try:
+ new_cert_file = os.path.join(tmpdir, cert_id + '.crt')
+
+ # Check if the request is for permanent certificate creation
+ if is_permanent_cert:
+ # Serial number for permanent certificate must be auto-generated
+ if serial:
+ raise Exception('--serial not allowed for permanent cert')
+ # Fixme: Get the serial from LDAP DB (Method 3a)
+ else:
+ if not serial:
+ # Fixme: Get the highest serial number from NSS DB and add 1 (Method 2b)
+ # If admin doesn't provide a serial number, generate one
+ serial = str(random.randint(
+ int(subsystem.config.get('dbs.beginSerialNumber', '1')),
+ int(subsystem.config.get('dbs.endSerialNumber', '10000000'))))
+
+ if cert_id == 'sslserver':
+ self.renew_ssl_cert(subsystem=subsystem, is_permanent_cert=is_permanent_cert, tmpdir=tmpdir,
+ new_cert_file=new_cert_file, nssdb=nssdb, serial=serial)
+
+ elif cert_id == 'ca_ocsp_signing':
+ self.renew_ocsp_cert(is_permanent_cert=is_permanent_cert)
+
+ elif cert_id == 'ca_audit_signing':
+ self.renew_audit_cert(is_permanent_cert=is_permanent_cert)
+
+ elif cert_id == 'subsystem':
+ self.renew_subsystem_cert(is_permanent_cert=is_permanent_cert)
+
+ else:
+ # renewal not yet supported
+ raise Exception('Renewal for %s not yet supported.' % cert_id)
+
+ # Import cert into NSS db
+ if self.verbose:
+ print('Removing old %s certificate from NSS database.' % cert_id)
+ nssdb.remove_cert(cert['nickname'])
+
+ if self.verbose:
+ print('Adding new %s certificate into NSS database.' % cert_id)
+ nssdb.add_cert(
+ nickname=cert['nickname'],
+ cert_file=new_cert_file)
+
+ # Update CS.cfg with the new certificate
+ if self.verbose:
+ print('Updating CS.cfg')
+ data = nssdb.get_cert(
+ nickname=cert['nickname'],
+ output_format='base64')
+ cert['data'] = data
+ subsystem.update_subsystem_cert(cert)
+ subsystem.save()
+
+ finally:
+ nssdb.close()
+ shutil.rmtree(tmpdir)
+
+ @staticmethod
+ def setup_temp_renewal(subsystem, cert_id, tmpdir):
+
+ csr_file = os.path.join(tmpdir, cert_id + '.csr')
+ ca_cert_file = os.path.join(tmpdir, 'ca_certificate.crt')
+
+ # Export the CSR for the cert
+ cert_request = subsystem.get_subsystem_cert(cert_id).get('request', None)
+ if cert_request is None:
+ print("ERROR: Unable to find certificate request for %s" % cert_id)
+ sys.exit(1)
+
+ csr_data = pki.nssdb.convert_csr(cert_request, 'base64', 'pem')
+ with open(csr_file, 'w') as f:
+ f.write(csr_data)
+
+ # Extract SKI
+ # 1. Get the CA certificate
+ # 2. Then get the SKI from it
+ ca_signing_cert = subsystem.get_subsystem_cert('signing')
+ ca_cert_data = ca_signing_cert.get('data', None)
+ if ca_cert_data is None:
+ print("ERROR: Unable to find certificate data for CA signing certificate.")
+ sys.exit(1)
+
+ ca_cert = pki.nssdb.convert_cert(ca_cert_data, 'base64', 'pem')
+ with open(ca_cert_file, 'w') as f:
+ f.write(ca_cert)
+
+ ca_cert_retrieve_cmd = [
+ 'openssl',
+ 'x509',
+ '-in', ca_cert_file,
+ '-noout',
+ '-text'
+ ]
+
+ ca_cert_details = subprocess.check_output(ca_cert_retrieve_cmd)
+ aki = re.search(r'Subject Key Identifier.*\n.*?(.*?)\n', ca_cert_details).group(1)
+
+ # Add 0x to represent this is a Hex
+ aki = '0x' + aki.strip().replace(':', '')
+
+ return ca_signing_cert, aki, csr_file
+
+ def renew_ssl_cert(self, subsystem, serial, tmpdir, is_permanent_cert, new_cert_file, nssdb):
+ if self.verbose:
+ print('Creating SSL server certificate.')
+
+ if is_permanent_cert:
+ # TODO: Online renewal
+ print('SSL cert online renewal not yet supported.')
+ else:
+ # Generate temp SSL Certificate signed by CA
+
+ ca_signing_cert, aki, csr_file = self.setup_temp_renewal(
+ subsystem=subsystem, tmpdir=tmpdir, cert_id='sslserver')
+
+ # --keyUsage
+ key_usage_ext = {
+ 'digitalSignature': True,
+ 'nonRepudiation': True,
+ 'keyEncipherment': True,
+ 'dataEncipherment': True,
+ 'critical': True
+ }
+
+ # -3
+ aki_ext = {
+ 'auth_key_id': aki
+ }
+
+ # --extKeyUsage
+ ext_key_usage_ext = {
+ 'serverAuth': True
+ }
+
+ rc = nssdb.create_cert(
+ issuer=ca_signing_cert['nickname'],
+ request_file=csr_file,
+ cert_file=new_cert_file,
+ serial=serial,
+ key_usage_ext=key_usage_ext,
+ aki_ext=aki_ext,
+ ext_key_usage_ext=ext_key_usage_ext)
+ if rc:
+ raise Exception('Failed to generate CA-signed temp SSL certificate. RC: %d' % rc)
+
+ def renew_ocsp_cert(self, is_permanent_cert):
+ if is_permanent_cert:
+ # TODO: Online renewal
+ raise Exception('OCSP cert online renewal not yet supported.')
+ else:
+ raise Exception('Temp certificate for OCSP is not supported.')
+
+ def renew_subsystem_cert(self, is_permanent_cert):
+ if is_permanent_cert:
+ # TODO: Online renewal
+ raise Exception('Subsystem cert online renewal not yet supported.')
+ else:
+ raise Exception('Temp certificate for subsystem is not supported.')
+
+ def renew_audit_cert(self, is_permanent_cert):
+ if is_permanent_cert:
+ # TODO: Online renewal
+ raise Exception('Audit signing cert online renewal not yet supported.')
+ else:
+ raise Exception('Temp certificate for audit signing is not supported.')
generated by cgit (git 2.34.1) at 2024-04-24 04:00:21 +0000