diff --git a/ChangeLog.rst b/ChangeLog.rst index 465543b..7046f3a 100644 --- a/ChangeLog.rst +++ b/ChangeLog.rst @@ -1,7 +1,7 @@ -wolfCrypt-py Release next (TBD) -========================================== +wolfCrypt-py Release next (TBD, 2026) * Drop support for end-of-life Python versions (<= 3.9) +* Add extra nonce parameter to Random generator wolfCrypt-py Release 5.8.4 (Jan 7, 2026) diff --git a/requirements/test.txt b/requirements/test.txt index 8c01ca8..53c4efe 100644 --- a/requirements/test.txt +++ b/requirements/test.txt @@ -1,3 +1,4 @@ -r prod.txt tox pytest +types-cffi diff --git a/scripts/build_ffi.py b/scripts/build_ffi.py index 01b3361..7e588a6 100644 --- a/scripts/build_ffi.py +++ b/scripts/build_ffi.py @@ -101,10 +101,10 @@ def wolfssl_lib_dir(local_wolfssl=None, fips=False): return lib_dir def call(cmd): - print("Calling: '{}' from working directory {}".format(cmd, os.getcwd())) + print(f"Calling: '{cmd}' from working directory {os.getcwd()}") old_env = os.environ["PATH"] - os.environ["PATH"] = "{}:{}".format(WOLFSSL_SRC_PATH, old_env) + os.environ["PATH"] = f"{WOLFSSL_SRC_PATH}:{old_env}" subprocess.check_call(cmd, shell=True, env=os.environ) os.environ["PATH"] = old_env @@ -133,7 +133,7 @@ def checkout_version(version): current = subprocess.check_output( ["git", "describe", "--all", "--exact-match"] ).strip().decode().split('/')[-1] - except: + except subprocess.CalledProcessError: pass if current != version: @@ -142,9 +142,9 @@ def checkout_version(version): ).strip().decode().split("\n") if version != "master" and version not in tags: - call("git fetch --depth=1 origin tag {}".format(version)) + call(f"git fetch --depth=1 origin tag {version}") - call("git checkout --force {}".format(version)) + call(f"git checkout --force {version}") return True # rebuild needed @@ -171,7 +171,7 @@ def make_flags(prefix, fips): """ if sys.platform == "win32": flags = [] - flags.append("-DCMAKE_INSTALL_PREFIX={}".format(prefix)) + flags.append(f"-DCMAKE_INSTALL_PREFIX={prefix}") flags.append("-DWOLFSSL_CRYPT_TESTS=no") flags.append("-DWOLFSSL_EXAMPLES=no") flags.append("-DBUILD_SHARED_LIBS=no") @@ -188,7 +188,7 @@ def make_flags(prefix, fips): flags.append("CFLAGS=-fPIC") # install location - flags.append("--prefix={}".format(prefix)) + flags.append(f"--prefix={prefix}") # crypt only, lib only flags.append("--enable-cryptonly") @@ -261,7 +261,7 @@ def make(configure_flags, fips=False): raise Exception("Cannot build wolfSSL FIPS from git repo.") with chdir(build_path): - call("cmake {} ..".format(configure_flags)) + call(f"cmake {configure_flags} ..") call("cmake --build . --config Release") call("cmake --install . --config Release") else: @@ -274,7 +274,7 @@ def make(configure_flags, fips=False): call("libtoolize") call("./autogen.sh") - call("./configure {}".format(configure_flags)) + call(f"./configure {configure_flags}") call("make") call("make install") @@ -458,45 +458,45 @@ def build_ffi(local_wolfssl, features): #include """ - init_source_string = """ + init_source_string = f""" #ifdef __cplusplus - extern "C" { + extern "C" {{ #endif - """ + includes_string + """ + {includes_string} #ifdef __cplusplus - } + }} #endif - int ERROR_STRINGS_ENABLED = """ + str(features["ERROR_STRINGS"]) + """; - int MPAPI_ENABLED = """ + str(features["MPAPI"]) + """; - int SHA_ENABLED = """ + str(features["SHA"]) + """; - int SHA256_ENABLED = """ + str(features["SHA256"]) + """; - int SHA384_ENABLED = """ + str(features["SHA384"]) + """; - int SHA512_ENABLED = """ + str(features["SHA512"]) + """; - int SHA3_ENABLED = """ + str(features["SHA3"]) + """; - int DES3_ENABLED = """ + str(features["DES3"]) + """; - int AES_ENABLED = """ + str(features["AES"]) + """; - int AES_SIV_ENABLED = """ + str(features["AES_SIV"]) + """; - int CHACHA_ENABLED = """ + str(features["CHACHA"]) + """; - int HMAC_ENABLED = """ + str(features["HMAC"]) + """; - int RSA_ENABLED = """ + str(features["RSA"]) + """; - int RSA_BLINDING_ENABLED = """ + str(features["RSA_BLINDING"]) + """; - int ECC_TIMING_RESISTANCE_ENABLED = """ + str(features["ECC_TIMING_RESISTANCE"]) + """; - int ECC_ENABLED = """ + str(features["ECC"]) + """; - int ED25519_ENABLED = """ + str(features["ED25519"]) + """; - int ED448_ENABLED = """ + str(features["ED448"]) + """; - int KEYGEN_ENABLED = """ + str(features["KEYGEN"]) + """; - int PWDBASED_ENABLED = """ + str(features["PWDBASED"]) + """; - int FIPS_ENABLED = """ + str(features["FIPS"]) + """; - int FIPS_VERSION = """ + str(features["FIPS_VERSION"]) + """; - int ASN_ENABLED = """ + str(features["ASN"]) + """; - int WC_RNG_SEED_CB_ENABLED = """ + str(features["WC_RNG_SEED_CB"]) + """; - int AESGCM_STREAM_ENABLED = """ + str(features["AESGCM_STREAM"]) + """; - int RSA_PSS_ENABLED = """ + str(features["RSA_PSS"]) + """; - int CHACHA20_POLY1305_ENABLED = """ + str(features["CHACHA20_POLY1305"]) + """; - int ML_KEM_ENABLED = """ + str(features["ML_KEM"]) + """; - int ML_DSA_ENABLED = """ + str(features["ML_DSA"]) + """; - int HKDF_ENABLED = """ + str(features["HKDF"]) + """; + int ERROR_STRINGS_ENABLED = {features["ERROR_STRINGS"]}; + int MPAPI_ENABLED = {features["MPAPI"]}; + int SHA_ENABLED = {features["SHA"]}; + int SHA256_ENABLED = {features["SHA256"]}; + int SHA384_ENABLED = {features["SHA384"]}; + int SHA512_ENABLED = {features["SHA512"]}; + int SHA3_ENABLED = {features["SHA3"]}; + int DES3_ENABLED = {features["DES3"]}; + int AES_ENABLED = {features["AES"]}; + int AES_SIV_ENABLED = {features["AES_SIV"]}; + int CHACHA_ENABLED = {features["CHACHA"]}; + int HMAC_ENABLED = {features["HMAC"]}; + int RSA_ENABLED = {features["RSA"]}; + int RSA_BLINDING_ENABLED = {features["RSA_BLINDING"]}; + int ECC_TIMING_RESISTANCE_ENABLED = {features["ECC_TIMING_RESISTANCE"]}; + int ECC_ENABLED = {features["ECC"]}; + int ED25519_ENABLED = {features["ED25519"]}; + int ED448_ENABLED = {features["ED448"]}; + int KEYGEN_ENABLED = {features["KEYGEN"]}; + int PWDBASED_ENABLED = {features["PWDBASED"]}; + int FIPS_ENABLED = {features["FIPS"]}; + int FIPS_VERSION = {features["FIPS_VERSION"]}; + int ASN_ENABLED = {features["ASN"]}; + int WC_RNG_SEED_CB_ENABLED = {features["WC_RNG_SEED_CB"]}; + int AESGCM_STREAM_ENABLED = {features["AESGCM_STREAM"]}; + int RSA_PSS_ENABLED = {features["RSA_PSS"]}; + int CHACHA20_POLY1305_ENABLED = {features["CHACHA20_POLY1305"]}; + int ML_KEM_ENABLED = {features["ML_KEM"]}; + int ML_DSA_ENABLED = {features["ML_DSA"]}; + int HKDF_ENABLED = {features["HKDF"]}; """ ffibuilder.set_source( "wolfcrypt._ffi", init_source_string, @@ -1306,6 +1306,7 @@ def build_ffi(local_wolfssl, features): if features["ML_DSA"]: cdef += """ + static const int DILITHIUM_SEED_SZ; static const int WC_ML_DSA_44; static const int WC_ML_DSA_65; static const int WC_ML_DSA_87; @@ -1375,9 +1376,9 @@ def main(ffibuilder): local_wolfssl = os.environ.get("USE_LOCAL_WOLFSSL") if local_wolfssl: - print("Using local wolfSSL at {}.".format(local_wolfssl)) + print(f"Using local wolfSSL at {local_wolfssl}.") if not os.path.exists(local_wolfssl): - e = "Local wolfssl installation path {} doesn't exist.".format(local_wolfssl) + e = f"Local wolfssl installation path {local_wolfssl} doesn't exist." raise FileNotFoundError(e) if not local_wolfssl: diff --git a/setup.py b/setup.py index d6f3d64..290c19c 100755 --- a/setup.py +++ b/setup.py @@ -64,5 +64,5 @@ setup( install_requires=["cffi>=1.0.0"], cffi_modules=["./scripts/build_ffi.py:ffibuilder"], - package_data={"wolfcrypt": ["*.dll"]} + package_data={"wolfcrypt": ["*.dll", "**/*.pyi"]} ) diff --git a/tests/test_ciphers.py b/tests/test_ciphers.py index 4647109..c8b1e25 100644 --- a/tests/test_ciphers.py +++ b/tests/test_ciphers.py @@ -25,7 +25,9 @@ import random import pytest from wolfcrypt._ffi import lib as _lib from wolfcrypt.ciphers import MODE_CTR, MODE_ECB, MODE_CBC, WolfCryptError +from wolfcrypt.random import Random from wolfcrypt.utils import t2b, h2b +from wolfcrypt.random import Random import os certs_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "certs") @@ -325,10 +327,18 @@ if _lib.CHACHA_ENABLED: assert plaintext == dec if _lib.RSA_ENABLED: + @pytest.fixture + def rng(): + return Random() + @pytest.fixture def rsa_private(vectors): return RsaPrivate(vectors[RsaPrivate].key) + @pytest.fixture + def rsa_private_rng(vectors, rng): + return RsaPrivate(vectors[RsaPrivate].key, rng=rng) + @pytest.fixture def rsa_private_oaep(vectors): return RsaPrivate(vectors[RsaPrivate].key, hash_type=HASH_TYPE_SHA) @@ -345,6 +355,10 @@ if _lib.RSA_ENABLED: def rsa_public(vectors): return RsaPublic(vectors[RsaPublic].key) + @pytest.fixture + def rsa_public_rng(vectors, rng): + return RsaPublic(vectors[RsaPublic].key, rng=rng) + @pytest.fixture def rsa_public_oaep(vectors): return RsaPublic(vectors[RsaPublic].key, hash_type=HASH_TYPE_SHA) @@ -365,6 +379,17 @@ if _lib.RSA_ENABLED: pem = f.read() return RsaPublic.from_pem(pem) + @pytest.fixture + def rsa_private_pem_rng(vectors, rng): + with open(vectors[RsaPrivate].pem, "rb") as f: + pem = f.read() + return RsaPrivate.from_pem(pem, rng=rng) + + @pytest.fixture + def rsa_public_pem_rng(vectors, rng): + with open(vectors[RsaPublic].pem, "rb") as f: + pem = f.read() + return RsaPublic.from_pem(pem, rng=rng) def test_new_rsa_raises(vectors): with pytest.raises(WolfCryptError): @@ -394,6 +419,22 @@ if _lib.RSA_ENABLED: assert 1024 / 8 == len(ciphertext) == rsa_private.output_size assert plaintext == rsa_private.decrypt(ciphertext) + def test_rsa_encrypt_decrypt_rng(rsa_private_rng, rsa_public_rng): + plaintext = t2b("Everyone gets Friday off.") + + # normal usage, encrypt with public, decrypt with private + ciphertext = rsa_public_rng.encrypt(plaintext) + + assert 1024 / 8 == len(ciphertext) == rsa_public_rng.output_size + assert plaintext == rsa_private_rng.decrypt(ciphertext) + + # private object holds both private and public info, so it can also encrypt + # using the known public key. + ciphertext = rsa_private_rng.encrypt(plaintext) + + assert 1024 / 8 == len(ciphertext) == rsa_private_rng.output_size + assert plaintext == rsa_private_rng.decrypt(ciphertext) + def test_rsa_encrypt_decrypt_pad_oaep(rsa_private_oaep, rsa_public_oaep): plaintext = t2b("Everyone gets Friday off.") @@ -477,6 +518,22 @@ if _lib.RSA_ENABLED: assert 256 == len(signature) == rsa_private_pem.output_size assert plaintext == rsa_private_pem.verify(signature) + def test_rsa_sign_verify_pem_rng(rsa_private_pem_rng, rsa_public_pem_rng): + plaintext = t2b("Everyone gets Friday off.") + + # normal usage, sign with private, verify with public + signature = rsa_private_pem_rng.sign(plaintext) + + assert 256 == len(signature) == rsa_private_pem_rng.output_size + assert plaintext == rsa_public_pem_rng.verify(signature) + + # private object holds both private and public info, so it can also verify + # using the known public key. + signature = rsa_private_pem_rng.sign(plaintext) + + assert 256 == len(signature) == rsa_private_pem_rng.output_size + assert plaintext == rsa_private_pem_rng.verify(signature) + def test_rsa_pkcs8_sign_verify(rsa_private_pkcs8, rsa_public): plaintext = t2b("Everyone gets Friday off.") @@ -613,11 +670,11 @@ if _lib.ECC_ENABLED: def test_ecc_make_shared_secret(): - a = EccPrivate.make_key(32) + a = EccPrivate.make_key(32, rng=Random()) a_pub = EccPublic() a_pub.import_x963(a.export_x963()) - b = EccPrivate.make_key(32) + b = EccPrivate.make_key(32, rng=Random()) b_pub = EccPublic() b_pub.import_x963(b.export_x963()) @@ -626,6 +683,13 @@ if _lib.ECC_ENABLED: == a.shared_secret(b_pub) \ == b.shared_secret(a_pub) + def test_ecc_make_key_no_rng(): + key = EccPrivate.make_key(32) + pub_key = EccPublic() + pub_key.import_x963(key.export_x963()) + + assert key.shared_secret(pub_key) + if _lib.ED25519_ENABLED: @pytest.fixture def ed25519_private(vectors): diff --git a/tests/test_delete_descriptor_binding.py b/tests/test_delete_descriptor_binding.py new file mode 100644 index 0000000..fb5bc94 --- /dev/null +++ b/tests/test_delete_descriptor_binding.py @@ -0,0 +1,185 @@ +# test_delete_descriptor_binding.py +# +# Copyright (C) 2026 wolfSSL Inc. +# +# This file is part of wolfSSL. (formerly known as CyaSSL) +# +# wolfSSL is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# wolfSSL is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA + +""" +Regression tests guarding against the Python descriptor-binding bug on +``_delete`` / ``_copy`` class attributes. + +Historically these were written as bare references to ``_lib`` functions:: + + class Random: + _delete = _lib.wc_FreeRng + + def __del__(self): + self._delete(self.native_object) + +If the underlying callable is ever a plain Python function (e.g. a mock, +wrapper, or future CFFI change), the descriptor protocol turns +``self._delete`` into a *bound method*, and ``self._delete(native)`` then +calls ``fn(self, native)`` - passing ``self`` as an extra C argument. + +The fix wraps the callable in ``staticmethod(...)`` at the class level so +that attribute lookup never binds ``self``. These tests assert the fix +stays in place and document the Python semantics it relies on. +""" + +# pylint: disable=redefined-outer-name + +import inspect + +import pytest + +from wolfcrypt._ffi import lib as _lib + + +def _static_attrs(): + """Yield (cls, attr_name) pairs that must be staticmethod-wrapped.""" + from wolfcrypt.random import Random + yield Random, "_delete" + + if _lib.SHA_ENABLED: + from wolfcrypt.hashes import Sha + yield Sha, "_delete" + yield Sha, "_copy" + if _lib.SHA256_ENABLED: + from wolfcrypt.hashes import Sha256 + yield Sha256, "_delete" + yield Sha256, "_copy" + if _lib.SHA384_ENABLED: + from wolfcrypt.hashes import Sha384 + yield Sha384, "_delete" + yield Sha384, "_copy" + if _lib.SHA512_ENABLED: + from wolfcrypt.hashes import Sha512 + yield Sha512, "_delete" + yield Sha512, "_copy" + if _lib.HMAC_ENABLED: + from wolfcrypt.hashes import _Hmac + yield _Hmac, "_delete" + + if _lib.AESGCM_STREAM_ENABLED: + from wolfcrypt.ciphers import AesGcmStream + yield AesGcmStream, "_delete" + if _lib.RSA_ENABLED: + from wolfcrypt.ciphers import _Rsa + yield _Rsa, "_delete" + if _lib.ECC_ENABLED: + from wolfcrypt.ciphers import _Ecc + yield _Ecc, "_delete" + if _lib.ED25519_ENABLED: + from wolfcrypt.ciphers import _Ed25519 + yield _Ed25519, "_delete" + if _lib.ED448_ENABLED: + from wolfcrypt.ciphers import _Ed448 + yield _Ed448, "_delete" + + +@pytest.mark.parametrize( + "cls,attr", + list(_static_attrs()), + ids=lambda v: v if isinstance(v, str) else v.__name__, +) +def test_lib_fn_class_attr_is_staticmethod(cls, attr): + """The class attribute must be a ``staticmethod`` so that attribute + access via an instance never triggers descriptor binding. + + ``inspect.getattr_static`` walks the MRO without invoking descriptors, + so it returns the raw object (the ``staticmethod`` wrapper itself). + """ + raw = inspect.getattr_static(cls, attr) + assert isinstance(raw, staticmethod), ( + "%s.%s must be wrapped in staticmethod(...) to prevent Python's " + "descriptor protocol from injecting `self` as an extra positional " + "argument when the underlying callable is a plain Python function " + "(e.g. a test mock). Got %r." % (cls.__name__, attr, type(raw)) + ) + + +def test_descriptor_binding_semantics_documentation(): + """Document the exact Python behavior the fix relies on. + + Without ``staticmethod``, a Python-function class attribute becomes a + bound method and leaks ``self`` into the call. ``staticmethod`` makes + the descriptor return the underlying callable unchanged. + """ + received = [] + + def recorder(*args, **kwargs): + received.append((args, kwargs)) + + class Buggy: + _delete = recorder + + def run(self): + self._delete("native") + + class Fixed: + _delete = staticmethod(recorder) + + def run(self): + self._delete("native") + + Buggy().run() + buggy_args, _ = received[-1] + assert len(buggy_args) == 2 and buggy_args[1] == "native", ( + "Sanity check failed: plain class-attribute Python function " + "should have been bound and passed self as the first arg." + ) + + Fixed().run() + fixed_args, _ = received[-1] + assert fixed_args == ("native",), ( + "staticmethod-wrapping should prevent self from being bound, " + "so the callable receives only the intended positional argument." + ) + + +def test_random_delete_receives_only_native_object(): + """End-to-end behavioral check on the real ``Random`` class. + + We substitute a plain Python recorder in place of the CFFI free + function (wrapped in staticmethod, mirroring how the class itself + stores it) and trigger the code path that calls ``self._delete``. + The recorder must see exactly one positional argument - the + ``native_object`` - and never ``self``. + """ + from wolfcrypt.random import Random + + received = [] + + def recorder(*args, **kwargs): + received.append((args, kwargs)) + + original = inspect.getattr_static(Random, "_delete") + try: + Random._delete = staticmethod(recorder) + r = Random() + native = r.native_object + r.__del__() + r.native_object = None # prevent real cleanup on the way out + assert received, "recorder was never called" + args, kwargs = received[-1] + assert kwargs == {} + assert args == (native,), ( + "Random.__del__ must call _delete with only native_object, " + "but got args=%r" % (args,) + ) + finally: + Random._delete = original diff --git a/tests/test_mldsa.py b/tests/test_mldsa.py index f04c80c..023b734 100644 --- a/tests/test_mldsa.py +++ b/tests/test_mldsa.py @@ -203,3 +203,14 @@ if _lib.ML_DSA_ENABLED: with pytest.raises(ValueError): _ = mldsa_priv.sign_with_seed(message, signature_seed[:-1], ctx=bytes(1000)) + def test_make_key_from_seed(mldsa_type): + seed = bytes(MlDsaPrivate.ML_DSA_KEYGEN_SEED_LENGTH) + assert MlDsaPrivate.make_key_from_seed(mldsa_type, seed) + + @pytest.mark.parametrize( + "seed_length", [MlDsaPrivate.ML_DSA_KEYGEN_SEED_LENGTH - 1, MlDsaPrivate.ML_DSA_KEYGEN_SEED_LENGTH + 1] + ) + def test_make_key_from_seed_bad_length(mldsa_type, seed_length): + seed = bytes(seed_length) + with pytest.raises(ValueError): + MlDsaPrivate.make_key_from_seed(mldsa_type, seed) diff --git a/wolfcrypt/__init__.py b/wolfcrypt/__init__.py index d7e8721..a3f893b 100644 --- a/wolfcrypt/__init__.py +++ b/wolfcrypt/__init__.py @@ -31,7 +31,7 @@ __license__ = "GPLv2 or Commercial License" __copyright__ = "Copyright (C) 2006-2022 wolfSSL Inc" __all__ = [ - "__title__", "__summary__", "__uri__", "__version__", + "__title__", "__summary__", "__uri__", "__version__", "__wolfssl_version__", "__author__", "__email__", "__license__", "__copyright__", "ciphers", "hashes", "random", "pwdbased" ] diff --git a/wolfcrypt/_ffi/__init__.pyi b/wolfcrypt/_ffi/__init__.pyi new file mode 100644 index 0000000..f3a1ba1 --- /dev/null +++ b/wolfcrypt/_ffi/__init__.pyi @@ -0,0 +1,6 @@ +import _cffi_backend +import wolfcrypt._ffi.lib as lib + +ffi: _cffi_backend.FFI + +__all__ = ["ffi", "lib"] diff --git a/wolfcrypt/_ffi/lib.pyi b/wolfcrypt/_ffi/lib.pyi new file mode 100644 index 0000000..047e5cf --- /dev/null +++ b/wolfcrypt/_ffi/lib.pyi @@ -0,0 +1,76 @@ + +from _cffi_backend import FFI +from typing import TypeAlias + +INVALID_DEVID: int + +AES_ENABLED: int +AES_SIV_ENABLED: int +AESGCM_STREAM_ENABLED: int +ASN_ENABLED: int +CHACHA_ENABLED: int +CHACHA_STREAM_ENABLED: int +CHACHA20_POLY1305_ENABLED: int +DES3_ENABLED: int +ECC_ENABLED: int +ED25519_ENABLED: int +ED448_ENABLED: int +FIPS_ENABLED: int +HMAC_ENABLED: int +KEYGEN_ENABLED: int +HKDF_ENABLED: int +ML_DSA_ENABLED: int +ML_KEM_ENABLED: int +MPAPI_ENABLED: int +PWDBASED_ENABLED: int +RSA_ENABLED: int +RSA_PSS_ENABLED: int +SHA_ENABLED: int +SHA3_ENABLED: int +SHA256_ENABLED: int +SHA384_ENABLED: int +SHA512_ENABLED: int +WC_RNG_SEED_CB_ENABLED: int + +FIPS_VERSION: int + +WC_MGF1NONE: int +WC_MGF1SHA1: int +WC_MGF1SHA224: int +WC_MGF1SHA256: int +WC_MGF1SHA384: int +WC_MGF1SHA512: int + +WC_HASH_TYPE_NONE: int +WC_HASH_TYPE_MD2: int +WC_HASH_TYPE_MD4: int +WC_HASH_TYPE_MD5: int +WC_HASH_TYPE_SHA: int +WC_HASH_TYPE_SHA224: int +WC_HASH_TYPE_SHA256: int +WC_HASH_TYPE_SHA384: int +WC_HASH_TYPE_SHA512: int +WC_HASH_TYPE_MD5_SHA: int +WC_HASH_TYPE_SHA3_224: int +WC_HASH_TYPE_SHA3_256: int +WC_HASH_TYPE_SHA3_384: int +WC_HASH_TYPE_SHA3_512: int +WC_HASH_TYPE_BLAKE2B: int +WC_HASH_TYPE_BLAKE2S: int + +WC_ML_KEM_512: int +WC_ML_KEM_768: int +WC_ML_KEM_1024: int + +WC_ML_DSA_44: int +WC_ML_DSA_65: int +WC_ML_DSA_87: int + +WC_KEYTYPE_ALL: int + +RNG: TypeAlias = FFI.CData + +def wc_InitRngNonce_ex(rng: RNG, nonce: bytes, nonce_size: int, heap: FFI.CData, device_id: int) -> int: ... +def wc_RNG_GenerateByte(rng: RNG, buffer: FFI.CData) -> int: ... +def wc_RNG_GenerateBlock(rng: RNG, buffer: FFI.CData, len: int) -> int: ... +def wc_FreeRng(rng: RNG) -> None: ... diff --git a/wolfcrypt/ciphers.py b/wolfcrypt/ciphers.py index 77ea987..308b443 100644 --- a/wolfcrypt/ciphers.py +++ b/wolfcrypt/ciphers.py @@ -364,7 +364,7 @@ if _lib.AES_SIV_ENABLED: C function has been called, in order to make sure that the memory is not freed by the FFI garbage collector before the data is read. """ - if (isinstance(associated_data, str) or isinstance(associated_data, bytes)): + if isinstance(associated_data, str) or isinstance(associated_data, bytes): # A single block is provided. # Make sure we have bytes. associated_data = t2b(associated_data) @@ -374,7 +374,7 @@ if _lib.AES_SIV_ENABLED: else: # It is assumed that a list is provided. num_blocks = len(associated_data) - if (num_blocks > 126): + if num_blocks > 126: raise WolfCryptError("AES-SIV does not support more than 126 blocks " "of associated data, got: %d" % num_blocks) # Make sure we have bytes. @@ -397,7 +397,7 @@ if _lib.AESGCM_STREAM_ENABLED: _key_sizes = [16, 24, 32] _native_type = "Aes *" # making sure _lib.wc_AesFree outlives Aes instances - _delete = _lib.wc_AesFree + _delete = staticmethod(_lib.wc_AesFree) def __init__(self, key, IV, tag_bytes=16): """ @@ -410,7 +410,7 @@ if _lib.AESGCM_STREAM_ENABLED: raise ValueError( "tag_bytes must be one of 4, 8, 12, 13, 14, 15, or 16") # Per-instance state: AAD, tag length, and current mode (enc/dec). - self._aad = bytes() + self._aad = b"" self._tag_bytes = tag_bytes self._mode = None if len(key) not in self._key_sizes: @@ -447,7 +447,7 @@ if _lib.AESGCM_STREAM_ENABLED: Add more data to the encryption stream """ data = t2b(data) - aad = bytes() + aad = b"" if self._mode is None: self._mode = _ENCRYPTION aad = self._aad @@ -463,7 +463,7 @@ if _lib.AESGCM_STREAM_ENABLED: """ Add more data to the decryption stream """ - aad = bytes() + aad = b"" data = t2b(data) if self._mode is None: self._mode = _DECRYPTION @@ -684,13 +684,16 @@ if _lib.RSA_ENABLED: _mgf = None _hash_type = None - def __init__(self): + def __init__(self, rng=None): + if rng is None: + rng = Random() + self.native_object = _ffi.new("RsaKey *") ret = _lib.wc_InitRsaKey(self.native_object, _ffi.NULL) if ret < 0: # pragma: no cover raise WolfCryptError("Invalid key error (%d)" % ret) - self._random = Random() + self._random = rng if _lib.RSA_BLINDING_ENABLED: ret = _lib.wc_RsaSetRNG(self.native_object, self._random.native_object) @@ -698,7 +701,7 @@ if _lib.RSA_ENABLED: raise WolfCryptError("Key initialization error (%d)" % ret) # making sure _lib.wc_FreeRsaKey outlives RsaKey instances - _delete = _lib.wc_FreeRsaKey + _delete = staticmethod(_lib.wc_FreeRsaKey) def __del__(self): if self.native_object: @@ -724,13 +727,13 @@ if _lib.RSA_ENABLED: class RsaPublic(_Rsa): - def __init__(self, key=None, hash_type=None): + def __init__(self, key=None, hash_type=None, rng=None): + super().__init__(rng) + if key is not None: key = t2b(key) self._hash_type = hash_type - _Rsa.__init__(self) - idx = _ffi.new("word32*") idx[0] = 0 @@ -747,9 +750,9 @@ if _lib.RSA_ENABLED: if _lib.ASN_ENABLED: @classmethod - def from_pem(cls, file, hash_type=None): + def from_pem(cls, file, hash_type=None, rng=None): der = pem_to_der(file, _lib.PUBLICKEY_TYPE) - return cls(key=der, hash_type=hash_type) + return cls(key=der, hash_type=hash_type, rng=rng) def encrypt(self, plaintext): """ @@ -826,8 +829,7 @@ if _lib.RSA_ENABLED: Returns a string containing the plaintext. """ if not self._hash_type: - raise WolfCryptError(("Hash type not set. Cannot verify a " - "PSS signature without a hash type.")) + raise WolfCryptError("Hash type not set. Cannot verify a PSS signature without a hash type.") hash_cls = hash_type_to_cls(self._hash_type) if not hash_cls: @@ -883,9 +885,9 @@ if _lib.RSA_ENABLED: return rsa - def __init__(self, key=None, hash_type=None): # pylint: disable=super-init-not-called + def __init__(self, key=None, hash_type=None, rng=None): # pylint: disable=super-init-not-called - _Rsa.__init__(self) # pylint: disable=non-parent-init-called + _Rsa.__init__(self, rng) # pylint: disable=non-parent-init-called self._hash_type = hash_type idx = _ffi.new("word32*") idx[0] = 0 @@ -913,9 +915,9 @@ if _lib.RSA_ENABLED: if _lib.ASN_ENABLED: @classmethod - def from_pem(cls, file, hash_type=None): + def from_pem(cls, file, hash_type=None, rng=None): der = pem_to_der(file, _lib.PRIVATEKEY_TYPE) - return cls(key=der, hash_type=hash_type) + return cls(key=der, hash_type=hash_type, rng=rng) if _lib.KEYGEN_ENABLED: def encode_key(self): @@ -1022,8 +1024,7 @@ if _lib.RSA_ENABLED: Returns a string containing the signature. """ if not self._hash_type: - raise WolfCryptError(("Hash type not set. Cannot verify a " - "PSS signature without a hash type.")) + raise WolfCryptError("Hash type not set. Cannot verify a PSS signature without a hash type.") hash_cls = hash_type_to_cls(self._hash_type) if not hash_cls: @@ -1057,7 +1058,7 @@ if _lib.ECC_ENABLED: raise WolfCryptError("Invalid key error (%d)" % ret) # making sure _lib.wc_ecc_free outlives ecc_key instances - _delete = _lib.wc_ecc_free + _delete = staticmethod(_lib.wc_ecc_free) def __del__(self): if self.native_object: @@ -1128,8 +1129,8 @@ if _lib.ECC_ENABLED: Returns (Qx, Qy) """ - Qx = _ffi.new("byte[%d]" % (self.size)) - Qy = _ffi.new("byte[%d]" % (self.size)) + Qx = _ffi.new("byte[%d]" % self.size) + Qy = _ffi.new("byte[%d]" % self.size) qx_size = _ffi.new("word32[1]") qy_size = _ffi.new("word32[1]") qx_size[0] = self.size @@ -1229,6 +1230,11 @@ if _lib.ECC_ENABLED: class EccPrivate(EccPublic): + + def __init__(self, key=None, rng=None): + super().__init__(key) + self._rng = rng + @classmethod def make_key(cls, size, rng=None): """ @@ -1236,24 +1242,19 @@ if _lib.ECC_ENABLED: """ if rng is None: rng = Random() - ecc = cls() + ecc = cls(rng=rng) - ret = _lib.wc_ecc_make_key(rng.native_object, size, + ret = _lib.wc_ecc_make_key(ecc._rng.native_object, size, ecc.native_object) if ret < 0: raise WolfCryptError("Key generation error (%d)" % ret) if _lib.ECC_TIMING_RESISTANCE_ENABLED and (not _lib.FIPS_ENABLED or _lib.FIPS_VERSION > 2): - ret = _lib.wc_ecc_set_rng(ecc.native_object, rng.native_object) + ret = _lib.wc_ecc_set_rng(ecc.native_object, ecc._rng.native_object) if ret < 0: raise WolfCryptError("Error setting ECC RNG (%d)" % ret) - # Retain the RNG so it outlives the ECC key. Even outside the - # timing-resistance path, wolfSSL internals may retain a pointer - # to the RNG; keeping the reference avoids any UAF risk. - ecc._rng = rng - return ecc def decode_key(self, key): @@ -1305,9 +1306,9 @@ if _lib.ECC_ENABLED: Returns (Qx, Qy, d) """ - Qx = _ffi.new("byte[%d]" % (self.size)) - Qy = _ffi.new("byte[%d]" % (self.size)) - d = _ffi.new("byte[%d]" % (self.size)) + Qx = _ffi.new("byte[%d]" % self.size) + Qy = _ffi.new("byte[%d]" % self.size) + d = _ffi.new("byte[%d]" % self.size) qx_size = _ffi.new("word32[1]") qy_size = _ffi.new("word32[1]") d_size = _ffi.new("word32[1]") @@ -1423,7 +1424,7 @@ if _lib.ED25519_ENABLED: raise WolfCryptError("Invalid key error (%d)" % ret) # making sure _lib.wc_ed25519_free outlives ed25519_key instances - _delete = _lib.wc_ed25519_free + _delete = staticmethod(_lib.wc_ed25519_free) def __del__(self): if self.native_object: @@ -1450,7 +1451,7 @@ if _lib.ED25519_ENABLED: Decodes an ED25519 public key """ key = t2b(key) - if (len(key) < _lib.wc_ed25519_pub_size(self.native_object)): + if len(key) < _lib.wc_ed25519_pub_size(self.native_object): raise WolfCryptError("Key decode error: key too short") idx = _ffi.new("word32*") @@ -1537,7 +1538,7 @@ if _lib.ED25519_ENABLED: """ key = t2b(key) - if (len(key) < _lib.wc_ed25519_priv_size(self.native_object)/2): + if len(key) < _lib.wc_ed25519_priv_size(self.native_object)/2: raise WolfCryptError("Key decode error: key too short") idx = _ffi.new("word32*") @@ -1623,7 +1624,7 @@ if _lib.ED448_ENABLED: raise WolfCryptError("Invalid key error (%d)" % ret) # making sure _lib.wc_ed448_free outlives ed448_key instances - _delete = _lib.wc_ed448_free + _delete = staticmethod(_lib.wc_ed448_free) def __del__(self): if self.native_object: @@ -1650,7 +1651,7 @@ if _lib.ED448_ENABLED: Decodes an ED448 public key """ key = t2b(key) - if (len(key) < _lib.wc_ed448_pub_size(self.native_object)): + if len(key) < _lib.wc_ed448_pub_size(self.native_object): raise WolfCryptError("Key decode error: key too short") idx = _ffi.new("word32*") @@ -1743,7 +1744,7 @@ if _lib.ED448_ENABLED: """ key = t2b(key) - if (len(key) < _lib.wc_ed448_priv_size(self.native_object)/2): + if len(key) < _lib.wc_ed448_priv_size(self.native_object)/2: raise WolfCryptError("Key decode error: key too short") idx = _ffi.new("word32*") @@ -2139,6 +2140,7 @@ if _lib.ML_DSA_ENABLED: class _MlDsaBase: INVALID_DEVID = _lib.INVALID_DEVID + ML_DSA_KEYGEN_SEED_LENGTH = _lib.DILITHIUM_SEED_SZ def __init__(self, mldsa_type): self._init_done = False @@ -2300,9 +2302,9 @@ if _lib.ML_DSA_ENABLED: raise TypeError( "seed must support the buffer protocol, such as `bytes` or `bytearray`" ) from exception - if len(seed_view) != ML_DSA_KEYGEN_SEED_LENGTH: + if len(seed_view) != cls.ML_DSA_KEYGEN_SEED_LENGTH: raise ValueError( - f"Seed for generating ML-DSA key must be {ML_DSA_KEYGEN_SEED_LENGTH} bytes" + f"Seed for generating ML-DSA key must be {cls.ML_DSA_KEYGEN_SEED_LENGTH} bytes" ) ret = _lib.wc_dilithium_make_key_from_seed(mldsa_priv.native_object, diff --git a/wolfcrypt/hashes.py b/wolfcrypt/hashes.py index 92adac3..d4a3d19 100644 --- a/wolfcrypt/hashes.py +++ b/wolfcrypt/hashes.py @@ -157,8 +157,8 @@ if _lib.SHA_ENABLED: digest_size = 20 _native_type = "wc_Sha *" _native_size = _ffi.sizeof("wc_Sha") - _delete = _lib.wc_ShaFree - _copy = _lib.wc_ShaCopy + _delete = staticmethod(_lib.wc_ShaFree) + _copy = staticmethod(_lib.wc_ShaCopy) def __del__(self): if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False): @@ -185,8 +185,8 @@ if _lib.SHA256_ENABLED: digest_size = 32 _native_type = "wc_Sha256 *" _native_size = _ffi.sizeof("wc_Sha256") - _delete = _lib.wc_Sha256Free - _copy = _lib.wc_Sha256Copy + _delete = staticmethod(_lib.wc_Sha256Free) + _copy = staticmethod(_lib.wc_Sha256Copy) def __del__(self): if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False): @@ -213,8 +213,8 @@ if _lib.SHA384_ENABLED: digest_size = 48 _native_type = "wc_Sha384 *" _native_size = _ffi.sizeof("wc_Sha384") - _delete = _lib.wc_Sha384Free - _copy = _lib.wc_Sha384Copy + _delete = staticmethod(_lib.wc_Sha384Free) + _copy = staticmethod(_lib.wc_Sha384Copy) def __del__(self): if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False): @@ -241,8 +241,8 @@ if _lib.SHA512_ENABLED: digest_size = 64 _native_type = "wc_Sha512 *" _native_size = _ffi.sizeof("wc_Sha512") - _delete = _lib.wc_Sha512Free - _copy = _lib.wc_Sha512Copy + _delete = staticmethod(_lib.wc_Sha512Free) + _copy = staticmethod(_lib.wc_Sha512Copy) def __del__(self): if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False): @@ -403,7 +403,7 @@ if _lib.HMAC_ENABLED: digest_size = None _native_type = "Hmac *" _native_size = _ffi.sizeof("Hmac") - _delete = _lib.wc_HmacFree + _delete = staticmethod(_lib.wc_HmacFree) def __del__(self): if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False): diff --git a/wolfcrypt/random.py b/wolfcrypt/random.py index 17d3ae7..a12fb7c 100644 --- a/wolfcrypt/random.py +++ b/wolfcrypt/random.py @@ -20,6 +20,8 @@ # pylint: disable=no-member,no-name-in-module +from __future__ import annotations + from wolfcrypt._ffi import ffi as _ffi from wolfcrypt._ffi import lib as _lib @@ -31,22 +33,18 @@ class Random: A Cryptographically Secure Pseudo Random Number Generator - CSPRNG """ - def __init__(self, nonce=_ffi.NULL, device_id=_lib.INVALID_DEVID): - self.native_object = _ffi.new("WC_RNG *") + def __init__(self, nonce: __builtins__.bytes = b"", device_id: int = _lib.INVALID_DEVID) -> None: + self.native_object: _lib.RNG | None = _ffi.new("WC_RNG *") - if nonce == _ffi.NULL: - nonce_size = 0 - else: - nonce_size = len(nonce) - ret = _lib.wc_InitRngNonce_ex(self.native_object, nonce, nonce_size, _ffi.NULL, device_id) + ret = _lib.wc_InitRngNonce_ex(self.native_object, nonce, len(nonce), _ffi.NULL, device_id) if ret < 0: # pragma: no cover self.native_object = None raise WolfCryptError("RNG init error (%d)" % ret) # making sure _lib.wc_FreeRng outlives WC_RNG instances - _delete = _lib.wc_FreeRng + _delete = staticmethod(_lib.wc_FreeRng) - def __del__(self): + def __del__(self) -> None: if self.native_object: try: self._delete(self.native_object) @@ -54,24 +52,26 @@ class Random: # Can occur during interpreter shutdown pass - def byte(self): + def byte(self) -> __builtins__.bytes: """ Generate and return a random byte. """ - result = _ffi.new('byte[1]') + result = _ffi.new("byte[1]") + assert self.native_object is not None ret = _lib.wc_RNG_GenerateByte(self.native_object, result) if ret < 0: # pragma: no cover raise WolfCryptError("RNG generate byte error (%d)" % ret) return _ffi.buffer(result, 1)[:] - def bytes(self, length): + def bytes(self, length: int) -> __builtins__.bytes: """ Generate and return a random sequence of length bytes. """ - result = _ffi.new('byte[%d]' % length) + result = _ffi.new("byte[%d]" % length) + assert self.native_object is not None ret = _lib.wc_RNG_GenerateBlock(self.native_object, result, length) if ret < 0: # pragma: no cover raise WolfCryptError("RNG generate block error (%d)" % ret) diff --git a/wolfcrypt/utils.py b/wolfcrypt/utils.py index b5274c8..5de6363 100644 --- a/wolfcrypt/utils.py +++ b/wolfcrypt/utils.py @@ -20,16 +20,26 @@ # pylint: disable=unused-import +from __future__ import annotations + from binascii import hexlify as b2h, unhexlify as h2b # noqa: F401 -def t2b(string): +def t2b(string: bytes | bytearray | memoryview | str) -> bytes: """ - Converts text to binary. + Converts text to bytes. - Passes through bytes, bytearray, and memoryview unchanged. + Passes through bytes unchanged. + Objects of type bytearray or memoryview are converted to bytes. Encodes str to UTF-8 bytes. + + :param string: text to convert to bytes. + :raises TypeError: if string is not one of the supported types. """ - if isinstance(string, (bytes, bytearray, memoryview)): + if isinstance(string, bytes): return string - return str(string).encode("utf-8") + if isinstance(string, (bytearray, memoryview)): + return bytes(string) + if isinstance(string, str): + return str(string).encode("utf-8") + raise TypeError(f"String parameter of wrong type {type(string).__name__}, expected bytes, bytearray, memoryview or str")