WolfSSL
/
wolfcrypt-py
mirror of https://github.com/wolfSSL/wolfcrypt-py.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Merge pull request #91 from JeremiahM37/fenrir-fixes-2 

Fenrir fixes
pull/106/head
David Garske 2026-04-16 15:40:07 -07:00 committed by GitHub
parent 3a675bb91a 664e887f09
commit 8d00f5f19b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 456 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
scripts/build_ffi.py
View File

@ -304,13 +304,14 @@ def generate_libwolfssl(fips):
def get_features(local_wolfssl, features):
fips = False
fips_file = None
if sys.platform == "win32":
if local_wolfssl and sys.platform == "win32":
# On Windows, we assume the local_wolfssl path is to a wolfSSL source
# directory where the library has been built.
fips_file = os.path.join(local_wolfssl, "wolfssl", "wolfcrypt",
"fips.h")
else:
elif local_wolfssl:
# On non-Windows platforms, first assume local_wolfssl is an
# installation directory with an include subdirectory.
fips_file = os.path.join(local_wolfssl, "include", "wolfssl",
@ -320,7 +321,7 @@ def get_features(local_wolfssl, features):
fips_file = os.path.join(local_wolfssl, "wolfssl", "wolfcrypt",
"fips.h")
if os.path.exists(fips_file):
if fips_file and os.path.exists(fips_file):
with open(fips_file, "r") as f:
contents = f.read()
if not contents.isspace():
@ -495,6 +496,7 @@ def build_ffi(local_wolfssl, features):
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 = """ + str(features["ERROR_STRINGS"]) + """;
"""
ffibuilder.set_source( "wolfcrypt._ffi", init_source_string,
@ -534,6 +536,7 @@ def build_ffi(local_wolfssl, features):
extern int ML_KEM_ENABLED;
extern int ML_DSA_ENABLED;
extern int HKDF_ENABLED;
extern int ERROR_STRINGS_ENABLED;
typedef unsigned char byte;
typedef unsigned int word32;
@ -559,6 +562,7 @@ def build_ffi(local_wolfssl, features):
typedef struct { ...; } mp_int;
int mp_init (mp_int * a);
void mp_clear (mp_int * a);
int mp_to_unsigned_bin (mp_int * a, unsigned char *b);
int mp_to_unsigned_bin_len (mp_int * a, unsigned char *b, int c);
int mp_read_unsigned_bin (mp_int * a, const unsigned char *b, int c);
@ -570,6 +574,8 @@ def build_ffi(local_wolfssl, features):
int wc_InitSha(wc_Sha*);
int wc_ShaUpdate(wc_Sha*, const byte*, word32);
int wc_ShaFinal(wc_Sha*, byte*);
void wc_ShaFree(wc_Sha*);
int wc_ShaCopy(wc_Sha*, wc_Sha*);
"""
if features["SHA256"]:
@ -578,6 +584,8 @@ def build_ffi(local_wolfssl, features):
int wc_InitSha256(wc_Sha256*);
int wc_Sha256Update(wc_Sha256*, const byte*, word32);
int wc_Sha256Final(wc_Sha256*, byte*);
void wc_Sha256Free(wc_Sha256*);
int wc_Sha256Copy(wc_Sha256*, wc_Sha256*);
"""
if features["SHA384"]:
@ -586,6 +594,8 @@ def build_ffi(local_wolfssl, features):
int wc_InitSha384(wc_Sha384*);
int wc_Sha384Update(wc_Sha384*, const byte*, word32);
int wc_Sha384Final(wc_Sha384*, byte*);
void wc_Sha384Free(wc_Sha384*);
int wc_Sha384Copy(wc_Sha384*, wc_Sha384*);
"""
if features["SHA512"]:
@ -595,6 +605,8 @@ def build_ffi(local_wolfssl, features):
int wc_InitSha512(wc_Sha512*);
int wc_Sha512Update(wc_Sha512*, const byte*, word32);
int wc_Sha512Final(wc_Sha512*, byte*);
void wc_Sha512Free(wc_Sha512*);
int wc_Sha512Copy(wc_Sha512*, wc_Sha512*);
"""
if features["SHA3"]:
cdef += """
@ -611,6 +623,14 @@ def build_ffi(local_wolfssl, features):
int wc_Sha3_256_Final(wc_Sha3*, byte*);
int wc_Sha3_384_Final(wc_Sha3*, byte*);
int wc_Sha3_512_Final(wc_Sha3*, byte*);
void wc_Sha3_224_Free(wc_Sha3*);
void wc_Sha3_256_Free(wc_Sha3*);
void wc_Sha3_384_Free(wc_Sha3*);
void wc_Sha3_512_Free(wc_Sha3*);
int wc_Sha3_224_Copy(wc_Sha3*, wc_Sha3*);
int wc_Sha3_256_Copy(wc_Sha3*, wc_Sha3*);
int wc_Sha3_384_Copy(wc_Sha3*, wc_Sha3*);
int wc_Sha3_512_Copy(wc_Sha3*, wc_Sha3*);
"""
if features["DES3"]:
@ -650,6 +670,7 @@ def build_ffi(local_wolfssl, features):
word32 sz, const byte* authIn, word32 authInSz);
int wc_AesGcmDecryptFinal(Aes* aes, const byte* authTag,
word32 authTagSz);
void wc_AesFree(Aes* aes);
"""
if features["AES"] and features["AES_SIV"]:
@ -706,6 +727,7 @@ def build_ffi(local_wolfssl, features):
int wc_HmacSetKey(Hmac*, int, const byte*, word32);
int wc_HmacUpdate(Hmac*, const byte*, word32);
int wc_HmacFinal(Hmac*, byte*);
void wc_HmacFree(Hmac*);
"""
if features["RSA"]:
@ -962,6 +984,7 @@ def build_ffi(local_wolfssl, features):
int wc_PemToDer(const unsigned char* buff, long longSz, int type,
DerBuffer** pDer, void* heap, EncryptedInfo* info,
int* keyFormat);
void wc_FreeDer(DerBuffer** pDer);
int wc_DerToPemEx(const byte* der, word32 derSz, byte* output, word32 outSz,
byte *cipher_info, int type);
"""
@ -989,6 +1012,11 @@ def build_ffi(local_wolfssl, features):
int wolfCrypt_GetPrivateKeyReadEnable_fips(enum wc_KeyType);
"""
if features["ERROR_STRINGS"]:
cdef += """
const char* wc_GetErrorString(int error);
"""
if features["ML_KEM"] or features["ML_DSA"]:
cdef += """
static const int INVALID_DEVID;
@ -1093,17 +1121,17 @@ def main(ffibuilder):
e = "Local wolfssl installation path {} doesn't exist.".format(local_wolfssl)
raise FileNotFoundError(e)
get_features(local_wolfssl, features)
if features["RSA_BLINDING"] and features["FIPS"]:
# These settings can't coexist. See settings.h.
features["RSA_BLINDING"] = 0
if not local_wolfssl:
print("Building wolfSSL...")
if not get_libwolfssl():
generate_libwolfssl(features["FIPS"])
get_features(local_wolfssl, features)
if features["RSA_BLINDING"] and features["FIPS"]:
# These settings can't coexist. See settings.h.
features["RSA_BLINDING"] = 0
build_ffi(local_wolfssl, features)

32
tests/test_aesgcmstream.py
View File

@ -122,3 +122,35 @@ if _lib.AESGCM_STREAM_ENABLED:
gcmdec.decrypt(buf)
with pytest.raises(WolfCryptError):
gcmdec.final(authTag)
def test_invalid_tag_bytes():
key = "fedcba9876543210"
iv = "0123456789abcdef"
# Out of range
with pytest.raises(ValueError, match="tag_bytes must be one of"):
AesGcmStream(key, iv, tag_bytes=0)
with pytest.raises(ValueError, match="tag_bytes must be one of"):
AesGcmStream(key, iv, tag_bytes=3)
with pytest.raises(ValueError, match="tag_bytes must be one of"):
AesGcmStream(key, iv, tag_bytes=17)
# Non-NIST sizes within 4-16 range
for bad in (5, 6, 7, 9, 10, 11):
with pytest.raises(ValueError, match="tag_bytes must be one of"):
AesGcmStream(key, iv, tag_bytes=bad)
# Valid NIST sizes: verify the resulting tag has the requested length.
for good in (4, 8, 12, 13, 14, 15, 16):
gcm = AesGcmStream(key, iv, tag_bytes=good)
gcm.encrypt("hello world")
tag = gcm.final()
assert len(tag) == good
def test_repeated_construction_destruction():
import gc
key = "fedcba9876543210"
iv = "0123456789abcdef"
for _ in range(1000):
gcm = AesGcmStream(key, iv)
gcm.encrypt("hello world")
gcm.final()
del gcm
gc.collect()

44
tests/test_ciphers.py
View File

@ -872,3 +872,47 @@ def test_aessiv_decrypt_kat_openssl():
TEST_VECTOR_CIPHERTEXT_OPENSSL
)
assert plaintext == TEST_VECTOR_PLAINTEXT_OPENSSL
if _lib.DES3_ENABLED:
def test_des3_rejects_mode_ctr():
key = b"\x01\x23\x45\x67\x89\xab\xcd\xef" * 3
iv = b"\xfe\xdc\xba\x98\x76\x54\x32\x10"
with pytest.raises(ValueError, match="Des3 only supports MODE_CBC"):
Des3.new(key, MODE_CTR, iv)
def test_des3_rejects_mode_ecb():
key = b"\x01\x23\x45\x67\x89\xab\xcd\xef" * 3
iv = b"\xfe\xdc\xba\x98\x76\x54\x32\x10"
with pytest.raises(ValueError, match="Des3 only supports MODE_CBC"):
Des3.new(key, MODE_ECB, iv)
if _lib.CHACHA_ENABLED:
def test_chacha_non_block_aligned():
key = b"\x00" * 32
chacha = ChaCha(key)
chacha.set_iv(b"\x00" * 12)
plaintext = b"This is 25 bytes of text!"
assert len(plaintext) == 25
ciphertext = chacha.encrypt(plaintext)
assert len(ciphertext) == 25
chacha2 = ChaCha(key)
chacha2.set_iv(b"\x00" * 12)
assert chacha2.decrypt(ciphertext) == plaintext
def test_chacha_invalid_key_length():
with pytest.raises(ValueError, match="key must be"):
ChaCha(b"\x00" * 20)
if _lib.RSA_ENABLED:
def test_encrypt_oaep_requires_hash_type(vectors):
rsa = RsaPublic(vectors[RsaPublic].key)
with pytest.raises(WolfCryptError, match="Hash type not set"):
rsa.encrypt_oaep(b"plaintext")
def test_decrypt_oaep_requires_hash_type(vectors):
rsa = RsaPrivate(vectors[RsaPrivate].key)
with pytest.raises(WolfCryptError, match="Hash type not set"):
rsa.decrypt_oaep(b"\x00" * rsa.output_size)

