| from time import time |
| |
| from OpenSSL.xcrypto import * |
| |
| from tls.c import api as _api |
| |
| FILETYPE_PEM = _api.SSL_FILETYPE_PEM |
| FILETYPE_ASN1 = _api.SSL_FILETYPE_ASN1 |
| |
| # TODO This was an API mistake. OpenSSL has no such constant. |
| FILETYPE_TEXT = 2 ** 16 - 1 |
| |
| TYPE_RSA = _api.EVP_PKEY_RSA |
| TYPE_DSA = _api.EVP_PKEY_DSA |
| |
| |
| def _bio_to_string(bio): |
| """ |
| Copy the contents of an OpenSSL BIO object into a Python byte string. |
| """ |
| result_buffer = _api.new('char**') |
| buffer_length = _api.BIO_get_mem_data(bio, result_buffer) |
| return _api.buffer(result_buffer[0], buffer_length)[:] |
| |
| |
| |
| def _raise_current_error(): |
| errors = [] |
| while True: |
| error = _api.ERR_get_error() |
| if error == 0: |
| break |
| errors.append(( |
| _api.string(_api.ERR_lib_error_string(error)), |
| _api.string(_api.ERR_func_error_string(error)), |
| _api.string(_api.ERR_reason_error_string(error)))) |
| |
| raise Error(errors) |
| |
| _exception_from_error_queue = _raise_current_error |
| |
| |
| class Error(Exception): |
| pass |
| |
| |
| |
| class PKey(object): |
| _only_public = False |
| _initialized = True |
| |
| def __init__(self): |
| self._pkey = _api.EVP_PKEY_new() |
| self._initialized = False |
| |
| |
| def generate_key(self, type, bits): |
| """ |
| Generate a key of a given type, with a given number of a bits |
| |
| :param type: The key type (TYPE_RSA or TYPE_DSA) |
| :param bits: The number of bits |
| |
| :return: None |
| """ |
| if not isinstance(type, int): |
| raise TypeError("type must be an integer") |
| |
| if not isinstance(bits, int): |
| raise TypeError("bits must be an integer") |
| |
| exponent = _api.new("BIGNUM**") |
| # TODO Check error return |
| # TODO Free the exponent[0] |
| _api.BN_hex2bn(exponent, "10001") |
| |
| if type == TYPE_RSA: |
| if bits <= 0: |
| raise ValueError("Invalid number of bits") |
| |
| rsa = _api.RSA_new(); |
| |
| # TODO Release GIL? |
| result = _api.RSA_generate_key_ex(rsa, bits, exponent[0], _api.NULL) |
| if result == -1: |
| 1/0 |
| |
| result = _api.EVP_PKEY_assign_RSA(self._pkey, rsa) |
| if not result: |
| 1/0 |
| |
| elif type == TYPE_DSA: |
| pass |
| else: |
| raise Error("No such key type") |
| |
| self._initialized = True |
| |
| |
| def check(self): |
| """ |
| Check the consistency of an RSA private key. |
| |
| :return: True if key is consistent. |
| :raise Error: if the key is inconsistent. |
| :raise TypeError: if the key is of a type which cannot be checked. |
| Only RSA keys can currently be checked. |
| """ |
| if _api.EVP_PKEY_type(self._pkey.type) != _api.EVP_PKEY_RSA: |
| raise TypeError("key type unsupported") |
| |
| rsa = _api.EVP_PKEY_get1_RSA(self._pkey) |
| result = _api.RSA_check_key(rsa) |
| if result: |
| return True |
| _raise_current_error() |
| |
| |
| |
| class X509(object): |
| def __init__(self): |
| # TODO Allocation failure? And why not __new__ instead of __init__? |
| self._x509 = _api.X509_new() |
| |
| |
| def set_version(self, version): |
| """ |
| Set version number of the certificate |
| |
| :param version: The version number |
| :type version: :py:class:`int` |
| |
| :return: None |
| """ |
| if not isinstance(version, int): |
| raise TypeError("version must be an integer") |
| |
| _api.X509_set_version(self._x509, version) |
| |
| |
| def get_version(self): |
| """ |
| Return version number of the certificate |
| |
| :return: Version number as a Python integer |
| """ |
| return _api.X509_get_version(self._x509) |
| |
| |
| def get_pubkey(self): |
| """ |
| Get the public key of the certificate |
| |
| :return: The public key |
| """ |
| pkey = PKey.__new__(PKey) |
| pkey._pkey = _api.X509_get_pubkey(self._x509) |
| if pkey._pkey == _api.NULL: |
| _raise_current_error() |
| pkey._only_public = True |
| return pkey |
| |
| |
| def set_pubkey(self, pkey): |
| """ |
| Set the public key of the certificate |
| |
| :param pkey: The public key |
| |
| :return: None |
| """ |
| if not isinstance(pkey, PKey): |
| raise TypeError("pkey must be a PKey instance") |
| |
| set_result = _api.X509_set_pubkey(self._x509, pkey._pkey) |
| if not set_result: |
| _raise_current_error() |
| |
| |
| def sign(self, pkey, digest): |
| """ |
| Sign the certificate using the supplied key and digest |
| |
| :param pkey: The key to sign with |
| :param digest: The message digest to use |
| :return: None |
| """ |
| if not isinstance(pkey, PKey): |
| raise TypeError("pkey must be a PKey instance") |
| |
| if pkey._only_public: |
| raise ValueError("Key only has public part") |
| |
| if not pkey._initialized: |
| raise ValueError("Key is uninitialized") |
| |
| evp_md = _api.EVP_get_digestbyname(digest) |
| if evp_md == _api.NULL: |
| raise ValueError("No such digest method") |
| |
| sign_result = _api.X509_sign(self._x509, pkey._pkey, evp_md) |
| if not sign_result: |
| _raise_current_error() |
| |
| |
| def get_signature_algorithm(self): |
| """ |
| Retrieve the signature algorithm used in the certificate |
| |
| :return: A byte string giving the name of the signature algorithm used in |
| the certificate. |
| :raise ValueError: If the signature algorithm is undefined. |
| """ |
| alg = self._x509.cert_info.signature.algorithm |
| nid = _api.OBJ_obj2nid(alg) |
| if nid == _api.NID_undef: |
| raise ValueError("Undefined signature algorithm") |
| return _api.string(_api.OBJ_nid2ln(nid)) |
| |
| |
| def digest(self, digest_name): |
| """ |
| Return the digest of the X509 object. |
| |
| :param digest_name: The name of the digest algorithm to use. |
| :type digest_name: :py:class:`bytes` |
| |
| :return: The digest of the object |
| """ |
| digest = _api.EVP_get_digestbyname(digest_name) |
| if digest == _api.NULL: |
| raise ValueError("No such digest method") |
| |
| result_buffer = _api.new("char[]", _api.EVP_MAX_MD_SIZE) |
| result_length = _api.new("unsigned int[]", 1) |
| result_length[0] = len(result_buffer) |
| |
| digest_result = _api.X509_digest( |
| self._x509, digest, result_buffer, result_length) |
| |
| if not digest_result: |
| 1/0 |
| |
| return ':'.join([ |
| ch.encode('hex').upper() for ch |
| in _api.buffer(result_buffer, result_length[0])]) |
| |
| |
| def subject_name_hash(self): |
| """ |
| Return the hash of the X509 subject. |
| |
| :return: The hash of the subject. |
| """ |
| return _api.X509_subject_name_hash(self._x509) |
| |
| |
| def set_serial_number(self, serial): |
| """ |
| Set serial number of the certificate |
| |
| :param serial: The serial number |
| :type serial: :py:class:`int` |
| |
| :return: None |
| """ |
| if not isinstance(serial, (int, long)): |
| raise TypeError("serial must be an integer") |
| |
| hex_serial = hex(serial)[2:] |
| if not isinstance(hex_serial, bytes): |
| hex_serial = hex_serial.encode('ascii') |
| |
| bignum_serial = _api.new("BIGNUM**") |
| |
| # BN_hex2bn stores the result in &bignum. Unless it doesn't feel like |
| # it. If bignum is still NULL after this call, then the return value is |
| # actually the result. I hope. -exarkun |
| small_serial = _api.BN_hex2bn(bignum_serial, hex_serial) |
| |
| if bignum_serial[0] == _api.NULL: |
| set_result = ASN1_INTEGER_set( |
| _api.X509_get_serialNumber(self._x509), small_serial) |
| if set_result: |
| # TODO Not tested |
| _raise_current_error() |
| else: |
| asn1_serial = _api.BN_to_ASN1_INTEGER(bignum_serial[0], _api.NULL) |
| _api.BN_free(bignum_serial[0]) |
| if asn1_serial == _api.NULL: |
| # TODO Not tested |
| _raise_current_error() |
| set_result = _api.X509_set_serialNumber(self._x509, asn1_serial) |
| if not set_result: |
| # TODO Not tested |
| _raise_current_error() |
| |
| |
| def get_serial_number(self): |
| """ |
| Return serial number of the certificate |
| |
| :return: Serial number as a Python integer |
| """ |
| asn1_serial = _api.X509_get_serialNumber(self._x509) |
| bignum_serial = _api.ASN1_INTEGER_to_BN(asn1_serial, _api.NULL) |
| try: |
| hex_serial = _api.BN_bn2hex(bignum_serial) |
| try: |
| hexstring_serial = _api.string(hex_serial) |
| serial = int(hexstring_serial, 16) |
| return serial |
| finally: |
| _api.OPENSSL_free(hex_serial) |
| finally: |
| _api.BN_free(bignum_serial) |
| |
| |
| def gmtime_adj_notAfter(self, amount): |
| """ |
| Adjust the time stamp for when the certificate stops being valid |
| |
| :param amount: The number of seconds by which to adjust the ending |
| validity time. |
| :type amount: :py:class:`int` |
| |
| :return: None |
| """ |
| if not isinstance(amount, int): |
| raise TypeError("amount must be an integer") |
| |
| notAfter = _api.X509_get_notAfter(self._x509) |
| _api.X509_gmtime_adj(notAfter, amount) |
| |
| |
| def has_expired(self): |
| """ |
| Check whether the certificate has expired. |
| |
| :return: True if the certificate has expired, false otherwise |
| """ |
| now = int(time()) |
| notAfter = _api.X509_get_notAfter(self._x509) |
| return _api.ASN1_UTCTIME_cmp_time_t( |
| _api.cast('ASN1_UTCTIME*', notAfter), now) < 0 |
| |
| |
| def get_notBefore(self): |
| """ |
| Retrieve the time stamp for when the certificate starts being valid |
| |
| :return: A string giving the timestamp, in the format:: |
| |
| YYYYMMDDhhmmssZ |
| YYYYMMDDhhmmss+hhmm |
| YYYYMMDDhhmmss-hhmm |
| |
| or None if there is no value set. |
| """ |
| timestamp = _api.X509_get_notBefore(self._x509) |
| string_timestamp = _api.cast('ASN1_STRING*', timestamp) |
| if _api.ASN1_STRING_length(string_timestamp) == 0: |
| return None |
| elif _api.ASN1_STRING_type(string_timestamp) == _api.V_ASN1_GENERALIZEDTIME: |
| return _api.string(_api.ASN1_STRING_data(string_timestamp)) |
| else: |
| generalized_timestamp = _api.new("ASN1_GENERALIZEDTIME**") |
| _api.ASN1_TIME_to_generalizedtime(timestamp, generalized_timestamp) |
| if generalized_timestamp[0] == _api.NULL: |
| 1/0 |
| else: |
| string_timestamp = _api.string( |
| _api.cast("char*", generalized_timestamp[0].data)) |
| _api.ASN1_GENERALIZEDTIME_free(generalized_timestamp[0]) |
| return string_timestamp |
| |
| |
| def set_notBefore(self, when): |
| """ |
| Set the time stamp for when the certificate starts being valid |
| |
| :param when: A string giving the timestamp, in the format: |
| |
| YYYYMMDDhhmmssZ |
| YYYYMMDDhhmmss+hhmm |
| YYYYMMDDhhmmss-hhmm |
| :type when: :py:class:`bytes` |
| |
| :return: None |
| """ |
| if not isinstance(when, bytes): |
| raise TypeError("when must be a byte string") |
| |
| notBefore = _api.X509_get_notBefore(self._x509) |
| set_result = _api.ASN1_GENERALIZEDTIME_set_string( |
| _api.cast('ASN1_GENERALIZEDTIME*', notBefore), when) |
| if set_result == 0: |
| dummy = _api.ASN1_STRING_new() |
| _api.ASN1_STRING_set(dummy, when, len(when)) |
| check_result = _api.ASN1_GENERALIZEDTIME_check( |
| _api.cast('ASN1_GENERALIZEDTIME*', dummy)) |
| if not check_result: |
| raise ValueError("Invalid string") |
| else: |
| # TODO No tests for this case |
| raise RuntimeError("Unknown ASN1_GENERALIZEDTIME_set_string failure") |
| |
| |
| def set_notAfter(self, when): |
| """ |
| Set the time stamp for when the certificate stops being valid |
| |
| :param when: A string giving the timestamp, in the format: |
| |
| YYYYMMDDhhmmssZ |
| YYYYMMDDhhmmss+hhmm |
| YYYYMMDDhhmmss-hhmm |
| :type when: :py:class:`bytes` |
| |
| :return: None |
| """ |
| notAfter = _api.X509_get_notAfter(self._x509) |
| _api.ASN1_GENERALIZEDTIME_set_string(notAfter, when) |
| X509Type = X509 |
| |
| |
| |
| def load_certificate(type, buffer): |
| """ |
| Load a certificate from a buffer |
| |
| :param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1) |
| |
| :param buffer: The buffer the certificate is stored in |
| :type buffer: :py:class:`bytes` |
| |
| :return: The X509 object |
| """ |
| bio = _api.BIO_new_mem_buf(buffer, len(buffer)) |
| |
| try: |
| if type == FILETYPE_PEM: |
| x509 = _api.PEM_read_bio_X509(bio, _api.NULL, _api.NULL, _api.NULL) |
| elif type == FILETYPE_ASN1: |
| x509 = _api.d2i_X509_bio(bio, _api.NULL); |
| else: |
| raise ValueError( |
| "type argument must be FILETYPE_PEM or FILETYPE_ASN1") |
| finally: |
| _api.BIO_free(bio) |
| |
| if x509 == _api.NULL: |
| _raise_current_error() |
| |
| cert = X509.__new__(X509) |
| cert._x509 = x509 |
| return cert |
| |
| |
| def dump_certificate(type, cert): |
| """ |
| Dump a certificate to a buffer |
| |
| :param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1) |
| :param cert: The certificate to dump |
| :return: The buffer with the dumped certificate in |
| """ |
| bio = _api.BIO_new(_api.BIO_s_mem()) |
| if type == FILETYPE_PEM: |
| result_code = _api.PEM_write_bio_X509(bio, cert._x509) |
| elif type == FILETYPE_ASN1: |
| result_code = _api.i2d_X509_bio(bio, cert._x509) |
| elif type == FILETYPE_TEXT: |
| result_code = _api.X509_print_ex(bio, cert._x509, 0, 0) |
| else: |
| raise ValueError( |
| "type argument must be FILETYPE_PEM, FILETYPE_ASN1, or " |
| "FILETYPE_TEXT") |
| |
| return _bio_to_string(bio) |
| |
| |
| |
| def dump_privatekey(type, pkey, cipher=None, passphrase=None): |
| """ |
| Dump a private key to a buffer |
| |
| :param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1) |
| :param pkey: The PKey to dump |
| :param cipher: (optional) if encrypted PEM format, the cipher to |
| use |
| :param passphrase: (optional) if encrypted PEM format, this can be either |
| the passphrase to use, or a callback for providing the |
| passphrase. |
| :return: The buffer with the dumped key in |
| :rtype: :py:data:`str` |
| """ |
| # TODO incomplete |
| bio = _api.BIO_new(_api.BIO_s_mem()) |
| |
| if type == FILETYPE_PEM: |
| result_code = _api.PEM_write_bio_PrivateKey( |
| bio, pkey._pkey, _api.NULL, _api.NULL, 0, _api.NULL, _api.NULL) |
| elif type == FILETYPE_ASN1: |
| result_code = _api.i2d_PrivateKey_bio(bio, pkey._pkey) |
| elif type == FILETYPE_TEXT: |
| rsa = _api.EVP_PKEY_get1_RSA(pkey._pkey) |
| result_code = _api.RSA_print(bio, rsa, 0) |
| # TODO RSA_free(rsa)? |
| else: |
| raise ValueError( |
| "type argument must be FILETYPE_PEM, FILETYPE_ASN1, or " |
| "FILETYPE_TEXT") |
| |
| if result_code == 0: |
| _raise_current_error() |
| |
| return _bio_to_string(bio) |
| |
| |
| |
| def load_privatekey(type, buffer, passphrase=None): |
| """ |
| Load a private key from a buffer |
| |
| :param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1) |
| :param buffer: The buffer the key is stored in |
| :param passphrase: (optional) if encrypted PEM format, this can be |
| either the passphrase to use, or a callback for |
| providing the passphrase. |
| |
| :return: The PKey object |
| """ |
| # TODO incomplete |
| bio = _api.BIO_new_mem_buf(buffer, len(buffer)) |
| |
| if type == FILETYPE_PEM: |
| evp_pkey = _api.PEM_read_bio_PrivateKey(bio, _api.NULL, _api.NULL, _api.NULL) |
| elif type == FILETYPE_ASN1: |
| evp_pkey = _api.d2i_PrivateKey_bio(bio, _api.NULL) |
| else: |
| raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1") |
| |
| pkey = PKey.__new__(PKey) |
| pkey._pkey = evp_pkey |
| return pkey |
| |
| |
| |
| def dump_certificate_request(type, req): |
| """ |
| Dump a certificate request to a buffer |
| |
| :param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1) |
| :param req: The certificate request to dump |
| :return: The buffer with the dumped certificate request in |
| """ |
| |
| |
| |
| def load_certificate_request(type, buffer): |
| """ |
| Load a certificate request from a buffer |
| |
| :param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1) |
| :param buffer: The buffer the certificate request is stored in |
| :return: The X509Req object |
| """ |