diff --git a/src/cryptography/hazmat/backends/openssl/backend.py b/src/cryptography/hazmat/backends/openssl/backend.py index b4f9e9df4e17e..fa0dd81f793c0 100644 --- a/src/cryptography/hazmat/backends/openssl/backend.py +++ b/src/cryptography/hazmat/backends/openssl/backend.py @@ -8,17 +8,12 @@ import contextlib import itertools import typing -from contextlib import contextmanager from cryptography import utils, x509 from cryptography.exceptions import UnsupportedAlgorithm, _Reasons from cryptography.hazmat.backends.openssl import aead from cryptography.hazmat.backends.openssl.ciphers import _CipherContext from cryptography.hazmat.backends.openssl.cmac import _CMACContext -from cryptography.hazmat.backends.openssl.ec import ( - _EllipticCurvePrivateKey, - _EllipticCurvePublicKey, -) from cryptography.hazmat.backends.openssl.rsa import ( _RSAPrivateKey, _RSAPublicKey, @@ -542,10 +537,9 @@ def _evp_pkey_to_private_key( int(self._ffi.cast("uintptr_t", evp_pkey)) ) elif key_type == self._lib.EVP_PKEY_EC: - ec_cdata = self._lib.EVP_PKEY_get1_EC_KEY(evp_pkey) - self.openssl_assert(ec_cdata != self._ffi.NULL) - ec_cdata = self._ffi.gc(ec_cdata, self._lib.EC_KEY_free) - return _EllipticCurvePrivateKey(self, ec_cdata, evp_pkey) + return rust_openssl.ec.private_key_from_ptr( + int(self._ffi.cast("uintptr_t", evp_pkey)) + ) elif key_type in self._dh_types: return rust_openssl.dh.private_key_from_ptr( int(self._ffi.cast("uintptr_t", evp_pkey)) @@ -603,12 +597,9 @@ def _evp_pkey_to_public_key(self, evp_pkey) -> PublicKeyTypes: int(self._ffi.cast("uintptr_t", evp_pkey)) ) elif key_type == self._lib.EVP_PKEY_EC: - ec_cdata = self._lib.EVP_PKEY_get1_EC_KEY(evp_pkey) - if ec_cdata == self._ffi.NULL: - errors = self._consume_errors() - raise ValueError("Unable to load EC key", errors) - ec_cdata = self._ffi.gc(ec_cdata, self._lib.EC_KEY_free) - return _EllipticCurvePublicKey(self, ec_cdata, evp_pkey) + return rust_openssl.ec.public_key_from_ptr( + int(self._ffi.cast("uintptr_t", evp_pkey)) + ) elif key_type in self._dh_types: return rust_openssl.dh.public_key_from_ptr( int(self._ffi.cast("uintptr_t", evp_pkey)) @@ -944,20 +935,7 @@ def elliptic_curve_supported(self, curve: ec.EllipticCurve) -> bool: ): return False - try: - curve_nid = self._elliptic_curve_to_nid(curve) - except UnsupportedAlgorithm: - curve_nid = self._lib.NID_undef - - group = self._lib.EC_GROUP_new_by_curve_name(curve_nid) - - if group == self._ffi.NULL: - self._consume_errors() - return False - else: - self.openssl_assert(curve_nid != self._lib.NID_undef) - self._lib.EC_GROUP_free(group) - return True + return rust_openssl.ec.curve_supported(curve) def elliptic_curve_signature_algorithm_supported( self, @@ -979,158 +957,27 @@ def generate_elliptic_curve_private_key( """ Generate a new private key on the named curve. """ - - if self.elliptic_curve_supported(curve): - ec_cdata = self._ec_key_new_by_curve(curve) - - res = self._lib.EC_KEY_generate_key(ec_cdata) - self.openssl_assert(res == 1) - - evp_pkey = self._ec_cdata_to_evp_pkey(ec_cdata) - - return _EllipticCurvePrivateKey(self, ec_cdata, evp_pkey) - else: - raise UnsupportedAlgorithm( - f"Backend object does not support {curve.name}.", - _Reasons.UNSUPPORTED_ELLIPTIC_CURVE, - ) + return rust_openssl.ec.generate_private_key(curve) def load_elliptic_curve_private_numbers( self, numbers: ec.EllipticCurvePrivateNumbers ) -> ec.EllipticCurvePrivateKey: - public = numbers.public_numbers - - ec_cdata = self._ec_key_new_by_curve(public.curve) - - private_value = self._ffi.gc( - self._int_to_bn(numbers.private_value), self._lib.BN_clear_free - ) - res = self._lib.EC_KEY_set_private_key(ec_cdata, private_value) - if res != 1: - self._consume_errors() - raise ValueError("Invalid EC key.") - - with self._tmp_bn_ctx() as bn_ctx: - self._ec_key_set_public_key_affine_coordinates( - ec_cdata, public.x, public.y, bn_ctx - ) - # derive the expected public point and compare it to the one we - # just set based on the values we were given. If they don't match - # this isn't a valid key pair. - group = self._lib.EC_KEY_get0_group(ec_cdata) - self.openssl_assert(group != self._ffi.NULL) - set_point = backend._lib.EC_KEY_get0_public_key(ec_cdata) - self.openssl_assert(set_point != self._ffi.NULL) - computed_point = self._lib.EC_POINT_new(group) - self.openssl_assert(computed_point != self._ffi.NULL) - computed_point = self._ffi.gc( - computed_point, self._lib.EC_POINT_free - ) - res = self._lib.EC_POINT_mul( - group, - computed_point, - private_value, - self._ffi.NULL, - self._ffi.NULL, - bn_ctx, - ) - self.openssl_assert(res == 1) - if ( - self._lib.EC_POINT_cmp( - group, set_point, computed_point, bn_ctx - ) - != 0 - ): - raise ValueError("Invalid EC key.") - - evp_pkey = self._ec_cdata_to_evp_pkey(ec_cdata) - - return _EllipticCurvePrivateKey(self, ec_cdata, evp_pkey) + return rust_openssl.ec.from_private_numbers(numbers) def load_elliptic_curve_public_numbers( self, numbers: ec.EllipticCurvePublicNumbers ) -> ec.EllipticCurvePublicKey: - ec_cdata = self._ec_key_new_by_curve(numbers.curve) - with self._tmp_bn_ctx() as bn_ctx: - self._ec_key_set_public_key_affine_coordinates( - ec_cdata, numbers.x, numbers.y, bn_ctx - ) - evp_pkey = self._ec_cdata_to_evp_pkey(ec_cdata) - - return _EllipticCurvePublicKey(self, ec_cdata, evp_pkey) + return rust_openssl.ec.from_public_numbers(numbers) def load_elliptic_curve_public_bytes( self, curve: ec.EllipticCurve, point_bytes: bytes ) -> ec.EllipticCurvePublicKey: - ec_cdata = self._ec_key_new_by_curve(curve) - group = self._lib.EC_KEY_get0_group(ec_cdata) - self.openssl_assert(group != self._ffi.NULL) - point = self._lib.EC_POINT_new(group) - self.openssl_assert(point != self._ffi.NULL) - point = self._ffi.gc(point, self._lib.EC_POINT_free) - with self._tmp_bn_ctx() as bn_ctx: - res = self._lib.EC_POINT_oct2point( - group, point, point_bytes, len(point_bytes), bn_ctx - ) - if res != 1: - self._consume_errors() - raise ValueError("Invalid public bytes for the given curve") - - res = self._lib.EC_KEY_set_public_key(ec_cdata, point) - self.openssl_assert(res == 1) - evp_pkey = self._ec_cdata_to_evp_pkey(ec_cdata) - return _EllipticCurvePublicKey(self, ec_cdata, evp_pkey) + return rust_openssl.ec.from_public_bytes(curve, point_bytes) def derive_elliptic_curve_private_key( self, private_value: int, curve: ec.EllipticCurve ) -> ec.EllipticCurvePrivateKey: - ec_cdata = self._ec_key_new_by_curve(curve) - - group = self._lib.EC_KEY_get0_group(ec_cdata) - self.openssl_assert(group != self._ffi.NULL) - - point = self._lib.EC_POINT_new(group) - self.openssl_assert(point != self._ffi.NULL) - point = self._ffi.gc(point, self._lib.EC_POINT_free) - - value = self._int_to_bn(private_value) - value = self._ffi.gc(value, self._lib.BN_clear_free) - - with self._tmp_bn_ctx() as bn_ctx: - res = self._lib.EC_POINT_mul( - group, point, value, self._ffi.NULL, self._ffi.NULL, bn_ctx - ) - self.openssl_assert(res == 1) - - bn_x = self._lib.BN_CTX_get(bn_ctx) - bn_y = self._lib.BN_CTX_get(bn_ctx) - - res = self._lib.EC_POINT_get_affine_coordinates( - group, point, bn_x, bn_y, bn_ctx - ) - if res != 1: - self._consume_errors() - raise ValueError("Unable to derive key from private_value") - - res = self._lib.EC_KEY_set_public_key(ec_cdata, point) - self.openssl_assert(res == 1) - private = self._int_to_bn(private_value) - private = self._ffi.gc(private, self._lib.BN_clear_free) - res = self._lib.EC_KEY_set_private_key(ec_cdata, private) - self.openssl_assert(res == 1) - - evp_pkey = self._ec_cdata_to_evp_pkey(ec_cdata) - - return _EllipticCurvePrivateKey(self, ec_cdata, evp_pkey) - - def _ec_key_new_by_curve(self, curve: ec.EllipticCurve): - curve_nid = self._elliptic_curve_to_nid(curve) - return self._ec_key_new_by_curve_nid(curve_nid) - - def _ec_key_new_by_curve_nid(self, curve_nid: int): - ec_cdata = self._lib.EC_KEY_new_by_curve_name(curve_nid) - self.openssl_assert(ec_cdata != self._ffi.NULL) - return self._ffi.gc(ec_cdata, self._lib.EC_KEY_free) + return rust_openssl.ec.derive_private_key(private_value, curve) def elliptic_curve_exchange_algorithm_supported( self, algorithm: ec.ECDH, curve: ec.EllipticCurve @@ -1139,73 +986,6 @@ def elliptic_curve_exchange_algorithm_supported( algorithm, ec.ECDH ) - def _ec_cdata_to_evp_pkey(self, ec_cdata): - evp_pkey = self._create_evp_pkey_gc() - res = self._lib.EVP_PKEY_set1_EC_KEY(evp_pkey, ec_cdata) - self.openssl_assert(res == 1) - return evp_pkey - - def _elliptic_curve_to_nid(self, curve: ec.EllipticCurve) -> int: - """ - Get the NID for a curve name. - """ - - curve_aliases = {"secp192r1": "prime192v1", "secp256r1": "prime256v1"} - - curve_name = curve_aliases.get(curve.name, curve.name) - - curve_nid = self._lib.OBJ_sn2nid(curve_name.encode()) - if curve_nid == self._lib.NID_undef: - raise UnsupportedAlgorithm( - f"{curve.name} is not a supported elliptic curve", - _Reasons.UNSUPPORTED_ELLIPTIC_CURVE, - ) - return curve_nid - - @contextmanager - def _tmp_bn_ctx(self): - bn_ctx = self._lib.BN_CTX_new() - self.openssl_assert(bn_ctx != self._ffi.NULL) - bn_ctx = self._ffi.gc(bn_ctx, self._lib.BN_CTX_free) - self._lib.BN_CTX_start(bn_ctx) - try: - yield bn_ctx - finally: - self._lib.BN_CTX_end(bn_ctx) - - def _ec_key_set_public_key_affine_coordinates( - self, - ec_cdata, - x: int, - y: int, - bn_ctx, - ) -> None: - """ - Sets the public key point in the EC_KEY context to the affine x and y - values. - """ - - if x < 0 or y < 0: - raise ValueError( - "Invalid EC key. Both x and y must be non-negative." - ) - - x = self._ffi.gc(self._int_to_bn(x), self._lib.BN_free) - y = self._ffi.gc(self._int_to_bn(y), self._lib.BN_free) - group = self._lib.EC_KEY_get0_group(ec_cdata) - self.openssl_assert(group != self._ffi.NULL) - point = self._lib.EC_POINT_new(group) - self.openssl_assert(point != self._ffi.NULL) - point = self._ffi.gc(point, self._lib.EC_POINT_free) - res = self._lib.EC_POINT_set_affine_coordinates( - group, point, x, y, bn_ctx - ) - if res != 1: - self._consume_errors() - raise ValueError("Invalid EC key.") - res = self._lib.EC_KEY_set_public_key(ec_cdata, point) - self.openssl_assert(res == 1) - def _private_key_bytes( self, encoding: serialization.Encoding, @@ -1278,11 +1058,8 @@ def _private_key_bytes( key_type = self._lib.EVP_PKEY_id(evp_pkey) if encoding is serialization.Encoding.PEM: - if key_type == self._lib.EVP_PKEY_RSA: - write_bio = self._lib.PEM_write_bio_RSAPrivateKey - else: - assert key_type == self._lib.EVP_PKEY_EC - write_bio = self._lib.PEM_write_bio_ECPrivateKey + assert key_type == self._lib.EVP_PKEY_RSA + write_bio = self._lib.PEM_write_bio_RSAPrivateKey return self._private_key_bytes_via_bio( write_bio, cdata, password ) @@ -1293,11 +1070,8 @@ def _private_key_bytes( "Encryption is not supported for DER encoded " "traditional OpenSSL keys" ) - if key_type == self._lib.EVP_PKEY_RSA: - write_bio = self._lib.i2d_RSAPrivateKey_bio - else: - assert key_type == self._lib.EVP_PKEY_EC - write_bio = self._lib.i2d_ECPrivateKey_bio + assert key_type == self._lib.EVP_PKEY_RSA + write_bio = self._lib.i2d_RSAPrivateKey_bio return self._bio_func_output(write_bio, cdata) raise ValueError("Unsupported encoding for TraditionalOpenSSL") @@ -1374,8 +1148,7 @@ def _public_key_bytes( if format is serialization.PublicFormat.PKCS1: # Only RSA is supported here. key_type = self._lib.EVP_PKEY_id(evp_pkey) - if key_type != self._lib.EVP_PKEY_RSA: - raise ValueError("PKCS1 format is supported only for RSA keys") + self.openssl_assert(key_type == self._lib.EVP_PKEY_RSA) if encoding is serialization.Encoding.PEM: write_bio = self._lib.PEM_write_bio_RSAPublicKey diff --git a/src/cryptography/hazmat/backends/openssl/ec.py b/src/cryptography/hazmat/backends/openssl/ec.py deleted file mode 100644 index 9821bd193e295..0000000000000 --- a/src/cryptography/hazmat/backends/openssl/ec.py +++ /dev/null @@ -1,328 +0,0 @@ -# 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 annotations - -import typing - -from cryptography.exceptions import ( - InvalidSignature, - UnsupportedAlgorithm, - _Reasons, -) -from cryptography.hazmat.backends.openssl.utils import ( - _calculate_digest_and_algorithm, - _evp_pkey_derive, -) -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import ec - -if typing.TYPE_CHECKING: - from cryptography.hazmat.backends.openssl.backend import Backend - - -def _check_signature_algorithm( - signature_algorithm: ec.EllipticCurveSignatureAlgorithm, -) -> None: - 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: Backend, ec_key) -> str: - 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 ValueError( - "ECDSA keys with explicit parameters 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 ValueError( - "ECDSA keys with explicit parameters 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: 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 _check_key_infinity(backend: Backend, ec_cdata) -> None: - point = backend._lib.EC_KEY_get0_public_key(ec_cdata) - backend.openssl_assert(point != backend._ffi.NULL) - group = backend._lib.EC_KEY_get0_group(ec_cdata) - backend.openssl_assert(group != backend._ffi.NULL) - if backend._lib.EC_POINT_is_at_infinity(group, point): - raise ValueError( - "Cannot load an EC public key where the point is at infinity" - ) - - -def _sn_to_elliptic_curve(backend: Backend, sn: str) -> ec.EllipticCurve: - try: - return ec._CURVE_TYPES[sn]() - except KeyError: - raise UnsupportedAlgorithm( - f"{sn} is not a supported elliptic curve", - _Reasons.UNSUPPORTED_ELLIPTIC_CURVE, - ) - - -def _ecdsa_sig_sign( - backend: Backend, private_key: _EllipticCurvePrivateKey, data: bytes -) -> bytes: - 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: Backend, - public_key: _EllipticCurvePublicKey, - signature: bytes, - data: bytes, -) -> None: - res = backend._lib.ECDSA_verify( - 0, data, len(data), signature, len(signature), public_key._ec_key - ) - if res != 1: - backend._consume_errors() - raise InvalidSignature - - -class _EllipticCurvePrivateKey(ec.EllipticCurvePrivateKey): - def __init__(self, backend: 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) - _check_key_infinity(backend, ec_key_cdata) - - @property - def curve(self) -> ec.EllipticCurve: - return self._curve - - @property - def key_size(self) -> int: - return self.curve.key_size - - def exchange( - self, algorithm: ec.ECDH, peer_public_key: ec.EllipticCurvePublicKey - ) -> bytes: - 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" - ) - - return _evp_pkey_derive(self._backend, self._evp_pkey, peer_public_key) - - def public_key(self) -> ec.EllipticCurvePublicKey: - 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) -> ec.EllipticCurvePrivateNumbers: - 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: serialization.Encoding, - format: serialization.PrivateFormat, - encryption_algorithm: serialization.KeySerializationEncryption, - ) -> bytes: - return self._backend._private_key_bytes( - encoding, - format, - encryption_algorithm, - self, - self._evp_pkey, - self._ec_key, - ) - - def sign( - self, - data: bytes, - signature_algorithm: ec.EllipticCurveSignatureAlgorithm, - ) -> bytes: - _check_signature_algorithm(signature_algorithm) - data, _ = _calculate_digest_and_algorithm( - data, - signature_algorithm.algorithm, - ) - return _ecdsa_sig_sign(self._backend, self, data) - - -class _EllipticCurvePublicKey(ec.EllipticCurvePublicKey): - def __init__(self, backend: 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) - _check_key_infinity(backend, ec_key_cdata) - - @property - def curve(self) -> ec.EllipticCurve: - return self._curve - - @property - def key_size(self) -> int: - return self.curve.key_size - - def __eq__(self, other: object) -> bool: - if not isinstance(other, _EllipticCurvePublicKey): - return NotImplemented - - return ( - self._backend._lib.EVP_PKEY_cmp(self._evp_pkey, other._evp_pkey) - == 1 - ) - - def public_numbers(self) -> ec.EllipticCurvePublicNumbers: - 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: - bn_x = self._backend._lib.BN_CTX_get(bn_ctx) - bn_y = self._backend._lib.BN_CTX_get(bn_ctx) - - res = self._backend._lib.EC_POINT_get_affine_coordinates( - 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: serialization.PublicFormat) -> bytes: - 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: serialization.Encoding, - format: serialization.PublicFormat, - ) -> bytes: - 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: bytes, - data: bytes, - signature_algorithm: ec.EllipticCurveSignatureAlgorithm, - ) -> None: - _check_signature_algorithm(signature_algorithm) - data, _ = _calculate_digest_and_algorithm( - data, - signature_algorithm.algorithm, - ) - _ecdsa_sig_verify(self._backend, self, signature, data) diff --git a/src/cryptography/hazmat/backends/openssl/utils.py b/src/cryptography/hazmat/backends/openssl/utils.py index 5b404defde334..570b776ef57d9 100644 --- a/src/cryptography/hazmat/backends/openssl/utils.py +++ b/src/cryptography/hazmat/backends/openssl/utils.py @@ -9,39 +9,6 @@ from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric.utils import Prehashed -if typing.TYPE_CHECKING: - from cryptography.hazmat.backends.openssl.backend import Backend - - -def _evp_pkey_derive(backend: Backend, evp_pkey, peer_public_key) -> bytes: - ctx = backend._lib.EVP_PKEY_CTX_new(evp_pkey, backend._ffi.NULL) - backend.openssl_assert(ctx != backend._ffi.NULL) - ctx = backend._ffi.gc(ctx, backend._lib.EVP_PKEY_CTX_free) - res = backend._lib.EVP_PKEY_derive_init(ctx) - backend.openssl_assert(res == 1) - - if backend._lib.Cryptography_HAS_EVP_PKEY_SET_PEER_EX: - res = backend._lib.EVP_PKEY_derive_set_peer_ex( - ctx, peer_public_key._evp_pkey, 0 - ) - else: - res = backend._lib.EVP_PKEY_derive_set_peer( - ctx, peer_public_key._evp_pkey - ) - backend.openssl_assert(res == 1) - - keylen = backend._ffi.new("size_t *") - res = backend._lib.EVP_PKEY_derive(ctx, backend._ffi.NULL, keylen) - backend.openssl_assert(res == 1) - backend.openssl_assert(keylen[0] > 0) - buf = backend._ffi.new("unsigned char[]", keylen[0]) - res = backend._lib.EVP_PKEY_derive(ctx, buf, keylen) - if res != 1: - errors = backend._consume_errors() - raise ValueError("Error computing shared key.", errors) - - return backend._ffi.buffer(buf, keylen[0])[:] - def _calculate_digest_and_algorithm( data: bytes, diff --git a/src/cryptography/hazmat/bindings/_rust/openssl/__init__.pyi b/src/cryptography/hazmat/bindings/_rust/openssl/__init__.pyi index 82f30d20b0ab2..d0e6ccaed238c 100644 --- a/src/cryptography/hazmat/bindings/_rust/openssl/__init__.pyi +++ b/src/cryptography/hazmat/bindings/_rust/openssl/__init__.pyi @@ -7,6 +7,7 @@ import typing from cryptography.hazmat.bindings._rust.openssl import ( dh, dsa, + ec, ed448, ed25519, hashes, @@ -22,6 +23,7 @@ __all__ = [ "raise_openssl_error", "dh", "dsa", + "ec", "hashes", "hmac", "kdf", diff --git a/src/cryptography/hazmat/bindings/_rust/openssl/ec.pyi b/src/cryptography/hazmat/bindings/_rust/openssl/ec.pyi new file mode 100644 index 0000000000000..f4fdf3856fc3a --- /dev/null +++ b/src/cryptography/hazmat/bindings/_rust/openssl/ec.pyi @@ -0,0 +1,27 @@ +# 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 cryptography.hazmat.primitives.asymmetric import ec + +class ECPrivateKey: ... +class ECPublicKey: ... + +def curve_supported(curve: ec.EllipticCurve) -> bool: ... +def private_key_from_ptr(ptr: int) -> ec.EllipticCurvePrivateKey: ... +def public_key_from_ptr(ptr: int) -> ec.EllipticCurvePublicKey: ... +def generate_private_key( + curve: ec.EllipticCurve, +) -> ec.EllipticCurvePrivateKey: ... +def from_private_numbers( + numbers: ec.EllipticCurvePrivateNumbers, +) -> ec.EllipticCurvePrivateKey: ... +def from_public_numbers( + numbers: ec.EllipticCurvePublicNumbers, +) -> ec.EllipticCurvePublicKey: ... +def from_public_bytes( + curve: ec.EllipticCurve, data: bytes +) -> ec.EllipticCurvePublicKey: ... +def derive_private_key( + private_value: int, curve: ec.EllipticCurve +) -> ec.EllipticCurvePrivateKey: ... diff --git a/src/cryptography/hazmat/primitives/asymmetric/ec.py b/src/cryptography/hazmat/primitives/asymmetric/ec.py index ddfaabf4f3e4e..3a5eb62573e09 100644 --- a/src/cryptography/hazmat/primitives/asymmetric/ec.py +++ b/src/cryptography/hazmat/primitives/asymmetric/ec.py @@ -9,6 +9,7 @@ from cryptography import utils from cryptography.hazmat._oid import ObjectIdentifier +from cryptography.hazmat.bindings._rust import openssl as rust_openssl from cryptography.hazmat.primitives import _serialization, hashes from cryptography.hazmat.primitives.asymmetric import utils as asym_utils @@ -121,6 +122,7 @@ def private_bytes( EllipticCurvePrivateKeyWithSerialization = EllipticCurvePrivateKey +EllipticCurvePrivateKey.register(rust_openssl.ec.ECPrivateKey) class EllipticCurvePublicKey(metaclass=abc.ABCMeta): @@ -192,6 +194,7 @@ def __eq__(self, other: object) -> bool: EllipticCurvePublicKeyWithSerialization = EllipticCurvePublicKey +EllipticCurvePublicKey.register(rust_openssl.ec.ECPublicKey) class SECT571R1(EllipticCurve): diff --git a/src/rust/build.rs b/src/rust/build.rs index 574560394d88e..49740fccecfbf 100644 --- a/src/rust/build.rs +++ b/src/rust/build.rs @@ -6,6 +6,14 @@ use std::env; #[allow(clippy::unusual_byte_groupings)] fn main() { + if let Ok(version) = env::var("DEP_OPENSSL_VERSION_NUMBER") { + let version = u64::from_str_radix(&version, 16).unwrap(); + + if version >= 0x3_00_00_00_0 { + println!("cargo:rustc-cfg=CRYPTOGRAPHY_OPENSSL_300_OR_GREATER"); + } + } + if let Ok(version) = env::var("DEP_OPENSSL_LIBRESSL_VERSION_NUMBER") { let version = u64::from_str_radix(&version, 16).unwrap(); diff --git a/src/rust/src/backend/ec.rs b/src/rust/src/backend/ec.rs new file mode 100644 index 0000000000000..52a16bd9a8750 --- /dev/null +++ b/src/rust/src/backend/ec.rs @@ -0,0 +1,570 @@ +// 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. + +use crate::backend::utils; +use crate::error::{CryptographyError, CryptographyResult}; +use crate::exceptions; +use foreign_types_shared::ForeignTypeRef; +use pyo3::basic::CompareOp; +use pyo3::ToPyObject; + +#[pyo3::prelude::pyclass(module = "cryptography.hazmat.bindings._rust.openssl.ec")] +struct ECPrivateKey { + pkey: openssl::pkey::PKey, + #[pyo3(get)] + curve: pyo3::Py, +} + +#[pyo3::prelude::pyclass(module = "cryptography.hazmat.bindings._rust.openssl.ec")] +struct ECPublicKey { + pkey: openssl::pkey::PKey, + #[pyo3(get)] + curve: pyo3::Py, +} + +fn curve_from_py_curve( + py: pyo3::Python<'_>, + py_curve: &pyo3::PyAny, +) -> CryptographyResult { + let curve_name = py_curve.getattr(pyo3::intern!(py, "name"))?.extract()?; + let nid = match curve_name { + "secp192r1" => openssl::nid::Nid::X9_62_PRIME192V1, + "secp224r1" => openssl::nid::Nid::SECP224R1, + "secp256r1" => openssl::nid::Nid::X9_62_PRIME256V1, + "secp384r1" => openssl::nid::Nid::SECP384R1, + "secp521r1" => openssl::nid::Nid::SECP521R1, + + "secp256k1" => openssl::nid::Nid::SECP256K1, + + "sect233r1" => openssl::nid::Nid::SECT233R1, + "sect283r1" => openssl::nid::Nid::SECT283R1, + "sect409r1" => openssl::nid::Nid::SECT409R1, + "sect571r1" => openssl::nid::Nid::SECT571R1, + + "sect163r2" => openssl::nid::Nid::SECT163R2, + + "sect163k1" => openssl::nid::Nid::SECT163K1, + "sect233k1" => openssl::nid::Nid::SECT233K1, + "sect283k1" => openssl::nid::Nid::SECT283K1, + "sect409k1" => openssl::nid::Nid::SECT409K1, + "sect571k1" => openssl::nid::Nid::SECT571K1, + + #[cfg(not(any(CRYPTOGRAPHY_IS_LIBRESSL, CRYPTOGRAPHY_IS_BORINGSSL)))] + "brainpoolP256r1" => openssl::nid::Nid::BRAINPOOL_P256R1, + #[cfg(not(any(CRYPTOGRAPHY_IS_LIBRESSL, CRYPTOGRAPHY_IS_BORINGSSL)))] + "brainpoolP384r1" => openssl::nid::Nid::BRAINPOOL_P384R1, + #[cfg(not(any(CRYPTOGRAPHY_IS_LIBRESSL, CRYPTOGRAPHY_IS_BORINGSSL)))] + "brainpoolP512r1" => openssl::nid::Nid::BRAINPOOL_P512R1, + + _ => { + return Err(CryptographyError::from( + exceptions::UnsupportedAlgorithm::new_err(( + format!("Curve {} is not supported", curve_name), + exceptions::Reasons::UNSUPPORTED_ELLIPTIC_CURVE, + )), + )); + } + }; + + Ok(openssl::ec::EcGroup::from_curve_name(nid)?) +} + +fn py_curve_from_curve<'p>( + py: pyo3::Python<'p>, + curve: &openssl::ec::EcGroupRef, +) -> CryptographyResult<&'p pyo3::PyAny> { + let name = curve + .curve_name() + .ok_or_else(|| { + pyo3::exceptions::PyValueError::new_err( + "ECDSA keys with explicit parameters are unsupported at this time", + ) + })? + .short_name()?; + + if curve.asn1_flag() == openssl::ec::Asn1Flag::EXPLICIT_CURVE { + return Err(CryptographyError::from( + pyo3::exceptions::PyValueError::new_err( + "ECDSA keys with explicit parameters are unsupported at this time", + ), + )); + } + + Ok(py + .import(pyo3::intern!( + py, + "cryptography.hazmat.primitives.asymmetric.ec" + ))? + .getattr(pyo3::intern!(py, "_CURVE_TYPES"))? + .extract::<&pyo3::types::PyDict>()? + .get_item(name) + .ok_or_else(|| { + CryptographyError::from(exceptions::UnsupportedAlgorithm::new_err(( + format!("{} is not a supported elliptic curve", name), + exceptions::Reasons::UNSUPPORTED_ELLIPTIC_CURVE, + ))) + })? + .call0()?) +} + +fn check_key_infinity( + ec: &openssl::ec::EcKeyRef, +) -> CryptographyResult<()> { + if ec.public_key().is_infinity(ec.group()) { + return Err(CryptographyError::from( + pyo3::exceptions::PyValueError::new_err( + "Cannot load an EC public key where the point is at infinity", + ), + )); + } + Ok(()) +} + +#[pyo3::prelude::pyfunction] +fn curve_supported(py: pyo3::Python<'_>, py_curve: &pyo3::PyAny) -> bool { + curve_from_py_curve(py, py_curve).is_ok() +} + +#[pyo3::prelude::pyfunction] +fn private_key_from_ptr(py: pyo3::Python<'_>, ptr: usize) -> CryptographyResult { + let pkey = unsafe { openssl::pkey::PKeyRef::from_ptr(ptr as *mut _) }; + let curve = py_curve_from_curve(py, pkey.ec_key().unwrap().group())?; + check_key_infinity(&pkey.ec_key().unwrap())?; + Ok(ECPrivateKey { + pkey: pkey.to_owned(), + curve: curve.into(), + }) +} + +#[pyo3::prelude::pyfunction] +fn public_key_from_ptr(py: pyo3::Python<'_>, ptr: usize) -> CryptographyResult { + let pkey = unsafe { openssl::pkey::PKeyRef::from_ptr(ptr as *mut _) }; + let ec = pkey.ec_key().map_err(|e| { + pyo3::exceptions::PyValueError::new_err(format!("Unable to load EC key: {}", e)) + })?; + let curve = py_curve_from_curve(py, ec.group())?; + check_key_infinity(&ec)?; + Ok(ECPublicKey { + pkey: pkey.to_owned(), + curve: curve.into(), + }) +} +#[pyo3::prelude::pyfunction] +fn generate_private_key( + py: pyo3::Python<'_>, + py_curve: &pyo3::PyAny, +) -> CryptographyResult { + let curve = curve_from_py_curve(py, py_curve)?; + let key = openssl::ec::EcKey::generate(&curve)?; + + Ok(ECPrivateKey { + pkey: openssl::pkey::PKey::from_ec_key(key)?, + curve: py_curve.into(), + }) +} + +#[pyo3::prelude::pyfunction] +fn derive_private_key( + py: pyo3::Python<'_>, + py_private_value: &pyo3::types::PyLong, + py_curve: &pyo3::PyAny, +) -> CryptographyResult { + let curve = curve_from_py_curve(py, py_curve)?; + let private_value = utils::py_int_to_bn(py, py_private_value)?; + + let mut point = openssl::ec::EcPoint::new(&curve)?; + let bn_ctx = openssl::bn::BigNumContext::new()?; + point.mul_generator(&curve, &private_value, &bn_ctx)?; + let ec = openssl::ec::EcKey::from_private_components(&curve, &private_value, &point) + .map_err(|_| pyo3::exceptions::PyValueError::new_err("Invalid EC key"))?; + check_key_infinity(&ec)?; + let pkey = openssl::pkey::PKey::from_ec_key(ec)?; + + Ok(ECPrivateKey { + pkey, + curve: py_curve.into(), + }) +} + +#[pyo3::prelude::pyfunction] +fn from_public_bytes( + py: pyo3::Python<'_>, + py_curve: &pyo3::PyAny, + data: &[u8], +) -> CryptographyResult { + let curve = curve_from_py_curve(py, py_curve)?; + + let mut bn_ctx = openssl::bn::BigNumContext::new()?; + let point = openssl::ec::EcPoint::from_bytes(&curve, data, &mut bn_ctx) + .map_err(|_| pyo3::exceptions::PyValueError::new_err("Invalid EC key."))?; + let ec = openssl::ec::EcKey::from_public_key(&curve, &point)?; + let pkey = openssl::pkey::PKey::from_ec_key(ec)?; + + Ok(ECPublicKey { + pkey, + curve: py_curve.into(), + }) +} + +fn public_key_from_numbers( + py: pyo3::Python<'_>, + numbers: &pyo3::PyAny, + curve: &openssl::ec::EcGroupRef, +) -> CryptographyResult> { + let py_x = numbers.getattr(pyo3::intern!(py, "x"))?; + let py_y = numbers.getattr(pyo3::intern!(py, "y"))?; + + let zero = (0).to_object(py); + if py_x.rich_compare(&zero, CompareOp::Lt)?.is_true()? + || py_y.rich_compare(&zero, CompareOp::Lt)?.is_true()? + { + return Err(CryptographyError::from( + pyo3::exceptions::PyValueError::new_err( + "Invalid EC key. Both x and y must be non-negative.", + ), + )); + } + + let x = utils::py_int_to_bn(py, py_x)?; + let y = utils::py_int_to_bn(py, py_y)?; + + let mut point = openssl::ec::EcPoint::new(curve)?; + let mut bn_ctx = openssl::bn::BigNumContext::new()?; + point + .set_affine_coordinates_gfp(curve, &x, &y, &mut bn_ctx) + .map_err(|_| { + pyo3::exceptions::PyValueError::new_err( + "Invalid EC key. Point is not on the curve specified.", + ) + })?; + + Ok(openssl::ec::EcKey::from_public_key(curve, &point)?) +} + +#[pyo3::prelude::pyfunction] +fn from_private_numbers( + py: pyo3::Python<'_>, + numbers: &pyo3::PyAny, +) -> CryptographyResult { + let public_numbers = numbers.getattr(pyo3::intern!(py, "public_numbers"))?; + let py_curve = public_numbers.getattr(pyo3::intern!(py, "curve"))?; + + let curve = curve_from_py_curve(py, py_curve)?; + let public_key = public_key_from_numbers(py, public_numbers, &curve)?; + let private_value = + utils::py_int_to_bn(py, numbers.getattr(pyo3::intern!(py, "private_value"))?)?; + + let mut bn_ctx = openssl::bn::BigNumContext::new()?; + let mut expected_pub = openssl::ec::EcPoint::new(&curve)?; + expected_pub.mul_generator(&curve, &private_value, &bn_ctx)?; + if !expected_pub.eq(&curve, public_key.public_key(), &mut bn_ctx)? { + return Err(CryptographyError::from( + pyo3::exceptions::PyValueError::new_err("Invalid EC key."), + )); + } + + let private_key = openssl::ec::EcKey::from_private_components( + &curve, + &private_value, + public_key.public_key(), + ) + .map_err(|_| pyo3::exceptions::PyValueError::new_err("Invalid EC key."))?; + + let pkey = openssl::pkey::PKey::from_ec_key(private_key)?; + + Ok(ECPrivateKey { + pkey, + curve: py_curve.into(), + }) +} + +#[pyo3::prelude::pyfunction] +fn from_public_numbers( + py: pyo3::Python<'_>, + numbers: &pyo3::PyAny, +) -> CryptographyResult { + let py_curve = numbers.getattr(pyo3::intern!(py, "curve"))?; + + let curve = curve_from_py_curve(py, py_curve)?; + let public_key = public_key_from_numbers(py, numbers, &curve)?; + + let pkey = openssl::pkey::PKey::from_ec_key(public_key)?; + + Ok(ECPublicKey { + pkey, + curve: py_curve.into(), + }) +} + +#[pyo3::prelude::pymethods] +impl ECPrivateKey { + #[getter] + fn key_size<'p>(&'p self, py: pyo3::Python<'p>) -> pyo3::PyResult<&'p pyo3::PyAny> { + self.curve.as_ref(py).getattr(pyo3::intern!(py, "key_size")) + } + + fn exchange<'p>( + &self, + py: pyo3::Python<'p>, + algorithm: &pyo3::PyAny, + public_key: &ECPublicKey, + ) -> CryptographyResult<&'p pyo3::types::PyBytes> { + let ecdh_class: &pyo3::types::PyType = py + .import(pyo3::intern!( + py, + "cryptography.hazmat.primitives.asymmetric.ec" + ))? + .getattr(pyo3::intern!(py, "ECDH"))? + .extract()?; + + if !algorithm.is_instance(ecdh_class)? { + return Err(CryptographyError::from( + exceptions::UnsupportedAlgorithm::new_err(( + "Unsupported EC exchange algorithm", + exceptions::Reasons::UNSUPPORTED_EXCHANGE_ALGORITHM, + )), + )); + } + + let mut deriver = openssl::derive::Deriver::new(&self.pkey)?; + #[cfg(CRYPTOGRAPHY_OPENSSL_300_OR_GREATER)] + deriver + .set_peer_ex(&public_key.pkey, false) + .map_err(|_| pyo3::exceptions::PyValueError::new_err("Error computing shared key."))?; + + #[cfg(not(CRYPTOGRAPHY_OPENSSL_300_OR_GREATER))] + deriver + .set_peer(&public_key.pkey) + .map_err(|_| pyo3::exceptions::PyValueError::new_err("Error computing shared key."))?; + + Ok(pyo3::types::PyBytes::new_with(py, deriver.len()?, |b| { + let n = deriver.derive(b).map_err(|_| { + pyo3::exceptions::PyValueError::new_err("Error computing shared key.") + })?; + assert_eq!(n, b.len()); + Ok(()) + })?) + } + + fn sign<'p>( + &self, + py: pyo3::Python<'p>, + data: &pyo3::types::PyBytes, + algorithm: &pyo3::PyAny, + ) -> CryptographyResult<&'p pyo3::types::PyBytes> { + let ecdsa_class: &pyo3::types::PyType = py + .import(pyo3::intern!( + py, + "cryptography.hazmat.primitives.asymmetric.ec" + ))? + .getattr(pyo3::intern!(py, "ECDSA"))? + .extract()?; + + if !algorithm.is_instance(ecdsa_class)? { + return Err(CryptographyError::from( + exceptions::UnsupportedAlgorithm::new_err(( + "Unsupported elliptic curve signature algorithm", + exceptions::Reasons::UNSUPPORTED_PUBLIC_KEY_ALGORITHM, + )), + )); + } + + let (data, _): (&[u8], &pyo3::PyAny) = py + .import(pyo3::intern!( + py, + "cryptography.hazmat.backends.openssl.utils" + ))? + .call_method1( + pyo3::intern!(py, "_calculate_digest_and_algorithm"), + (data, algorithm.getattr(pyo3::intern!(py, "algorithm"))?), + )? + .extract()?; + + let mut signer = openssl::pkey_ctx::PkeyCtx::new(&self.pkey)?; + signer.sign_init()?; + // XXX: single allocation + let mut sig = vec![]; + signer.sign_to_vec(data, &mut sig)?; + Ok(pyo3::types::PyBytes::new(py, &sig)) + } + + fn public_key(&self, py: pyo3::Python<'_>) -> CryptographyResult { + let orig_ec = self.pkey.ec_key().unwrap(); + let ec = openssl::ec::EcKey::from_public_key(orig_ec.group(), orig_ec.public_key())?; + let pkey = openssl::pkey::PKey::from_ec_key(ec)?; + + Ok(ECPublicKey { + pkey, + curve: self.curve.clone_ref(py), + }) + } + + fn private_numbers<'p>(&self, py: pyo3::Python<'p>) -> CryptographyResult<&'p pyo3::PyAny> { + let ec = self.pkey.ec_key().unwrap(); + + let mut bn_ctx = openssl::bn::BigNumContext::new()?; + let mut x = openssl::bn::BigNum::new()?; + let mut y = openssl::bn::BigNum::new()?; + ec.public_key() + .affine_coordinates(ec.group(), &mut x, &mut y, &mut bn_ctx)?; + let py_x = utils::bn_to_py_int(py, &x)?; + let py_y = utils::bn_to_py_int(py, &y)?; + + let py_private_key = utils::bn_to_py_int(py, ec.private_key())?; + + let ec_mod = py.import(pyo3::intern!( + py, + "cryptography.hazmat.primitives.asymmetric.ec" + ))?; + + let public_numbers = ec_mod.call_method1( + pyo3::intern!(py, "EllipticCurvePublicNumbers"), + (py_x, py_y, self.curve.clone_ref(py)), + )?; + + Ok(ec_mod.call_method1( + pyo3::intern!(py, "EllipticCurvePrivateNumbers"), + (py_private_key, public_numbers), + )?) + } + + fn private_bytes<'p>( + slf: &pyo3::PyCell, + py: pyo3::Python<'p>, + encoding: &pyo3::PyAny, + format: &pyo3::PyAny, + encryption_algorithm: &pyo3::PyAny, + ) -> CryptographyResult<&'p pyo3::types::PyBytes> { + utils::pkey_private_bytes( + py, + slf, + &slf.borrow().pkey, + encoding, + format, + encryption_algorithm, + true, + false, + ) + } +} + +#[pyo3::prelude::pymethods] +impl ECPublicKey { + #[getter] + fn key_size<'p>(&'p self, py: pyo3::Python<'p>) -> pyo3::PyResult<&'p pyo3::PyAny> { + self.curve.as_ref(py).getattr(pyo3::intern!(py, "key_size")) + } + + fn verify( + &self, + py: pyo3::Python<'_>, + signature: &[u8], + data: &pyo3::types::PyBytes, + signature_algorithm: &pyo3::PyAny, + ) -> CryptographyResult<()> { + let ecdsa_class: &pyo3::types::PyType = py + .import(pyo3::intern!( + py, + "cryptography.hazmat.primitives.asymmetric.ec" + ))? + .getattr(pyo3::intern!(py, "ECDSA"))? + .extract()?; + + if !signature_algorithm.is_instance(ecdsa_class)? { + return Err(CryptographyError::from( + exceptions::UnsupportedAlgorithm::new_err(( + "Unsupported elliptic curve signature algorithm", + exceptions::Reasons::UNSUPPORTED_PUBLIC_KEY_ALGORITHM, + )), + )); + } + + let (data, _): (&[u8], &pyo3::PyAny) = py + .import(pyo3::intern!( + py, + "cryptography.hazmat.backends.openssl.utils" + ))? + .call_method1( + pyo3::intern!(py, "_calculate_digest_and_algorithm"), + ( + data, + signature_algorithm.getattr(pyo3::intern!(py, "algorithm"))?, + ), + )? + .extract()?; + + let mut verifier = openssl::pkey_ctx::PkeyCtx::new(&self.pkey)?; + verifier.verify_init()?; + let valid = verifier.verify(data, signature).unwrap_or(false); + // TODO: Empty the error stack. BoringSSL leaves one in the event of + // signature validation failure. Upstream to rust-openssl? + openssl::error::ErrorStack::get(); + if !valid { + return Err(CryptographyError::from( + exceptions::InvalidSignature::new_err(()), + )); + } + + Ok(()) + } + + fn public_numbers<'p>(&self, py: pyo3::Python<'p>) -> CryptographyResult<&'p pyo3::PyAny> { + let ec = self.pkey.ec_key().unwrap(); + + let mut bn_ctx = openssl::bn::BigNumContext::new()?; + let mut x = openssl::bn::BigNum::new()?; + let mut y = openssl::bn::BigNum::new()?; + ec.public_key() + .affine_coordinates(ec.group(), &mut x, &mut y, &mut bn_ctx)?; + let py_x = utils::bn_to_py_int(py, &x)?; + let py_y = utils::bn_to_py_int(py, &y)?; + + let ec_mod = py.import(pyo3::intern!( + py, + "cryptography.hazmat.primitives.asymmetric.ec" + ))?; + + Ok(ec_mod.call_method1( + pyo3::intern!(py, "EllipticCurvePublicNumbers"), + (py_x, py_y, self.curve.clone_ref(py)), + )?) + } + + fn public_bytes<'p>( + slf: &pyo3::PyCell, + py: pyo3::Python<'p>, + encoding: &pyo3::PyAny, + format: &pyo3::PyAny, + ) -> CryptographyResult<&'p pyo3::types::PyBytes> { + utils::pkey_public_bytes(py, slf, &slf.borrow().pkey, encoding, format, true, false) + } + + fn __richcmp__( + &self, + other: pyo3::PyRef<'_, ECPublicKey>, + op: pyo3::basic::CompareOp, + ) -> pyo3::PyResult { + match op { + pyo3::basic::CompareOp::Eq => Ok(self.pkey.public_eq(&other.pkey)), + pyo3::basic::CompareOp::Ne => Ok(!self.pkey.public_eq(&other.pkey)), + _ => Err(pyo3::exceptions::PyTypeError::new_err("Cannot be ordered")), + } + } +} +pub(crate) fn create_module(py: pyo3::Python<'_>) -> pyo3::PyResult<&pyo3::prelude::PyModule> { + let m = pyo3::prelude::PyModule::new(py, "ec")?; + m.add_function(pyo3::wrap_pyfunction!(curve_supported, m)?)?; + m.add_function(pyo3::wrap_pyfunction!(private_key_from_ptr, m)?)?; + m.add_function(pyo3::wrap_pyfunction!(public_key_from_ptr, m)?)?; + m.add_function(pyo3::wrap_pyfunction!(generate_private_key, m)?)?; + m.add_function(pyo3::wrap_pyfunction!(derive_private_key, m)?)?; + m.add_function(pyo3::wrap_pyfunction!(from_public_bytes, m)?)?; + m.add_function(pyo3::wrap_pyfunction!(from_private_numbers, m)?)?; + m.add_function(pyo3::wrap_pyfunction!(from_public_numbers, m)?)?; + + m.add_class::()?; + m.add_class::()?; + + Ok(m) +} diff --git a/src/rust/src/backend/mod.rs b/src/rust/src/backend/mod.rs index 765b0ab199f43..b032aaac44047 100644 --- a/src/rust/src/backend/mod.rs +++ b/src/rust/src/backend/mod.rs @@ -4,6 +4,7 @@ pub(crate) mod dh; pub(crate) mod dsa; +pub(crate) mod ec; #[cfg(any(not(CRYPTOGRAPHY_IS_LIBRESSL), CRYPTOGRAPHY_LIBRESSL_370_OR_GREATER))] pub(crate) mod ed25519; #[cfg(all(not(CRYPTOGRAPHY_IS_LIBRESSL), not(CRYPTOGRAPHY_IS_BORINGSSL)))] @@ -21,6 +22,7 @@ pub(crate) mod x448; pub(crate) fn add_to_module(module: &pyo3::prelude::PyModule) -> pyo3::PyResult<()> { module.add_submodule(dh::create_module(module.py())?)?; module.add_submodule(dsa::create_module(module.py())?)?; + module.add_submodule(ec::create_module(module.py())?)?; #[cfg(any(not(CRYPTOGRAPHY_IS_LIBRESSL), CRYPTOGRAPHY_LIBRESSL_370_OR_GREATER))] module.add_submodule(ed25519::create_module(module.py())?)?; diff --git a/src/rust/src/backend/utils.rs b/src/rust/src/backend/utils.rs index dea36117182b8..086f88ab93608 100644 --- a/src/rust/src/backend/utils.rs +++ b/src/rust/src/backend/utils.rs @@ -67,6 +67,9 @@ pub(crate) fn pkey_private_bytes<'p>( let best_available_encryption_class: &pyo3::types::PyType = serialization_mod .getattr(pyo3::intern!(py, "BestAvailableEncryption"))? .extract()?; + let encryption_builder_class: &pyo3::types::PyType = serialization_mod + .getattr(pyo3::intern!(py, "_KeySerializationEncryption"))? + .extract()?; if !encoding.is_instance(encoding_class)? { return Err(CryptographyError::from( @@ -109,7 +112,12 @@ pub(crate) fn pkey_private_bytes<'p>( let password = if encryption_algorithm.is_instance(no_encryption_class)? { b"" - } else if encryption_algorithm.is_instance(best_available_encryption_class)? { + } else if encryption_algorithm.is_instance(best_available_encryption_class)? + || (encryption_algorithm.is_instance(encryption_builder_class)? + && encryption_algorithm + .getattr(pyo3::intern!(py, "_format"))? + .is(format)) + { encryption_algorithm .getattr(pyo3::intern!(py, "password"))? .extract::<&[u8]>()? @@ -178,6 +186,29 @@ pub(crate) fn pkey_private_bytes<'p>( let der_bytes = dsa.private_key_to_der()?; return Ok(pyo3::types::PyBytes::new(py, &der_bytes)); } + } else if let Ok(ec) = pkey.ec_key() { + if encoding.is(encoding_class.getattr(pyo3::intern!(py, "PEM"))?) { + let pem_bytes = if password.is_empty() { + ec.private_key_to_pem()? + } else { + ec.private_key_to_pem_passphrase( + openssl::symm::Cipher::aes_256_cbc(), + password, + )? + }; + return Ok(pyo3::types::PyBytes::new(py, &pem_bytes)); + } else if encoding.is(encoding_class.getattr(pyo3::intern!(py, "DER"))?) { + if !password.is_empty() { + return Err(CryptographyError::from( + pyo3::exceptions::PyValueError::new_err( + "Encryption is not supported for DER encoded traditional OpenSSL keys", + ), + )); + } + + let der_bytes = ec.private_key_to_der()?; + return Ok(pyo3::types::PyBytes::new(py, &der_bytes)); + } } } @@ -277,6 +308,30 @@ pub(crate) fn pkey_public_bytes<'p>( )); } + if let Ok(ec) = pkey.ec_key() { + if encoding.is(encoding_class.getattr(pyo3::intern!(py, "X962"))?) { + let point_form = if format + .is(public_format_class.getattr(pyo3::intern!(py, "UncompressedPoint"))?) + { + openssl::ec::PointConversionForm::UNCOMPRESSED + } else if format.is(public_format_class.getattr(pyo3::intern!(py, "CompressedPoint"))?) + { + openssl::ec::PointConversionForm::COMPRESSED + } else { + return Err(CryptographyError::from( + pyo3::exceptions::PyValueError::new_err( + "X962 encoding must be used with CompressedPoint or UncompressedPoint format" + ) + )); + }; + let mut bn_ctx = openssl::bn::BigNumContext::new()?; + let data = ec + .public_key() + .to_bytes(ec.group(), point_form, &mut bn_ctx)?; + return Ok(pyo3::types::PyBytes::new(py, &data)); + } + } + // OpenSSH + OpenSSH if openssh_allowed && format.is(public_format_class.getattr(pyo3::intern!(py, "OpenSSH"))?) { if encoding.is(encoding_class.getattr(pyo3::intern!(py, "OpenSSH"))?) { diff --git a/tests/hazmat/backends/test_openssl.py b/tests/hazmat/backends/test_openssl.py index c8fa1efa21f56..b0058e8a4bacb 100644 --- a/tests/hazmat/backends/test_openssl.py +++ b/tests/hazmat/backends/test_openssl.py @@ -11,7 +11,6 @@ from cryptography.exceptions import InternalError, _Reasons from cryptography.hazmat.backends import default_backend from cryptography.hazmat.backends.openssl.backend import backend -from cryptography.hazmat.backends.openssl.ec import _sn_to_elliptic_curve from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.primitives.ciphers import Cipher @@ -346,12 +345,6 @@ def test_very_long_pem_serialization_password(self): ) -class TestOpenSSLEllipticCurve: - def test_sn_to_elliptic_curve_not_supported(self): - with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_ELLIPTIC_CURVE): - _sn_to_elliptic_curve(backend, "fake") - - class TestRSAPEMSerialization: def test_password_length_limit(self, rsa_key_2048): password = b"x" * 1024 diff --git a/tests/hazmat/primitives/test_ec.py b/tests/hazmat/primitives/test_ec.py index 1120fa4be3a09..beb5739b22c05 100644 --- a/tests/hazmat/primitives/test_ec.py +++ b/tests/hazmat/primitives/test_ec.py @@ -134,7 +134,12 @@ def test_derive_point_at_infinity(backend): _skip_curve_unsupported(backend, curve) # order of the curve q = 0xFFFFFFFF00000000FFFFFFFFFFFFFFFFBCE6FAADA7179E84F3B9CAC2FC632551 - with pytest.raises(ValueError, match="Unable to derive"): + # BoringSSL rejects infinity points before it ever gets to us, so it + # uses a more generic error message. + match = ( + "infinity" if not backend._lib.CRYPTOGRAPHY_IS_BORINGSSL else "Invalid" + ) + with pytest.raises(ValueError, match=match): ec.derive_private_key(q, ec.SECP256R1())