24
tests/test_hashes.py
View File

@ -184,3 +184,27 @@ def test_hash(hash_cls, vectors):
copy.update("wolfcrypt")
assert hash_obj.hexdigest() == copy.hexdigest() == digest
def test_hash_repeated_construction_destruction(hash_cls, vectors):
import gc
digest = vectors[hash_cls].digest
for _ in range(1000):
h = hash_new(hash_cls, "wolfcrypt")
assert h.hexdigest() == digest
del h
gc.collect()
def test_hash_copy_destroy_lifecycle(hash_cls, vectors):
import gc
digest = vectors[hash_cls].digest
for _ in range(100):
h = hash_new(hash_cls, "wolfcrypt")
c = h.copy()
# Destroy original first, then verify copy still produces correct digest.
del h
gc.collect()
assert c.hexdigest() == digest
del c
gc.collect()

18
wolfcrypt/asn.py
View File

@ -20,6 +20,8 @@
# pylint: disable=no-member,no-name-in-module
import hmac as _hmac
from wolfcrypt._ffi import ffi as _ffi
from wolfcrypt._ffi import lib as _lib
from wolfcrypt.exceptions import WolfCryptError
@ -42,7 +44,11 @@ if _lib.ASN_ENABLED:
err = "Error converting from PEM to DER. ({})".format(ret)
raise WolfCryptError(err)
return _ffi.buffer(der[0][0].buffer, der[0][0].length)[:]
try:
result = _ffi.buffer(der[0][0].buffer, der[0][0].length)[:]
finally:
_lib.wc_FreeDer(der)
return result
def der_to_pem(der, pem_type):
pem_length = _lib.wc_DerToPemEx(der, len(der), _ffi.NULL, 0, _ffi.NULL,
@ -61,13 +67,13 @@ if _lib.ASN_ENABLED:
return _ffi.buffer(pem, pem_length)[:]
def hash_oid_from_class(hash_cls):
if hash_cls == Sha:
if _lib.SHA_ENABLED and hash_cls == Sha:
return _lib.SHAh
elif hash_cls == Sha256:
elif _lib.SHA256_ENABLED and hash_cls == Sha256:
return _lib.SHA256h
elif hash_cls == Sha384:
elif _lib.SHA384_ENABLED and hash_cls == Sha384:
return _lib.SHA384h
elif hash_cls == Sha512:
elif _lib.SHA512_ENABLED and hash_cls == Sha512:
return _lib.SHA512h
else:
err = "Unknown hash class {}.".format(hash_cls.__name__)
@ -95,4 +101,4 @@ if _lib.ASN_ENABLED:
def check_signature(signature, data, hash_cls, pub_key):
computed_signature = make_signature(data, hash_cls)
decrypted_signature = pub_key.verify(signature)
return computed_signature == decrypted_signature
return _hmac.compare_digest(computed_signature, decrypted_signature)

246
wolfcrypt/ciphers.py
View File

@ -129,6 +129,10 @@ class _Cipher:
self.mode = mode
key = t2b(key)
if IV is not None:
IV = t2b(IV)
if self.key_size:
if self.key_size != len(key):
raise ValueError("key must be %d in length, not %d" %
@ -147,10 +151,10 @@ class _Cipher:
self._native_object = _ffi.new(self._native_type)
self._enc = None
self._dec = None
self._key = t2b(key)
self._key = key
if IV:
self._IV = t2b(IV)
self._IV = IV
else: # pragma: no cover
self._IV = _ffi.new("byte[%d]" % self.block_size)
@ -183,7 +187,7 @@ class _Cipher:
raise ValueError(
"empty string not allowed")
if len(string) % self.block_size and not self.mode == MODE_CTR and "ChaCha" not in self._native_type:
if len(string) % self.block_size and "ChaCha" not in self._native_type and self.mode != MODE_CTR:
raise ValueError(
"string must be a multiple of %d in length" % self.block_size)
@ -191,6 +195,7 @@ class _Cipher:
self._enc = _ffi.new(self._native_type)
ret = self._set_key(_ENCRYPTION)
if ret < 0: # pragma: no cover
self._enc = None
raise WolfCryptError("Invalid key error (%d)" % ret)
result = _ffi.new("byte[%d]" % len(string))
@ -215,7 +220,7 @@ class _Cipher:
if not string:
raise ValueError("empty string not allowed")
if len(string) % self.block_size and self.mode != MODE_CTR and "ChaCha" not in self._native_type:
if len(string) % self.block_size and "ChaCha" not in self._native_type and self.mode != MODE_CTR:
raise ValueError(
"string must be a multiple of %d in length" % self.block_size)
@ -223,6 +228,7 @@ class _Cipher:
self._dec = _ffi.new(self._native_type)
ret = self._set_key(_DECRYPTION)
if ret < 0: # pragma: no cover
self._dec = None
raise WolfCryptError("Invalid key error (%d)" % ret)
result = _ffi.new("byte[%d]" % len(string))
@ -390,9 +396,8 @@ if _lib.AESGCM_STREAM_ENABLED:
block_size = 16
_key_sizes = [16, 24, 32]
_native_type = "Aes *"
_aad = bytes()
_tag_bytes = 16
_mode = None
# making sure _lib.wc_AesFree outlives Aes instances
_delete = _lib.wc_AesFree
def __init__(self, key, IV, tag_bytes=16):
"""
@ -400,16 +405,32 @@ if _lib.AESGCM_STREAM_ENABLED:
"""
key = t2b(key)
IV = t2b(IV)
# NIST SP 800-38D valid GCM tag lengths: 16, 15, 14, 13, 12, 8, 4 bytes.
if tag_bytes not in (4, 8, 12, 13, 14, 15, 16):
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._tag_bytes = tag_bytes
self._mode = None
if len(key) not in self._key_sizes:
raise ValueError("key must be %s in length, not %d" %
(self._key_sizes, len(key)))
self._init_done = False
self._native_object = _ffi.new(self._native_type)
_lib.wc_AesInit(self._native_object, _ffi.NULL, -2)
ret = _lib.wc_AesInit(self._native_object, _ffi.NULL, -2)
if ret < 0:
raise WolfCryptError("AES init error (%d)" % ret)
self._init_done = True
ret = _lib.wc_AesGcmInit(self._native_object, key, len(key), IV, len(IV))
if ret < 0:
raise WolfCryptError("Init error (%d)" % ret)
def __del__(self):
if getattr(self, '_init_done', False):
self._delete(self._native_object)
self._init_done = False
def set_aad(self, data):
"""
Set the additional authentication data for the stream
@ -432,11 +453,11 @@ if _lib.AESGCM_STREAM_ENABLED:
aad = self._aad
elif self._mode == _DECRYPTION:
raise WolfCryptError("Class instance already in use for decryption")
self._buf = _ffi.new("byte[%d]" % (len(data)))
ret = _lib.wc_AesGcmEncryptUpdate(self._native_object, self._buf, data, len(data), aad, len(aad))
buf = _ffi.new("byte[%d]" % (len(data)))
ret = _lib.wc_AesGcmEncryptUpdate(self._native_object, buf, data, len(data), aad, len(aad))
if ret < 0:
raise WolfCryptError("Decryption error (%d)" % ret)
return bytes(self._buf)
raise WolfCryptError("Encryption error (%d)" % ret)
return bytes(buf)
def decrypt(self, data):
"""
@ -448,12 +469,12 @@ if _lib.AESGCM_STREAM_ENABLED:
self._mode = _DECRYPTION
aad = self._aad
elif self._mode == _ENCRYPTION:
raise WolfCryptError("Class instance already in use for decryption")
self._buf = _ffi.new("byte[%d]" % (len(data)))
ret = _lib.wc_AesGcmDecryptUpdate(self._native_object, self._buf, data, len(data), aad, len(aad))
raise WolfCryptError("Class instance already in use for encryption")
buf = _ffi.new("byte[%d]" % (len(data)))
ret = _lib.wc_AesGcmDecryptUpdate(self._native_object, buf, data, len(data), aad, len(aad))
if ret < 0:
raise WolfCryptError("Decryption error (%d)" % ret)
return bytes(self._buf)
return bytes(buf)
def final(self, authTag=None):
"""
@ -488,20 +509,23 @@ if _lib.CHACHA_ENABLED:
key_size = None # 16, 24, 32
_key_sizes = [16, 32]
_native_type = "ChaCha *"
_IV_nonce = []
_IV_nonce = b""
_IV_counter = 0
def __init__(self, key="", size=32):
def __init__(self, key="", size=32): # pylint: disable=unused-argument
# size is kept for backwards compatibility; key length is now
# derived from the actual key and validated against _key_sizes.
self._native_object = _ffi.new(self._native_type)
self._enc = None
self._dec = None
self._key = None
if len(key) > 0:
if size not in self._key_sizes:
raise ValueError("Invalid key size %d" % size)
self._key = t2b(key)
self.key_size = size
self._IV_nonce = []
if len(self._key) not in self._key_sizes:
raise ValueError("key must be %s in length, not %d" %
(self._key_sizes, len(self._key)))
self.key_size = len(self._key)
self._IV_nonce = b""
self._IV_counter = 0
def _set_key(self, direction):
@ -510,13 +534,13 @@ if _lib.CHACHA_ENABLED:
if self._enc:
ret = _lib.wc_Chacha_SetKey(self._enc, self._key, len(self._key))
if ret == 0:
_lib.wc_Chacha_SetIV(self._enc, self._IV_nonce, self._IV_counter)
ret = _lib.wc_Chacha_SetIV(self._enc, self._IV_nonce, self._IV_counter)
if ret != 0:
return ret
if self._dec:
ret = _lib.wc_Chacha_SetKey(self._dec, self._key, len(self._key))
if ret == 0:
_lib.wc_Chacha_SetIV(self._dec, self._IV_nonce, self._IV_counter)
ret = _lib.wc_Chacha_SetIV(self._dec, self._IV_nonce, self._IV_counter)
if ret != 0:
return ret
return 0
@ -537,7 +561,9 @@ if _lib.CHACHA_ENABLED:
raise ValueError("nonce must be %d bytes, got %d" %
(self._NONCE_SIZE, len(self._IV_nonce)))
self._IV_counter = counter
self._set_key(0)
ret = self._set_key(0)
if ret < 0:
raise WolfCryptError("ChaCha set_iv error (%d)" % ret)
if _lib.CHACHA20_POLY1305_ENABLED:
class ChaCha20Poly1305:
@ -627,6 +653,14 @@ if _lib.DES3_ENABLED:
key_size = 24
_native_type = "Des3 *"
def __init__(self, key, mode, IV=None):
# Intentionally stricter than _Cipher.__init__, which accepts both
# CBC and CTR. wolfCrypt has no 3DES-CTR implementation, so reject
# MODE_CTR here with a clearer error before delegating.
if mode != MODE_CBC:
raise ValueError("Des3 only supports MODE_CBC")
super().__init__(key, mode, IV)
def _set_key(self, direction):
if direction == _ENCRYPTION:
return _lib.wc_Des3_SetKey(self._enc, self._key,
@ -741,6 +775,8 @@ if _lib.RSA_ENABLED:
return _ffi.buffer(ciphertext)[:]
def encrypt_oaep(self, plaintext, label=""):
if not self._hash_type:
raise WolfCryptError("Hash type not set. Cannot use OAEP padding without a hash type.")
plaintext = t2b(plaintext)
label = t2b(label)
ciphertext = _ffi.new("byte[%d]" % self.output_size)
@ -824,10 +860,12 @@ if _lib.RSA_ENABLED:
class RsaPrivate(RsaPublic):
if _lib.KEYGEN_ENABLED:
@classmethod
def make_key(cls, size, rng=Random(), hash_type=None):
def make_key(cls, size, rng=None, hash_type=None):
"""
Generates a new key pair of desired length **size**.
"""
if rng is None:
rng = Random()
rsa = cls(hash_type=hash_type)
ret = _lib.wc_MakeRsaKey(rsa.native_object, size, 65537,
@ -840,6 +878,9 @@ if _lib.RSA_ENABLED:
if rsa.output_size <= 0: # pragma: no cover
raise WolfCryptError("Invalid key size error (%d)" % ret)
# Retain RNG reference defensively.
rsa._rng = rng
return rsa
def __init__(self, key=None, hash_type=None): # pylint: disable=super-init-not-called
@ -930,6 +971,8 @@ if _lib.RSA_ENABLED:
Returns a string containing the plaintext.
"""
if not self._hash_type:
raise WolfCryptError("Hash type not set. Cannot use OAEP padding without a hash type.")
ciphertext = t2b(ciphertext)
label = t2b(label)
plaintext = _ffi.new("byte[%d]" % self.output_size)
@ -1160,33 +1203,39 @@ if _lib.ECC_ENABLED:
raise WolfCryptError("wolfCrypt error (%d)" % ret)
ret = _lib.mp_init(mpS)
if ret != 0: # pragma: no cover
_lib.mp_clear(mpR)
raise WolfCryptError("wolfCrypt error (%d)" % ret)
ret = _lib.mp_read_unsigned_bin(mpR, R, len(R))
if ret != 0: # pragma: no cover
raise WolfCryptError("wolfCrypt error (%d)" % ret)
try:
ret = _lib.mp_read_unsigned_bin(mpR, R, len(R))
if ret != 0: # pragma: no cover
raise WolfCryptError("wolfCrypt error (%d)" % ret)
ret = _lib.mp_read_unsigned_bin(mpS, S, len(S))
if ret != 0: # pragma: no cover
raise WolfCryptError("wolfCrypt error (%d)" % ret)
ret = _lib.mp_read_unsigned_bin(mpS, S, len(S))
if ret != 0: # pragma: no cover
raise WolfCryptError("wolfCrypt error (%d)" % ret)
ret = _lib.wc_ecc_verify_hash_ex(mpR, mpS,
data, len(data),
status, self.native_object)
ret = _lib.wc_ecc_verify_hash_ex(mpR, mpS,
data, len(data),
status, self.native_object)
if ret < 0:
raise WolfCryptError("Verify error (%d)" % ret)
if ret < 0:
raise WolfCryptError("Verify error (%d)" % ret)
return status[0] == 1
return status[0] == 1
finally:
_lib.mp_clear(mpR)
_lib.mp_clear(mpS)
class EccPrivate(EccPublic):
@classmethod
def make_key(cls, size, rng=Random()):
def make_key(cls, size, rng=None):
"""
Generates a new key pair of desired length **size**.
"""
if rng is None:
rng = Random()
ecc = cls()
ret = _lib.wc_ecc_make_key(rng.native_object, size,
@ -1200,6 +1249,11 @@ if _lib.ECC_ENABLED:
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):
@ -1289,12 +1343,14 @@ if _lib.ECC_ENABLED:
return _ffi.buffer(shared_secret, secret_size[0])[:]
def sign(self, plaintext, rng=Random()):
def sign(self, plaintext, rng=None):
"""
Signs **plaintext**, using the private key data in the object.
Returns the signature.
"""
if rng is None:
rng = Random()
plaintext = t2b(plaintext)
signature = _ffi.new("byte[%d]" % self.max_signature_size)
@ -1312,12 +1368,14 @@ if _lib.ECC_ENABLED:
return _ffi.buffer(signature, signature_size[0])[:]
if _lib.MPAPI_ENABLED:
def sign_raw(self, plaintext, rng=Random()):
def sign_raw(self, plaintext, rng=None):
"""
Signs **plaintext**, using the private key data in the object.
Returns the signature in its two raw components r, s
"""
if rng is None:
rng = Random()
plaintext = t2b(plaintext)
R = _ffi.new("mp_int[1]")
S = _ffi.new("mp_int[1]")
@ -1330,25 +1388,30 @@ if _lib.ECC_ENABLED:
raise WolfCryptError("wolfCrypt error (%d)" % ret)
ret = _lib.mp_init(S)
if ret != 0: # pragma: no cover
_lib.mp_clear(R)
raise WolfCryptError("wolfCrypt error (%d)" % ret)
ret = _lib.wc_ecc_sign_hash_ex(plaintext, len(plaintext),
rng.native_object,
self.native_object,
R, S)
if ret != 0: # pragma: no cover
raise WolfCryptError("Signature error (%d)" % ret)
try:
ret = _lib.wc_ecc_sign_hash_ex(plaintext, len(plaintext),
rng.native_object,
self.native_object,
R, S)
if ret != 0: # pragma: no cover
raise WolfCryptError("Signature error (%d)" % ret)
ret = _lib.mp_to_unsigned_bin_len(R, R_bin, self.size)
if ret != 0: # pragma: no cover
raise WolfCryptError("wolfCrypt error (%d)" % ret)
ret = _lib.mp_to_unsigned_bin_len(R, R_bin, self.size)
if ret != 0: # pragma: no cover
raise WolfCryptError("wolfCrypt error (%d)" % ret)
ret = _lib.mp_to_unsigned_bin_len(S, S_bin, self.size)
if ret != 0: # pragma: no cover
raise WolfCryptError("wolfCrypt error (%d)" % ret)
ret = _lib.mp_to_unsigned_bin_len(S, S_bin, self.size)
if ret != 0: # pragma: no cover
raise WolfCryptError("wolfCrypt error (%d)" % ret)
return _ffi.buffer(R_bin, self.size)[:], _ffi.buffer(S_bin,
self.size)[:]
return _ffi.buffer(R_bin, self.size)[:], _ffi.buffer(S_bin,
self.size)[:]
finally:
_lib.mp_clear(R)
_lib.mp_clear(S)
if _lib.ED25519_ENABLED:
@ -1449,10 +1512,12 @@ if _lib.ED25519_ENABLED:
self.decode_key(key,pub)
@classmethod
def make_key(cls, size, rng=Random()):
def make_key(cls, size, rng=None):
"""
Generates a new key pair of desired length **size**.
"""
if rng is None:
rng = Random()
ed25519 = cls()
ret = _lib.wc_ed25519_make_key(rng.native_object, size,
@ -1460,6 +1525,10 @@ if _lib.ED25519_ENABLED:
if ret < 0:
raise WolfCryptError("Key generation error (%d)" % ret)
# Retain RNG reference defensively; wolfSSL may retain a pointer
# internally on some builds.
ed25519._rng = rng
return ed25519
def decode_key(self, key, pub = None):
@ -1490,6 +1559,8 @@ if _lib.ED25519_ENABLED:
raise WolfCryptError("Public key generate error (%d)" % ret)
ret = _lib.wc_ed25519_import_public(pubkey, self.size,
self.native_object)
if ret < 0:
raise WolfCryptError("Public key import error (%d)" % ret)
if self.size <= 0: # pragma: no cover
raise WolfCryptError("Key decode error (%d)" % self.size)
@ -1505,20 +1576,22 @@ if _lib.ED25519_ENABLED:
"""
key = _ffi.new("byte[%d]" % (self.size * 4))
pubkey = _ffi.new("byte[%d]" % (self.size * 4))
size = _ffi.new("word32[1]")
priv_size = _ffi.new("word32[1]")
pub_size = _ffi.new("word32[1]")
size[0] = _lib.wc_ed25519_priv_size(self.native_object)
priv_size[0] = _lib.wc_ed25519_priv_size(self.native_object)
pub_size[0] = _lib.wc_ed25519_pub_size(self.native_object)
ret = _lib.wc_ed25519_export_private_only(self.native_object,
key, size)
key, priv_size)
if ret != 0: # pragma: no cover
raise WolfCryptError("Private key encode error (%d)" % ret)
ret = _lib.wc_ed25519_export_public(self.native_object, pubkey,
size)
pub_size)
if ret != 0: # pragma: no cover
raise WolfCryptError("Public key encode error (%d)" % ret)
return _ffi.buffer(key, size[0])[:], _ffi.buffer(pubkey, size[0])[:]
return _ffi.buffer(key, priv_size[0])[:], _ffi.buffer(pubkey, pub_size[0])[:]
def sign(self, plaintext):
"""
@ -1645,10 +1718,12 @@ if _lib.ED448_ENABLED:
self.decode_key(key,pub)
@classmethod
def make_key(cls, size, rng=Random()):
def make_key(cls, size, rng=None):
"""
Generates a new key pair of desired length **size**.
"""
if rng is None:
rng = Random()
ed448 = cls()
ret = _lib.wc_ed448_make_key(rng.native_object, size,
@ -1656,6 +1731,10 @@ if _lib.ED448_ENABLED:
if ret < 0:
raise WolfCryptError("Key generation error (%d)" % ret)
# Retain RNG reference defensively; wolfSSL may retain a pointer
# internally on some builds.
ed448._rng = rng
return ed448
def decode_key(self, key, pub = None):
@ -1686,6 +1765,8 @@ if _lib.ED448_ENABLED:
raise WolfCryptError("Public key generate error (%d)" % ret)
ret = _lib.wc_ed448_import_public(pubkey, self.size,
self.native_object)
if ret < 0:
raise WolfCryptError("Public key import error (%d)" % ret)
if self.size <= 0: # pragma: no cover
raise WolfCryptError("Key decode error (%d)" % self.size)
@ -1701,20 +1782,22 @@ if _lib.ED448_ENABLED:
"""
key = _ffi.new("byte[%d]" % (self.size * 4))
pubkey = _ffi.new("byte[%d]" % (self.size * 4))
size = _ffi.new("word32[1]")
priv_size = _ffi.new("word32[1]")
pub_size = _ffi.new("word32[1]")
size[0] = _lib.wc_ed448_priv_size(self.native_object)
priv_size[0] = _lib.wc_ed448_priv_size(self.native_object)
pub_size[0] = _lib.wc_ed448_pub_size(self.native_object)
ret = _lib.wc_ed448_export_private_only(self.native_object,
key, size)
key, priv_size)
if ret != 0: # pragma: no cover
raise WolfCryptError("Private key encode error (%d)" % ret)
ret = _lib.wc_ed448_export_public(self.native_object, pubkey,
size)
pub_size)
if ret != 0: # pragma: no cover
raise WolfCryptError("Public key encode error (%d)" % ret)
return _ffi.buffer(key, size[0])[:], _ffi.buffer(pubkey, size[0])[:]
return _ffi.buffer(key, priv_size[0])[:], _ffi.buffer(pubkey, pub_size[0])[:]
def sign(self, plaintext, ctx=None):
"""
@ -1862,13 +1945,15 @@ if _lib.ML_KEM_ENABLED:
if ret < 0: # pragma: no cover
raise WolfCryptError("wc_KyberKey_DecodePublicKey() error (%d)" % ret)
def encapsulate(self, rng=Random()):
def encapsulate(self, rng=None):
"""
:param rng: random number generator for an encupsulation
:type rng: Random
:return: tuple of a shared secret (first element) and the cipher text (second element)
:rtype: tuple[bytes, bytes]
"""
if rng is None:
rng = Random()
ct_size = self.ct_size
ss_size = self.ss_size
ct = _ffi.new(f"unsigned char[{ct_size}]")
@ -1906,7 +1991,7 @@ if _lib.ML_KEM_ENABLED:
class MlKemPrivate(_MlKemBase):
@classmethod
def make_key(cls, mlkem_type, rng=Random()):
def make_key(cls, mlkem_type, rng=None):
"""
:param mlkem_type: ML-KEM type
:type mlkem_type: MlKemType
@ -1915,12 +2000,17 @@ if _lib.ML_KEM_ENABLED:
:return: `MlKemPrivate` object
:rtype: MlKemPrivate
"""
if rng is None:
rng = Random()
mlkem_priv = cls(mlkem_type)
ret = _lib.wc_KyberKey_MakeKey(mlkem_priv.native_object, rng.native_object)
if ret < 0: # pragma: no cover
raise WolfCryptError("wc_KyberKey_MakeKey() error (%d)" % ret)
# Retain RNG reference defensively.
mlkem_priv._rng = rng
return mlkem_priv
@classmethod
@ -2021,7 +2111,6 @@ if _lib.ML_KEM_ENABLED:
)
if ret < 0: # pragma: no cover
self.native_object = None
raise WolfCryptError("wc_KyberKey_Decapsulate() error (%d)" % ret)
return _ffi.buffer(ss, ss_size)[:]
@ -2170,7 +2259,7 @@ if _lib.ML_DSA_ENABLED:
class MlDsaPrivate(_MlDsaBase):
@classmethod
def make_key(cls, mldsa_type, rng=Random()):
def make_key(cls, mldsa_type, rng=None):
"""
:param mldsa_type: ML-DSA type
:type mldsa_type: MlDsaType
@ -2179,6 +2268,8 @@ if _lib.ML_DSA_ENABLED:
:return: `MlDsaPrivate` object
:rtype: MlDsaPrivate
"""
if rng is None:
rng = Random()
mldsa_priv = cls(mldsa_type)
ret = _lib.wc_dilithium_make_key(
mldsa_priv.native_object, rng.native_object
@ -2187,6 +2278,9 @@ if _lib.ML_DSA_ENABLED:
if ret < 0: # pragma: no cover
raise WolfCryptError("wc_dilithium_make_key() error (%d)" % ret)
# Retain RNG reference defensively.
mldsa_priv._rng = rng
return mldsa_priv
@classmethod
@ -2239,9 +2333,7 @@ if _lib.ML_DSA_ENABLED:
if ret < 0: # pragma: no cover
raise WolfCryptError("wc_MlDsaKey_GetPrivLen() error (%d)" % ret)
key_pair_size = size[0]
return key_pair_size - self.pub_key_size
return size[0] - self.pub_key_size
def encode_pub_key(self):
"""
@ -2293,7 +2385,7 @@ if _lib.ML_DSA_ENABLED:
if pub_key is not None:
self._decode_pub_key(pub_key)
def sign(self, message, rng=Random(), ctx=None):
def sign(self, message, rng=None, ctx=None):
"""
:param message: message to be signed
:type message: bytes or str
@ -2304,6 +2396,8 @@ if _lib.ML_DSA_ENABLED:
:return: signature
:rtype: bytes
"""
if rng is None:
rng = Random()
msg_bytestype = t2b(message)
in_size = self.sig_size
signature = _ffi.new(f"byte[{in_size}]")

144
wolfcrypt/hashes.py
View File

@ -34,6 +34,7 @@ class _Hash:
"""
def __init__(self, string=None):
self._native_object = _ffi.new(self._native_type)
self._shallow_copy = False
ret = self._init()
if ret < 0: # pragma: no cover
raise WolfCryptError("Hash init error (%d)" % ret)
@ -56,11 +57,32 @@ class _Hash:
Returns a separate copy of this hashing object. An update
to this copy won't affect the original object.
"""
copy = self.new("")
# Bypass __init__ to avoid calling _init() on a state that _copy
# immediately overwrites (which would leak internal resources in
# async/HW-accelerated builds). Mark as shallow up front so __del__
# skips the free if we bail out before the copy completes.
copy = type(self).__new__(type(self))
copy._shallow_copy = True # pylint: disable=protected-access
copy._native_object = _ffi.new(self._native_type) # pylint: disable=protected-access
_ffi.memmove(copy._native_object, # pylint: disable=protected-access
self._native_object,
self._native_size)
copy_fn = getattr(self, '_copy', None)
if copy_fn:
ret = copy_fn(self._native_object,
copy._native_object) # pylint: disable=protected-access
if ret < 0: # pragma: no cover
# Free any partial allocation before raising; __del__ would
# skip it because _shallow_copy is still True.
delete = getattr(self, '_delete', None)
if delete:
delete(copy._native_object) # pylint: disable=protected-access
raise WolfCryptError("Hash copy error (%d)" % ret)
copy._shallow_copy = False # pylint: disable=protected-access
else:
_ffi.memmove(copy._native_object, # pylint: disable=protected-access
self._native_object,
self._native_size)
# Keep _shallow_copy = True: memmove shares internal state with
# self, so __del__ must not free it separately.
return copy
@ -87,12 +109,31 @@ class _Hash:
if self._native_object:
obj = _ffi.new(self._native_type)
# _copy and _delete are class attributes on Sha/Sha256/etc, but
# are set as instance attributes on Sha3 (because the SHA3 variant
# is selected by digest_size at __init__ time). getattr handles
# both cases.
copy_fn = getattr(self, '_copy', None)
_ffi.memmove(obj, self._native_object, self._native_size)
try:
if copy_fn:
ret = copy_fn(self._native_object, obj)
if ret < 0: # pragma: no cover
raise WolfCryptError("Hash copy error (%d)" % ret)
else:
_ffi.memmove(obj, self._native_object, self._native_size)
ret = self._final(obj, result)
if ret < 0: # pragma: no cover
raise WolfCryptError("Hash finalize error (%d)" % ret)
ret = self._final(obj, result)
if ret < 0: # pragma: no cover
raise WolfCryptError("Hash finalize error (%d)" % ret)
finally:
# Only free when we did a deep copy; memmove'd temps share
# internal resources with self and must not be separately freed.
# Runs even on failed copy to clean up any partial allocation.
if copy_fn:
delete = getattr(self, '_delete', None)
if delete:
delete(obj)
return _ffi.buffer(result, self.digest_size)[:]
@ -116,6 +157,12 @@ 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
def __del__(self):
if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False):
self._delete(self._native_object)
def _init(self):
return _lib.wc_InitSha(self._native_object)
@ -138,6 +185,12 @@ 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
def __del__(self):
if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False):
self._delete(self._native_object)
def _init(self):
return _lib.wc_InitSha256(self._native_object)
@ -160,6 +213,12 @@ 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
def __del__(self):
if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False):
self._delete(self._native_object)
def _init(self):
return _lib.wc_InitSha384(self._native_object)
@ -182,6 +241,12 @@ 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
def __del__(self):
if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False):
self._delete(self._native_object)
def _init(self):
return _lib.wc_InitSha512(self._native_object)
@ -209,9 +274,35 @@ if _lib.SHA3_ENABLED:
SHA3_384_DIGEST_SIZE = 48
SHA3_512_DIGEST_SIZE = 64
_SHA3_FREE = {
28: _lib.wc_Sha3_224_Free,
32: _lib.wc_Sha3_256_Free,
48: _lib.wc_Sha3_384_Free,
64: _lib.wc_Sha3_512_Free,
}
_SHA3_COPY = {
28: _lib.wc_Sha3_224_Copy,
32: _lib.wc_Sha3_256_Copy,
48: _lib.wc_Sha3_384_Copy,
64: _lib.wc_Sha3_512_Copy,
}
def __del__(self):
# Unlike the SHA-1/2 classes, Sha3's _delete is set per-instance
# from a size->function dict and is None for invalid sizes, so
# we need the extra truthiness check.
if (hasattr(self, '_native_object')
and not getattr(self, '_shallow_copy', False)
and getattr(self, '_delete', None)):
self._delete(self._native_object)
def __init__(self, string=None, size=SHA3_384_DIGEST_SIZE): # pylint: disable=W0231
self._native_object = _ffi.new(self._native_type)
self._shallow_copy = False
self.digest_size = size
self._delete = self._SHA3_FREE.get(size)
self._copy = self._SHA3_COPY.get(size)
ret = self._init()
if ret < 0: # pragma: no cover
raise WolfCryptError("Sha3 init error (%d)" % ret)
@ -223,8 +314,27 @@ if _lib.SHA3_ENABLED:
return cls(string, size)
def copy(self):
c = Sha3(size=self.digest_size)
_ffi.memmove(c._native_object, self._native_object, self._native_size)
# Bypass __init__ to avoid calling _init() on a state that _copy
# immediately overwrites (which would leak internal resources in
# async/HW-accelerated builds). Mark as shallow up front so
# __del__ skips the free if we bail out before the copy completes.
c = type(self).__new__(type(self))
c._shallow_copy = True
c._native_object = _ffi.new(self._native_type)
c.digest_size = self.digest_size
c._delete = self._delete
c._copy = self._copy
if self._copy:
ret = self._copy(self._native_object, c._native_object)
if ret < 0: # pragma: no cover
# Free any partial allocation before raising.
if self._delete:
self._delete(c._native_object)
raise WolfCryptError("Hash copy error (%d)" % ret)
c._shallow_copy = False
else:
_ffi.memmove(c._native_object, self._native_object, self._native_size)
# Keep _shallow_copy = True: memmove shares state with self.
return c
def _init(self):
@ -281,15 +391,29 @@ if _lib.HMAC_ENABLED:
"""
A **PEP 247: Cryptographic Hash Functions** compliant
**Keyed Hash Function Interface**.
Note: wolfSSL does not provide a `wc_HmacCopy` equivalent, so
`copy()` falls back to a byte-level memmove. In default builds the
Hmac struct is self-contained and this is safe. In async or
hardware-accelerated builds where the struct contains internal
pointers, the copy shares those pointers with the original; the
copy must not outlive the original or be used after the original
is freed.
"""
digest_size = None
_native_type = "Hmac *"
_native_size = _ffi.sizeof("Hmac")
_delete = _lib.wc_HmacFree
def __del__(self):
if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False):
self._delete(self._native_object)
def __init__(self, key, string=None): # pylint: disable=W0231
key = t2b(key)
self._native_object = _ffi.new(self._native_type)
self._shallow_copy = False
ret = self._init(self._type, key)
if ret < 0: # pragma: no cover
raise WolfCryptError("Hmac init error (%d)" % ret)

6
wolfcrypt/hkdf.py
View File

@ -34,7 +34,7 @@ if _lib.HKDF_ENABLED:
Perform HKDF Extract-and-Expand in one call (wraps wc_HKDF).
Parameters:
- hash_cls: hash class, see `wolfcrypt.hashes`.
- hash_cls: HMAC class, e.g. HmacSha256, see `wolfcrypt.hashes`.
- in_key: input key material (IKM) as bytes or str.
- salt: optional salt value (bytes or str). If None, treated as empty.
- info: optional context/application info (bytes or str). If None,
@ -79,7 +79,7 @@ if _lib.HKDF_ENABLED:
Wraps wc_HKDF_Extract.
Parameters:
- hash_cls: hash class, see `wolfcrypt.hashes`.
- hash_cls: HMAC class, e.g. HmacSha256, see `wolfcrypt.hashes`.
- salt: bytes/str (can be None -> treated as empty).
- in_key: input key material (IKM) as bytes/str.
@ -106,7 +106,7 @@ if _lib.HKDF_ENABLED:
Wraps wc_HKDF_Expand.
Parameters:
- hash_cls: hash class, see `wolfcrypt.hashes`.
- hash_cls: HMAC class, e.g. HmacSha256, see `wolfcrypt.hashes`.
- prk: pseudorandom key (output from HKDF-Extract) as bytes/str.
- info: optional context/application info (bytes/str). If None, treated as empty.
- out_len: length of output keying material in bytes.