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,
  41 # assuming it is split into pieces p1.pem p2.pem p3.pem
  42 # you can use openssl to verify the chain using this command
  43 # openssl verify -verbose -CAfile <(cat p2.pem p3.pem) p1.pem
  44 # also you can use sfax509 to invoke openssl x509 on all parts of the gid
  45 #
  46
  47
  48
  49
  50 import functools
  51 import os
  52 import tempfile
  53 import base64
  54 from tempfile import mkstemp
  55
  56 import OpenSSL
  57 # M2Crypto is imported on the fly to minimize crashes
  58 # import M2Crypto
  59
  60 from sfa.util.faults import (CertExpired, CertMissingParent,
  61                              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 = True
  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.encode())
 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             "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         with open(filename, 'wb') as output:
 184             output.write(self.as_pem())
 185         self.filename = filename
 186
 187     ##
 188     # Load the private key from a file. Implicity the private key includes the
 189     # public key.
 190
 191     def load_from_file(self, filename):
 192         self.filename = filename
 193         buffer = open(filename, 'r').read()
 194         self.load_from_string(buffer)
 195
 196     ##
 197     # Load the private key from a string. Implicitly the private key includes
 198     # the public key.
 199
 200     def load_from_string(self, string):
 201         import M2Crypto
 202         if glo_passphrase_callback:
 203             self.key = OpenSSL.crypto.load_privatekey(
 204                 OpenSSL.crypto.FILETYPE_PEM, string,
 205                 functools.partial(glo_passphrase_callback, self, string))
 206             self.m2key = M2Crypto.EVP.load_key_string(
 207                 string.encode(encoding="utf-8"),
 208                 functools.partial(glo_passphrase_callback, self, string))
 209         else:
 210             self.key = OpenSSL.crypto.load_privatekey(
 211                 OpenSSL.crypto.FILETYPE_PEM, string)
 212             self.m2key = M2Crypto.EVP.load_key_string(
 213                 string.encode(encoding="utf-8"))
 214
 215     ##
 216     #  Load the public key from a string. No private key is loaded.
 217
 218     def load_pubkey_from_file(self, filename):
 219         import M2Crypto
 220         # load the m2 public key
 221         m2rsakey = M2Crypto.RSA.load_pub_key(filename)
 222         self.m2key = M2Crypto.EVP.PKey()
 223         self.m2key.assign_rsa(m2rsakey)
 224
 225         # create an m2 x509 cert
 226         m2name = M2Crypto.X509.X509_Name()
 227         m2name.add_entry_by_txt(field="CN", type=0x1001,
 228                                 entry="junk", len=-1, loc=-1, set=0)
 229         m2x509 = M2Crypto.X509.X509()
 230         m2x509.set_pubkey(self.m2key)
 231         m2x509.set_serial_number(0)
 232         m2x509.set_issuer_name(m2name)
 233         m2x509.set_subject_name(m2name)
 234         ASN1 = M2Crypto.ASN1.ASN1_UTCTIME()
 235         ASN1.set_time(500)
 236         m2x509.set_not_before(ASN1)
 237         m2x509.set_not_after(ASN1)
 238         # x509v3 so it can have extensions
 239         # prob not necc since this cert itself is junk but still...
 240         m2x509.set_version(2)
 241         junk_key = Keypair(create=True)
 242         m2x509.sign(pkey=junk_key.get_m2_pubkey(), md="sha1")
 243
 244         # convert the m2 x509 cert to a pyopenssl x509
 245         m2pem = m2x509.as_pem()
 246         pyx509 = OpenSSL.crypto.load_certificate(
 247             OpenSSL.crypto.FILETYPE_PEM, m2pem)
 248
 249         # get the pyopenssl pkey from the pyopenssl x509
 250         self.key = pyx509.get_pubkey()
 251         self.filename = filename
 252
 253     ##
 254     # Load the public key from a string. No private key is loaded.
 255
 256     def load_pubkey_from_string(self, string):
 257         (f, fn) = tempfile.mkstemp()
 258         os.write(f, string)
 259         os.close(f)
 260         self.load_pubkey_from_file(fn)
 261         os.remove(fn)
 262
 263     ##
 264     # Return the private key in PEM format.
 265
 266     def as_pem(self):
 267         return OpenSSL.crypto.dump_privatekey(
 268             OpenSSL.crypto.FILETYPE_PEM, self.key)
 269
 270     ##
 271     # Return an M2Crypto key object
 272
 273     def get_m2_pubkey(self):
 274         import M2Crypto
 275         if not self.m2key:
 276             self.m2key = M2Crypto.EVP.load_key_string(self.as_pem())
 277         return self.m2key
 278
 279     ##
 280     # Returns a string containing the public key represented by this object.
 281
 282     def get_pubkey_string(self):
 283         m2pkey = self.get_m2_pubkey()
 284         return base64.b64encode(m2pkey.as_der())
 285
 286     ##
 287     # Return an OpenSSL pkey object
 288
 289     def get_openssl_pkey(self):
 290         return self.key
 291
 292     ##
 293     # Given another Keypair object, return TRUE if the two keys are the same.
 294
 295     def is_same(self, pkey):
 296         return self.as_pem() == pkey.as_pem()
 297
 298     def sign_string(self, data):
 299         k = self.get_m2_pubkey()
 300         k.sign_init()
 301         k.sign_update(data)
 302         return base64.b64encode(k.sign_final())
 303
 304     def verify_string(self, data, sig):
 305         import M2Crypto
 306         k = self.get_m2_pubkey()
 307         k.verify_init()
 308         k.verify_update(data)
 309         return M2Crypto.m2.verify_final(k.ctx, base64.b64decode(sig), k.pkey)
 310
 311     def compute_hash(self, value):
 312         return self.sign_string(str(value))
 313
 314     # only informative
 315     def get_filename(self):
 316         return getattr(self, 'filename', None)
 317
 318     def dump(self, *args, **kwargs):
 319         print(self.dump_string(*args, **kwargs))
 320
 321     def dump_string(self):
 322         result = ""
 323         result += "KEYPAIR: pubkey={:>40}...".format(self.get_pubkey_string())
 324         filename = self.get_filename()
 325         if filename:
 326             result += "Filename {}\n".format(filename)
 327         return result
 328
 329 ##
 330 # The certificate class implements a general purpose X509 certificate, making
 331 # use of the appropriate pyOpenSSL or M2Crypto abstractions. It also adds
 332 # several addition features, such as the ability to maintain a chain of
 333 # parent certificates, and storage of application-specific data.
 334 #
 335 # Certificates include the ability to maintain a chain of parents. Each
 336 # certificate includes a pointer to it's parent certificate. When loaded
 337 # from a file or a string, the parent chain will be automatically loaded.
 338 # When saving a certificate to a file or a string, the caller can choose
 339 # whether to save the parent certificates as well.
 340
 341
 342 class Certificate:
 343     digest = "sha256"
 344
 345 #    x509 = None
 346 #    issuerKey = None
 347 #    issuerSubject = None
 348 #    parent = None
 349     isCA = None  # will be a boolean once set
 350
 351     separator = "-----parent-----"
 352
 353     ##
 354     # Create a certificate object.
 355     #
 356     # @param lifeDays life of cert in days - default is 1825==5 years
 357     # @param create If create==True, then also create a blank X509 certificate.
 358     # @param subject If subject!=None, then create a blank certificate and set
 359     #     it's subject name.
 360     # @param string If string!=None, load the certificate from the string.
 361     # @param filename If filename!=None, load the certificate from the file.
 362     # @param isCA If !=None, set whether this cert is for a CA
 363
 364     def __init__(self, lifeDays=1825, create=False, subject=None, string=None,
 365                  filename=None, isCA=None):
 366         # these used to be defined in the class !
 367         self.x509 = None
 368         self.issuerKey = None
 369         self.issuerSubject = None
 370         self.parent = None
 371
 372         self.data = {}
 373         if create or subject:
 374             self.create(lifeDays)
 375         if subject:
 376             self.set_subject(subject)
 377         if string:
 378             self.load_from_string(string)
 379         if filename:
 380             self.load_from_file(filename)
 381
 382         # Set the CA bit if a value was supplied
 383         if isCA is not None:
 384             self.set_is_ca(isCA)
 385
 386     # Create a blank X509 certificate and store it in this object.
 387
 388     def create(self, lifeDays=1825):
 389         self.x509 = OpenSSL.crypto.X509()
 390         # FIXME: Use different serial #s
 391         self.x509.set_serial_number(3)
 392         self.x509.gmtime_adj_notBefore(0)  # 0 means now
 393         self.x509.gmtime_adj_notAfter(
 394             lifeDays * 60 * 60 * 24)  # five years is default
 395         self.x509.set_version(2)  # x509v3 so it can have extensions
 396
 397     ##
 398     # Given a pyOpenSSL X509 object, store that object inside of this
 399     # certificate object.
 400
 401     def load_from_pyopenssl_x509(self, x509):
 402         self.x509 = x509
 403
 404     ##
 405     # Load the certificate from a string
 406
 407     def load_from_string(self, string):
 408         # if it is a chain of multiple certs, then split off the first one and
 409         # load it (support for the ---parent--- tag as well as normal chained
 410         # certs)
 411
 412         if string is None or string.strip() == "":
 413             logger.warning("Empty string in load_from_string")
 414             return
 415
 416         string = string.strip()
 417
 418         # If it's not in proper PEM format, wrap it
 419         if string.count('-----BEGIN CERTIFICATE') == 0:
 420             string = '-----BEGIN CERTIFICATE-----'\
 421                      '\n{}\n-----END CERTIFICATE-----'\
 422                      .format(string)
 423
 424         # If there is a PEM cert in there, but there is some other text first
 425         # such as the text of the certificate, skip the text
 426         beg = string.find('-----BEGIN CERTIFICATE')
 427         if beg > 0:
 428             # skipping over non cert beginning
 429             string = string[beg:]
 430
 431         parts = []
 432
 433         if string.count('-----BEGIN CERTIFICATE-----') > 1 and \
 434                 string.count(Certificate.separator) == 0:
 435             parts = string.split('-----END CERTIFICATE-----', 1)
 436             parts[0] += '-----END CERTIFICATE-----'
 437         else:
 438             parts = string.split(Certificate.separator, 1)
 439
 440         self.x509 = OpenSSL.crypto.load_certificate(
 441             OpenSSL.crypto.FILETYPE_PEM, parts[0])
 442
 443         if self.x509 is None:
 444             logger.warning(
 445                 "Loaded from string but cert is None: {}".format(string))
 446
 447         # if there are more certs, then create a parent and let the parent load
 448         # itself from the remainder of the string
 449         if len(parts) > 1 and parts[1] != '':
 450             self.parent = self.__class__()
 451             self.parent.load_from_string(parts[1])
 452
 453     ##
 454     # Load the certificate from a file
 455
 456     def load_from_file(self, filename):
 457         file = open(filename)
 458         string = file.read()
 459         self.load_from_string(string)
 460         self.filename = filename
 461
 462     ##
 463     # Save the certificate to a string.
 464     #
 465     # @param save_parents If save_parents==True,
 466     # then also save the parent certificates.
 467
 468     def save_to_string(self, save_parents=True):
 469         if self.x509 is None:
 470             logger.warning("None cert in certificate.save_to_string")
 471             return ""
 472         string = OpenSSL.crypto.dump_certificate(
 473             OpenSSL.crypto.FILETYPE_PEM, self.x509)
 474         if isinstance(string, bytes):
 475             string = string.decode()
 476         if save_parents and self.parent:
 477             string = string + self.parent.save_to_string(save_parents)
 478         return string
 479
 480     ##
 481     # Save the certificate to a file.
 482     # @param save_parents If save_parents==True,
 483     #  then also save the parent certificates.
 484
 485     def save_to_file(self, filename, save_parents=True, filep=None):
 486         string = self.save_to_string(save_parents=save_parents)
 487         if filep:
 488             f = filep
 489         else:
 490             f = open(filename, 'w')
 491         if isinstance(string, bytes):
 492             string = string.decode()
 493         f.write(string)
 494         f.close()
 495         self.filename = filename
 496
 497     ##
 498     # Save the certificate to a random file in /tmp/
 499     # @param save_parents If save_parents==True,
 500     #  then also save the parent certificates.
 501     def save_to_random_tmp_file(self, save_parents=True):
 502         fp, filename = mkstemp(suffix='cert', text=True)
 503         fp = os.fdopen(fp, "w")
 504         self.save_to_file(filename, save_parents=True, filep=fp)
 505         return filename
 506
 507     ##
 508     # Sets the issuer private key and name
 509     # @param key Keypair object containing the private key of the issuer
 510     # @param subject String containing the name of the issuer
 511     # @param cert (optional)
 512     #  Certificate object containing the name of the issuer
 513
 514     def set_issuer(self, key, subject=None, cert=None):
 515         self.issuerKey = key
 516         if subject:
 517             # it's a mistake to use subject and cert params at the same time
 518             assert(not cert)
 519             if isinstance(subject, dict) or isinstance(subject, str):
 520                 req = OpenSSL.crypto.X509Req()
 521                 reqSubject = req.get_subject()
 522                 if isinstance(subject, dict):
 523                     for key in list(reqSubject.keys()):
 524                         setattr(reqSubject, key, subject[key])
 525                 else:
 526                     setattr(reqSubject, "CN", subject)
 527                 subject = reqSubject
 528                 # subject is not valid once req is out of scope, so save req
 529                 self.issuerReq = req
 530         if cert:
 531             # if a cert was supplied, then get the subject from the cert
 532             subject = cert.x509.get_subject()
 533         assert(subject)
 534         self.issuerSubject = subject
 535
 536     ##
 537     # Get the issuer name
 538
 539     def get_issuer(self, which="CN"):
 540         x = self.x509.get_issuer()
 541         return getattr(x, which)
 542
 543     ##
 544     # Set the subject name of the certificate
 545
 546     def set_subject(self, name):
 547         req = OpenSSL.crypto.X509Req()
 548         subj = req.get_subject()
 549         if isinstance(name, dict):
 550             for key in list(name.keys()):
 551                 setattr(subj, key, name[key])
 552         else:
 553             setattr(subj, "CN", name)
 554         self.x509.set_subject(subj)
 555
 556     ##
 557     # Get the subject name of the certificate
 558
 559     def get_subject(self, which="CN"):
 560         x = self.x509.get_subject()
 561         return getattr(x, which)
 562
 563     ##
 564     # Get a pretty-print subject name of the certificate
 565     # let's try to make this a little more usable as is makes logs hairy
 566     # FIXME: Consider adding 'urn:publicid' and 'uuid' back for GENI?
 567     pretty_fields = ['email']
 568
 569     def filter_chunk(self, chunk):
 570         for field in self.pretty_fields:
 571             if field in chunk:
 572                 return " " + chunk
 573
 574     def pretty_cert(self):
 575         message = "[Cert."
 576         x = self.x509.get_subject()
 577         ou = getattr(x, "OU")
 578         if ou:
 579             message += " OU: {}".format(ou)
 580         cn = getattr(x, "CN")
 581         if cn:
 582             message += " CN: {}".format(cn)
 583         data = self.get_data(field='subjectAltName')
 584         if data:
 585             message += " SubjectAltName:"
 586             filtered = [self.filter_chunk(chunk) for chunk in data.split()]
 587             message += " ".join([f for f in filtered if f])
 588             omitted = len([f for f in filtered if not f])
 589             if omitted:
 590                 message += "..+{} omitted".format(omitted)
 591         message += "]"
 592         return message
 593
 594     def pretty_chain(self):
 595         message = "{}".format(self.x509.get_subject())
 596         parent = self.parent
 597         while parent:
 598             message += "->{}".format(parent.x509.get_subject())
 599             parent = parent.parent
 600         return message
 601
 602     def pretty_name(self):
 603         return self.get_filename() or self.pretty_chain()
 604
 605     ##
 606     # Get the public key of the certificate.
 607     #
 608     # @param key Keypair object containing the public key
 609
 610     def set_pubkey(self, key):
 611         assert(isinstance(key, Keypair))
 612         self.x509.set_pubkey(key.get_openssl_pkey())
 613
 614     ##
 615     # Get the public key of the certificate.
 616     # It is returned in the form of a Keypair object.
 617
 618     def get_pubkey(self):
 619         import M2Crypto
 620         m2x509 = M2Crypto.X509.load_cert_string(self.save_to_string())
 621         pkey = Keypair()
 622         pkey.key = self.x509.get_pubkey()
 623         pkey.m2key = m2x509.get_pubkey()
 624         return pkey
 625
 626     def set_intermediate_ca(self, val):
 627         return self.set_is_ca(val)
 628
 629     # Set whether this cert is for a CA.
 630     # All signers and only signers should be CAs.
 631     # The local member starts unset, letting us check that you only set it once
 632     # @param val Boolean indicating whether this cert is for a CA
 633     def set_is_ca(self, val):
 634         if val is None:
 635             return
 636
 637         if self.isCA is not None:
 638             # Can't double set properties
 639             raise Exception(
 640                 "Cannot set basicConstraints CA:?? more than once. "
 641                 "Was {}, trying to set as {}".format(self.isCA, val))
 642
 643         self.isCA = val
 644         if val:
 645             self.add_extension('basicConstraints', 1, 'CA:TRUE')
 646         else:
 647             self.add_extension('basicConstraints', 1, 'CA:FALSE')
 648
 649     ##
 650     # Add an X509 extension to the certificate. Add_extension can only
 651     # be called once for a particular extension name, due to
 652     # limitations in the underlying library.
 653     #
 654     # @param name string containing name of extension
 655     # @param value string containing value of the extension
 656
 657     def add_extension(self, name, critical, value):
 658         oldExtVal = None
 659         try:
 660             oldExtVal = self.get_extension(name)
 661         except:
 662             # M2Crypto LookupError when the extension isn't there (yet)
 663             pass
 664
 665         # This code limits you from adding the extension with the same value
 666         # The method comment says you shouldn't do this with the same name
 667         # But actually it (m2crypto) appears to allow you to do this.
 668         if oldExtVal and oldExtVal == value:
 669             # don't add this extension again
 670             # just do nothing as here
 671             return
 672         # FIXME: What if they are trying to set with a different value?
 673         # Is this ever OK? Or should we raise an exception?
 674 #        elif oldExtVal:
 675 #            raise "Cannot add extension {} which had val {} with new val {}"\
 676 #                  .format(name, oldExtVal, value)
 677
 678         if isinstance(name, str):
 679             name = name.encode()
 680         if isinstance(value, str):
 681             value = value.encode()
 682
 683         ext = OpenSSL.crypto.X509Extension(name, critical, value)
 684         self.x509.add_extensions([ext])
 685
 686     ##
 687     # Get an X509 extension from the certificate
 688
 689     def get_extension(self, name):
 690
 691         import M2Crypto
 692         if name is None:
 693             return None
 694
 695         certstr = self.save_to_string()
 696         if certstr is None or certstr == "":
 697             return None
 698         # pyOpenSSL does not have a way to get extensions
 699         m2x509 = M2Crypto.X509.load_cert_string(certstr)
 700         if m2x509 is None:
 701             logger.warning("No cert loaded in get_extension")
 702             return None
 703         if m2x509.get_ext(name) is None:
 704             return None
 705         value = m2x509.get_ext(name).get_value()
 706
 707         return value
 708
 709     ##
 710     # Set_data is a wrapper around add_extension. It stores the
 711     # parameter str in the X509 subject_alt_name extension. Set_data
 712     # can only be called once, due to limitations in the underlying
 713     # library.
 714
 715     def set_data(self, string, field='subjectAltName'):
 716         # pyOpenSSL only allows us to add extensions, so if we try to set the
 717         # same extension more than once, it will not work
 718         if field in self.data:
 719             raise Exception("Cannot set {} more than once".format(field))
 720         self.data[field] = string
 721         # call str() because we've seen unicode there
 722         # and the underlying C code doesn't like it
 723         self.add_extension(field, 0, str(string))
 724
 725     ##
 726     # Return the data string that was previously set with set_data
 727
 728     def get_data(self, field='subjectAltName'):
 729         if field in self.data:
 730             return self.data[field]
 731
 732         try:
 733             uri = self.get_extension(field)
 734             self.data[field] = uri
 735         except LookupError:
 736             return None
 737
 738         return self.data[field]
 739
 740     ##
 741     # Sign the certificate using the issuer private key and issuer subject
 742     # previous set with set_issuer().
 743
 744     def sign(self):
 745         logger.debug('certificate.sign')
 746         assert self.x509 is not None
 747         assert self.issuerSubject is not None
 748         assert self.issuerKey is not None
 749         self.x509.set_issuer(self.issuerSubject)
 750         self.x509.sign(self.issuerKey.get_openssl_pkey(), self.digest)
 751
 752     ##
 753     # Verify the authenticity of a certificate.
 754     # @param pkey is a Keypair object representing a public key. If Pkey
 755     #     did not sign the certificate, then an exception will be thrown.
 756
 757     def verify(self, pubkey):
 758         import M2Crypto
 759         # pyOpenSSL does not have a way to verify signatures
 760         m2x509 = M2Crypto.X509.load_cert_string(self.save_to_string())
 761         m2pubkey = pubkey.get_m2_pubkey()
 762         # verify it
 763         # https://www.openssl.org/docs/man1.1.0/crypto/X509_verify.html
 764         # verify returns
 765         # 1  if it checks out
 766         # 0  if if does not
 767         # -1 if it could not be checked 'for some reason'
 768         m2result = m2x509.verify(m2pubkey)
 769         result = m2result == 1
 770         if debug_verify_chain:
 771             logger.debug("Certificate.verify: <- {} (m2={}) ({} x {})"
 772                          .format(result, m2result,
 773                                  self.pretty_cert(), m2pubkey))
 774         return result
 775
 776         # XXX alternatively, if openssl has been patched, do the much simpler:
 777         # try:
 778         #   self.x509.verify(pkey.get_openssl_key())
 779         #   return 1
 780         # except:
 781         #   return 0
 782
 783     ##
 784     # Return True if pkey is identical to the public key that is
 785     # contained in the certificate.
 786     # @param pkey Keypair object
 787
 788     def is_pubkey(self, pkey):
 789         return self.get_pubkey().is_same(pkey)
 790
 791     ##
 792     # Given a certificate cert, verify that this certificate was signed by the
 793     # public key contained in cert. Throw an exception otherwise.
 794     #
 795     # @param cert certificate object
 796
 797     def is_signed_by_cert(self, cert):
 798         key = cert.get_pubkey()
 799         logger.debug("Certificate.is_signed_by_cert -> verify on {}\n"
 800                      "with pubkey {}"
 801                      .format(self, key))
 802         result = self.verify(key)
 803         return result
 804
 805     ##
 806     # Set the parent certificate.
 807     #
 808     # @param p certificate object.
 809
 810     def set_parent(self, p):
 811         self.parent = p
 812
 813     ##
 814     # Return the certificate object of the parent of this certificate.
 815
 816     def get_parent(self):
 817         return self.parent
 818
 819     ##
 820     # Verification examines a chain of certificates to ensure that each parent
 821     # signs the child, and that some certificate in the chain is signed by a
 822     # trusted certificate.
 823     #
 824     # Verification is a basic recursion: <pre>
 825     #     if this_certificate was signed by trusted_certs:
 826     #         return
 827     #     else
 828     #         return verify_chain(parent, trusted_certs)
 829     # </pre>
 830     #
 831     # At each recursion, the parent is tested to ensure that it did sign the
 832     # child. If a parent did not sign a child, then an exception is thrown. If
 833     # the bottom of the recursion is reached and the certificate does not match
 834     # a trusted root, then an exception is thrown.
 835     # Also require that parents are CAs.
 836     #
 837     # @param Trusted_certs is a list of certificates that are trusted.
 838     #
 839
 840     def verify_chain(self, trusted_certs=None):
 841         # Verify a chain of certificates. Each certificate must be signed by
 842         # the public key contained in it's parent. The chain is recursed
 843         # until a certificate is found that is signed by a trusted root.
 844
 845         # verify expiration time
 846         if self.x509.has_expired():
 847             if debug_verify_chain:
 848                 logger.debug("verify_chain: NO, Certificate {} has expired"
 849                              .format(self.pretty_cert()))
 850             raise CertExpired(self.pretty_cert(), "client cert")
 851
 852         # if this cert is signed by a trusted_cert, then we are set
 853         for i, trusted_cert in enumerate(trusted_certs, 1):
 854             logger.debug(5*'-' +
 855                          " Certificate.verify_chain - trying trusted #{} : {}"
 856                          .format(i, trusted_cert.pretty_name()))
 857             if self.is_signed_by_cert(trusted_cert):
 858                 # verify expiration of trusted_cert ?
 859                 if not trusted_cert.x509.has_expired():
 860                     if debug_verify_chain:
 861                         logger.debug("verify_chain: YES."
 862                                      " Cert {} signed by trusted cert {}"
 863                                      .format(self.pretty_name(),
 864                                              trusted_cert.pretty_name()))
 865                     return trusted_cert
 866                 else:
 867                     if debug_verify_chain:
 868                         logger.debug("verify_chain: NO. Cert {} "
 869                                      "is signed by trusted_cert {}, "
 870                                      "but that signer is expired..."
 871                                      .format(self.pretty_cert(),
 872                                              trusted_cert.pretty_cert()))
 873                     raise CertExpired("{} signer trusted_cert {}"
 874                                       .format(self.pretty_name(),
 875                                               trusted_cert.pretty_name()))
 876             else:
 877                 logger.debug("verify_chain: not a direct"
 878                              " descendant of trusted root #{}".format(i))
 879
 880         # if there is no parent, then no way to verify the chain
 881         if not self.parent:
 882             if debug_verify_chain:
 883                 logger.debug("verify_chain: NO. {} has no parent "
 884                              "and issuer {} is not in {} trusted roots"
 885                              .format(self.pretty_name(), self.get_issuer(),
 886                                      len(trusted_certs)))
 887             raise CertMissingParent("{}: Issuer {} is not "
 888                                     "one of the {} trusted roots, "
 889                                     "and cert has no parent."
 890                                     .format(self.pretty_name(),
 891                                             self.get_issuer(),
 892                                             len(trusted_certs)))
 893
 894         # if it wasn't signed by the parent...
 895         if not self.is_signed_by_cert(self.parent):
 896             if debug_verify_chain:
 897                 logger.debug("verify_chain: NO. {} is not signed by parent {}"
 898                              .format(self.pretty_name(),
 899                                      self.parent.pretty_name()))
 900                 self.save_to_file("/tmp/xxx-capture.pem", save_parents=True)
 901             raise CertNotSignedByParent("{}: Parent {}, issuer {}"
 902                                         .format(self.pretty_name(),
 903                                                 self.parent.pretty_name(),
 904                                                 self.get_issuer()))
 905
 906         # Confirm that the parent is a CA. Only CAs can be trusted as
 907         # signers.
 908         # Note that trusted roots are not parents, so don't need to be
 909         # CAs.
 910         # Ugly - cert objects aren't parsed so we need to read the
 911         # extension and hope there are no other basicConstraints
 912         if not self.parent.isCA and not (
 913                 self.parent.get_extension('basicConstraints') == 'CA:TRUE'):
 914             logger.warning("verify_chain: cert {}'s parent {} is not a CA"
 915                             .format(self.pretty_name(), self.parent.pretty_name()))
 916             raise CertNotSignedByParent("{}: Parent {} not a CA"
 917                                         .format(self.pretty_name(),
 918                                                 self.parent.pretty_name()))
 919
 920         # if the parent isn't verified...
 921         if debug_verify_chain:
 922             logger.debug("verify_chain: .. {}, -> verifying parent {}"
 923                          .format(self.pretty_name(),
 924                                  self.parent.pretty_name()))
 925         self.parent.verify_chain(trusted_certs)
 926
 927         return
 928
 929     # more introspection
 930     def get_extensions(self):
 931         import M2Crypto
 932         # pyOpenSSL does not have a way to get extensions
 933         triples = []
 934         m2x509 = M2Crypto.X509.load_cert_string(self.save_to_string())
 935         nb_extensions = m2x509.get_ext_count()
 936         logger.debug("X509 had {} extensions".format(nb_extensions))
 937         for i in range(nb_extensions):
 938             ext = m2x509.get_ext_at(i)
 939             triples.append(
 940                 (ext.get_name(), ext.get_value(), ext.get_critical(),))
 941         return triples
 942
 943     def get_data_names(self):
 944         return list(self.data.keys())
 945
 946     def get_all_datas(self):
 947         triples = self.get_extensions()
 948         for name in self.get_data_names():
 949             triples.append((name, self.get_data(name), 'data',))
 950         return triples
 951
 952     # only informative
 953     def get_filename(self):
 954         return getattr(self, 'filename', None)
 955
 956     def dump(self, *args, **kwargs):
 957         print(self.dump_string(*args, **kwargs))
 958
 959     def dump_string(self, show_extensions=False):
 960         result = ""
 961         result += "CERTIFICATE for {}\n".format(self.pretty_cert())
 962         result += "Issued by {}\n".format(self.get_issuer())
 963         filename = self.get_filename()
 964         if filename:
 965             result += "Filename {}\n".format(filename)
 966         if show_extensions:
 967             all_datas = self.get_all_datas()
 968             result += " has {} extensions/data attached".format(len(all_datas))
 969             for n, v, c in all_datas:
 970                 if c == 'data':
 971                     result += "   data: {}={}\n".format(n, v)
 972                 else:
 973                     result += "    ext: {} (crit={})=<<<{}>>>\n".format(n, c, v)
 974         return result