| # This file is dual licensed under the terms of the Apache License, Version |
| # 2.0, and the BSD License. See the LICENSE file in the root of this repository |
| # for complete details. |
| |
| from __future__ import absolute_import, division, print_function |
| |
| from cryptography import utils |
| from cryptography.exceptions import ( |
| InvalidSignature, |
| UnsupportedAlgorithm, |
| _Reasons, |
| ) |
| from cryptography.hazmat.backends.openssl.utils import ( |
| _calculate_digest_and_algorithm, |
| _check_not_prehashed, |
| _warn_sign_verify_deprecated, |
| ) |
| from cryptography.hazmat.primitives import hashes, serialization |
| from cryptography.hazmat.primitives.asymmetric import ( |
| AsymmetricSignatureContext, |
| AsymmetricVerificationContext, |
| ec, |
| ) |
| |
| |
| def _check_signature_algorithm(signature_algorithm): |
| if not isinstance(signature_algorithm, ec.ECDSA): |
| raise UnsupportedAlgorithm( |
| "Unsupported elliptic curve signature algorithm.", |
| _Reasons.UNSUPPORTED_PUBLIC_KEY_ALGORITHM, |
| ) |
| |
| |
| def _ec_key_curve_sn(backend, ec_key): |
| group = backend._lib.EC_KEY_get0_group(ec_key) |
| backend.openssl_assert(group != backend._ffi.NULL) |
| |
| nid = backend._lib.EC_GROUP_get_curve_name(group) |
| # The following check is to find EC keys with unnamed curves and raise |
| # an error for now. |
| if nid == backend._lib.NID_undef: |
| raise NotImplementedError( |
| "ECDSA keys with unnamed curves are unsupported at this time" |
| ) |
| |
| # This is like the above check, but it also catches the case where you |
| # explicitly encoded a curve with the same parameters as a named curve. |
| # Don't do that. |
| if ( |
| not backend._lib.CRYPTOGRAPHY_IS_LIBRESSL |
| and backend._lib.EC_GROUP_get_asn1_flag(group) == 0 |
| ): |
| raise NotImplementedError( |
| "ECDSA keys with unnamed curves are unsupported at this time" |
| ) |
| |
| curve_name = backend._lib.OBJ_nid2sn(nid) |
| backend.openssl_assert(curve_name != backend._ffi.NULL) |
| |
| sn = backend._ffi.string(curve_name).decode("ascii") |
| return sn |
| |
| |
| def _mark_asn1_named_ec_curve(backend, ec_cdata): |
| """ |
| Set the named curve flag on the EC_KEY. This causes OpenSSL to |
| serialize EC keys along with their curve OID which makes |
| deserialization easier. |
| """ |
| |
| backend._lib.EC_KEY_set_asn1_flag( |
| ec_cdata, backend._lib.OPENSSL_EC_NAMED_CURVE |
| ) |
| |
| |
| def _sn_to_elliptic_curve(backend, sn): |
| try: |
| return ec._CURVE_TYPES[sn]() |
| except KeyError: |
| raise UnsupportedAlgorithm( |
| "{} is not a supported elliptic curve".format(sn), |
| _Reasons.UNSUPPORTED_ELLIPTIC_CURVE, |
| ) |
| |
| |
| def _ecdsa_sig_sign(backend, private_key, data): |
| max_size = backend._lib.ECDSA_size(private_key._ec_key) |
| backend.openssl_assert(max_size > 0) |
| |
| sigbuf = backend._ffi.new("unsigned char[]", max_size) |
| siglen_ptr = backend._ffi.new("unsigned int[]", 1) |
| res = backend._lib.ECDSA_sign( |
| 0, data, len(data), sigbuf, siglen_ptr, private_key._ec_key |
| ) |
| backend.openssl_assert(res == 1) |
| return backend._ffi.buffer(sigbuf)[: siglen_ptr[0]] |
| |
| |
| def _ecdsa_sig_verify(backend, public_key, signature, data): |
| res = backend._lib.ECDSA_verify( |
| 0, data, len(data), signature, len(signature), public_key._ec_key |
| ) |
| if res != 1: |
| backend._consume_errors() |
| raise InvalidSignature |
| |
| |
| @utils.register_interface(AsymmetricSignatureContext) |
| class _ECDSASignatureContext(object): |
| def __init__(self, backend, private_key, algorithm): |
| self._backend = backend |
| self._private_key = private_key |
| self._digest = hashes.Hash(algorithm, backend) |
| |
| def update(self, data): |
| self._digest.update(data) |
| |
| def finalize(self): |
| digest = self._digest.finalize() |
| |
| return _ecdsa_sig_sign(self._backend, self._private_key, digest) |
| |
| |
| @utils.register_interface(AsymmetricVerificationContext) |
| class _ECDSAVerificationContext(object): |
| def __init__(self, backend, public_key, signature, algorithm): |
| self._backend = backend |
| self._public_key = public_key |
| self._signature = signature |
| self._digest = hashes.Hash(algorithm, backend) |
| |
| def update(self, data): |
| self._digest.update(data) |
| |
| def verify(self): |
| digest = self._digest.finalize() |
| _ecdsa_sig_verify( |
| self._backend, self._public_key, self._signature, digest |
| ) |
| |
| |
| @utils.register_interface(ec.EllipticCurvePrivateKeyWithSerialization) |
| class _EllipticCurvePrivateKey(object): |
| def __init__(self, backend, ec_key_cdata, evp_pkey): |
| self._backend = backend |
| self._ec_key = ec_key_cdata |
| self._evp_pkey = evp_pkey |
| |
| sn = _ec_key_curve_sn(backend, ec_key_cdata) |
| self._curve = _sn_to_elliptic_curve(backend, sn) |
| _mark_asn1_named_ec_curve(backend, ec_key_cdata) |
| |
| curve = utils.read_only_property("_curve") |
| |
| @property |
| def key_size(self): |
| return self.curve.key_size |
| |
| def signer(self, signature_algorithm): |
| _warn_sign_verify_deprecated() |
| _check_signature_algorithm(signature_algorithm) |
| _check_not_prehashed(signature_algorithm.algorithm) |
| return _ECDSASignatureContext( |
| self._backend, self, signature_algorithm.algorithm |
| ) |
| |
| def exchange(self, algorithm, peer_public_key): |
| if not ( |
| self._backend.elliptic_curve_exchange_algorithm_supported( |
| algorithm, self.curve |
| ) |
| ): |
| raise UnsupportedAlgorithm( |
| "This backend does not support the ECDH algorithm.", |
| _Reasons.UNSUPPORTED_EXCHANGE_ALGORITHM, |
| ) |
| |
| if peer_public_key.curve.name != self.curve.name: |
| raise ValueError( |
| "peer_public_key and self are not on the same curve" |
| ) |
| |
| group = self._backend._lib.EC_KEY_get0_group(self._ec_key) |
| z_len = (self._backend._lib.EC_GROUP_get_degree(group) + 7) // 8 |
| self._backend.openssl_assert(z_len > 0) |
| z_buf = self._backend._ffi.new("uint8_t[]", z_len) |
| peer_key = self._backend._lib.EC_KEY_get0_public_key( |
| peer_public_key._ec_key |
| ) |
| |
| r = self._backend._lib.ECDH_compute_key( |
| z_buf, z_len, peer_key, self._ec_key, self._backend._ffi.NULL |
| ) |
| self._backend.openssl_assert(r > 0) |
| return self._backend._ffi.buffer(z_buf)[:z_len] |
| |
| def public_key(self): |
| group = self._backend._lib.EC_KEY_get0_group(self._ec_key) |
| self._backend.openssl_assert(group != self._backend._ffi.NULL) |
| |
| curve_nid = self._backend._lib.EC_GROUP_get_curve_name(group) |
| public_ec_key = self._backend._ec_key_new_by_curve_nid(curve_nid) |
| |
| point = self._backend._lib.EC_KEY_get0_public_key(self._ec_key) |
| self._backend.openssl_assert(point != self._backend._ffi.NULL) |
| |
| res = self._backend._lib.EC_KEY_set_public_key(public_ec_key, point) |
| self._backend.openssl_assert(res == 1) |
| |
| evp_pkey = self._backend._ec_cdata_to_evp_pkey(public_ec_key) |
| |
| return _EllipticCurvePublicKey(self._backend, public_ec_key, evp_pkey) |
| |
| def private_numbers(self): |
| bn = self._backend._lib.EC_KEY_get0_private_key(self._ec_key) |
| private_value = self._backend._bn_to_int(bn) |
| return ec.EllipticCurvePrivateNumbers( |
| private_value=private_value, |
| public_numbers=self.public_key().public_numbers(), |
| ) |
| |
| def private_bytes(self, encoding, format, encryption_algorithm): |
| return self._backend._private_key_bytes( |
| encoding, |
| format, |
| encryption_algorithm, |
| self, |
| self._evp_pkey, |
| self._ec_key, |
| ) |
| |
| def sign(self, data, signature_algorithm): |
| _check_signature_algorithm(signature_algorithm) |
| data, algorithm = _calculate_digest_and_algorithm( |
| self._backend, data, signature_algorithm._algorithm |
| ) |
| return _ecdsa_sig_sign(self._backend, self, data) |
| |
| |
| @utils.register_interface(ec.EllipticCurvePublicKeyWithSerialization) |
| class _EllipticCurvePublicKey(object): |
| def __init__(self, backend, ec_key_cdata, evp_pkey): |
| self._backend = backend |
| self._ec_key = ec_key_cdata |
| self._evp_pkey = evp_pkey |
| |
| sn = _ec_key_curve_sn(backend, ec_key_cdata) |
| self._curve = _sn_to_elliptic_curve(backend, sn) |
| _mark_asn1_named_ec_curve(backend, ec_key_cdata) |
| |
| curve = utils.read_only_property("_curve") |
| |
| @property |
| def key_size(self): |
| return self.curve.key_size |
| |
| def verifier(self, signature, signature_algorithm): |
| _warn_sign_verify_deprecated() |
| utils._check_bytes("signature", signature) |
| |
| _check_signature_algorithm(signature_algorithm) |
| _check_not_prehashed(signature_algorithm.algorithm) |
| return _ECDSAVerificationContext( |
| self._backend, self, signature, signature_algorithm.algorithm |
| ) |
| |
| def public_numbers(self): |
| get_func, group = self._backend._ec_key_determine_group_get_func( |
| self._ec_key |
| ) |
| point = self._backend._lib.EC_KEY_get0_public_key(self._ec_key) |
| self._backend.openssl_assert(point != self._backend._ffi.NULL) |
| |
| with self._backend._tmp_bn_ctx() as bn_ctx: |
| bn_x = self._backend._lib.BN_CTX_get(bn_ctx) |
| bn_y = self._backend._lib.BN_CTX_get(bn_ctx) |
| |
| res = get_func(group, point, bn_x, bn_y, bn_ctx) |
| self._backend.openssl_assert(res == 1) |
| |
| x = self._backend._bn_to_int(bn_x) |
| y = self._backend._bn_to_int(bn_y) |
| |
| return ec.EllipticCurvePublicNumbers(x=x, y=y, curve=self._curve) |
| |
| def _encode_point(self, format): |
| if format is serialization.PublicFormat.CompressedPoint: |
| conversion = self._backend._lib.POINT_CONVERSION_COMPRESSED |
| else: |
| assert format is serialization.PublicFormat.UncompressedPoint |
| conversion = self._backend._lib.POINT_CONVERSION_UNCOMPRESSED |
| |
| group = self._backend._lib.EC_KEY_get0_group(self._ec_key) |
| self._backend.openssl_assert(group != self._backend._ffi.NULL) |
| point = self._backend._lib.EC_KEY_get0_public_key(self._ec_key) |
| self._backend.openssl_assert(point != self._backend._ffi.NULL) |
| with self._backend._tmp_bn_ctx() as bn_ctx: |
| buflen = self._backend._lib.EC_POINT_point2oct( |
| group, point, conversion, self._backend._ffi.NULL, 0, bn_ctx |
| ) |
| self._backend.openssl_assert(buflen > 0) |
| buf = self._backend._ffi.new("char[]", buflen) |
| res = self._backend._lib.EC_POINT_point2oct( |
| group, point, conversion, buf, buflen, bn_ctx |
| ) |
| self._backend.openssl_assert(buflen == res) |
| |
| return self._backend._ffi.buffer(buf)[:] |
| |
| def public_bytes(self, encoding, format): |
| |
| if ( |
| encoding is serialization.Encoding.X962 |
| or format is serialization.PublicFormat.CompressedPoint |
| or format is serialization.PublicFormat.UncompressedPoint |
| ): |
| if encoding is not serialization.Encoding.X962 or format not in ( |
| serialization.PublicFormat.CompressedPoint, |
| serialization.PublicFormat.UncompressedPoint, |
| ): |
| raise ValueError( |
| "X962 encoding must be used with CompressedPoint or " |
| "UncompressedPoint format" |
| ) |
| |
| return self._encode_point(format) |
| else: |
| return self._backend._public_key_bytes( |
| encoding, format, self, self._evp_pkey, None |
| ) |
| |
| def verify(self, signature, data, signature_algorithm): |
| _check_signature_algorithm(signature_algorithm) |
| data, algorithm = _calculate_digest_and_algorithm( |
| self._backend, data, signature_algorithm._algorithm |
| ) |
| _ecdsa_sig_verify(self._backend, self, signature, data) |