X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=sfa%2Ftrust%2Fcertificate.py;h=e05d3a1c30e1019b5aaf0e9cd9b41e87254c3784;hb=847e5b53e0030190f9d16ccfa52597eeb9a260ef;hp=b5296a6a6d6cb98f9e4d77c5ee397b8356653a61;hpb=7eb34251548a271ae964b2f7d7e0fad7a1e41f5a;p=sfa.git diff --git a/sfa/trust/certificate.py b/sfa/trust/certificate.py index b5296a6a..e05d3a1c 100644 --- a/sfa/trust/certificate.py +++ b/sfa/trust/certificate.py @@ -45,7 +45,7 @@ # -from __future__ import print_function + import functools import os @@ -64,7 +64,7 @@ from sfa.util.faults import (CertExpired, CertMissingParent, from sfa.util.sfalogging import logger # this tends to generate quite some logs for little or no value -debug_verify_chain = False +debug_verify_chain = True glo_passphrase_callback = None @@ -182,7 +182,8 @@ class Keypair: # @param filename name of file to store the keypair in def save_to_file(self, filename): - open(filename, 'w').write(self.as_pem()) + with open(filename, 'wb') as output: + output.write(self.as_pem()) self.filename = filename ## @@ -190,6 +191,7 @@ class Keypair: # public key. def load_from_file(self, filename): + logger.info(f"opening {filename} from certficate.load_from_file") self.filename = filename buffer = open(filename, 'r').read() self.load_from_string(buffer) @@ -205,12 +207,13 @@ class Keypair: OpenSSL.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)) + string.encode(encoding="utf-8"), + functools.partial(glo_passphrase_callback, self, string)) else: self.key = OpenSSL.crypto.load_privatekey( OpenSSL.crypto.FILETYPE_PEM, string) - self.m2key = M2Crypto.EVP.load_key_string(string) + self.m2key = M2Crypto.EVP.load_key_string( + string.encode(encoding="utf-8")) ## # Load the public key from a string. No private key is loaded. @@ -273,7 +276,8 @@ class Keypair: def get_m2_pubkey(self): import M2Crypto if not self.m2key: - self.m2key = M2Crypto.EVP.load_key_string(self.as_pem()) + self.m2key = M2Crypto.EVP.load_key_string( + self.as_pem().encode(encoding="utf-8")) return self.m2key ## @@ -520,7 +524,7 @@ class Certificate: req = OpenSSL.crypto.X509Req() reqSubject = req.get_subject() if isinstance(subject, dict): - for key in reqSubject.keys(): + for key in list(reqSubject.keys()): setattr(reqSubject, key, subject[key]) else: setattr(reqSubject, "CN", subject) @@ -547,7 +551,7 @@ class Certificate: req = OpenSSL.crypto.X509Req() subj = req.get_subject() if isinstance(name, dict): - for key in name.keys(): + for key in list(name.keys()): setattr(subj, key, name[key]) else: setattr(subj, "CN", name) @@ -583,7 +587,6 @@ class Certificate: 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]) @@ -676,6 +679,11 @@ class Certificate: # raise "Cannot add extension {} which had val {} with new val {}"\ # .format(name, oldExtVal, value) + if isinstance(name, str): + name = name.encode() + if isinstance(value, str): + value = value.encode() + ext = OpenSSL.crypto.X509Extension(name, critical, value) self.x509.add_extensions([ext]) @@ -714,7 +722,9 @@ class Certificate: if field in self.data: raise Exception("Cannot set {} more than once".format(field)) self.data[field] = string - self.add_extension(field, 0, string) + # call str() because we've seen unicode there + # and the underlying C code doesn't like it + self.add_extension(field, 0, str(string)) ## # Return the data string that was previously set with set_data @@ -789,9 +799,11 @@ class Certificate: # @param cert certificate object def is_signed_by_cert(self, cert): - logger.debug("Certificate.is_signed_by_cert -> invoking verify") - k = cert.get_pubkey() - result = self.verify(k) + key = cert.get_pubkey() + logger.debug("Certificate.is_signed_by_cert -> verify on {}\n" + "with pubkey {}" + .format(self, key)) + result = self.verify(key) return result ## @@ -834,7 +846,6 @@ class Certificate: # the public key contained in it's parent. The chain is recursed # until a certificate is found that is signed by a trusted root. - logger.debug("Certificate.verify_chain {}".format(self.pretty_name())) # verify expiration time if self.x509.has_expired(): if debug_verify_chain: @@ -844,7 +855,8 @@ class Certificate: # if this cert is signed by a trusted_cert, then we are set for i, trusted_cert in enumerate(trusted_certs, 1): - logger.debug("Certificate.verify_chain - trying trusted #{} : {}" + logger.debug(5*'-' + + " Certificate.verify_chain - trying trusted #{} : {}" .format(i, trusted_cert.pretty_name())) if self.is_signed_by_cert(trusted_cert): # verify expiration of trusted_cert ? @@ -867,7 +879,7 @@ class Certificate: trusted_cert.pretty_name())) else: logger.debug("verify_chain: not a direct" - " descendant of a trusted root") + " descendant of trusted root #{}".format(i)) # if there is no parent, then no way to verify the chain if not self.parent: @@ -933,7 +945,7 @@ class Certificate: return triples def get_data_names(self): - return self.data.keys() + return list(self.data.keys()) def get_all_datas(self): triples = self.get_extensions()