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 import functools
  39 import os
  40 import tempfile
  41 import base64
  42 from tempfile import mkstemp
  43
  44 from OpenSSL import crypto
  45 import M2Crypto
  46 from M2Crypto import X509
  47
  48 from sfa.util.faults import CertExpired, CertMissingParent, CertNotSignedByParent
  49 from sfa.util.sfalogging import logger
  50
  51 glo_passphrase_callback = None
  52
  53 ##
  54 # A global callback msy be implemented for requesting passphrases from the
  55 # user. The function will be called with three arguments:
  56 #
  57 #    keypair_obj: the keypair object that is calling the passphrase
  58 #    string: the string containing the private key that's being loaded
  59 #    x: unknown, appears to be 0, comes from pyOpenSSL and/or m2crypto
  60 #
  61 # The callback should return a string containing the passphrase.
  62
  63 def set_passphrase_callback(callback_func):
  64     global glo_passphrase_callback
  65
  66     glo_passphrase_callback = callback_func
  67
  68 ##
  69 # Sets a fixed passphrase.
  70
  71 def set_passphrase(passphrase):
  72     set_passphrase_callback( lambda k,s,x: passphrase )
  73
  74 ##
  75 # Check to see if a passphrase works for a particular private key string.
  76 # Intended to be used by passphrase callbacks for input validation.
  77
  78 def test_passphrase(string, passphrase):
  79     try:
  80         crypto.load_privatekey(crypto.FILETYPE_PEM, string, (lambda x: passphrase))
  81         return True
  82     except:
  83         return False
  84
  85 def convert_public_key(key):
  86     keyconvert_path = "/usr/bin/keyconvert.py"
  87     if not os.path.isfile(keyconvert_path):
  88         raise IOError, "Could not find keyconvert in %s" % keyconvert_path
  89
  90     # we can only convert rsa keys
  91     if "ssh-dss" in key:
  92         raise Exception, "keyconvert: dss keys are not supported"
  93
  94     (ssh_f, ssh_fn) = tempfile.mkstemp()
  95     ssl_fn = tempfile.mktemp()
  96     os.write(ssh_f, key)
  97     os.close(ssh_f)
  98
  99     cmd = keyconvert_path + " " + ssh_fn + " " + ssl_fn
 100     os.system(cmd)
 101
 102     # this check leaves the temporary file containing the public key so
 103     # that it can be expected to see why it failed.
 104     # TODO: for production, cleanup the temporary files
 105     if not os.path.exists(ssl_fn):
 106         raise Exception, "keyconvert: generated certificate not found. keyconvert may have failed."
 107
 108     k = Keypair()
 109     try:
 110         k.load_pubkey_from_file(ssl_fn)
 111         return k
 112     except:
 113         logger.log_exc("convert_public_key caught exception")
 114         raise
 115     finally:
 116         # remove the temporary files
 117         if os.path.exists(ssh_fn):
 118             os.remove(ssh_fn)
 119         if os.path.exists(ssl_fn):
 120             os.remove(ssl_fn)
 121
 122
 123 ##
 124 # Public-private key pairs are implemented by the Keypair class.
 125 # A Keypair object may represent both a public and private key pair, or it
 126 # may represent only a public key (this usage is consistent with OpenSSL).
 127
 128 class Keypair:
 129     key = None       # public/private keypair
 130     m2key = None     # public key (m2crypto format)
 131
 132     ##
 133     # Creates a Keypair object
 134     # @param create If create==True, creates a new public/private key and
 135     #     stores it in the object
 136     # @param string If string!=None, load the keypair from the string (PEM)
 137     # @param filename If filename!=None, load the keypair from the file
 138
 139     def __init__(self, create=False, string=None, filename=None):
 140         if create:
 141             self.create()
 142         if string:
 143             self.load_from_string(string)
 144         if filename:
 145             self.load_from_file(filename)
 146
 147     ##
 148     # Create a RSA public/private key pair and store it inside the keypair object
 149
 150     def create(self):
 151         self.key = crypto.PKey()
 152         self.key.generate_key(crypto.TYPE_RSA, 1024)
 153
 154     ##
 155     # Save the private key to a file
 156     # @param filename name of file to store the keypair in
 157
 158     def save_to_file(self, filename):
 159         open(filename, 'w').write(self.as_pem())
 160         self.filename=filename
 161
 162     ##
 163     # Load the private key from a file. Implicity the private key includes the public key.
 164
 165     def load_from_file(self, filename):
 166         self.filename=filename
 167         buffer = open(filename, 'r').read()
 168         self.load_from_string(buffer)
 169
 170     ##
 171     # Load the private key from a string. Implicitly the private key includes the public key.
 172
 173     def load_from_string(self, string):
 174         if glo_passphrase_callback:
 175             self.key = crypto.load_privatekey(crypto.FILETYPE_PEM, string, functools.partial(glo_passphrase_callback, self, string) )
 176             self.m2key = M2Crypto.EVP.load_key_string(string, functools.partial(glo_passphrase_callback, self, string) )
 177         else:
 178             self.key = crypto.load_privatekey(crypto.FILETYPE_PEM, string)
 179             self.m2key = M2Crypto.EVP.load_key_string(string)
 180
 181     ##
 182     #  Load the public key from a string. No private key is loaded.
 183
 184     def load_pubkey_from_file(self, filename):
 185         # load the m2 public key
 186         m2rsakey = M2Crypto.RSA.load_pub_key(filename)
 187         self.m2key = M2Crypto.EVP.PKey()
 188         self.m2key.assign_rsa(m2rsakey)
 189
 190         # create an m2 x509 cert
 191         m2name = M2Crypto.X509.X509_Name()
 192         m2name.add_entry_by_txt(field="CN", type=0x1001, entry="junk", len=-1, loc=-1, set=0)
 193         m2x509 = M2Crypto.X509.X509()
 194         m2x509.set_pubkey(self.m2key)
 195         m2x509.set_serial_number(0)
 196         m2x509.set_issuer_name(m2name)
 197         m2x509.set_subject_name(m2name)
 198         ASN1 = M2Crypto.ASN1.ASN1_UTCTIME()
 199         ASN1.set_time(500)
 200         m2x509.set_not_before(ASN1)
 201         m2x509.set_not_after(ASN1)
 202         # x509v3 so it can have extensions
 203         # prob not necc since this cert itself is junk but still...
 204         m2x509.set_version(2)
 205         junk_key = Keypair(create=True)
 206         m2x509.sign(pkey=junk_key.get_m2_pkey(), md="sha1")
 207
 208         # convert the m2 x509 cert to a pyopenssl x509
 209         m2pem = m2x509.as_pem()
 210         pyx509 = crypto.load_certificate(crypto.FILETYPE_PEM, m2pem)
 211
 212         # get the pyopenssl pkey from the pyopenssl x509
 213         self.key = pyx509.get_pubkey()
 214         self.filename=filename
 215
 216     ##
 217     # Load the public key from a string. No private key is loaded.
 218
 219     def load_pubkey_from_string(self, string):
 220         (f, fn) = tempfile.mkstemp()
 221         os.write(f, string)
 222         os.close(f)
 223         self.load_pubkey_from_file(fn)
 224         os.remove(fn)
 225
 226     ##
 227     # Return the private key in PEM format.
 228
 229     def as_pem(self):
 230         return crypto.dump_privatekey(crypto.FILETYPE_PEM, self.key)
 231
 232     ##
 233     # Return an M2Crypto key object
 234
 235     def get_m2_pkey(self):
 236         if not self.m2key:
 237             self.m2key = M2Crypto.EVP.load_key_string(self.as_pem())
 238         return self.m2key
 239
 240     ##
 241     # Returns a string containing the public key represented by this object.
 242
 243     def get_pubkey_string(self):
 244         m2pkey = self.get_m2_pkey()
 245         return base64.b64encode(m2pkey.as_der())
 246
 247     ##
 248     # Return an OpenSSL pkey object
 249
 250     def get_openssl_pkey(self):
 251         return self.key
 252
 253     ##
 254     # Given another Keypair object, return TRUE if the two keys are the same.
 255
 256     def is_same(self, pkey):
 257         return self.as_pem() == pkey.as_pem()
 258
 259     def sign_string(self, data):
 260         k = self.get_m2_pkey()
 261         k.sign_init()
 262         k.sign_update(data)
 263         return base64.b64encode(k.sign_final())
 264
 265     def verify_string(self, data, sig):
 266         k = self.get_m2_pkey()
 267         k.verify_init()
 268         k.verify_update(data)
 269         return M2Crypto.m2.verify_final(k.ctx, base64.b64decode(sig), k.pkey)
 270
 271     def compute_hash(self, value):
 272         return self.sign_string(str(value))
 273
 274     # only informative
 275     def get_filename(self):
 276         return getattr(self,'filename',None)
 277
 278     def dump (self, *args, **kwargs):
 279         print self.dump_string(*args, **kwargs)
 280
 281     def dump_string (self):
 282         result=""
 283         result += "KEYPAIR: pubkey=%40s..."%self.get_pubkey_string()
 284         filename=self.get_filename()
 285         if filename: result += "Filename %s\n"%filename
 286         return result
 287
 288 ##
 289 # The certificate class implements a general purpose X509 certificate, making
 290 # use of the appropriate pyOpenSSL or M2Crypto abstractions. It also adds
 291 # several addition features, such as the ability to maintain a chain of
 292 # parent certificates, and storage of application-specific data.
 293 #
 294 # Certificates include the ability to maintain a chain of parents. Each
 295 # certificate includes a pointer to it's parent certificate. When loaded
 296 # from a file or a string, the parent chain will be automatically loaded.
 297 # When saving a certificate to a file or a string, the caller can choose
 298 # whether to save the parent certificates as well.
 299
 300 class Certificate:
 301     digest = "md5"
 302
 303     cert = None
 304     issuerKey = None
 305     issuerSubject = None
 306     parent = None
 307     isCA = None # will be a boolean once set
 308
 309     separator="-----parent-----"
 310
 311     ##
 312     # Create a certificate object.
 313     #
 314     # @param lifeDays life of cert in days - default is 1825==5 years
 315     # @param create If create==True, then also create a blank X509 certificate.
 316     # @param subject If subject!=None, then create a blank certificate and set
 317     #     it's subject name.
 318     # @param string If string!=None, load the certficate from the string.
 319     # @param filename If filename!=None, load the certficiate from the file.
 320     # @param isCA If !=None, set whether this cert is for a CA
 321
 322     def __init__(self, lifeDays=1825, create=False, subject=None, string=None, filename=None, isCA=None):
 323         self.data = {}
 324         if create or subject:
 325             self.create(lifeDays)
 326         if subject:
 327             self.set_subject(subject)
 328         if string:
 329             self.load_from_string(string)
 330         if filename:
 331             self.load_from_file(filename)
 332
 333         # Set the CA bit if a value was supplied
 334         if isCA != None:
 335             self.set_is_ca(isCA)
 336
 337     # Create a blank X509 certificate and store it in this object.
 338
 339     def create(self, lifeDays=1825):
 340         self.cert = crypto.X509()
 341         # FIXME: Use different serial #s
 342         self.cert.set_serial_number(3)
 343         self.cert.gmtime_adj_notBefore(0) # 0 means now
 344         self.cert.gmtime_adj_notAfter(lifeDays*60*60*24) # five years is default
 345         self.cert.set_version(2) # x509v3 so it can have extensions
 346
 347
 348     ##
 349     # Given a pyOpenSSL X509 object, store that object inside of this
 350     # certificate object.
 351
 352     def load_from_pyopenssl_x509(self, x509):
 353         self.cert = x509
 354
 355     ##
 356     # Load the certificate from a string
 357
 358     def load_from_string(self, string):
 359         # if it is a chain of multiple certs, then split off the first one and
 360         # load it (support for the ---parent--- tag as well as normal chained certs)
 361
 362         string = string.strip()
 363
 364         # If it's not in proper PEM format, wrap it
 365         if string.count('-----BEGIN CERTIFICATE') == 0:
 366             string = '-----BEGIN CERTIFICATE-----\n%s\n-----END CERTIFICATE-----' % string
 367
 368         # If there is a PEM cert in there, but there is some other text first
 369         # such as the text of the certificate, skip the text
 370         beg = string.find('-----BEGIN CERTIFICATE')
 371         if beg > 0:
 372             # skipping over non cert beginning
 373             string = string[beg:]
 374
 375         parts = []
 376
 377         if string.count('-----BEGIN CERTIFICATE-----') > 1 and \
 378                string.count(Certificate.separator) == 0:
 379             parts = string.split('-----END CERTIFICATE-----',1)
 380             parts[0] += '-----END CERTIFICATE-----'
 381         else:
 382             parts = string.split(Certificate.separator, 1)
 383
 384         self.cert = crypto.load_certificate(crypto.FILETYPE_PEM, parts[0])
 385
 386         # if there are more certs, then create a parent and let the parent load
 387         # itself from the remainder of the string
 388         if len(parts) > 1 and parts[1] != '':
 389             self.parent = self.__class__()
 390             self.parent.load_from_string(parts[1])
 391
 392     ##
 393     # Load the certificate from a file
 394
 395     def load_from_file(self, filename):
 396         file = open(filename)
 397         string = file.read()
 398         self.load_from_string(string)
 399         self.filename=filename
 400
 401     ##
 402     # Save the certificate to a string.
 403     #
 404     # @param save_parents If save_parents==True, then also save the parent certificates.
 405
 406     def save_to_string(self, save_parents=True):
 407         string = crypto.dump_certificate(crypto.FILETYPE_PEM, self.cert)
 408         if save_parents and self.parent:
 409             string = string + self.parent.save_to_string(save_parents)
 410         return string
 411
 412     ##
 413     # Save the certificate to a file.
 414     # @param save_parents If save_parents==True, then also save the parent certificates.
 415
 416     def save_to_file(self, filename, save_parents=True, filep=None):
 417         string = self.save_to_string(save_parents=save_parents)
 418         if filep:
 419             f = filep
 420         else:
 421             f = open(filename, 'w')
 422         f.write(string)
 423         f.close()
 424         self.filename=filename
 425
 426     ##
 427     # Save the certificate to a random file in /tmp/
 428     # @param save_parents If save_parents==True, then also save the parent certificates.
 429     def save_to_random_tmp_file(self, save_parents=True):
 430         fp, filename = mkstemp(suffix='cert', text=True)
 431         fp = os.fdopen(fp, "w")
 432         self.save_to_file(filename, save_parents=True, filep=fp)
 433         return filename
 434
 435     ##
 436     # Sets the issuer private key and name
 437     # @param key Keypair object containing the private key of the issuer
 438     # @param subject String containing the name of the issuer
 439     # @param cert (optional) Certificate object containing the name of the issuer
 440
 441     def set_issuer(self, key, subject=None, cert=None):
 442         self.issuerKey = key
 443         if subject:
 444             # it's a mistake to use subject and cert params at the same time
 445             assert(not cert)
 446             if isinstance(subject, dict) or isinstance(subject, str):
 447                 req = crypto.X509Req()
 448                 reqSubject = req.get_subject()
 449                 if (isinstance(subject, dict)):
 450                     for key in reqSubject.keys():
 451                         setattr(reqSubject, key, subject[key])
 452                 else:
 453                     setattr(reqSubject, "CN", subject)
 454                 subject = reqSubject
 455                 # subject is not valid once req is out of scope, so save req
 456                 self.issuerReq = req
 457         if cert:
 458             # if a cert was supplied, then get the subject from the cert
 459             subject = cert.cert.get_subject()
 460         assert(subject)
 461         self.issuerSubject = subject
 462
 463     ##
 464     # Get the issuer name
 465
 466     def get_issuer(self, which="CN"):
 467         x = self.cert.get_issuer()
 468         return getattr(x, which)
 469
 470     ##
 471     # Set the subject name of the certificate
 472
 473     def set_subject(self, name):
 474         req = crypto.X509Req()
 475         subj = req.get_subject()
 476         if (isinstance(name, dict)):
 477             for key in name.keys():
 478                 setattr(subj, key, name[key])
 479         else:
 480             setattr(subj, "CN", name)
 481         self.cert.set_subject(subj)
 482
 483     ##
 484     # Get the subject name of the certificate
 485
 486     def get_subject(self, which="CN"):
 487         x = self.cert.get_subject()
 488         return getattr(x, which)
 489
 490     ##
 491     # Get a pretty-print subject name of the certificate
 492
 493     def get_printable_subject(self):
 494         x = self.cert.get_subject()
 495         return "[ OU: %s, CN: %s, SubjectAltName: %s ]" % (getattr(x, "OU"), getattr(x, "CN"), self.get_data())
 496
 497     ##
 498     # Get the public key of the certificate.
 499     #
 500     # @param key Keypair object containing the public key
 501
 502     def set_pubkey(self, key):
 503         assert(isinstance(key, Keypair))
 504         self.cert.set_pubkey(key.get_openssl_pkey())
 505
 506     ##
 507     # Get the public key of the certificate.
 508     # It is returned in the form of a Keypair object.
 509
 510     def get_pubkey(self):
 511         m2x509 = X509.load_cert_string(self.save_to_string())
 512         pkey = Keypair()
 513         pkey.key = self.cert.get_pubkey()
 514         pkey.m2key = m2x509.get_pubkey()
 515         return pkey
 516
 517     def set_intermediate_ca(self, val):
 518         return self.set_is_ca(val)
 519
 520     # Set whether this cert is for a CA. All signers and only signers should be CAs.
 521     # The local member starts unset, letting us check that you only set it once
 522     # @param val Boolean indicating whether this cert is for a CA
 523     def set_is_ca(self, val):
 524         if val is None:
 525             return
 526
 527         if self.isCA != None:
 528             # Can't double set properties
 529             raise Exception, "Cannot set basicConstraints CA:?? more than once. Was %s, trying to set as %s" % (self.isCA, val)
 530
 531         self.isCA = val
 532         if val:
 533             self.add_extension('basicConstraints', 1, 'CA:TRUE')
 534         else:
 535             self.add_extension('basicConstraints', 1, 'CA:FALSE')
 536
 537
 538
 539     ##
 540     # Add an X509 extension to the certificate. Add_extension can only be called
 541     # once for a particular extension name, due to limitations in the underlying
 542     # library.
 543     #
 544     # @param name string containing name of extension
 545     # @param value string containing value of the extension
 546
 547     def add_extension(self, name, critical, value):
 548         oldExtVal = None
 549         try:
 550             oldExtVal = self.get_extension(name)
 551         except:
 552             # M2Crypto LookupError when the extension isn't there (yet)
 553             pass
 554
 555         # This code limits you from adding the extension with the same value
 556         # The method comment says you shouldn't do this with the same name
 557         # But actually it (m2crypto) appears to allow you to do this.
 558         if oldExtVal and oldExtVal == value:
 559             # don't add this extension again
 560             # just do nothing as here
 561             return
 562         # FIXME: What if they are trying to set with a different value?
 563         # Is this ever OK? Or should we raise an exception?
 564 #        elif oldExtVal:
 565 #            raise "Cannot add extension %s which had val %s with new val %s" % (name, oldExtVal, value)
 566
 567         ext = crypto.X509Extension (name, critical, value)
 568         self.cert.add_extensions([ext])
 569
 570     ##
 571     # Get an X509 extension from the certificate
 572
 573     def get_extension(self, name):
 574
 575         # pyOpenSSL does not have a way to get extensions
 576         m2x509 = X509.load_cert_string(self.save_to_string())
 577         value = m2x509.get_ext(name).get_value()
 578
 579         return value
 580
 581     ##
 582     # Set_data is a wrapper around add_extension. It stores the parameter str in
 583     # the X509 subject_alt_name extension. Set_data can only be called once, due
 584     # to limitations in the underlying library.
 585
 586     def set_data(self, str, field='subjectAltName'):
 587         # pyOpenSSL only allows us to add extensions, so if we try to set the
 588         # same extension more than once, it will not work
 589         if self.data.has_key(field):
 590             raise "Cannot set ", field, " more than once"
 591         self.data[field] = str
 592         self.add_extension(field, 0, str)
 593
 594     ##
 595     # Return the data string that was previously set with set_data
 596
 597     def get_data(self, field='subjectAltName'):
 598         if self.data.has_key(field):
 599             return self.data[field]
 600
 601         try:
 602             uri = self.get_extension(field)
 603             self.data[field] = uri
 604         except LookupError:
 605             return None
 606
 607         return self.data[field]
 608
 609     ##
 610     # Sign the certificate using the issuer private key and issuer subject previous set with set_issuer().
 611
 612     def sign(self):
 613         logger.debug('certificate.sign')
 614         assert self.cert != None
 615         assert self.issuerSubject != None
 616         assert self.issuerKey != None
 617         self.cert.set_issuer(self.issuerSubject)
 618         self.cert.sign(self.issuerKey.get_openssl_pkey(), self.digest)
 619
 620     ##
 621     # Verify the authenticity of a certificate.
 622     # @param pkey is a Keypair object representing a public key. If Pkey
 623     #     did not sign the certificate, then an exception will be thrown.
 624
 625     def verify(self, pkey):
 626         # pyOpenSSL does not have a way to verify signatures
 627         m2x509 = X509.load_cert_string(self.save_to_string())
 628         m2pkey = pkey.get_m2_pkey()
 629         # verify it
 630         return m2x509.verify(m2pkey)
 631
 632         # XXX alternatively, if openssl has been patched, do the much simpler:
 633         # try:
 634         #   self.cert.verify(pkey.get_openssl_key())
 635         #   return 1
 636         # except:
 637         #   return 0
 638
 639     ##
 640     # Return True if pkey is identical to the public key that is contained in the certificate.
 641     # @param pkey Keypair object
 642
 643     def is_pubkey(self, pkey):
 644         return self.get_pubkey().is_same(pkey)
 645
 646     ##
 647     # Given a certificate cert, verify that this certificate was signed by the
 648     # public key contained in cert. Throw an exception otherwise.
 649     #
 650     # @param cert certificate object
 651
 652     def is_signed_by_cert(self, cert):
 653         k = cert.get_pubkey()
 654         result = self.verify(k)
 655         return result
 656
 657     ##
 658     # Set the parent certficiate.
 659     #
 660     # @param p certificate object.
 661
 662     def set_parent(self, p):
 663         self.parent = p
 664
 665     ##
 666     # Return the certificate object of the parent of this certificate.
 667
 668     def get_parent(self):
 669         return self.parent
 670
 671     ##
 672     # Verification examines a chain of certificates to ensure that each parent
 673     # signs the child, and that some certificate in the chain is signed by a
 674     # trusted certificate.
 675     #
 676     # Verification is a basic recursion: <pre>
 677     #     if this_certificate was signed by trusted_certs:
 678     #         return
 679     #     else
 680     #         return verify_chain(parent, trusted_certs)
 681     # </pre>
 682     #
 683     # At each recursion, the parent is tested to ensure that it did sign the
 684     # child. If a parent did not sign a child, then an exception is thrown. If
 685     # the bottom of the recursion is reached and the certificate does not match
 686     # a trusted root, then an exception is thrown.
 687     # Also require that parents are CAs.
 688     #
 689     # @param Trusted_certs is a list of certificates that are trusted.
 690     #
 691
 692     def verify_chain(self, trusted_certs = None):
 693         # Verify a chain of certificates. Each certificate must be signed by
 694         # the public key contained in it's parent. The chain is recursed
 695         # until a certificate is found that is signed by a trusted root.
 696
 697         # verify expiration time
 698         if self.cert.has_expired():
 699             logger.debug("verify_chain: NO, Certificate %s has expired" % self.get_printable_subject())
 700             raise CertExpired(self.get_printable_subject(), "client cert")
 701
 702         # if this cert is signed by a trusted_cert, then we are set
 703         for trusted_cert in trusted_certs:
 704             if self.is_signed_by_cert(trusted_cert):
 705                 # verify expiration of trusted_cert ?
 706                 if not trusted_cert.cert.has_expired():
 707                     logger.debug("verify_chain: YES. Cert %s signed by trusted cert %s"%(
 708                             self.get_printable_subject(), trusted_cert.get_printable_subject()))
 709                     return trusted_cert
 710                 else:
 711                     logger.debug("verify_chain: NO. Cert %s is signed by trusted_cert %s, but that signer is expired..."%(
 712                             self.get_printable_subject(),trusted_cert.get_printable_subject()))
 713                     raise CertExpired(self.get_printable_subject()," signer trusted_cert %s"%trusted_cert.get_printable_subject())
 714
 715         # if there is no parent, then no way to verify the chain
 716         if not self.parent:
 717             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)))
 718             raise CertMissingParent(self.get_printable_subject() + ": Issuer %s not trusted by any of %d trusted roots, and cert has no parent." % (self.get_issuer(), len(trusted_certs)))
 719
 720         # if it wasn't signed by the parent...
 721         if not self.is_signed_by_cert(self.parent):
 722             logger.debug("verify_chain: NO. %s is not signed by parent %s, but by %s"%\
 723                              (self.get_printable_subject(),
 724                               self.parent.get_printable_subject(),
 725                               self.get_issuer()))
 726             raise CertNotSignedByParent("%s: Parent %s, issuer %s"\
 727                                             % (self.get_printable_subject(),
 728                                                self.parent.get_printable_subject(),
 729                                                self.get_issuer()))
 730
 731         # Confirm that the parent is a CA. Only CAs can be trusted as
 732         # signers.
 733         # Note that trusted roots are not parents, so don't need to be
 734         # CAs.
 735         # Ugly - cert objects aren't parsed so we need to read the
 736         # extension and hope there are no other basicConstraints
 737         if not self.parent.isCA and not (self.parent.get_extension('basicConstraints') == 'CA:TRUE'):
 738             logger.warn("verify_chain: cert %s's parent %s is not a CA" % \
 739                             (self.get_printable_subject(), self.parent.get_printable_subject()))
 740             raise CertNotSignedByParent("%s: Parent %s not a CA" % (self.get_printable_subject(),
 741                                                                     self.parent.get_printable_subject()))
 742
 743         # if the parent isn't verified...
 744         logger.debug("verify_chain: .. %s, -> verifying parent %s"%\
 745                          (self.get_printable_subject(),self.parent.get_printable_subject()))
 746         self.parent.verify_chain(trusted_certs)
 747
 748         return
 749
 750     ### more introspection
 751     def get_extensions(self):
 752         # pyOpenSSL does not have a way to get extensions
 753         triples=[]
 754         m2x509 = X509.load_cert_string(self.save_to_string())
 755         nb_extensions=m2x509.get_ext_count()
 756         logger.debug("X509 had %d extensions"%nb_extensions)
 757         for i in range(nb_extensions):
 758             ext=m2x509.get_ext_at(i)
 759             triples.append( (ext.get_name(), ext.get_value(), ext.get_critical(),) )
 760         return triples
 761
 762     def get_data_names(self):
 763         return self.data.keys()
 764
 765     def get_all_datas (self):
 766         triples=self.get_extensions()
 767         for name in self.get_data_names():
 768             triples.append( (name,self.get_data(name),'data',) )
 769         return triples
 770
 771     # only informative
 772     def get_filename(self):
 773         return getattr(self,'filename',None)
 774
 775     def dump (self, *args, **kwargs):
 776         print self.dump_string(*args, **kwargs)
 777
 778     def dump_string (self,show_extensions=False):
 779         result = ""
 780         result += "CERTIFICATE for %s\n"%self.get_printable_subject()
 781         result += "Issued by %s\n"%self.get_issuer()
 782         filename=self.get_filename()
 783         if filename: result += "Filename %s\n"%filename
 784         if show_extensions:
 785             all_datas=self.get_all_datas()
 786             result += " has %d extensions/data attached"%len(all_datas)
 787             for (n,v,c) in all_datas:
 788                 if c=='data':
 789                     result += "   data: %s=%s\n"%(n,v)
 790                 else:
 791                     result += "    ext: %s (crit=%s)=<<<%s>>>\n"%(n,c,v)
 792         return result