X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=sfa%2Ftrust%2Fcertificate.py;h=758198ecae651a7c5ccb1dc845e5be6cbec7d2e7;hb=dfafd9825fecddbbd1ea4c7093766c386f660c29;hp=59ec4427fd929aa113b843d8e7986544414ade96;hpb=3353d88a6b96e2bc33d1dcec16d15d845d7df40b;p=sfa.git diff --git a/sfa/trust/certificate.py b/sfa/trust/certificate.py index 59ec4427..758198ec 100644 --- a/sfa/trust/certificate.py +++ b/sfa/trust/certificate.py @@ -35,6 +35,8 @@ ## # +from __future__ import print_function + import functools import os import tempfile @@ -48,6 +50,9 @@ from M2Crypto import X509 from sfa.util.faults import CertExpired, CertMissingParent, CertNotSignedByParent from sfa.util.sfalogging import logger +# this tends to generate quite some logs for little or no value +debug_verify_chain = False + glo_passphrase_callback = None ## @@ -85,11 +90,11 @@ def test_passphrase(string, passphrase): def convert_public_key(key): keyconvert_path = "/usr/bin/keyconvert.py" if not os.path.isfile(keyconvert_path): - raise IOError, "Could not find keyconvert in %s" % keyconvert_path + raise IOError("Could not find keyconvert in {}".format(keyconvert_path)) # we can only convert rsa keys if "ssh-dss" in key: - raise Exception, "keyconvert: dss keys are not supported" + raise Exception("keyconvert: dss keys are not supported") (ssh_f, ssh_fn) = tempfile.mkstemp() ssl_fn = tempfile.mktemp() @@ -103,7 +108,7 @@ def convert_public_key(key): # that it can be expected to see why it failed. # TODO: for production, cleanup the temporary files if not os.path.exists(ssl_fn): - raise Exception, "keyconvert: generated certificate not found. keyconvert may have failed." + raise Exception("keyconvert: generated certificate not found. keyconvert may have failed.") k = Keypair() try: @@ -132,8 +137,8 @@ class Keypair: # Creates a Keypair object # @param create If create==True, creates a new public/private key and # stores it in the object - # @param string If string!=None, load the keypair from the string (PEM) - # @param filename If filename!=None, load the keypair from the file + # @param string If string != None, load the keypair from the string (PEM) + # @param filename If filename != None, load the keypair from the file def __init__(self, create=False, string=None, filename=None): if create: @@ -148,7 +153,7 @@ class Keypair: def create(self): self.key = crypto.PKey() - self.key.generate_key(crypto.TYPE_RSA, 1024) + self.key.generate_key(crypto.TYPE_RSA, 2048) ## # Save the private key to a file @@ -156,13 +161,13 @@ class Keypair: def save_to_file(self, filename): open(filename, 'w').write(self.as_pem()) - self.filename=filename + self.filename = filename ## # Load the private key from a file. Implicity the private key includes the public key. def load_from_file(self, filename): - self.filename=filename + self.filename = filename buffer = open(filename, 'r').read() self.load_from_string(buffer) @@ -171,8 +176,10 @@ class Keypair: def load_from_string(self, string): if glo_passphrase_callback: - self.key = crypto.load_privatekey(crypto.FILETYPE_PEM, string, functools.partial(glo_passphrase_callback, self, string) ) - self.m2key = M2Crypto.EVP.load_key_string(string, functools.partial(glo_passphrase_callback, self, string) ) + self.key = crypto.load_privatekey( + crypto.FILETYPE_PEM, string, functools.partial(glo_passphrase_callback, self, string)) + self.m2key = M2Crypto.EVP.load_key_string( + string, functools.partial(glo_passphrase_callback, self, string)) else: self.key = crypto.load_privatekey(crypto.FILETYPE_PEM, string) self.m2key = M2Crypto.EVP.load_key_string(string) @@ -202,7 +209,7 @@ class Keypair: # prob not necc since this cert itself is junk but still... m2x509.set_version(2) junk_key = Keypair(create=True) - m2x509.sign(pkey=junk_key.get_m2_pkey(), md="sha1") + m2x509.sign(pkey=junk_key.get_m2_pubkey(), md="sha1") # convert the m2 x509 cert to a pyopenssl x509 m2pem = m2x509.as_pem() @@ -210,7 +217,7 @@ class Keypair: # get the pyopenssl pkey from the pyopenssl x509 self.key = pyx509.get_pubkey() - self.filename=filename + self.filename = filename ## # Load the public key from a string. No private key is loaded. @@ -231,7 +238,7 @@ class Keypair: ## # Return an M2Crypto key object - def get_m2_pkey(self): + def get_m2_pubkey(self): if not self.m2key: self.m2key = M2Crypto.EVP.load_key_string(self.as_pem()) return self.m2key @@ -240,7 +247,7 @@ class Keypair: # Returns a string containing the public key represented by this object. def get_pubkey_string(self): - m2pkey = self.get_m2_pkey() + m2pkey = self.get_m2_pubkey() return base64.b64encode(m2pkey.as_der()) ## @@ -256,13 +263,13 @@ class Keypair: return self.as_pem() == pkey.as_pem() def sign_string(self, data): - k = self.get_m2_pkey() + k = self.get_m2_pubkey() k.sign_init() k.sign_update(data) return base64.b64encode(k.sign_final()) def verify_string(self, data, sig): - k = self.get_m2_pkey() + k = self.get_m2_pubkey() k.verify_init() k.verify_update(data) return M2Crypto.m2.verify_final(k.ctx, base64.b64decode(sig), k.pkey) @@ -274,14 +281,14 @@ class Keypair: def get_filename(self): return getattr(self,'filename',None) - def dump (self, *args, **kwargs): - print self.dump_string(*args, **kwargs) + def dump(self, *args, **kwargs): + print(self.dump_string(*args, **kwargs)) - def dump_string (self): - result="" - result += "KEYPAIR: pubkey=%40s..."%self.get_pubkey_string() - filename=self.get_filename() - if filename: result += "Filename %s\n"%filename + def dump_string(self): + result = "" + result += "KEYPAIR: pubkey={:>40}...".format(self.get_pubkey_string()) + filename = self.get_filename() + if filename: result += "Filename {}\n".format(filename) return result ## @@ -297,15 +304,15 @@ class Keypair: # whether to save the parent certificates as well. class Certificate: - digest = "md5" + digest = "sha256" - cert = None - issuerKey = None - issuerSubject = None - parent = None +# x509 = None +# issuerKey = None +# issuerSubject = None +# parent = None isCA = None # will be a boolean once set - separator="-----parent-----" + separator = "-----parent-----" ## # Create a certificate object. @@ -319,6 +326,12 @@ class Certificate: # @param isCA If !=None, set whether this cert is for a CA def __init__(self, lifeDays=1825, create=False, subject=None, string=None, filename=None, isCA=None): + # these used to be defined in the class ! + self.x509 = None + self.issuerKey = None + self.issuerSubject = None + self.parent = None + self.data = {} if create or subject: self.create(lifeDays) @@ -336,12 +349,12 @@ class Certificate: # Create a blank X509 certificate and store it in this object. def create(self, lifeDays=1825): - self.cert = crypto.X509() + self.x509 = crypto.X509() # FIXME: Use different serial #s - self.cert.set_serial_number(3) - self.cert.gmtime_adj_notBefore(0) # 0 means now - self.cert.gmtime_adj_notAfter(lifeDays*60*60*24) # five years is default - self.cert.set_version(2) # x509v3 so it can have extensions + self.x509.set_serial_number(3) + self.x509.gmtime_adj_notBefore(0) # 0 means now + self.x509.gmtime_adj_notAfter(lifeDays*60*60*24) # five years is default + self.x509.set_version(2) # x509v3 so it can have extensions ## @@ -349,7 +362,7 @@ class Certificate: # certificate object. def load_from_pyopenssl_x509(self, x509): - self.cert = x509 + self.x509 = x509 ## # Load the certificate from a string @@ -358,11 +371,16 @@ class Certificate: # if it is a chain of multiple certs, then split off the first one and # load it (support for the ---parent--- tag as well as normal chained certs) + if string is None or string.strip() == "": + logger.warn("Empty string in load_from_string") + return + string = string.strip() # If it's not in proper PEM format, wrap it if string.count('-----BEGIN CERTIFICATE') == 0: - string = '-----BEGIN CERTIFICATE-----\n%s\n-----END CERTIFICATE-----' % string + string = '-----BEGIN CERTIFICATE-----\n{}\n-----END CERTIFICATE-----'\ + .format(string) # If there is a PEM cert in there, but there is some other text first # such as the text of the certificate, skip the text @@ -380,7 +398,10 @@ class Certificate: else: parts = string.split(Certificate.separator, 1) - self.cert = crypto.load_certificate(crypto.FILETYPE_PEM, parts[0]) + self.x509 = crypto.load_certificate(crypto.FILETYPE_PEM, parts[0]) + + if self.x509 is None: + logger.warn("Loaded from string but cert is None: {}".format(string)) # if there are more certs, then create a parent and let the parent load # itself from the remainder of the string @@ -395,7 +416,7 @@ class Certificate: file = open(filename) string = file.read() self.load_from_string(string) - self.filename=filename + self.filename = filename ## # Save the certificate to a string. @@ -403,7 +424,10 @@ class Certificate: # @param save_parents If save_parents==True, then also save the parent certificates. def save_to_string(self, save_parents=True): - string = crypto.dump_certificate(crypto.FILETYPE_PEM, self.cert) + if self.x509 is None: + logger.warn("None cert in certificate.save_to_string") + return "" + string = crypto.dump_certificate(crypto.FILETYPE_PEM, self.x509) if save_parents and self.parent: string = string + self.parent.save_to_string(save_parents) return string @@ -420,7 +444,7 @@ class Certificate: f = open(filename, 'w') f.write(string) f.close() - self.filename=filename + self.filename = filename ## # Save the certificate to a random file in /tmp/ @@ -445,7 +469,7 @@ class Certificate: if isinstance(subject, dict) or isinstance(subject, str): req = crypto.X509Req() reqSubject = req.get_subject() - if (isinstance(subject, dict)): + if isinstance(subject, dict): for key in reqSubject.keys(): setattr(reqSubject, key, subject[key]) else: @@ -455,7 +479,7 @@ class Certificate: self.issuerReq = req if cert: # if a cert was supplied, then get the subject from the cert - subject = cert.cert.get_subject() + subject = cert.x509.get_subject() assert(subject) self.issuerSubject = subject @@ -463,7 +487,7 @@ class Certificate: # Get the issuer name def get_issuer(self, which="CN"): - x = self.cert.get_issuer() + x = self.x509.get_issuer() return getattr(x, which) ## @@ -472,26 +496,48 @@ class Certificate: def set_subject(self, name): req = crypto.X509Req() subj = req.get_subject() - if (isinstance(name, dict)): + if isinstance(name, dict): for key in name.keys(): setattr(subj, key, name[key]) else: setattr(subj, "CN", name) - self.cert.set_subject(subj) + self.x509.set_subject(subj) ## # Get the subject name of the certificate def get_subject(self, which="CN"): - x = self.cert.get_subject() + x = self.x509.get_subject() return getattr(x, which) ## # Get a pretty-print subject name of the certificate - - def get_printable_subject(self): - x = self.cert.get_subject() - return "[ OU: %s, CN: %s, SubjectAltName: %s ]" % (getattr(x, "OU"), getattr(x, "CN"), self.get_data()) + # let's try to make this a little more usable as is makes logs hairy + # FIXME: Consider adding 'urn:publicid' and 'uuid' back for GENI? + pretty_fields = ['email'] + def filter_chunk(self, chunk): + for field in self.pretty_fields: + if field in chunk: + return " "+chunk + + def pretty_cert(self): + message = "[Cert." + x = self.x509.get_subject() + ou = getattr(x, "OU") + if ou: message += " OU: {}".format(ou) + cn = getattr(x, "CN") + if cn: message += " CN: {}".format(cn) + data = self.get_data(field='subjectAltName') + if data: + message += " SubjectAltName:" + counter = 0 + filtered = [self.filter_chunk(chunk) for chunk in data.split()] + message += " ".join( [f for f in filtered if f]) + omitted = len([f for f in filtered if not f]) + if omitted: + message += "..+{} omitted".format(omitted) + message += "]" + return message ## # Get the public key of the certificate. @@ -500,7 +546,7 @@ class Certificate: def set_pubkey(self, key): assert(isinstance(key, Keypair)) - self.cert.set_pubkey(key.get_openssl_pkey()) + self.x509.set_pubkey(key.get_openssl_pkey()) ## # Get the public key of the certificate. @@ -509,7 +555,7 @@ class Certificate: def get_pubkey(self): m2x509 = X509.load_cert_string(self.save_to_string()) pkey = Keypair() - pkey.key = self.cert.get_pubkey() + pkey.key = self.x509.get_pubkey() pkey.m2key = m2x509.get_pubkey() return pkey @@ -525,7 +571,8 @@ class Certificate: if self.isCA != None: # Can't double set properties - raise Exception, "Cannot set basicConstraints CA:?? more than once. Was %s, trying to set as %s" % (self.isCA, val) + raise Exception("Cannot set basicConstraints CA:?? more than once. " + "Was {}, trying to set as {}%s".format(self.isCA, val)) self.isCA = val if val: @@ -561,18 +608,29 @@ class Certificate: # FIXME: What if they are trying to set with a different value? # Is this ever OK? Or should we raise an exception? # elif oldExtVal: -# raise "Cannot add extension %s which had val %s with new val %s" % (name, oldExtVal, value) +# raise "Cannot add extension {} which had val {} with new val {}".format(name, oldExtVal, value) - ext = crypto.X509Extension (name, critical, value) - self.cert.add_extensions([ext]) + ext = crypto.X509Extension(name, critical, value) + self.x509.add_extensions([ext]) ## # Get an X509 extension from the certificate def get_extension(self, name): + if name is None: + return None + + certstr = self.save_to_string() + if certstr is None or certstr == "": + return None # pyOpenSSL does not have a way to get extensions - m2x509 = X509.load_cert_string(self.save_to_string()) + m2x509 = X509.load_cert_string(certstr) + if m2x509 is None: + logger.warn("No cert loaded in get_extension") + return None + if m2x509.get_ext(name) is None: + return None value = m2x509.get_ext(name).get_value() return value @@ -585,8 +643,8 @@ class Certificate: def set_data(self, str, field='subjectAltName'): # pyOpenSSL only allows us to add extensions, so if we try to set the # same extension more than once, it will not work - if self.data.has_key(field): - raise "Cannot set ", field, " more than once" + if field in self.data: + raise Exception("Cannot set {} more than once".format(field)) self.data[field] = str self.add_extension(field, 0, str) @@ -594,7 +652,7 @@ class Certificate: # Return the data string that was previously set with set_data def get_data(self, field='subjectAltName'): - if self.data.has_key(field): + if field in self.data: return self.data[field] try: @@ -610,27 +668,29 @@ class Certificate: def sign(self): logger.debug('certificate.sign') - assert self.cert != None + assert self.x509 != None assert self.issuerSubject != None assert self.issuerKey != None - self.cert.set_issuer(self.issuerSubject) - self.cert.sign(self.issuerKey.get_openssl_pkey(), self.digest) + self.x509.set_issuer(self.issuerSubject) + self.x509.sign(self.issuerKey.get_openssl_pkey(), self.digest) ## # Verify the authenticity of a certificate. # @param pkey is a Keypair object representing a public key. If Pkey # did not sign the certificate, then an exception will be thrown. - def verify(self, pkey): + def verify(self, pubkey): # pyOpenSSL does not have a way to verify signatures m2x509 = X509.load_cert_string(self.save_to_string()) - m2pkey = pkey.get_m2_pkey() + m2pubkey = pubkey.get_m2_pubkey() # verify it - return m2x509.verify(m2pkey) + # verify returns -1 or 0 on failure depending on how serious the + # error conditions are + return m2x509.verify(m2pubkey) == 1 # XXX alternatively, if openssl has been patched, do the much simpler: # try: - # self.cert.verify(pkey.get_openssl_key()) + # self.x509.verify(pkey.get_openssl_key()) # return 1 # except: # return 0 @@ -694,38 +754,50 @@ class Certificate: # until a certificate is found that is signed by a trusted root. # verify expiration time - if self.cert.has_expired(): - logger.debug("verify_chain: NO, Certificate %s has expired" % self.get_printable_subject()) - raise CertExpired(self.get_printable_subject(), "client cert") + if self.x509.has_expired(): + if debug_verify_chain: + logger.debug("verify_chain: NO, Certificate {} has expired" + .format(self.pretty_cert())) + raise CertExpired(self.pretty_cert(), "client cert") # if this cert is signed by a trusted_cert, then we are set for trusted_cert in trusted_certs: if self.is_signed_by_cert(trusted_cert): # verify expiration of trusted_cert ? - if not trusted_cert.cert.has_expired(): - logger.debug("verify_chain: YES. Cert %s signed by trusted cert %s"%( - self.get_printable_subject(), trusted_cert.get_printable_subject())) + if not trusted_cert.x509.has_expired(): + if debug_verify_chain: + logger.debug("verify_chain: YES. Cert {} signed by trusted cert {}" + .format(self.pretty_cert(), trusted_cert.pretty_cert())) return trusted_cert else: - logger.debug("verify_chain: NO. Cert %s is signed by trusted_cert %s, but that signer is expired..."%( - self.get_printable_subject(),trusted_cert.get_printable_subject())) - raise CertExpired(self.get_printable_subject()," signer trusted_cert %s"%trusted_cert.get_printable_subject()) + if debug_verify_chain: + logger.debug("verify_chain: NO. Cert {} is signed by trusted_cert {}, " + "but that signer is expired..." + .format(self.pretty_cert(),trusted_cert.pretty_cert())) + raise CertExpired("{} signer trusted_cert {}" + .format(self.pretty_cert(), trusted_cert.pretty_cert())) # if there is no parent, then no way to verify the chain if not self.parent: - logger.debug("verify_chain: NO. %s has no parent and issuer %s is not in %d trusted roots"%(self.get_printable_subject(), self.get_issuer(), len(trusted_certs))) - raise CertMissingParent(self.get_printable_subject() + ": Issuer %s is not one of the %d trusted roots, and cert has no parent." % (self.get_issuer(), len(trusted_certs))) + if debug_verify_chain: + logger.debug("verify_chain: NO. {} has no parent " + "and issuer {} is not in {} trusted roots" + .format(self.pretty_cert(), self.get_issuer(), len(trusted_certs))) + raise CertMissingParent("{}: Issuer {} is not one of the {} trusted roots, " + "and cert has no parent." + .format(self.pretty_cert(), self.get_issuer(), len(trusted_certs))) # if it wasn't signed by the parent... if not self.is_signed_by_cert(self.parent): - logger.debug("verify_chain: NO. %s is not signed by parent %s, but by %s"%\ - (self.get_printable_subject(), - self.parent.get_printable_subject(), - self.get_issuer())) - raise CertNotSignedByParent("%s: Parent %s, issuer %s"\ - % (self.get_printable_subject(), - self.parent.get_printable_subject(), - self.get_issuer())) + if debug_verify_chain: + logger.debug("verify_chain: NO. {} is not signed by parent {}, but by {}" + .format(self.pretty_cert(), + self.parent.pretty_cert(), + self.get_issuer())) + raise CertNotSignedByParent("{}: Parent {}, issuer {}" + .format(self.pretty_cert(), + self.parent.pretty_cert(), + self.get_issuer())) # Confirm that the parent is a CA. Only CAs can be trusted as # signers. @@ -734,14 +806,15 @@ class Certificate: # Ugly - cert objects aren't parsed so we need to read the # extension and hope there are no other basicConstraints if not self.parent.isCA and not (self.parent.get_extension('basicConstraints') == 'CA:TRUE'): - logger.warn("verify_chain: cert %s's parent %s is not a CA" % \ - (self.get_printable_subject(), self.parent.get_printable_subject())) - raise CertNotSignedByParent("%s: Parent %s not a CA" % (self.get_printable_subject(), - self.parent.get_printable_subject())) + logger.warn("verify_chain: cert {}'s parent {} is not a CA" + .format(self.pretty_cert(), self.parent.pretty_cert())) + raise CertNotSignedByParent("{}: Parent {} not a CA" + .format(self.pretty_cert(), self.parent.pretty_cert())) # if the parent isn't verified... - logger.debug("verify_chain: .. %s, -> verifying parent %s"%\ - (self.get_printable_subject(),self.parent.get_printable_subject())) + if debug_verify_chain: + logger.debug("verify_chain: .. {}, -> verifying parent {}" + .format(self.pretty_cert(),self.parent.pretty_cert())) self.parent.verify_chain(trusted_certs) return @@ -749,20 +822,20 @@ class Certificate: ### more introspection def get_extensions(self): # pyOpenSSL does not have a way to get extensions - triples=[] + triples = [] m2x509 = X509.load_cert_string(self.save_to_string()) - nb_extensions=m2x509.get_ext_count() - logger.debug("X509 had %d extensions"%nb_extensions) + nb_extensions = m2x509.get_ext_count() + logger.debug("X509 had {} extensions".format(nb_extensions)) for i in range(nb_extensions): - ext=m2x509.get_ext_at(i) + ext = m2x509.get_ext_at(i) triples.append( (ext.get_name(), ext.get_value(), ext.get_critical(),) ) return triples def get_data_names(self): return self.data.keys() - def get_all_datas (self): - triples=self.get_extensions() + def get_all_datas(self): + triples = self.get_extensions() for name in self.get_data_names(): triples.append( (name,self.get_data(name),'data',) ) return triples @@ -771,21 +844,22 @@ class Certificate: def get_filename(self): return getattr(self,'filename',None) - def dump (self, *args, **kwargs): - print self.dump_string(*args, **kwargs) + def dump(self, *args, **kwargs): + print(self.dump_string(*args, **kwargs)) - def dump_string (self,show_extensions=False): + def dump_string(self, show_extensions=False): result = "" - result += "CERTIFICATE for %s\n"%self.get_printable_subject() - result += "Issued by %s\n"%self.get_issuer() - filename=self.get_filename() - if filename: result += "Filename %s\n"%filename + result += "CERTIFICATE for {}\n".format(self.pretty_cert()) + result += "Issued by {}\n".format(self.get_issuer()) + filename = self.get_filename() + if filename: + result += "Filename {}\n".format(filename) if show_extensions: - all_datas=self.get_all_datas() - result += " has %d extensions/data attached"%len(all_datas) - for (n,v,c) in all_datas: - if c=='data': - result += " data: %s=%s\n"%(n,v) + all_datas = self.get_all_datas() + result += " has {} extensions/data attached".format(len(all_datas)) + for n, v, c in all_datas: + if c == 'data': + result += " data: {}={}\n".format(n, v) else: - result += " ext: %s (crit=%s)=<<<%s>>>\n"%(n,c,v) + result += " ext: {} (crit={})=<<<{}>>>\n".format(n, c, v) return result