sfa/trust/certificate.py

   1 #----------------------------------------------------------------------
   2 # Copyright (c) 2008 Board of Trustees, Princeton University
   3 #
   4 # Permission is hereby granted, free of charge, to any person obtaining
   5 # a copy of this software and/or hardware specification (the "Work") to
   6 # deal in the Work without restriction, including without limitation the
   7 # rights to use, copy, modify, merge, publish, distribute, sublicense,
   8 # and/or sell copies of the Work, and to permit persons to whom the Work
   9 # is furnished to do so, subject to the following conditions:
  10 #
  11 # The above copyright notice and this permission notice shall be
  12 # included in all copies or substantial portions of the Work.
  13 #
  14 # THE WORK IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
  15 # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  16 # MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  17 # NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
  18 # HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
  19 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  20 # OUT OF OR IN CONNECTION WITH THE WORK OR THE USE OR OTHER DEALINGS
  21 # IN THE WORK.
  22 #----------------------------------------------------------------------
  23
  24 ##
  25 # SFA uses two crypto libraries: pyOpenSSL and M2Crypto to implement
  26 # the necessary crypto functionality. Ideally just one of these libraries
  27 # would be used, but unfortunately each of these libraries is independently
  28 # lacking. The pyOpenSSL library is missing many necessary functions, and
  29 # the M2Crypto library has crashed inside of some of the functions. The
  30 # design decision is to use pyOpenSSL whenever possible as it seems more
  31 # stable, and only use M2Crypto for those functions that are not possible
  32 # in pyOpenSSL.
  33 #
  34 # This module exports two classes: Keypair and Certificate.
  35 ##
  36 #
  37
  38 # Notes on using the openssl command line
  39 #
  40 # for verifying the chain in a gid, assuming it is split into pieces p1.pem p2.pem p3.pem
  41 # you can use openssl to verify the chain using this command
  42 # openssl verify -verbose -CAfile <(cat p2.pem p3.pem) p1.pem
  43 # also you can use sfax509 to invoke openssl x509 on all parts of the gid
  44 #
  45
  46
  47 from __future__ import print_function
  48
  49 import functools
  50 import os
  51 import tempfile
  52 import base64
  53 from tempfile import mkstemp
  54
  55 import OpenSSL
  56 # M2Crypto is imported on the fly to minimize crashes
  57 #import M2Crypto
  58
  59 from sfa.util.py23 import PY3
  60
  61 from sfa.util.faults import CertExpired, CertMissingParent, CertNotSignedByParent
  62 from sfa.util.sfalogging import logger
  63
  64 # this tends to generate quite some logs for little or no value
  65 debug_verify_chain = False
  66
  67 glo_passphrase_callback = None
  68
  69 ##
  70 # A global callback may be implemented for requesting passphrases from the
  71 # user. The function will be called with three arguments:
  72 #
  73 #    keypair_obj: the keypair object that is calling the passphrase
  74 #    string: the string containing the private key that's being loaded
  75 #    x: unknown, appears to be 0, comes from pyOpenSSL and/or m2crypto
  76 #
  77 # The callback should return a string containing the passphrase.
  78
  79
  80 def set_passphrase_callback(callback_func):
  81     global glo_passphrase_callback
  82
  83     glo_passphrase_callback = callback_func
  84
  85 ##
  86 # Sets a fixed passphrase.
  87
  88
  89 def set_passphrase(passphrase):
  90     set_passphrase_callback(lambda k, s, x: passphrase)
  91
  92 ##
  93 # Check to see if a passphrase works for a particular private key string.
  94 # Intended to be used by passphrase callbacks for input validation.
  95
  96
  97 def test_passphrase(string, passphrase):
  98     try:
  99         OpenSSL.crypto.load_privatekey(
 100             OpenSSL.crypto.FILETYPE_PEM, string, (lambda x: passphrase))
 101         return True
 102     except:
 103         return False
 104
 105
 106 def convert_public_key(key):
 107     keyconvert_path = "/usr/bin/keyconvert.py"
 108     if not os.path.isfile(keyconvert_path):
 109         raise IOError(
 110             "Could not find keyconvert in {}".format(keyconvert_path))
 111
 112     # we can only convert rsa keys
 113     if "ssh-dss" in key:
 114         raise Exception("keyconvert: dss keys are not supported")
 115
 116     (ssh_f, ssh_fn) = tempfile.mkstemp()
 117     ssl_fn = tempfile.mktemp()
 118     os.write(ssh_f, key)
 119     os.close(ssh_f)
 120
 121     cmd = keyconvert_path + " " + ssh_fn + " " + ssl_fn
 122     os.system(cmd)
 123
 124     # this check leaves the temporary file containing the public key so
 125     # that it can be expected to see why it failed.
 126     # TODO: for production, cleanup the temporary files
 127     if not os.path.exists(ssl_fn):
 128         raise Exception(
 129             "keyconvert: generated certificate not found. keyconvert may have failed.")
 130
 131     k = Keypair()
 132     try:
 133         k.load_pubkey_from_file(ssl_fn)
 134         return k
 135     except:
 136         logger.log_exc("convert_public_key caught exception")
 137         raise
 138     finally:
 139         # remove the temporary files
 140         if os.path.exists(ssh_fn):
 141             os.remove(ssh_fn)
 142         if os.path.exists(ssl_fn):
 143             os.remove(ssl_fn)
 144
 145 ##
 146 # Public-private key pairs are implemented by the Keypair class.
 147 # A Keypair object may represent both a public and private key pair, or it
 148 # may represent only a public key (this usage is consistent with OpenSSL).
 149
 150
 151 class Keypair:
 152     key = None       # public/private keypair
 153     m2key = None     # public key (m2crypto format)
 154
 155     ##
 156     # Creates a Keypair object
 157     # @param create If create==True, creates a new public/private key and
 158     #     stores it in the object
 159     # @param string If string != None, load the keypair from the string (PEM)
 160     # @param filename If filename != None, load the keypair from the file
 161
 162     def __init__(self, create=False, string=None, filename=None):
 163         if create:
 164             self.create()
 165         if string:
 166             self.load_from_string(string)
 167         if filename:
 168             self.load_from_file(filename)
 169
 170     ##
 171     # Create a RSA public/private key pair and store it inside the keypair
 172     # object
 173
 174     def create(self):
 175         self.key = OpenSSL.crypto.PKey()
 176         self.key.generate_key(OpenSSL.crypto.TYPE_RSA, 2048)
 177
 178     ##
 179     # Save the private key to a file
 180     # @param filename name of file to store the keypair in
 181
 182     def save_to_file(self, filename):
 183         open(filename, 'w').write(self.as_pem())
 184         self.filename = filename
 185
 186     ##
 187     # Load the private key from a file. Implicity the private key includes the
 188     # public key.
 189
 190     def load_from_file(self, filename):
 191         self.filename = filename
 192         buffer = open(filename, 'r').read()
 193         self.load_from_string(buffer)
 194
 195     ##
 196     # Load the private key from a string. Implicitly the private key includes
 197     # the public key.
 198
 199     def load_from_string(self, string):
 200         import M2Crypto
 201         if glo_passphrase_callback:
 202             self.key = OpenSSL.crypto.load_privatekey(
 203                 OpenSSL.crypto.FILETYPE_PEM, string, functools.partial(glo_passphrase_callback, self, string))
 204             self.m2key = M2Crypto.EVP.load_key_string(
 205                 string, functools.partial(glo_passphrase_callback, self, string))
 206         else:
 207             self.key = OpenSSL.crypto.load_privatekey(
 208                 OpenSSL.crypto.FILETYPE_PEM, string)
 209             self.m2key = M2Crypto.EVP.load_key_string(string)
 210
 211     ##
 212     #  Load the public key from a string. No private key is loaded.
 213
 214     def load_pubkey_from_file(self, filename):
 215         import M2Crypto
 216         # load the m2 public key
 217         m2rsakey = M2Crypto.RSA.load_pub_key(filename)
 218         self.m2key = M2Crypto.EVP.PKey()
 219         self.m2key.assign_rsa(m2rsakey)
 220
 221         # create an m2 x509 cert
 222         m2name = M2Crypto.X509.X509_Name()
 223         m2name.add_entry_by_txt(field="CN", type=0x1001,
 224                                 entry="junk", len=-1, loc=-1, set=0)
 225         m2x509 = M2Crypto.X509.X509()
 226         m2x509.set_pubkey(self.m2key)
 227         m2x509.set_serial_number(0)
 228         m2x509.set_issuer_name(m2name)
 229         m2x509.set_subject_name(m2name)
 230         ASN1 = M2Crypto.ASN1.ASN1_UTCTIME()
 231         ASN1.set_time(500)
 232         m2x509.set_not_before(ASN1)
 233         m2x509.set_not_after(ASN1)
 234         # x509v3 so it can have extensions
 235         # prob not necc since this cert itself is junk but still...
 236         m2x509.set_version(2)
 237         junk_key = Keypair(create=True)
 238         m2x509.sign(pkey=junk_key.get_m2_pubkey(), md="sha1")
 239
 240         # convert the m2 x509 cert to a pyopenssl x509
 241         m2pem = m2x509.as_pem()
 242         pyx509 = OpenSSL.crypto.load_certificate(
 243             OpenSSL.crypto.FILETYPE_PEM, m2pem)
 244
 245         # get the pyopenssl pkey from the pyopenssl x509
 246         self.key = pyx509.get_pubkey()
 247         self.filename = filename
 248
 249     ##
 250     # Load the public key from a string. No private key is loaded.
 251
 252     def load_pubkey_from_string(self, string):
 253         (f, fn) = tempfile.mkstemp()
 254         os.write(f, string)
 255         os.close(f)
 256         self.load_pubkey_from_file(fn)
 257         os.remove(fn)
 258
 259     ##
 260     # Return the private key in PEM format.
 261
 262     def as_pem(self):
 263         return OpenSSL.crypto.dump_privatekey(OpenSSL.crypto.FILETYPE_PEM, self.key)
 264
 265     ##
 266     # Return an M2Crypto key object
 267
 268     def get_m2_pubkey(self):
 269         import M2Crypto
 270         if not self.m2key:
 271             self.m2key = M2Crypto.EVP.load_key_string(self.as_pem())
 272         return self.m2key
 273
 274     ##
 275     # Returns a string containing the public key represented by this object.
 276
 277     def get_pubkey_string(self):
 278         m2pkey = self.get_m2_pubkey()
 279         return base64.b64encode(m2pkey.as_der())
 280
 281     ##
 282     # Return an OpenSSL pkey object
 283
 284     def get_openssl_pkey(self):
 285         return self.key
 286
 287     ##
 288     # Given another Keypair object, return TRUE if the two keys are the same.
 289
 290     def is_same(self, pkey):
 291         return self.as_pem() == pkey.as_pem()
 292
 293     def sign_string(self, data):
 294         k = self.get_m2_pubkey()
 295         k.sign_init()
 296         k.sign_update(data)
 297         return base64.b64encode(k.sign_final())
 298
 299     def verify_string(self, data, sig):
 300         import M2Crypto
 301         k = self.get_m2_pubkey()
 302         k.verify_init()
 303         k.verify_update(data)
 304         return M2Crypto.m2.verify_final(k.ctx, base64.b64decode(sig), k.pkey)
 305
 306     def compute_hash(self, value):
 307         return self.sign_string(str(value))
 308
 309     # only informative
 310     def get_filename(self):
 311         return getattr(self, 'filename', None)
 312
 313     def dump(self, *args, **kwargs):
 314         print(self.dump_string(*args, **kwargs))
 315
 316     def dump_string(self):
 317         result = ""
 318         result += "KEYPAIR: pubkey={:>40}...".format(self.get_pubkey_string())
 319         filename = self.get_filename()
 320         if filename:
 321             result += "Filename {}\n".format(filename)
 322         return result
 323
 324 ##
 325 # The certificate class implements a general purpose X509 certificate, making
 326 # use of the appropriate pyOpenSSL or M2Crypto abstractions. It also adds
 327 # several addition features, such as the ability to maintain a chain of
 328 # parent certificates, and storage of application-specific data.
 329 #
 330 # Certificates include the ability to maintain a chain of parents. Each
 331 # certificate includes a pointer to it's parent certificate. When loaded
 332 # from a file or a string, the parent chain will be automatically loaded.
 333 # When saving a certificate to a file or a string, the caller can choose
 334 # whether to save the parent certificates as well.
 335
 336
 337 class Certificate:
 338     digest = "sha256"
 339
 340 #    x509 = None
 341 #    issuerKey = None
 342 #    issuerSubject = None
 343 #    parent = None
 344     isCA = None  # will be a boolean once set
 345
 346     separator = "-----parent-----"
 347
 348     ##
 349     # Create a certificate object.
 350     #
 351     # @param lifeDays life of cert in days - default is 1825==5 years
 352     # @param create If create==True, then also create a blank X509 certificate.
 353     # @param subject If subject!=None, then create a blank certificate and set
 354     #     it's subject name.
 355     # @param string If string!=None, load the certficate from the string.
 356     # @param filename If filename!=None, load the certficiate from the file.
 357     # @param isCA If !=None, set whether this cert is for a CA
 358
 359     def __init__(self, lifeDays=1825, create=False, subject=None, string=None, filename=None, isCA=None):
 360         # these used to be defined in the class !
 361         self.x509 = None
 362         self.issuerKey = None
 363         self.issuerSubject = None
 364         self.parent = None
 365
 366         self.data = {}
 367         if create or subject:
 368             self.create(lifeDays)
 369         if subject:
 370             self.set_subject(subject)
 371         if string:
 372             self.load_from_string(string)
 373         if filename:
 374             self.load_from_file(filename)
 375
 376         # Set the CA bit if a value was supplied
 377         if isCA != None:
 378             self.set_is_ca(isCA)
 379
 380     # Create a blank X509 certificate and store it in this object.
 381
 382     def create(self, lifeDays=1825):
 383         self.x509 = OpenSSL.crypto.X509()
 384         # FIXME: Use different serial #s
 385         self.x509.set_serial_number(3)
 386         self.x509.gmtime_adj_notBefore(0)  # 0 means now
 387         self.x509.gmtime_adj_notAfter(
 388             lifeDays * 60 * 60 * 24)  # five years is default
 389         self.x509.set_version(2)  # x509v3 so it can have extensions
 390
 391     ##
 392     # Given a pyOpenSSL X509 object, store that object inside of this
 393     # certificate object.
 394
 395     def load_from_pyopenssl_x509(self, x509):
 396         self.x509 = x509
 397
 398     ##
 399     # Load the certificate from a string
 400
 401     def load_from_string(self, string):
 402         # if it is a chain of multiple certs, then split off the first one and
 403         # load it (support for the ---parent--- tag as well as normal chained
 404         # certs)
 405
 406         if string is None or string.strip() == "":
 407             logger.warn("Empty string in load_from_string")
 408             return
 409
 410         string = string.strip()
 411
 412         # If it's not in proper PEM format, wrap it
 413         if string.count('-----BEGIN CERTIFICATE') == 0:
 414             string = '-----BEGIN CERTIFICATE-----\n{}\n-----END CERTIFICATE-----'\
 415                      .format(string)
 416
 417         # If there is a PEM cert in there, but there is some other text first
 418         # such as the text of the certificate, skip the text
 419         beg = string.find('-----BEGIN CERTIFICATE')
 420         if beg > 0:
 421             # skipping over non cert beginning
 422             string = string[beg:]
 423
 424         parts = []
 425
 426         if string.count('-----BEGIN CERTIFICATE-----') > 1 and \
 427                 string.count(Certificate.separator) == 0:
 428             parts = string.split('-----END CERTIFICATE-----', 1)
 429             parts[0] += '-----END CERTIFICATE-----'
 430         else:
 431             parts = string.split(Certificate.separator, 1)
 432
 433         self.x509 = OpenSSL.crypto.load_certificate(
 434             OpenSSL.crypto.FILETYPE_PEM, parts[0])
 435
 436         if self.x509 is None:
 437             logger.warn(
 438                 "Loaded from string but cert is None: {}".format(string))
 439
 440         # if there are more certs, then create a parent and let the parent load
 441         # itself from the remainder of the string
 442         if len(parts) > 1 and parts[1] != '':
 443             self.parent = self.__class__()
 444             self.parent.load_from_string(parts[1])
 445
 446     ##
 447     # Load the certificate from a file
 448
 449     def load_from_file(self, filename):
 450         file = open(filename)
 451         string = file.read()
 452         self.load_from_string(string)
 453         self.filename = filename
 454
 455     ##
 456     # Save the certificate to a string.
 457     #
 458     # @param save_parents If save_parents==True, then also save the parent certificates.
 459
 460     def save_to_string(self, save_parents=True):
 461         if self.x509 is None:
 462             logger.warn("None cert in certificate.save_to_string")
 463             return ""
 464         string = OpenSSL.crypto.dump_certificate(
 465             OpenSSL.crypto.FILETYPE_PEM, self.x509)
 466         if PY3 and isinstance(string, bytes):
 467             string = string.decode()
 468         if save_parents and self.parent:
 469             string = string + self.parent.save_to_string(save_parents)
 470         return string
 471
 472     ##
 473     # Save the certificate to a file.
 474     # @param save_parents If save_parents==True, then also save the parent certificates.
 475
 476     def save_to_file(self, filename, save_parents=True, filep=None):
 477         string = self.save_to_string(save_parents=save_parents)
 478         if filep:
 479             f = filep
 480         else:
 481             f = open(filename, 'w')
 482         if PY3 and isinstance(string, bytes):
 483             string = string.decode()
 484         f.write(string)
 485         f.close()
 486         self.filename = filename
 487
 488     ##
 489     # Save the certificate to a random file in /tmp/
 490     # @param save_parents If save_parents==True, then also save the parent certificates.
 491     def save_to_random_tmp_file(self, save_parents=True):
 492         fp, filename = mkstemp(suffix='cert', text=True)
 493         fp = os.fdopen(fp, "w")
 494         self.save_to_file(filename, save_parents=True, filep=fp)
 495         return filename
 496
 497     ##
 498     # Sets the issuer private key and name
 499     # @param key Keypair object containing the private key of the issuer
 500     # @param subject String containing the name of the issuer
 501     # @param cert (optional) Certificate object containing the name of the issuer
 502
 503     def set_issuer(self, key, subject=None, cert=None):
 504         self.issuerKey = key
 505         if subject:
 506             # it's a mistake to use subject and cert params at the same time
 507             assert(not cert)
 508             if isinstance(subject, dict) or isinstance(subject, str):
 509                 req = OpenSSL.crypto.X509Req()
 510                 reqSubject = req.get_subject()
 511                 if isinstance(subject, dict):
 512                     for key in reqSubject.keys():
 513                         setattr(reqSubject, key, subject[key])
 514                 else:
 515                     setattr(reqSubject, "CN", subject)
 516                 subject = reqSubject
 517                 # subject is not valid once req is out of scope, so save req
 518                 self.issuerReq = req
 519         if cert:
 520             # if a cert was supplied, then get the subject from the cert
 521             subject = cert.x509.get_subject()
 522         assert(subject)
 523         self.issuerSubject = subject
 524
 525     ##
 526     # Get the issuer name
 527
 528     def get_issuer(self, which="CN"):
 529         x = self.x509.get_issuer()
 530         return getattr(x, which)
 531
 532     ##
 533     # Set the subject name of the certificate
 534
 535     def set_subject(self, name):
 536         req = OpenSSL.crypto.X509Req()
 537         subj = req.get_subject()
 538         if isinstance(name, dict):
 539             for key in name.keys():
 540                 setattr(subj, key, name[key])
 541         else:
 542             setattr(subj, "CN", name)
 543         self.x509.set_subject(subj)
 544
 545     ##
 546     # Get the subject name of the certificate
 547
 548     def get_subject(self, which="CN"):
 549         x = self.x509.get_subject()
 550         return getattr(x, which)
 551
 552     ##
 553     # Get a pretty-print subject name of the certificate
 554     # let's try to make this a little more usable as is makes logs hairy
 555     # FIXME: Consider adding 'urn:publicid' and 'uuid' back for GENI?
 556     pretty_fields = ['email']
 557
 558     def filter_chunk(self, chunk):
 559         for field in self.pretty_fields:
 560             if field in chunk:
 561                 return " " + chunk
 562
 563     def pretty_cert(self):
 564         message = "[Cert."
 565         x = self.x509.get_subject()
 566         ou = getattr(x, "OU")
 567         if ou:
 568             message += " OU: {}".format(ou)
 569         cn = getattr(x, "CN")
 570         if cn:
 571             message += " CN: {}".format(cn)
 572         data = self.get_data(field='subjectAltName')
 573         if data:
 574             message += " SubjectAltName:"
 575             counter = 0
 576             filtered = [self.filter_chunk(chunk) for chunk in data.split()]
 577             message += " ".join([f for f in filtered if f])
 578             omitted = len([f for f in filtered if not f])
 579             if omitted:
 580                 message += "..+{} omitted".format(omitted)
 581         message += "]"
 582         return message
 583
 584     def pretty_chain(self):
 585         message = "{}".format(self.x509.get_subject())
 586         parent = self.parent
 587         while parent:
 588             message += "->{}".format(parent.x509.get_subject())
 589             parent = parent.parent
 590         return message
 591
 592     def pretty_name(self):
 593         return self.get_filename() or self.pretty_chain()
 594
 595     ##
 596     # Get the public key of the certificate.
 597     #
 598     # @param key Keypair object containing the public key
 599
 600     def set_pubkey(self, key):
 601         assert(isinstance(key, Keypair))
 602         self.x509.set_pubkey(key.get_openssl_pkey())
 603
 604     ##
 605     # Get the public key of the certificate.
 606     # It is returned in the form of a Keypair object.
 607
 608     def get_pubkey(self):
 609         import M2Crypto
 610         m2x509 = M2Crypto.X509.load_cert_string(self.save_to_string())
 611         pkey = Keypair()
 612         pkey.key = self.x509.get_pubkey()
 613         pkey.m2key = m2x509.get_pubkey()
 614         return pkey
 615
 616     def set_intermediate_ca(self, val):
 617         return self.set_is_ca(val)
 618
 619     # Set whether this cert is for a CA. All signers and only signers should be CAs.
 620     # The local member starts unset, letting us check that you only set it once
 621     # @param val Boolean indicating whether this cert is for a CA
 622     def set_is_ca(self, val):
 623         if val is None:
 624             return
 625
 626         if self.isCA != None:
 627             # Can't double set properties
 628             raise Exception("Cannot set basicConstraints CA:?? more than once. "
 629                             "Was {}, trying to set as {}%s".format(self.isCA, val))
 630
 631         self.isCA = val
 632         if val:
 633             self.add_extension('basicConstraints', 1, 'CA:TRUE')
 634         else:
 635             self.add_extension('basicConstraints', 1, 'CA:FALSE')
 636
 637     ##
 638     # Add an X509 extension to the certificate. Add_extension can only be called
 639     # once for a particular extension name, due to limitations in the underlying
 640     # library.
 641     #
 642     # @param name string containing name of extension
 643     # @param value string containing value of the extension
 644
 645     def add_extension(self, name, critical, value):
 646         oldExtVal = None
 647         try:
 648             oldExtVal = self.get_extension(name)
 649         except:
 650             # M2Crypto LookupError when the extension isn't there (yet)
 651             pass
 652
 653         # This code limits you from adding the extension with the same value
 654         # The method comment says you shouldn't do this with the same name
 655         # But actually it (m2crypto) appears to allow you to do this.
 656         if oldExtVal and oldExtVal == value:
 657             # don't add this extension again
 658             # just do nothing as here
 659             return
 660         # FIXME: What if they are trying to set with a different value?
 661         # Is this ever OK? Or should we raise an exception?
 662 #        elif oldExtVal:
 663 #            raise "Cannot add extension {} which had val {} with new val {}".format(name, oldExtVal, value)
 664
 665         ext = OpenSSL.crypto.X509Extension(name, critical, value)
 666         self.x509.add_extensions([ext])
 667
 668     ##
 669     # Get an X509 extension from the certificate
 670
 671     def get_extension(self, name):
 672
 673         import M2Crypto
 674         if name is None:
 675             return None
 676
 677         certstr = self.save_to_string()
 678         if certstr is None or certstr == "":
 679             return None
 680         # pyOpenSSL does not have a way to get extensions
 681         m2x509 = M2Crypto.X509.load_cert_string(certstr)
 682         if m2x509 is None:
 683             logger.warn("No cert loaded in get_extension")
 684             return None
 685         if m2x509.get_ext(name) is None:
 686             return None
 687         value = m2x509.get_ext(name).get_value()
 688
 689         return value
 690
 691     ##
 692     # Set_data is a wrapper around add_extension. It stores the parameter str in
 693     # the X509 subject_alt_name extension. Set_data can only be called once, due
 694     # to limitations in the underlying library.
 695
 696     def set_data(self, string, field='subjectAltName'):
 697         # pyOpenSSL only allows us to add extensions, so if we try to set the
 698         # same extension more than once, it will not work
 699         if field in self.data:
 700             raise Exception("Cannot set {} more than once".format(field))
 701         self.data[field] = string
 702         self.add_extension(field, 0, string)
 703
 704     ##
 705     # Return the data string that was previously set with set_data
 706
 707     def get_data(self, field='subjectAltName'):
 708         if field in self.data:
 709             return self.data[field]
 710
 711         try:
 712             uri = self.get_extension(field)
 713             self.data[field] = uri
 714         except LookupError:
 715             return None
 716
 717         return self.data[field]
 718
 719     ##
 720     # Sign the certificate using the issuer private key and issuer subject
 721     # previous set with set_issuer().
 722
 723     def sign(self):
 724         logger.debug('certificate.sign')
 725         assert self.x509 != None
 726         assert self.issuerSubject != None
 727         assert self.issuerKey != None
 728         self.x509.set_issuer(self.issuerSubject)
 729         self.x509.sign(self.issuerKey.get_openssl_pkey(), self.digest)
 730
 731     ##
 732     # Verify the authenticity of a certificate.
 733     # @param pkey is a Keypair object representing a public key. If Pkey
 734     #     did not sign the certificate, then an exception will be thrown.
 735
 736     def verify(self, pubkey):
 737         import M2Crypto
 738         # pyOpenSSL does not have a way to verify signatures
 739         m2x509 = M2Crypto.X509.load_cert_string(self.save_to_string())
 740         m2pubkey = pubkey.get_m2_pubkey()
 741         # verify it
 742         # https://www.openssl.org/docs/man1.1.0/crypto/X509_verify.html
 743         # verify returns
 744         # 1  if it checks out
 745         # 0  if if does not
 746         # -1 if it could not be checked 'for some reason'
 747         m2result = m2x509.verify(m2pubkey)
 748         result = m2result == 1
 749         if debug_verify_chain:
 750             logger.debug("Certificate.verify: <- {} (m2={}) ({} x {})"
 751                          .format(result, m2result, self.pretty_cert(), m2pubkey))
 752         return result
 753
 754         # XXX alternatively, if openssl has been patched, do the much simpler:
 755         # try:
 756         #   self.x509.verify(pkey.get_openssl_key())
 757         #   return 1
 758         # except:
 759         #   return 0
 760
 761     ##
 762     # Return True if pkey is identical to the public key that is contained in the certificate.
 763     # @param pkey Keypair object
 764
 765     def is_pubkey(self, pkey):
 766         return self.get_pubkey().is_same(pkey)
 767
 768     ##
 769     # Given a certificate cert, verify that this certificate was signed by the
 770     # public key contained in cert. Throw an exception otherwise.
 771     #
 772     # @param cert certificate object
 773
 774     def is_signed_by_cert(self, cert):
 775         logger.debug("Certificate.is_signed_by_cert -> invoking verify")
 776         k = cert.get_pubkey()
 777         result = self.verify(k)
 778         return result
 779
 780     ##
 781     # Set the parent certficiate.
 782     #
 783     # @param p certificate object.
 784
 785     def set_parent(self, p):
 786         self.parent = p
 787
 788     ##
 789     # Return the certificate object of the parent of this certificate.
 790
 791     def get_parent(self):
 792         return self.parent
 793
 794     ##
 795     # Verification examines a chain of certificates to ensure that each parent
 796     # signs the child, and that some certificate in the chain is signed by a
 797     # trusted certificate.
 798     #
 799     # Verification is a basic recursion: <pre>
 800     #     if this_certificate was signed by trusted_certs:
 801     #         return
 802     #     else
 803     #         return verify_chain(parent, trusted_certs)
 804     # </pre>
 805     #
 806     # At each recursion, the parent is tested to ensure that it did sign the
 807     # child. If a parent did not sign a child, then an exception is thrown. If
 808     # the bottom of the recursion is reached and the certificate does not match
 809     # a trusted root, then an exception is thrown.
 810     # Also require that parents are CAs.
 811     #
 812     # @param Trusted_certs is a list of certificates that are trusted.
 813     #
 814
 815     def verify_chain(self, trusted_certs=None):
 816         # Verify a chain of certificates. Each certificate must be signed by
 817         # the public key contained in it's parent. The chain is recursed
 818         # until a certificate is found that is signed by a trusted root.
 819
 820         logger.debug("Certificate.verify_chain {}".format(self.pretty_name()))
 821         # verify expiration time
 822         if self.x509.has_expired():
 823             if debug_verify_chain:
 824                 logger.debug("verify_chain: NO, Certificate {} has expired"
 825                              .format(self.pretty_cert()))
 826             raise CertExpired(self.pretty_cert(), "client cert")
 827
 828         # if this cert is signed by a trusted_cert, then we are set
 829         for i, trusted_cert in enumerate(trusted_certs, 1):
 830             logger.debug("Certificate.verify_chain - trying trusted #{} : {}"
 831                          .format(i, trusted_cert.pretty_name()))
 832             if self.is_signed_by_cert(trusted_cert):
 833                 # verify expiration of trusted_cert ?
 834                 if not trusted_cert.x509.has_expired():
 835                     if debug_verify_chain:
 836                         logger.debug("verify_chain: YES. Cert {} signed by trusted cert {}"
 837                                      .format(self.pretty_name(), trusted_cert.pretty_name()))
 838                     return trusted_cert
 839                 else:
 840                     if debug_verify_chain:
 841                         logger.debug("verify_chain: NO. Cert {} is signed by trusted_cert {}, "
 842                                      "but that signer is expired..."
 843                                      .format(self.pretty_cert(), trusted_cert.pretty_cert()))
 844                     raise CertExpired("{} signer trusted_cert {}"
 845                                       .format(self.pretty_name(), trusted_cert.pretty_name()))
 846             else:
 847                 logger.debug("verify_chain: not a direct descendant of a trusted root".
 848                              format(self.pretty_name(), trusted_cert))
 849
 850         # if there is no parent, then no way to verify the chain
 851         if not self.parent:
 852             if debug_verify_chain:
 853                 logger.debug("verify_chain: NO. {} has no parent "
 854                              "and issuer {} is not in {} trusted roots"
 855                              .format(self.pretty_name(), self.get_issuer(), len(trusted_certs)))
 856             raise CertMissingParent("{}: Issuer {} is not one of the {} trusted roots, "
 857                                     "and cert has no parent."
 858                                     .format(self.pretty_name(), self.get_issuer(), len(trusted_certs)))
 859
 860         # if it wasn't signed by the parent...
 861         if not self.is_signed_by_cert(self.parent):
 862             if debug_verify_chain:
 863                 logger.debug("verify_chain: NO. {} is not signed by parent {}"
 864                              .format(self.pretty_name(),
 865                                      self.parent.pretty_name()))
 866                 self.save_to_file("/tmp/xxx-capture.pem", save_parents=True)
 867             raise CertNotSignedByParent("{}: Parent {}, issuer {}"
 868                                         .format(self.pretty_name(),
 869                                                 self.parent.pretty_name(),
 870                                                 self.get_issuer()))
 871
 872         # Confirm that the parent is a CA. Only CAs can be trusted as
 873         # signers.
 874         # Note that trusted roots are not parents, so don't need to be
 875         # CAs.
 876         # Ugly - cert objects aren't parsed so we need to read the
 877         # extension and hope there are no other basicConstraints
 878         if not self.parent.isCA and not (self.parent.get_extension('basicConstraints') == 'CA:TRUE'):
 879             logger.warn("verify_chain: cert {}'s parent {} is not a CA"
 880                         .format(self.pretty_name(), self.parent.pretty_name()))
 881             raise CertNotSignedByParent("{}: Parent {} not a CA"
 882                                         .format(self.pretty_name(), self.parent.pretty_name()))
 883
 884         # if the parent isn't verified...
 885         if debug_verify_chain:
 886             logger.debug("verify_chain: .. {}, -> verifying parent {}"
 887                          .format(self.pretty_name(),self.parent.pretty_name()))
 888         self.parent.verify_chain(trusted_certs)
 889
 890         return
 891
 892     # more introspection
 893     def get_extensions(self):
 894         import M2Crypto
 895         # pyOpenSSL does not have a way to get extensions
 896         triples = []
 897         m2x509 = M2Crypto.X509.load_cert_string(self.save_to_string())
 898         nb_extensions = m2x509.get_ext_count()
 899         logger.debug("X509 had {} extensions".format(nb_extensions))
 900         for i in range(nb_extensions):
 901             ext = m2x509.get_ext_at(i)
 902             triples.append(
 903                 (ext.get_name(), ext.get_value(), ext.get_critical(),))
 904         return triples
 905
 906     def get_data_names(self):
 907         return self.data.keys()
 908
 909     def get_all_datas(self):
 910         triples = self.get_extensions()
 911         for name in self.get_data_names():
 912             triples.append((name, self.get_data(name), 'data',))
 913         return triples
 914
 915     # only informative
 916     def get_filename(self):
 917         return getattr(self, 'filename', None)
 918
 919     def dump(self, *args, **kwargs):
 920         print(self.dump_string(*args, **kwargs))
 921
 922     def dump_string(self, show_extensions=False):
 923         result = ""
 924         result += "CERTIFICATE for {}\n".format(self.pretty_cert())
 925         result += "Issued by {}\n".format(self.get_issuer())
 926         filename = self.get_filename()
 927         if filename:
 928             result += "Filename {}\n".format(filename)
 929         if show_extensions:
 930             all_datas = self.get_all_datas()
 931             result += " has {} extensions/data attached".format(len(all_datas))
 932             for n, v, c in all_datas:
 933                 if c == 'data':
 934                     result += "   data: {}={}\n".format(n, v)
 935                 else:
 936                     result += "    ext: {} (crit={})=<<<{}>>>\n".format(n, c, v)
 937         return result