mirror of https://github.com/nucypher/pyUmbral.git
Working secret & public keys
parent
f030fd3353
commit
f33431d92a
|
@ -1,8 +1,18 @@
|
|||
from umbral.__about__ import (
|
||||
from .__about__ import (
|
||||
__author__, __license__, __summary__, __title__, __version__, __copyright__, __email__, __url__
|
||||
)
|
||||
|
||||
from .keys import SecretKey, PublicKey
|
||||
|
||||
__all__ = [
|
||||
"__title__", "__summary__", "__version__", "__author__", "__license__", "__copyright__", "__email__", "__url__"
|
||||
"__title__",
|
||||
"__summary__",
|
||||
"__version__",
|
||||
"__author__",
|
||||
"__license__",
|
||||
"__copyright__",
|
||||
"__email__",
|
||||
"__url__",
|
||||
"SecretKey",
|
||||
"PublicKey",
|
||||
]
|
||||
|
|
|
@ -0,0 +1,120 @@
|
|||
from cryptography.hazmat.backends import default_backend
|
||||
|
||||
from . import openssl
|
||||
|
||||
|
||||
class Curve:
|
||||
"""
|
||||
Acts as a container to store constant variables such as the OpenSSL
|
||||
curve_nid, the EC_GROUP struct, and the order of the curve.
|
||||
|
||||
Contains a whitelist of supported elliptic curves used in pyUmbral.
|
||||
|
||||
"""
|
||||
|
||||
_supported_curves = {
|
||||
415: 'secp256r1',
|
||||
714: 'secp256k1',
|
||||
715: 'secp384r1'
|
||||
}
|
||||
|
||||
def __init__(self, nid: int) -> None:
|
||||
"""
|
||||
Instantiates an OpenSSL curve with the provided curve_nid and derives
|
||||
the proper EC_GROUP struct and order. You can _only_ instantiate curves
|
||||
with supported nids (see `Curve.supported_curves`).
|
||||
"""
|
||||
|
||||
try:
|
||||
self.__curve_name = self._supported_curves[nid]
|
||||
except KeyError:
|
||||
raise NotImplementedError("Curve NID {} is not supported.".format(nid))
|
||||
|
||||
# set only once
|
||||
self.__curve_nid = nid
|
||||
self.__ec_group = openssl._get_ec_group_by_curve_nid(self.__curve_nid)
|
||||
self.__order = openssl._get_ec_order_by_group(self.ec_group)
|
||||
self.__generator = openssl._get_ec_generator_by_group(self.ec_group)
|
||||
|
||||
# Init cache
|
||||
self.__field_order_size_in_bytes = 0
|
||||
self.__group_order_size_in_bytes = 0
|
||||
|
||||
@classmethod
|
||||
def from_name(cls, name: str) -> 'Curve':
|
||||
"""
|
||||
Alternate constructor to generate a curve instance by its name.
|
||||
|
||||
Raises NotImplementedError if the name cannot be mapped to a known
|
||||
supported curve NID.
|
||||
|
||||
"""
|
||||
|
||||
name = name.casefold() # normalize
|
||||
|
||||
for supported_nid, supported_name in cls._supported_curves.items():
|
||||
if name == supported_name:
|
||||
instance = cls(nid=supported_nid)
|
||||
break
|
||||
else:
|
||||
message = "{} is not supported curve name.".format(name)
|
||||
raise NotImplementedError(message)
|
||||
|
||||
return instance
|
||||
|
||||
def __eq__(self, other):
|
||||
return self.__curve_nid == other.curve_nid
|
||||
|
||||
def __str__(self):
|
||||
return "<OpenSSL Curve(nid={}, name={})>".format(self.__curve_nid, self.__curve_name)
|
||||
|
||||
#
|
||||
# Immutable Curve Data
|
||||
#
|
||||
|
||||
@property
|
||||
def field_order_size_in_bytes(self) -> int:
|
||||
if not self.__field_order_size_in_bytes:
|
||||
size_in_bits = openssl._get_ec_group_degree(self.__ec_group)
|
||||
self.__field_order_size_in_bytes = (size_in_bits + 7) // 8
|
||||
return self.__field_order_size_in_bytes
|
||||
|
||||
@property
|
||||
def group_order_size_in_bytes(self) -> int:
|
||||
if not self.__group_order_size_in_bytes:
|
||||
BN_num_bytes = default_backend()._lib.BN_num_bytes
|
||||
self.__group_order_size_in_bytes = BN_num_bytes(self.order)
|
||||
return self.__group_order_size_in_bytes
|
||||
|
||||
@property
|
||||
def curve_nid(self) -> int:
|
||||
return self.__curve_nid
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return self.__curve_name
|
||||
|
||||
@property
|
||||
def ec_group(self):
|
||||
return self.__ec_group
|
||||
|
||||
@property
|
||||
def order(self):
|
||||
return self.__order
|
||||
|
||||
@property
|
||||
def generator(self):
|
||||
return self.__generator
|
||||
|
||||
|
||||
#
|
||||
# Global Curve Instances
|
||||
#
|
||||
|
||||
SECP256R1 = Curve.from_name('secp256r1')
|
||||
SECP256K1 = Curve.from_name('secp256k1')
|
||||
SECP384R1 = Curve.from_name('secp384r1')
|
||||
|
||||
CURVES = (SECP256K1, SECP256R1, SECP384R1)
|
||||
|
||||
CURVE = SECP256K1
|
|
@ -0,0 +1,148 @@
|
|||
from typing import Optional, Tuple
|
||||
|
||||
from cryptography.hazmat.backends.openssl import backend
|
||||
|
||||
from . import openssl
|
||||
from .curve import CURVE
|
||||
from .curve_scalar import CurveScalar
|
||||
from .serializable import Serializable
|
||||
|
||||
|
||||
class CurvePoint(Serializable):
|
||||
"""
|
||||
Represents an OpenSSL EC_POINT except more Pythonic.
|
||||
"""
|
||||
|
||||
def __init__(self, backend_point) -> None:
|
||||
self._backend_point = backend_point
|
||||
|
||||
@classmethod
|
||||
def generator(cls) -> 'CurvePoint':
|
||||
return cls(CURVE.generator)
|
||||
|
||||
@classmethod
|
||||
def random(cls) -> 'CurvePoint':
|
||||
"""
|
||||
Returns a CurvePoint object with a cryptographically secure EC_POINT based
|
||||
on the provided curve.
|
||||
"""
|
||||
return cls.generator() * CurveScalar.random_nonzero()
|
||||
|
||||
@classmethod
|
||||
def from_affine(cls, coords: Tuple[int, int]) -> 'CurvePoint':
|
||||
"""
|
||||
Returns a CurvePoint object from the given affine coordinates in a tuple in
|
||||
the format of (x, y) and a given curve.
|
||||
"""
|
||||
affine_x, affine_y = coords
|
||||
if type(affine_x) == int:
|
||||
affine_x = openssl._int_to_bn(affine_x, curve=None)
|
||||
|
||||
if type(affine_y) == int:
|
||||
affine_y = openssl._int_to_bn(affine_y, curve=None)
|
||||
|
||||
backend_point = openssl._get_EC_POINT_via_affine(affine_x, affine_y, CURVE)
|
||||
return cls(backend_point)
|
||||
|
||||
def to_affine(self):
|
||||
"""
|
||||
Returns a tuple of Python ints in the format of (x, y) that represents
|
||||
the point in the curve.
|
||||
"""
|
||||
affine_x, affine_y = openssl._get_affine_coords_via_EC_POINT(self._backend_point, CURVE)
|
||||
return (backend._bn_to_int(affine_x), backend._bn_to_int(affine_y))
|
||||
|
||||
@classmethod
|
||||
def __take__(cls, data: bytes) -> Tuple['CurvePoint', bytes]:
|
||||
"""
|
||||
Returns a CurvePoint object from the given byte data on the curve provided.
|
||||
"""
|
||||
size = CURVE.field_order_size_in_bytes + 1 # compressed point size
|
||||
point_data, data = cls.__take_bytes__(data, size)
|
||||
|
||||
point = openssl._get_new_EC_POINT(CURVE)
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
res = backend._lib.EC_POINT_oct2point(
|
||||
CURVE.ec_group, point, point_data, len(point_data), bn_ctx);
|
||||
backend.openssl_assert(res == 1)
|
||||
|
||||
return cls(point), data
|
||||
|
||||
def __bytes__(self) -> bytes:
|
||||
"""
|
||||
Returns the CurvePoint serialized as bytes in the compressed form.
|
||||
"""
|
||||
point_conversion_form = backend._lib.POINT_CONVERSION_COMPRESSED
|
||||
size = CURVE.field_order_size_in_bytes + 1 # compressed point size
|
||||
|
||||
bin_ptr = backend._ffi.new("unsigned char[]", size)
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
bin_len = backend._lib.EC_POINT_point2oct(
|
||||
CURVE.ec_group, self._backend_point, point_conversion_form,
|
||||
bin_ptr, size, bn_ctx
|
||||
)
|
||||
backend.openssl_assert(bin_len != 0)
|
||||
|
||||
return bytes(backend._ffi.buffer(bin_ptr, bin_len)[:])
|
||||
|
||||
def __eq__(self, other):
|
||||
"""
|
||||
Compares two EC_POINTS for equality.
|
||||
"""
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
is_equal = backend._lib.EC_POINT_cmp(
|
||||
CURVE.ec_group, self._backend_point, other._backend_point, bn_ctx
|
||||
)
|
||||
backend.openssl_assert(is_equal != -1)
|
||||
|
||||
# 1 is not-equal, 0 is equal, -1 is error
|
||||
return not bool(is_equal)
|
||||
|
||||
def __mul__(self, other: CurveScalar) -> 'CurvePoint':
|
||||
"""
|
||||
Performs an EC_POINT_mul on an EC_POINT and a BIGNUM.
|
||||
"""
|
||||
# TODO: Check that both points use the same curve.
|
||||
prod = openssl._get_new_EC_POINT(CURVE)
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
res = backend._lib.EC_POINT_mul(
|
||||
CURVE.ec_group, prod, backend._ffi.NULL,
|
||||
self._backend_point, other._backend_bignum, bn_ctx
|
||||
)
|
||||
backend.openssl_assert(res == 1)
|
||||
|
||||
return CurvePoint(prod)
|
||||
|
||||
def __add__(self, other: 'CurvePoint') -> 'CurvePoint':
|
||||
"""
|
||||
Performs an EC_POINT_add on two EC_POINTS.
|
||||
"""
|
||||
op_sum = openssl._get_new_EC_POINT(CURVE)
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
res = backend._lib.EC_POINT_add(
|
||||
CURVE.ec_group, op_sum, self._backend_point, other._backend_point, bn_ctx
|
||||
)
|
||||
backend.openssl_assert(res == 1)
|
||||
return CurvePoint(op_sum)
|
||||
|
||||
def __sub__(self, other: 'CurvePoint') -> 'CurvePoint':
|
||||
"""
|
||||
Performs subtraction by adding the inverse of the `other` to the point.
|
||||
"""
|
||||
return (self + (-other))
|
||||
|
||||
def __neg__(self) -> 'CurvePoint':
|
||||
"""
|
||||
Computes the additive inverse of a CurvePoint, by performing an
|
||||
EC_POINT_invert on itself.
|
||||
"""
|
||||
inv = backend._lib.EC_POINT_dup(self._backend_point, CURVE.ec_group)
|
||||
backend.openssl_assert(inv != backend._ffi.NULL)
|
||||
inv = backend._ffi.gc(inv, backend._lib.EC_POINT_clear_free)
|
||||
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
res = backend._lib.EC_POINT_invert(
|
||||
CURVE.ec_group, inv, bn_ctx
|
||||
)
|
||||
backend.openssl_assert(res == 1)
|
||||
return CurvePoint(inv)
|
|
@ -0,0 +1,156 @@
|
|||
from typing import Optional, Union, Tuple
|
||||
|
||||
from cryptography.hazmat.backends.openssl import backend
|
||||
|
||||
from . import openssl
|
||||
from .curve import CURVE
|
||||
from .serializable import Serializable
|
||||
|
||||
|
||||
class CurveScalar(Serializable):
|
||||
"""
|
||||
Represents an OpenSSL Bignum modulo the order of a curve. Some of these
|
||||
operations will only work with prime numbers.
|
||||
By default, the underlying OpenSSL BIGNUM has BN_FLG_CONSTTIME set for
|
||||
constant time operations.
|
||||
"""
|
||||
|
||||
def __init__(self, backend_bignum):
|
||||
on_curve = openssl._bn_is_on_curve(backend_bignum, CURVE)
|
||||
if not on_curve:
|
||||
raise ValueError("The provided BIGNUM is not on the provided curve.")
|
||||
|
||||
self._backend_bignum = backend_bignum
|
||||
|
||||
@classmethod
|
||||
def random_nonzero(cls) -> 'CurveScalar':
|
||||
"""
|
||||
Returns a CurveScalar object with a cryptographically secure OpenSSL BIGNUM.
|
||||
"""
|
||||
one = backend._lib.BN_value_one()
|
||||
|
||||
# TODO: in most cases, we want this number to be secret.
|
||||
# OpenSSL 1.1.1 has `BN_priv_rand_range()`, but it is not
|
||||
# currently exported by `cryptography`.
|
||||
# Use when available.
|
||||
|
||||
# Calculate `order - 1`
|
||||
order_minus_1 = openssl._get_new_BN()
|
||||
res = backend._lib.BN_sub(order_minus_1, CURVE.order, one)
|
||||
backend.openssl_assert(res == 1)
|
||||
|
||||
# Get a random in range `[0, order - 1)`
|
||||
new_rand_bn = openssl._get_new_BN()
|
||||
res = backend._lib.BN_rand_range(new_rand_bn, order_minus_1)
|
||||
backend.openssl_assert(res == 1)
|
||||
|
||||
# Turn it into a random in range `[1, order)`
|
||||
op_sum = openssl._get_new_BN()
|
||||
res = backend._lib.BN_add(op_sum, new_rand_bn, one)
|
||||
backend.openssl_assert(res == 1)
|
||||
|
||||
return cls(op_sum)
|
||||
|
||||
@classmethod
|
||||
def from_int(cls, num: int) -> 'CurveScalar':
|
||||
"""
|
||||
Returns a CurveScalar object from a given integer on a curve.
|
||||
"""
|
||||
conv_bn = openssl._int_to_bn(num, CURVE)
|
||||
return cls(conv_bn)
|
||||
|
||||
@classmethod
|
||||
def __take__(cls, data: bytes) -> Tuple['CurveScalar', bytes]:
|
||||
size = backend._lib.BN_num_bytes(CURVE.order)
|
||||
scalar_data, data = cls.__take_bytes__(data, size)
|
||||
bignum = openssl._bytes_to_bn(scalar_data)
|
||||
return cls(bignum), data
|
||||
|
||||
def __bytes__(self) -> bytes:
|
||||
"""
|
||||
Returns the CurveScalar as bytes.
|
||||
"""
|
||||
size = backend._lib.BN_num_bytes(CURVE.order)
|
||||
return openssl._bn_to_bytes(self._backend_bignum, size)
|
||||
|
||||
def __int__(self) -> int:
|
||||
"""
|
||||
Converts the CurveScalar to a Python int.
|
||||
"""
|
||||
return backend._bn_to_int(self._backend_bignum)
|
||||
|
||||
def __eq__(self, other) -> bool:
|
||||
"""
|
||||
Compares the two BIGNUMS or int.
|
||||
"""
|
||||
if type(other) == int:
|
||||
other = CurveScalar.from_int(other)
|
||||
|
||||
# -1 less than, 0 is equal to, 1 is greater than
|
||||
return not bool(backend._lib.BN_cmp(self._backend_bignum, other._backend_bignum))
|
||||
|
||||
def is_zero(self):
|
||||
# BN_is_zero() is not exported, so this will have to do
|
||||
return self == 0
|
||||
|
||||
def __mul__(self, other: Union[int, 'CurveScalar']) -> 'CurveScalar':
|
||||
"""
|
||||
Performs a BN_mod_mul between two BIGNUMS.
|
||||
"""
|
||||
if isinstance(other, int):
|
||||
other = CurveScalar.from_int(other)
|
||||
|
||||
product = openssl._get_new_BN()
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
res = backend._lib.BN_mod_mul(
|
||||
product, self._backend_bignum, other._backend_bignum, CURVE.order, bn_ctx
|
||||
)
|
||||
backend.openssl_assert(res == 1)
|
||||
|
||||
return CurveScalar(product)
|
||||
|
||||
def __add__(self, other : Union[int, 'CurveScalar']) -> 'CurveScalar':
|
||||
"""
|
||||
Performs a BN_mod_add on two BIGNUMs.
|
||||
"""
|
||||
if isinstance(other, int):
|
||||
other = CurveScalar.from_int(other)
|
||||
|
||||
op_sum = openssl._get_new_BN()
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
res = backend._lib.BN_mod_add(
|
||||
op_sum, self._backend_bignum, other._backend_bignum, CURVE.order, bn_ctx
|
||||
)
|
||||
backend.openssl_assert(res == 1)
|
||||
|
||||
return CurveScalar(op_sum)
|
||||
|
||||
def __sub__(self, other : Union[int, 'CurveScalar']) -> 'CurveScalar':
|
||||
"""
|
||||
Performs a BN_mod_sub on two BIGNUMS.
|
||||
"""
|
||||
if isinstance(other, int):
|
||||
other = CurveScalar.from_int(other)
|
||||
|
||||
diff = openssl._get_new_BN()
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
res = backend._lib.BN_mod_sub(
|
||||
diff, self._backend_bignum, other._backend_bignum, CURVE.order, bn_ctx
|
||||
)
|
||||
backend.openssl_assert(res == 1)
|
||||
|
||||
return CurveScalar(diff)
|
||||
|
||||
def invert(self) -> 'CurveScalar':
|
||||
"""
|
||||
Performs a BN_mod_inverse.
|
||||
WARNING: Only in constant time if BN_FLG_CONSTTIME is set on the BN.
|
||||
"""
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
inv = backend._lib.BN_mod_inverse(
|
||||
backend._ffi.NULL, self._backend_bignum, CURVE.order, bn_ctx
|
||||
)
|
||||
backend.openssl_assert(inv != backend._ffi.NULL)
|
||||
inv = backend._ffi.gc(inv, backend._lib.BN_clear_free)
|
||||
|
||||
return CurveScalar(inv)
|
|
@ -0,0 +1,61 @@
|
|||
import os
|
||||
from typing import Optional
|
||||
|
||||
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
|
||||
from cryptography.hazmat.backends.openssl import backend
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
|
||||
from nacl.bindings.crypto_aead import (
|
||||
crypto_aead_xchacha20poly1305_ietf_encrypt as xchacha_encrypt,
|
||||
crypto_aead_xchacha20poly1305_ietf_decrypt as xchacha_decrypt,
|
||||
crypto_aead_xchacha20poly1305_ietf_KEYBYTES as XCHACHA_KEY_SIZE,
|
||||
crypto_aead_xchacha20poly1305_ietf_NPUBBYTES as XCHACHA_NONCE_SIZE,
|
||||
)
|
||||
|
||||
|
||||
def kdf(data: bytes,
|
||||
key_length: int,
|
||||
salt: Optional[bytes] = None,
|
||||
info: Optional[bytes] = None,
|
||||
) -> bytes:
|
||||
|
||||
hkdf = HKDF(algorithm=hashes.SHA256(),
|
||||
length=key_length,
|
||||
salt=salt,
|
||||
info=info,
|
||||
backend=backend)
|
||||
return hkdf.derive(data)
|
||||
|
||||
|
||||
class DEM:
|
||||
|
||||
KEY_SIZE = XCHACHA_KEY_SIZE
|
||||
NONCE_SIZE = XCHACHA_NONCE_SIZE
|
||||
|
||||
def __init__(self,
|
||||
key_material: bytes,
|
||||
salt: Optional[bytes] = None,
|
||||
info: Optional[bytes] = None,
|
||||
):
|
||||
self._key = kdf(key_material, self.KEY_SIZE, salt, info)
|
||||
|
||||
def encrypt(self, plaintext: bytes, nonce: Optional[bytes] = None) -> bytes:
|
||||
if nonce is None:
|
||||
nonce = os.urandom(self.NONCE_SIZE)
|
||||
|
||||
if len(nonce) != self.NONCE_SIZE:
|
||||
raise ValueError(f"The nonce must be exactly {self.NONCE_SIZE} bytes long")
|
||||
|
||||
ciphertext = xchacha_encrypt(plaintext, b"", nonce, self._key)
|
||||
return nonce + ciphertext
|
||||
|
||||
def decrypt(self, nonce_and_ciphertext: bytes) -> bytes:
|
||||
|
||||
if len(nonce_and_ciphertext) < self.NONCE_SIZE:
|
||||
raise ValueError(f"The ciphertext must include the nonce")
|
||||
|
||||
nonce = nonce_and_ciphertext[:self.NONCE_SIZE]
|
||||
ciphertext = nonce_and_ciphertext[self.NONCE_SIZE:]
|
||||
|
||||
# TODO: replace `nacl.exceptions.CryptoError` with our error?
|
||||
return xchacha_decrypt(ciphertext, b"", nonce, self._key)
|
|
@ -0,0 +1,62 @@
|
|||
from typing import Optional, Type
|
||||
|
||||
from cryptography.hazmat.backends.openssl import backend
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
from cryptography.exceptions import InternalError
|
||||
|
||||
from . import openssl
|
||||
from .curve import CURVE
|
||||
from .curve_scalar import CurveScalar
|
||||
from .curve_point import CurvePoint
|
||||
|
||||
|
||||
class Hash:
|
||||
def __init__(self, dst: bytes):
|
||||
self._sha256 = hashes.Hash(hashes.SHA256(), backend=backend)
|
||||
len_dst = len(dst).to_bytes(4, byteorder='big')
|
||||
self.update(len_dst + dst)
|
||||
|
||||
def update(self, data: bytes) -> None:
|
||||
self._sha256.update(data)
|
||||
|
||||
def finalize(self) -> bytes:
|
||||
return self._sha256.finalize()
|
||||
|
||||
|
||||
def unsafe_hash_to_point(dst: bytes, data: bytes) -> 'Point':
|
||||
"""
|
||||
Hashes arbitrary data into a valid EC point of the specified curve,
|
||||
using the try-and-increment method.
|
||||
|
||||
WARNING: Do not use when the input data is secret, as this implementation is not
|
||||
in constant time, and hence, it is not safe with respect to timing attacks.
|
||||
"""
|
||||
|
||||
len_data = len(data).to_bytes(4, byteorder='big')
|
||||
data_with_len = len_data + data
|
||||
sign = b'\x02'
|
||||
|
||||
# We use an internal 32-bit counter as additional input
|
||||
for i in range(2**32):
|
||||
ibytes = i.to_bytes(4, byteorder='big')
|
||||
digest = Hash(dst)
|
||||
digest.update(data_with_len + ibytes)
|
||||
point_data = digest.finalize()[:CURVE.field_order_size_in_bytes]
|
||||
|
||||
compressed_point = sign + point_data
|
||||
|
||||
try:
|
||||
return CurvePoint.from_bytes(compressed_point)
|
||||
except InternalError as e:
|
||||
# We want to catch specific InternalExceptions:
|
||||
# - Point not in the curve (code 107)
|
||||
# - Invalid compressed point (code 110)
|
||||
# https://github.com/openssl/openssl/blob/master/include/openssl/ecerr.h#L228
|
||||
if e.err_code[0].reason in (107, 110):
|
||||
pass
|
||||
else:
|
||||
# Any other exception, we raise it
|
||||
raise e
|
||||
|
||||
# Only happens with probability 2^(-32)
|
||||
raise ValueError('Could not hash input into the curve') # pragma: no cover
|
|
@ -0,0 +1,123 @@
|
|||
from typing import Tuple
|
||||
|
||||
from cryptography.hazmat.backends.openssl import backend
|
||||
from cryptography.hazmat.backends.openssl.ec import _EllipticCurvePrivateKey, _EllipticCurvePublicKey
|
||||
|
||||
from . import openssl
|
||||
from .curve import CURVE
|
||||
from .curve_scalar import CurveScalar
|
||||
from .curve_point import CurvePoint
|
||||
from .dem import DEM
|
||||
from .serializable import Serializable
|
||||
|
||||
|
||||
class SecretKey(Serializable):
|
||||
|
||||
__SERIALIZATION_INFO = b"SECRET_KEY"
|
||||
|
||||
def __init__(self, scalar_key: CurveScalar):
|
||||
self._scalar_key = scalar_key
|
||||
|
||||
@classmethod
|
||||
def random(cls) -> 'SecretKey':
|
||||
"""
|
||||
Generates a secret key and returns it.
|
||||
"""
|
||||
return cls(CurveScalar.random_nonzero())
|
||||
|
||||
def __eq__(self, other):
|
||||
return self._scalar_key == other._scalar_key
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.__class__.__name__}:..."
|
||||
|
||||
def __hash__(self):
|
||||
raise NotImplementedError("Hashing secret objects is insecure")
|
||||
|
||||
@classmethod
|
||||
def __take__(cls, data: bytes) -> Tuple['SecretKey', bytes]:
|
||||
(scalar_key,), data = cls.__take_types__(data, CurveScalar)
|
||||
return cls(scalar_key), data
|
||||
|
||||
def __bytes__(self) -> bytes:
|
||||
return bytes(self._scalar_key)
|
||||
|
||||
def to_cryptography_privkey(self) -> _EllipticCurvePrivateKey:
|
||||
"""
|
||||
Returns a cryptography.io EllipticCurvePrivateKey from the Umbral key.
|
||||
"""
|
||||
ec_key = backend._lib.EC_KEY_new()
|
||||
backend.openssl_assert(ec_key != backend._ffi.NULL)
|
||||
ec_key = backend._ffi.gc(ec_key, backend._lib.EC_KEY_free)
|
||||
|
||||
set_group_result = backend._lib.EC_KEY_set_group(ec_key, CURVE.ec_group)
|
||||
backend.openssl_assert(set_group_result == 1)
|
||||
|
||||
set_privkey_result = backend._lib.EC_KEY_set_private_key(
|
||||
ec_key, self._scalar_key._backend_bignum
|
||||
)
|
||||
backend.openssl_assert(set_privkey_result == 1)
|
||||
|
||||
# Get public key
|
||||
point = openssl._get_new_EC_POINT(CURVE)
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
mult_result = backend._lib.EC_POINT_mul(
|
||||
CURVE.ec_group, point, self._scalar_key._backend_bignum,
|
||||
backend._ffi.NULL, backend._ffi.NULL, bn_ctx
|
||||
)
|
||||
backend.openssl_assert(mult_result == 1)
|
||||
|
||||
set_pubkey_result = backend._lib.EC_KEY_set_public_key(ec_key, point)
|
||||
backend.openssl_assert(set_pubkey_result == 1)
|
||||
|
||||
evp_pkey = backend._ec_cdata_to_evp_pkey(ec_key)
|
||||
return _EllipticCurvePrivateKey(backend, ec_key, evp_pkey)
|
||||
|
||||
|
||||
class PublicKey(Serializable):
|
||||
|
||||
def __init__(self, point_key: CurvePoint):
|
||||
self._point_key = point_key
|
||||
|
||||
def point(self):
|
||||
return self._point_key
|
||||
|
||||
@classmethod
|
||||
def from_secret_key(cls, sk: SecretKey) -> 'PublicKey':
|
||||
return cls(CurvePoint.generator() * sk.secret_scalar())
|
||||
|
||||
@classmethod
|
||||
def __take__(cls, data: bytes) -> Tuple['PublicKey', bytes]:
|
||||
(point_key,), data = cls.__take_types__(data, CurvePoint)
|
||||
return cls(point_key), data
|
||||
|
||||
def __bytes__(self) -> bytes:
|
||||
return bytes(self._point_key)
|
||||
|
||||
def to_cryptography_pubkey(self) -> _EllipticCurvePublicKey:
|
||||
"""
|
||||
Returns a cryptography.io EllipticCurvePublicKey from the Umbral key.
|
||||
"""
|
||||
ec_key = backend._lib.EC_KEY_new()
|
||||
backend.openssl_assert(ec_key != backend._ffi.NULL)
|
||||
ec_key = backend._ffi.gc(ec_key, backend._lib.EC_KEY_free)
|
||||
|
||||
set_group_result = backend._lib.EC_KEY_set_group(ec_key, CURVE.ec_group)
|
||||
backend.openssl_assert(set_group_result == 1)
|
||||
|
||||
set_pubkey_result = backend._lib.EC_KEY_set_public_key(
|
||||
ec_key, self._point_key._backend_point
|
||||
)
|
||||
backend.openssl_assert(set_pubkey_result == 1)
|
||||
|
||||
evp_pkey = backend._ec_cdata_to_evp_pkey(ec_key)
|
||||
return _EllipticCurvePublicKey(backend, ec_key, evp_pkey)
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.__class__.__name__}:{bytes(self).hex()[:16]}"
|
||||
|
||||
def __eq__(self, other):
|
||||
return self._point_key == other._point_key
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash((self.__class__, bytes(self)))
|
|
@ -0,0 +1,201 @@
|
|||
from contextlib import contextmanager
|
||||
import typing
|
||||
|
||||
from cryptography.hazmat.backends.openssl import backend
|
||||
|
||||
|
||||
@typing.no_type_check
|
||||
def _get_new_BN(set_consttime_flag=True):
|
||||
"""
|
||||
Returns a new and initialized OpenSSL BIGNUM.
|
||||
The set_consttime_flag is set to True by default. When this instance of a
|
||||
CurveBN object has BN_FLG_CONSTTIME set, OpenSSL will use constant time
|
||||
operations whenever this CurveBN is passed.
|
||||
"""
|
||||
new_bn = backend._lib.BN_new()
|
||||
backend.openssl_assert(new_bn != backend._ffi.NULL)
|
||||
new_bn = backend._ffi.gc(new_bn, backend._lib.BN_clear_free)
|
||||
|
||||
if set_consttime_flag:
|
||||
backend._lib.BN_set_flags(new_bn, backend._lib.BN_FLG_CONSTTIME)
|
||||
return new_bn
|
||||
|
||||
|
||||
@typing.no_type_check
|
||||
def _get_ec_group_by_curve_nid(curve_nid: int):
|
||||
"""
|
||||
Returns the group of a given curve via its OpenSSL nid. This must be freed
|
||||
after each use otherwise it leaks memory.
|
||||
"""
|
||||
group = backend._lib.EC_GROUP_new_by_curve_name(curve_nid)
|
||||
backend.openssl_assert(group != backend._ffi.NULL)
|
||||
|
||||
return group
|
||||
|
||||
|
||||
@typing.no_type_check
|
||||
def _get_ec_order_by_group(ec_group):
|
||||
"""
|
||||
Returns the order of a given curve via its OpenSSL EC_GROUP.
|
||||
"""
|
||||
ec_order = _get_new_BN()
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
res = backend._lib.EC_GROUP_get_order(ec_group, ec_order, bn_ctx)
|
||||
backend.openssl_assert(res == 1)
|
||||
return ec_order
|
||||
|
||||
|
||||
@typing.no_type_check
|
||||
def _get_ec_generator_by_group(ec_group):
|
||||
"""
|
||||
Returns the generator point of a given curve via its OpenSSL EC_GROUP.
|
||||
"""
|
||||
generator = backend._lib.EC_GROUP_get0_generator(ec_group)
|
||||
backend.openssl_assert(generator != backend._ffi.NULL)
|
||||
generator = backend._ffi.gc(generator, backend._lib.EC_POINT_clear_free)
|
||||
|
||||
return generator
|
||||
|
||||
|
||||
@typing.no_type_check
|
||||
def _get_ec_group_degree(ec_group):
|
||||
"""
|
||||
Returns the number of bits needed to represent the order of the finite
|
||||
field upon the curve is based.
|
||||
"""
|
||||
return backend._lib.EC_GROUP_get_degree(ec_group)
|
||||
|
||||
|
||||
@typing.no_type_check
|
||||
def _bn_is_on_curve(check_bn, curve: 'Curve'):
|
||||
"""
|
||||
Checks if a given OpenSSL BIGNUM is within the provided curve's order.
|
||||
Returns True if the provided BN is on the curve, that is in the range `[0, curve_order)`.
|
||||
"""
|
||||
zero = backend._int_to_bn(0)
|
||||
zero = backend._ffi.gc(zero, backend._lib.BN_clear_free)
|
||||
|
||||
check_sign = backend._lib.BN_cmp(check_bn, zero)
|
||||
range_check = backend._lib.BN_cmp(check_bn, curve.order)
|
||||
return (check_sign == 1 or check_sign == 0) and range_check == -1
|
||||
|
||||
|
||||
@typing.no_type_check
|
||||
def _int_to_bn(py_int: int, curve: 'Curve'=None, set_consttime_flag=True):
|
||||
"""
|
||||
Converts the given Python int to an OpenSSL BIGNUM. If a curve is
|
||||
provided, it will check if the Python integer is within the order of that
|
||||
curve. If it's not within the order, it will raise a ValueError.
|
||||
|
||||
If set_consttime_flag is set to True, OpenSSL will use constant time
|
||||
operations when using this CurveBN.
|
||||
"""
|
||||
conv_bn = backend._int_to_bn(py_int)
|
||||
conv_bn = backend._ffi.gc(conv_bn, backend._lib.BN_clear_free)
|
||||
|
||||
if curve:
|
||||
on_curve = _bn_is_on_curve(conv_bn, curve)
|
||||
if not on_curve:
|
||||
raise ValueError("The Python integer given is not on the provided curve.")
|
||||
|
||||
if set_consttime_flag:
|
||||
backend._lib.BN_set_flags(conv_bn, backend._lib.BN_FLG_CONSTTIME)
|
||||
return conv_bn
|
||||
|
||||
@typing.no_type_check
|
||||
def _bytes_to_bn(bytes_seq: bytes, set_consttime_flag=True):
|
||||
"""
|
||||
Converts the given byte sequence to an OpenSSL BIGNUM.
|
||||
If set_consttime_flag is set to True, OpenSSL will use constant time
|
||||
operations when using this BIGNUM.
|
||||
"""
|
||||
bn = _get_new_BN(set_consttime_flag)
|
||||
backend._lib.BN_bin2bn(bytes_seq, len(bytes_seq), bn)
|
||||
backend.openssl_assert(bn != backend._ffi.NULL)
|
||||
return bn
|
||||
|
||||
@typing.no_type_check
|
||||
def _bn_to_bytes(bignum, length : int = None):
|
||||
"""
|
||||
Converts the given OpenSSL BIGNUM into a Python bytes sequence.
|
||||
If length is given, the return bytes will have such length.
|
||||
If the BIGNUM doesn't fit, it raises a ValueError.
|
||||
"""
|
||||
|
||||
if bignum is None or bignum == backend._ffi.NULL:
|
||||
raise ValueError("Input BIGNUM must have a value")
|
||||
|
||||
bn_num_bytes = backend._lib.BN_num_bytes(bignum)
|
||||
if length is None:
|
||||
length = bn_num_bytes
|
||||
elif bn_num_bytes > length:
|
||||
raise ValueError("Input BIGNUM doesn't fit in {} B".format(length))
|
||||
|
||||
bin_ptr = backend._ffi.new("unsigned char []", length)
|
||||
bin_len = backend._lib.BN_bn2bin(bignum, bin_ptr)
|
||||
return bytes.rjust(backend._ffi.buffer(bin_ptr, bin_len)[:], length, b'\0')
|
||||
|
||||
|
||||
@typing.no_type_check
|
||||
def _get_new_EC_POINT(curve: 'Curve'):
|
||||
"""
|
||||
Returns a new and initialized OpenSSL EC_POINT given the group of a curve.
|
||||
If __curve_nid is provided, it retrieves the group from the curve provided.
|
||||
"""
|
||||
new_point = backend._lib.EC_POINT_new(curve.ec_group)
|
||||
backend.openssl_assert(new_point != backend._ffi.NULL)
|
||||
new_point = backend._ffi.gc(new_point, backend._lib.EC_POINT_clear_free)
|
||||
|
||||
return new_point
|
||||
|
||||
|
||||
@typing.no_type_check
|
||||
def _get_EC_POINT_via_affine(affine_x, affine_y, curve: 'Curve'):
|
||||
"""
|
||||
Returns an EC_POINT given the group of a curve and the affine coordinates
|
||||
provided.
|
||||
"""
|
||||
new_point = _get_new_EC_POINT(curve)
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
res = backend._lib.EC_POINT_set_affine_coordinates_GFp(
|
||||
curve.ec_group, new_point, affine_x, affine_y, bn_ctx
|
||||
)
|
||||
backend.openssl_assert(res == 1)
|
||||
return new_point
|
||||
|
||||
|
||||
@typing.no_type_check
|
||||
def _get_affine_coords_via_EC_POINT(ec_point, curve: 'Curve'):
|
||||
"""
|
||||
Returns the affine coordinates of a given point on the provided ec_group.
|
||||
"""
|
||||
affine_x = _get_new_BN()
|
||||
affine_y = _get_new_BN()
|
||||
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
res = backend._lib.EC_POINT_get_affine_coordinates_GFp(
|
||||
curve.ec_group, ec_point, affine_x, affine_y, bn_ctx
|
||||
)
|
||||
backend.openssl_assert(res == 1)
|
||||
return (affine_x, affine_y)
|
||||
|
||||
|
||||
@typing.no_type_check
|
||||
@contextmanager
|
||||
def _tmp_bn_mont_ctx(modulus):
|
||||
"""
|
||||
Initializes and returns a BN_MONT_CTX for Montgomery ops.
|
||||
Requires a modulus to place in the Montgomery structure.
|
||||
"""
|
||||
bn_mont_ctx = backend._lib.BN_MONT_CTX_new()
|
||||
backend.openssl_assert(bn_mont_ctx != backend._ffi.NULL)
|
||||
# Don't set the garbage collector. Only free it when the context is done
|
||||
# or else you'll get a null pointer error.
|
||||
|
||||
try:
|
||||
with backend._tmp_bn_ctx() as bn_ctx:
|
||||
res = backend._lib.BN_MONT_CTX_set(bn_mont_ctx, modulus, bn_ctx)
|
||||
backend.openssl_assert(res == 1)
|
||||
yield bn_mont_ctx
|
||||
finally:
|
||||
backend._lib.BN_MONT_CTX_free(bn_mont_ctx)
|
|
@ -0,0 +1,10 @@
|
|||
from .hashing import unsafe_hash_to_point
|
||||
|
||||
|
||||
class Parameters:
|
||||
|
||||
def __init__(self):
|
||||
self.u = unsafe_hash_to_point(b'PARAMETERS', b'POINT_U')
|
||||
|
||||
|
||||
PARAMETERS = Parameters()
|
Loading…
Reference in New Issue