Merge branch 'master' into add-pyproject-toml

pull/115/head
Robert de Vries 2026-05-06 00:51:03 +02:00 committed by GitHub
commit ddf707b9b8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 478 additions and 122 deletions

View File

@ -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)

View File

@ -1,3 +1,4 @@
-r prod.txt
tox
pytest
types-cffi

View File

@ -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 <wolfssl/wolfcrypt/dilithium.h>
"""
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:

View File

@ -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"]}
)

View File

@ -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):

View File

@ -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

View File

@ -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)

View File

@ -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"
]

View File

@ -0,0 +1,6 @@
import _cffi_backend
import wolfcrypt._ffi.lib as lib
ffi: _cffi_backend.FFI
__all__ = ["ffi", "lib"]

View File

@ -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: ...

View File

@ -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,

View File

@ -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):

View File

@ -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)

View File

@ -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")