geni/util/cert.py

   1 ##
   2 # Geniwrapper uses two crypto libraries: pyOpenSSL and M2Crypto to implement
   3 # the necessary crypto functionality. Ideally just one of these libraries
   4 # would be used, but unfortunately each of these libraries is independently
   5 # lacking. The pyOpenSSL library is missing many necessary functions, and
   6 # the M2Crypto library has crashed inside of some of the functions. The
   7 # design decision is to use pyOpenSSL whenever possible as it seems more
   8 # stable, and only use M2Crypto for those functions that are not possible
   9 # in pyOpenSSL.
  10 #
  11 # This module exports two classes: Keypair and Certificate.
  12 ##
  13
  14 import os
  15 import tempfile
  16 import base64
  17 from OpenSSL import crypto
  18 import M2Crypto
  19 from M2Crypto import X509
  20 from M2Crypto import EVP
  21
  22 from excep import *
  23
  24 ##
  25 # Public-private key pairs are implemented by the Keypair class.
  26 # A Keypair object may represent both a public and private key pair, or it
  27 # may represent only a public key (this usage is consistent with OpenSSL).
  28
  29 class Keypair:
  30    key = None       # public/private keypair
  31    m2key = None     # public key (m2crypto format)
  32
  33    ##
  34    # Creates a Keypair object
  35    # @param create If create==True, creates a new public/private key and
  36    #     stores it in the object
  37    # @param string If string!=None, load the keypair from the string (PEM)
  38    # @param filename If filename!=None, load the keypair from the file
  39
  40    def __init__(self, create=False, string=None, filename=None):
  41       if create:
  42          self.create()
  43       if string:
  44          self.load_from_string(string)
  45       if filename:
  46          self.load_from_file(filename)
  47
  48    ##
  49    # Create a RSA public/private key pair and store it inside the keypair object
  50
  51    def create(self):
  52       self.key = crypto.PKey()
  53       self.key.generate_key(crypto.TYPE_RSA, 1024)
  54
  55    ##
  56    # Save the private key to a file
  57    # @param filename name of file to store the keypair in
  58
  59    def save_to_file(self, filename):
  60       open(filename, 'w').write(self.as_pem())
  61
  62    ##
  63    # Load the private key from a file. Implicity the private key includes the public key.
  64
  65    def load_from_file(self, filename):
  66       buffer = open(filename, 'r').read()
  67       self.load_from_string(buffer)
  68
  69    ##
  70    # Load the private key from a string. Implicitly the private key includes the public key.
  71
  72    def load_from_string(self, string):
  73       self.key = crypto.load_privatekey(crypto.FILETYPE_PEM, string)
  74       self.m2key = M2Crypto.EVP.load_key_string(string)
  75
  76    ##
  77    #  Load the public key from a string. No private key is loaded.
  78
  79    def load_pubkey_from_file(self, filename):
  80       # load the m2 public key
  81       m2rsakey = M2Crypto.RSA.load_pub_key(filename)
  82       self.m2key = M2Crypto.EVP.PKey()
  83       self.m2key.assign_rsa(m2rsakey)
  84
  85       # create an m2 x509 cert
  86       m2name = M2Crypto.X509.X509_Name()
  87       m2name.add_entry_by_txt(field="CN", type=0x1001, entry="junk", len=-1, loc=-1, set=0)
  88       m2x509 = M2Crypto.X509.X509()
  89       m2x509.set_pubkey(self.m2key)
  90       m2x509.set_serial_number(0)
  91       m2x509.set_issuer_name(m2name)
  92       m2x509.set_subject_name(m2name)
  93       ASN1 = M2Crypto.ASN1.ASN1_UTCTIME()
  94       ASN1.set_time(500)
  95       m2x509.set_not_before(ASN1)
  96       m2x509.set_not_after(ASN1)
  97       junk_key = Keypair(create=True)
  98       m2x509.sign(pkey=junk_key.get_m2_pkey(), md="sha1")
  99
 100       # convert the m2 x509 cert to a pyopenssl x509
 101       m2pem = m2x509.as_pem()
 102       pyx509 = crypto.load_certificate(crypto.FILETYPE_PEM, m2pem)
 103
 104       # get the pyopenssl pkey from the pyopenssl x509
 105       self.key = pyx509.get_pubkey()
 106
 107    ##
 108    # Load the public key from a string. No private key is loaded.
 109
 110    def load_pubkey_from_string(self, string):
 111       (f, fn) = tempfile.mkstemp()
 112       os.write(f, string)
 113       os.close(f)
 114       self.load_pubkey_from_file(fn)
 115       os.remove(fn)
 116
 117    ##
 118    # Return the private key in PEM format.
 119
 120    def as_pem(self):
 121       return crypto.dump_privatekey(crypto.FILETYPE_PEM, self.key)
 122
 123    def get_m2_pkey(self):
 124       if not self.m2key:
 125          self.m2key = M2Crypto.EVP.load_key_string(self.as_pem())
 126       return self.m2key
 127
 128    ##
 129    # Return an OpenSSL pkey object
 130
 131    def get_openssl_pkey(self):
 132       return self.key
 133
 134    ##
 135    # Given another Keypair object, return TRUE if the two keys are the same.
 136
 137    def is_same(self, pkey):
 138       return self.as_pem() == pkey.as_pem()
 139
 140    def sign_string(self, data):
 141       k = self.get_m2_pkey()
 142       k.sign_init()
 143       k.sign_update(data)
 144       return base64.b64encode(k.sign_final())
 145
 146    def verify_string(self, data, sig):
 147       k = self.get_m2_pkey()
 148       k.verify_init()
 149       k.verify_update(data)
 150       return M2Crypto.m2.verify_final(k.ctx, base64.b64decode(sig), k.pkey)
 151
 152 ##
 153 # The certificate class implements a general purpose X509 certificate, making
 154 # use of the appropriate pyOpenSSL or M2Crypto abstractions. It also adds
 155 # several addition features, such as the ability to maintain a chain of
 156 # parent certificates, and storage of application-specific data.
 157 #
 158 # Certificates include the ability to maintain a chain of parents. Each
 159 # certificate includes a pointer to it's parent certificate. When loaded
 160 # from a file or a string, the parent chain will be automatically loaded.
 161 # When saving a certificate to a file or a string, the caller can choose
 162 # whether to save the parent certificates as well.
 163
 164 class Certificate:
 165    digest = "md5"
 166
 167    data = None
 168    cert = None
 169    issuerKey = None
 170    issuerSubject = None
 171    parent = None
 172
 173    separator="-----parent-----\n"
 174
 175    ##
 176    # Create a certificate object.
 177    #
 178    # @param create If create==True, then also create a blank X509 certificate.
 179    # @param subject If subject!=None, then create a blank certificate and set
 180    #     it's subject name.
 181    # @param string If string!=None, load the certficate from the string.
 182    # @param filename If filename!=None, load the certficiate from the file.
 183
 184    def __init__(self, create=False, subject=None, string=None, filename=None):
 185        if create or subject:
 186            self.create()
 187        if subject:
 188            self.set_subject(subject)
 189        if string:
 190            self.load_from_string(string)
 191        if filename:
 192            self.load_from_file(filename)
 193
 194    ##
 195    # Create a blank X509 certificate and store it in this object.
 196
 197    def create(self):
 198        self.cert = crypto.X509()
 199        self.cert.set_serial_number(1)
 200        self.cert.gmtime_adj_notBefore(0)
 201        self.cert.gmtime_adj_notAfter(60*60*24*365*5) # five years
 202
 203    ##
 204    # Given a pyOpenSSL X509 object, store that object inside of this
 205    # certificate object.
 206
 207    def load_from_pyopenssl_x509(self, x509):
 208        self.cert = x509
 209
 210    ##
 211    # Load the certificate from a string
 212
 213    def load_from_string(self, string):
 214        # if it is a chain of multiple certs, then split off the first one and
 215        # load it
 216        parts = string.split(Certificate.separator, 1)
 217        self.cert = crypto.load_certificate(crypto.FILETYPE_PEM, parts[0])
 218
 219        # if there are more certs, then create a parent and let the parent load
 220        # itself from the remainder of the string
 221        if len(parts) > 1:
 222            self.parent = self.__class__()
 223            self.parent.load_from_string(parts[1])
 224
 225    ##
 226    # Load the certificate from a file
 227
 228    def load_from_file(self, filename):
 229        file = open(filename)
 230        string = file.read()
 231        self.load_from_string(string)
 232
 233    ##
 234    # Save the certificate to a string.
 235    #
 236    # @param save_parents If save_parents==True, then also save the parent certificates.
 237
 238    def save_to_string(self, save_parents=False):
 239        string = crypto.dump_certificate(crypto.FILETYPE_PEM, self.cert)
 240        if save_parents and self.parent:
 241           string = string + Certificate.separator + self.parent.save_to_string(save_parents)
 242        return string
 243
 244    ##
 245    # Save the certificate to a file.
 246    # @param save_parents If save_parents==True, then also save the parent certificates.
 247
 248    def save_to_file(self, filename, save_parents=False):
 249        string = self.save_to_string(save_parents=save_parents)
 250        open(filename, 'w').write(string)
 251
 252    ##
 253    # Sets the issuer private key and name
 254    # @param key Keypair object containing the private key of the issuer
 255    # @param subject String containing the name of the issuer
 256    # @param cert (optional) Certificate object containing the name of the issuer
 257
 258    def set_issuer(self, key, subject=None, cert=None):
 259        self.issuerKey = key
 260        if subject:
 261           # it's a mistake to use subject and cert params at the same time
 262           assert(not cert)
 263           if isinstance(subject, dict) or isinstance(subject, str):
 264              req = crypto.X509Req()
 265              reqSubject = req.get_subject()
 266              if (isinstance(subject, dict)):
 267                 for key in reqSubject.keys():
 268                     setattr(reqSubject, key, name[key])
 269              else:
 270                 setattr(reqSubject, "CN", subject)
 271              subject = reqSubject
 272              # subject is not valid once req is out of scope, so save req
 273              self.issuerReq = req
 274        if cert:
 275           # if a cert was supplied, then get the subject from the cert
 276           subject = cert.cert.get_issuer()
 277        assert(subject)
 278        self.issuerSubject = subject
 279
 280    ##
 281    # Get the issuer name
 282
 283    def get_issuer(self, which="CN"):
 284        x = self.cert.get_issuer()
 285        return getattr(x, which)
 286
 287    ##
 288    # Set the subject name of the certificate
 289
 290    def set_subject(self, name):
 291        req = crypto.X509Req()
 292        subj = req.get_subject()
 293        if (isinstance(name, dict)):
 294            for key in name.keys():
 295                setattr(subj, key, name[key])
 296        else:
 297            setattr(subj, "CN", name)
 298        self.cert.set_subject(subj)
 299    ##
 300    # Get the subject name of the certificate
 301
 302    def get_subject(self, which="CN"):
 303        x = self.cert.get_subject()
 304        return getattr(x, which)
 305
 306    ##
 307    # Get the public key of the certificate.
 308    #
 309    # @param key Keypair object containing the public key
 310
 311    def set_pubkey(self, key):
 312        assert(isinstance(key, Keypair))
 313        self.cert.set_pubkey(key.get_openssl_pkey())
 314
 315    ##
 316    # Get the public key of the certificate.
 317    # It is returned in the form of a Keypair object.
 318
 319    def get_pubkey(self):
 320        m2x509 = X509.load_cert_string(self.save_to_string())
 321        pkey = Keypair()
 322        pkey.key = self.cert.get_pubkey()
 323        pkey.m2key = m2x509.get_pubkey()
 324        return pkey
 325
 326    ##
 327    # Add an X509 extension to the certificate. Add_extension can only be called
 328    # once for a particular extension name, due to limitations in the underlying
 329    # library.
 330    #
 331    # @param name string containing name of extension
 332    # @param value string containing value of the extension
 333
 334    def add_extension(self, name, critical, value):
 335        ext = crypto.X509Extension (name, critical, value)
 336        self.cert.add_extensions([ext])
 337
 338    ##
 339    # Get an X509 extension from the certificate
 340
 341    def get_extension(self, name):
 342        # pyOpenSSL does not have a way to get extensions
 343        m2x509 = X509.load_cert_string(self.save_to_string())
 344        value = m2x509.get_ext(name).get_value()
 345        return value
 346
 347    ##
 348    # Set_data is a wrapper around add_extension. It stores the parameter str in
 349    # the X509 subject_alt_name extension. Set_data can only be called once, due
 350    # to limitations in the underlying library.
 351
 352    def set_data(self, str):
 353        # pyOpenSSL only allows us to add extensions, so if we try to set the
 354        # same extension more than once, it will not work
 355        if self.data != None:
 356           raise "cannot set subjectAltName more than once"
 357        self.data = str
 358        self.add_extension("subjectAltName", 0, "URI:http://" + str)
 359
 360    ##
 361    # Return the data string that was previously set with set_data
 362
 363    def get_data(self):
 364        if self.data:
 365            return self.data
 366
 367        try:
 368            uri = self.get_extension("subjectAltName")
 369        except LookupError:
 370            self.data = None
 371            return self.data
 372
 373        if not uri.startswith("URI:http://"):
 374            raise "bad encoding in subjectAltName"
 375        self.data = uri[11:]
 376        return self.data
 377
 378    ##
 379    # Sign the certificate using the issuer private key and issuer subject previous set with set_issuer().
 380
 381    def sign(self):
 382        assert self.cert != None
 383        assert self.issuerSubject != None
 384        assert self.issuerKey != None
 385        self.cert.set_issuer(self.issuerSubject)
 386        self.cert.sign(self.issuerKey.get_openssl_pkey(), self.digest)
 387
 388     ##
 389     # Verify the authenticity of a certificate.
 390     # @param pkey is a Keypair object representing a public key. If Pkey
 391     #     did not sign the certificate, then an exception will be thrown.
 392
 393    def verify(self, pkey):
 394        # pyOpenSSL does not have a way to verify signatures
 395        m2x509 = X509.load_cert_string(self.save_to_string())
 396        m2pkey = pkey.get_m2_pkey()
 397        # verify it
 398        return m2x509.verify(m2pkey)
 399
 400        # XXX alternatively, if openssl has been patched, do the much simpler:
 401        # try:
 402        #   self.cert.verify(pkey.get_openssl_key())
 403        #   return 1
 404        # except:
 405        #   return 0
 406
 407    ##
 408    # Return True if pkey is identical to the public key that is contained in the certificate.
 409    # @param pkey Keypair object
 410
 411    def is_pubkey(self, pkey):
 412        return self.get_pubkey().is_same(pkey)
 413
 414    ##
 415    # Given a certificate cert, verify that this certificate was signed by the
 416    # public key contained in cert. Throw an exception otherwise.
 417    #
 418    # @param cert certificate object
 419
 420    def is_signed_by_cert(self, cert):
 421        k = cert.get_pubkey()
 422        result = self.verify(k)
 423        return result
 424
 425    ##
 426    # Set the parent certficiate.
 427    #
 428    # @param p certificate object.
 429
 430    def set_parent(self, p):
 431         self.parent = p
 432
 433    ##
 434    # Return the certificate object of the parent of this certificate.
 435
 436    def get_parent(self):
 437         return self.parent
 438
 439    ##
 440    # Verification examines a chain of certificates to ensure that each parent
 441    # signs the child, and that some certificate in the chain is signed by a
 442    # trusted certificate.
 443    #
 444    # Verification is a basic recursion: <pre>
 445    #     if this_certificate was signed by trusted_certs:
 446    #         return
 447    #     else
 448    #         return verify_chain(parent, trusted_certs)
 449    # </pre>
 450    #
 451    # At each recursion, the parent is tested to ensure that it did sign the
 452    # child. If a parent did not sign a child, then an exception is thrown. If
 453    # the bottom of the recursion is reached and the certificate does not match
 454    # a trusted root, then an exception is thrown.
 455    #
 456    # @param Trusted_certs is a list of certificates that are trusted.
 457    #
 458
 459    def verify_chain(self, trusted_certs = None):
 460         # Verify a chain of certificates. Each certificate must be signed by
 461         # the public key contained in it's parent. The chain is recursed
 462         # until a certificate is found that is signed by a trusted root.
 463
 464         # TODO: verify expiration time
 465
 466         # if this cert is signed by a trusted_cert, then we are set
 467         for trusted_cert in trusted_certs:
 468             # TODO: verify expiration of trusted_cert ?
 469             if self.is_signed_by_cert(trusted_cert):
 470                 #print self.get_subject(), "is signed by a root"
 471                 return
 472
 473         # if there is no parent, then no way to verify the chain
 474         if not self.parent:
 475             #print self.get_subject(), "has no parent"
 476             raise CertMissingParent(self.get_subject())
 477
 478         # if it wasn't signed by the parent...
 479         if not self.is_signed_by_cert(self.parent):
 480             #print self.get_subject(), "is not signed by parent"
 481             return CertNotSignedByParent(self.get_subject())
 482
 483         # if the parent isn't verified...
 484         self.parent.verify_chain(trusted_certs)
 485
 486         return