# Authors: Karl MacMillan # # Copyright (C) 2007 Red Hat # see file 'COPYING' for use and warranty information # # 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 the Free Software Foundation; version 2 only # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA # import os, stat, subprocess, re import sha import errno import tempfile import shutil from ipa import sysrestore from ipa import ipautil CA_SERIALNO="/var/lib/ipa/ca_serialno" class CertDB(object): def __init__(self, dir, fstore=None): self.secdir = dir self.noise_fname = self.secdir + "/noise.txt" self.passwd_fname = self.secdir + "/pwdfile.txt" self.certdb_fname = self.secdir + "/cert8.db" self.keydb_fname = self.secdir + "/key3.db" self.secmod_fname = self.secdir + "/secmod.db" self.cacert_fname = self.secdir + "/cacert.asc" self.pk12_fname = self.secdir + "/cacert.p12" self.pin_fname = self.secdir + "/pin.txt" self.reqdir = tempfile.mkdtemp('', 'ipa-', '/var/lib/ipa') self.certreq_fname = self.reqdir + "/tmpcertreq" self.certder_fname = self.reqdir + "/tmpcert.der" # Making this a starting value that will generate # unique values for the current DB is the # responsibility of the caller for now. In the # future we might automatically determine this # for a given db. self.cur_serial = -1 self.cacert_name = "CA certificate" self.valid_months = "120" self.keysize = "1024" # We are going to set the owner of all of the cert # files to the owner of the containing directory # instead of that of the process. This works when # this is called by root for a daemon that runs as # a normal user mode = os.stat(self.secdir) self.uid = mode[stat.ST_UID] self.gid = mode[stat.ST_GID] if fstore: self.fstore = fstore else: self.fstore = sysrestore.FileStore('/var/lib/ipa/sysrestore') def __del__(self): shutil.rmtree(self.reqdir, ignore_errors=True) def set_serial_from_pkcs12(self): """A CA cert was loaded from a PKCS#12 file. Set up our serial file""" self.cur_serial = self.find_cacert_serial() try: f=open(CA_SERIALNO,"w") f.write(str(self.cur_serial)) f.close() except IOError, e: raise RuntimeError("Unable to increment serial number: %s" % str(e)) def next_serial(self): try: f=open(CA_SERIALNO,"r") r = f.readline() try: self.cur_serial = int(r) + 1 except ValueError: raise RuntimeError("The value in %s is not an integer" % CA_SERIALNO) f.close() except IOError, e: if e.errno == errno.ENOENT: self.cur_serial = 1000 f=open(CA_SERIALNO,"w") f.write(str(self.cur_serial)) f.close() else: raise RuntimeError("Unable to determine serial number: %s" % str(e)) try: f=open(CA_SERIALNO,"w") f.write(str(self.cur_serial)) f.close() except IOError, e: raise RuntimeError("Unable to increment serial number: %s" % str(e)) return str(self.cur_serial) def set_perms(self, fname, write=False): os.chown(fname, self.uid, self.gid) perms = stat.S_IRUSR if write: perms |= stat.S_IWUSR os.chmod(fname, perms) def gen_password(self): return sha.sha(ipautil.ipa_generate_password()).hexdigest() def run_certutil(self, args, stdin=None): new_args = ["/usr/bin/certutil", "-d", self.secdir] new_args = new_args + args return ipautil.run(new_args, stdin) def run_signtool(self, args, stdin=None): new_args = ["/usr/bin/signtool", "-d", self.secdir] new_args = new_args + args ipautil.run(new_args, stdin) def create_noise_file(self): ipautil.backup_file(self.noise_fname) f = open(self.noise_fname, "w") f.write(self.gen_password()) self.set_perms(self.noise_fname) def create_passwd_file(self, passwd=None): ipautil.backup_file(self.passwd_fname) f = open(self.passwd_fname, "w") if passwd is not None: f.write("%s\n" % passwd) else: f.write(self.gen_password()) f.close() self.set_perms(self.passwd_fname) def create_certdbs(self): ipautil.backup_file(self.certdb_fname) ipautil.backup_file(self.keydb_fname) ipautil.backup_file(self.secmod_fname) self.run_certutil(["-N", "-f", self.passwd_fname]) self.set_perms(self.passwd_fname, write=True) def create_ca_cert(self): # Generate the encryption key self.run_certutil(["-G", "-z", self.noise_fname, "-f", self.passwd_fname]) # Generate the self-signed cert self.run_certutil(["-S", "-n", self.cacert_name, "-s", "cn=IPA Test Certificate Authority", "-x", "-t", "CT,,C", "-m", self.next_serial(), "-v", self.valid_months, "-z", self.noise_fname, "-f", self.passwd_fname]) def export_ca_cert(self, nickname, create_pkcs12=False): """create_pkcs12 tells us whether we should create a PKCS#12 file of the CA or not. If we are running on a replica then we won't have the private key to make a PKCS#12 file so we don't need to do that step.""" # export the CA cert for use with other apps ipautil.backup_file(self.cacert_fname) self.run_certutil(["-L", "-n", nickname, "-a", "-o", self.cacert_fname]) self.set_perms(self.cacert_fname) if create_pkcs12: ipautil.backup_file(self.pk12_fname) ipautil.run(["/usr/bin/pk12util", "-d", self.secdir, "-o", self.pk12_fname, "-n", self.cacert_name, "-w", self.passwd_fname, "-k", self.passwd_fname]) self.set_perms(self.pk12_fname) def load_cacert(self, cacert_fname): self.run_certutil(["-A", "-n", self.cacert_name, "-t", "CT,,C", "-a", "-i", cacert_fname]) def find_cacert_serial(self): (out,err) = self.run_certutil(["-L", "-n", self.cacert_name]) data = out.split('\n') for line in data: x = re.match(r'\s+Serial Number: (\d+) .*', line) if x is not None: return x.group(1) raise RuntimeError("Unable to find serial number") def create_server_cert(self, nickname, name, other_certdb=None): cdb = other_certdb if not cdb: cdb = self self.request_cert(name) cdb.issue_server_cert(self.certreq_fname, self.certder_fname) self.add_cert(self.certder_fname, nickname) os.unlink(self.certreq_fname) os.unlink(self.certder_fname) def create_signing_cert(self, nickname, name, other_certdb=None): cdb = other_certdb if not cdb: cdb = self self.request_cert(name) cdb.issue_signing_cert(self.certreq_fname, self.certder_fname) self.add_cert(self.certder_fname, nickname) os.unlink(self.certreq_fname) os.unlink(self.certder_fname) def request_cert(self, name): self.run_certutil(["-R", "-s", name, "-o", self.certreq_fname, "-g", self.keysize, "-z", self.noise_fname, "-f", self.passwd_fname]) def issue_server_cert(self, certreq_fname, cert_fname): p = subprocess.Popen(["/usr/bin/certutil", "-d", self.secdir, "-C", "-c", self.cacert_name, "-i", certreq_fname, "-o", cert_fname, "-m", self.next_serial(), "-v", self.valid_months, "-f", self.passwd_fname, "-1", "-5"], stdin=subprocess.PIPE, stdout=subprocess.PIPE) # Bah - this sucks, but I guess it isn't possible to fully # control this with command line arguments. # # What this is requesting is: # -1 (Create key usage extension) # 2 - Key encipherment # 9 - done # n - not critical # # -5 (Create netscape cert type extension) # 1 - SSL Server # 9 - done # n - not critical p.stdin.write("2\n9\nn\n1\n9\nn\n") p.wait() def issue_signing_cert(self, certreq_fname, cert_fname): p = subprocess.Popen(["/usr/bin/certutil", "-d", self.secdir, "-C", "-c", self.cacert_name, "-i", certreq_fname, "-o", cert_fname, "-m", self.next_serial(), "-v", self.valid_months, "-f", self.passwd_fname, "-1", "-5"], stdin=subprocess.PIPE, stdout=subprocess.PIPE) # Bah - this sucks, but I guess it isn't possible to fully # control this with command line arguments. # # What this is requesting is: # -1 (Create key usage extension) # 0 - Digital Signature # 5 - Cert signing key # 9 - done # n - not critical # # -5 (Create netscape cert type extension) # 3 - Object Signing # 9 - done # n - not critical p.stdin.write("0\n5\n9\nn\n3\n9\nn\n") p.wait() def add_cert(self, cert_fname, nickname): self.run_certutil(["-A", "-n", nickname, "-t", "u,u,u", "-i", cert_fname, "-f", cert_fname]) def create_pin_file(self): ipautil.backup_file(self.pin_fname) f = open(self.pin_fname, "w") f.write("Internal (Software) Token:") pwd = open(self.passwd_fname) f.write(pwd.read()) f.close() self.set_perms(self.pin_fname) def find_root_cert(self, nickname): p = subprocess.Popen(["/usr/bin/certutil", "-d", self.secdir, "-O", "-n", nickname], stdout=subprocess.PIPE) chain = p.stdout.read() chain = chain.split("\n") root_nickname = re.match('\ *"(.*)".*', chain[0]).groups()[0] return root_nickname def trust_root_cert(self, nickname): root_nickname = self.find_root_cert(nickname) self.run_certutil(["-M", "-n", root_nickname, "-t", "CT,CT,"]) def find_server_certs(self): p = subprocess.Popen(["/usr/bin/certutil", "-d", self.secdir, "-L"], stdout=subprocess.PIPE) certs = p.stdout.read() certs = certs.split("\n") server_certs = [] for cert in certs: fields = cert.split() if not len(fields): continue flags = fields[-1] if 'u' in flags: name = " ".join(fields[0:-1]) # NSS 3.12 added a header to the certutil output if name == "Certificate Nickname Trust": continue server_certs.append((name, flags)) return server_certs def import_pkcs12(self, pkcs12_fname, passwd_fname=None): args = ["/usr/bin/pk12util", "-d", self.secdir, "-i", pkcs12_fname, "-k", self.passwd_fname] if passwd_fname: args = args + ["-w", passwd_fname] try: ipautil.run(args) except ipautil.CalledProcessError, e: if e.returncode == 17: raise RuntimeError("incorrect password") else: raise RuntimeError("unknown error import pkcs#12 file") def export_pkcs12(self, pkcs12_fname, pkcs12_pwd_fname, nickname="CA certificate"): ipautil.run(["/usr/bin/pk12util", "-d", self.secdir, "-o", pkcs12_fname, "-n", nickname, "-k", self.passwd_fname, "-w", pkcs12_pwd_fname]) def create_self_signed(self, passwd=None): self.create_noise_file() self.create_passwd_file(passwd) self.create_certdbs() self.create_ca_cert() self.export_ca_cert(self.cacert_name, True) self.create_pin_file() def create_from_cacert(self, cacert_fname, passwd=""): self.create_noise_file() self.create_passwd_file(passwd) self.create_certdbs() self.load_cacert(cacert_fname) def create_from_pkcs12(self, pkcs12_fname, pkcs12_pwd_fname, passwd=None): """Create a new NSS database using the certificates in a PKCS#12 file. pkcs12_fname: the filename of the PKCS#12 file pkcs12_pwd_fname: the file containing the pin for the PKCS#12 file nickname: the nickname/friendly-name of the cert we are loading passwd: The password to use for the new NSS database we are creating """ self.create_noise_file() self.create_passwd_file(passwd) self.create_certdbs() self.import_pkcs12(pkcs12_fname, pkcs12_pwd_fname) server_certs = self.find_server_certs() if len(server_certs) == 0: raise RuntimeError("Could not find a suitable server cert in import in %s" % pkcs12_fname) # We only handle one server cert nickname = server_certs[0][0] self.cacert_name = self.find_root_cert(nickname) self.trust_root_cert(nickname) self.create_pin_file() self.export_ca_cert(self.cacert_name, False) # This file implies that we have our own self-signed CA. Ensure # that it no longer exists (from previous installs, for example). try: os.remove(CA_SERIALNO) except: pass def backup_files(self): self.fstore.backup_file(self.noise_fname) self.fstore.backup_file(self.passwd_fname) self.fstore.backup_file(self.certdb_fname) self.fstore.backup_file(self.keydb_fname) self.fstore.backup_file(self.secmod_fname) self.fstore.backup_file(self.cacert_fname) self.fstore.backup_file(self.pk12_fname) self.fstore.backup_file(self.pin_fname) self.fstore.backup_file(self.certreq_fname) self.fstore.backup_file(self.certder_fname)