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