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

Address review comments and annotate overrides.

pull/125/head
Robert de Vries 2026-06-10 17:17:00 +02:00
parent 2081edd6d7
commit 1fb34e5cd8
5 changed files with 78 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
tests/test_delete_descriptor_binding.py
View File

@ -173,7 +173,6 @@ def test_random_delete_receives_only_native_object():
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 == {}

6
wolfcrypt/_ffi/lib.pyi
View File

@ -318,13 +318,13 @@ def wc_dilithium_export_public(key: DilithiumKey, public: BytePtr, public_len: I
def wc_dilithium_verify_ctx_msg(sig: bytes, sig_len: int, ctx: bytes, ctx_len: int, msg: bytes, msg_len: int, res: IntPtr, key: DilithiumKey) -> int: ...
def wc_dilithium_verify_msg(sig: bytes, sig_len: int, msg: bytes, msg_len: int, res: IntPtr, key: DilithiumKey) -> int: ...
def wc_dilithium_make_key(key: DilithiumKey, rng: RNG) -> int: ...
def wc_dilithium_make_key_from_seed(key: DilithiumKey, seed: bytes) -> int: ...
def wc_dilithium_make_key_from_seed(key: DilithiumKey, seed: bytes | list[int] | tuple[int]) -> int: ...
def wc_dilithium_export_private(key: DilithiumKey, out: BytePtr, out_len: IntPtr) -> int: ...
def wc_dilithium_import_private(priv: bytes, priv_size: int, key: DilithiumKey) -> int: ...
def wc_dilithium_sign_ctx_msg(ctx: bytes, ctx_len: int, msg: bytes, msg_len: int, sig: BytePtr, sig_len: IntPtr, key: DilithiumKey, rng: RNG) -> int: ...
def wc_dilithium_sign_msg(msg: bytes, msg_len: int, sig: BytePtr, sig_len: IntPtr, key: DilithiumKey, rng: RNG) -> int: ...
def wc_dilithium_sign_ctx_msg_with_seed(ctx: bytes, ctx_len: int, msg: bytes, msg_len: int, sig: BytePtr, sig_len: IntPtr, key: DilithiumKey, seed: bytes) -> int: ...
def wc_dilithium_sign_msg_with_seed(msg: bytes, msg_len: int, sig: BytePtr, sig_len: IntPtr, key: DilithiumKey, seed: bytes) -> int: ...
def wc_dilithium_sign_ctx_msg_with_seed(ctx: bytes, ctx_len: int, msg: bytes, msg_len: int, sig: BytePtr, sig_len: IntPtr, key: DilithiumKey, seed: bytes | list[int] | tuple[int]) -> int: ...
def wc_dilithium_sign_msg_with_seed(msg: bytes, msg_len: int, sig: BytePtr, sig_len: IntPtr, key: DilithiumKey, seed: bytes | list[int] | tuple[int]) -> int: ...
def wc_MlDsaKey_GetPrivLen(key: DilithiumKey, len: IntPtr) -> int: ...
def wc_MlDsaKey_GetPubLen(key: DilithiumKey, len: IntPtr) -> int: ...
def wc_MlDsaKey_GetSigLen(key: DilithiumKey, len: IntPtr) -> int: ...

61
wolfcrypt/ciphers.py
View File

@ -25,14 +25,17 @@ from __future__ import annotations
from abc import ABC, abstractmethod
from enum import IntEnum
from typing_extensions import override
from wolfcrypt._ffi import ffi as _ffi
from wolfcrypt._ffi import lib as _lib
from wolfcrypt.utils import BytesOrStr, t2b
from wolfcrypt.random import Random
from wolfcrypt.asn import pem_to_der
from wolfcrypt.hashes import hash_type_to_cls
from wolfcrypt.exceptions import WolfCryptError, WolfCryptApiError
from wolfcrypt.hashes import hash_type_to_cls
from wolfcrypt.random import Random
from wolfcrypt.utils import BytesOrStr, t2b
if _lib.ASN_ENABLED:
from wolfcrypt.asn import pem_to_der # ty: ignore[possibly-missing-import]
# key direction flags
_ENCRYPTION = 0
@ -115,7 +118,6 @@ if _lib.RSA_ENABLED:
HASH_TYPE_BLAKE2S = _lib.WC_HASH_TYPE_BLAKE2S
class _Cipher(ABC):
"""
A **PEP 272: Block Encryption Algorithms** compliant
@ -273,6 +275,7 @@ if _lib.AES_ENABLED:
_key_sizes = [16, 24, 32]
_native_type = "Aes *"
@override
def _set_key(self, direction: int) -> int:
if direction == _ENCRYPTION:
assert self._enc is not None
@ -285,6 +288,7 @@ if _lib.AES_ENABLED:
return _lib.wc_AesSetKey(
self._dec, self._key, len(self._key), self._IV, _DECRYPTION)
@override
def _encrypt(self, destination: _ffi.CData, source: bytes) -> int:
assert self._enc is not None
if self.mode == MODE_CBC:
@ -296,6 +300,7 @@ if _lib.AES_ENABLED:
else:
raise ValueError("Invalid mode associated to cipher")
@override
def _decrypt(self, destination: _ffi.CData, source: bytes) -> int:
assert self._dec is not None
if self.mode == MODE_CBC:
@ -398,7 +403,7 @@ if _lib.AES_SIV_ENABLED:
associated_data_bytes = t2b(associated_data)
result = _ffi.new("AesSivAssoc[1]")
result[0].assoc = _ffi.from_buffer(associated_data_bytes)
result[0].assocSz = len(associated_data)
result[0].assocSz = len(associated_data_bytes)
else:
# It is assumed that a list is provided.
num_blocks = len(associated_data)
@ -547,7 +552,7 @@ if _lib.CHACHA_ENABLED:
_IV_nonce = b""
_IV_counter = 0
def __init__(self, key: BytesOrStr="", _size: int= 32) -> None: # pylint: disable=unused-argument
def __init__(self, key: BytesOrStr = "", size: int = 32) -> None: # 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)
@ -582,6 +587,7 @@ if _lib.CHACHA_ENABLED:
# collide with _ENCRYPTION (0) or _DECRYPTION (1).
_REKEY_BOTH = -1
@override
def _set_key(self, direction: int) -> int:
if self._key is None:
return -1
@ -608,11 +614,12 @@ if _lib.CHACHA_ENABLED:
return ret
return 0
@override
def _encrypt(self, destination: _ffi.CData, source: bytes) -> int:
assert self._enc is not None
return _lib.wc_Chacha_Process(self._enc, destination,
source, len(source))
@override
def _decrypt(self, destination: _ffi.CData, source: bytes) -> int:
assert self._dec is not None
return _lib.wc_Chacha_Process(self._dec,
@ -726,6 +733,7 @@ if _lib.DES3_ENABLED:
raise ValueError("Des3 only supports MODE_CBC")
super().__init__(key, mode, IV)
@override
def _set_key(self, direction: int) -> int:
if direction == _ENCRYPTION:
assert self._enc is not None
@ -734,10 +742,12 @@ if _lib.DES3_ENABLED:
assert self._dec is not None
return _lib.wc_Des3_SetKey(self._dec, self._key, self._IV, _DECRYPTION)
@override
def _encrypt(self, destination: _ffi.CData, source: bytes) -> int:
assert self._enc is not None
return _lib.wc_Des3_CbcEncrypt(self._enc, destination, source, len(source))
@override
def _decrypt(self, destination: _ffi.CData, source: bytes) -> int:
assert self._dec is not None
return _lib.wc_Des3_CbcDecrypt(self._dec, destination, source, len(source))
@ -795,15 +805,13 @@ if _lib.RSA_ENABLED:
def __init__(self, key: BytesOrStr, hash_type: int | None = None, rng: Random | None = None) -> None:
super().__init__(rng)
if key is not None:
key = t2b(key)
key = t2b(key)
self._hash_type = hash_type
idx = _ffi.new("word32*")
idx[0] = 0
ret = _lib.wc_RsaPublicKeyDecode(key, idx,
self.native_object, len(key))
ret = _lib.wc_RsaPublicKeyDecode(key, idx, self.native_object, len(key))
if ret < 0:
raise WolfCryptApiError("Invalid key error", ret)
@ -981,6 +989,7 @@ if _lib.RSA_ENABLED:
raise WolfCryptApiError("Invalid key size error", self.output_size)
if _lib.ASN_ENABLED:
@override
@classmethod
def from_pem(cls, file: bytes, hash_type: int | None = None, rng: Random | None = None) -> RsaPrivate:
der = pem_to_der(file, _lib.PRIVATEKEY_TYPE)
@ -1334,6 +1343,7 @@ if _lib.ECC_ENABLED:
return ecc
@override
def decode_key(self, key: BytesOrStr) -> None:
"""
Decodes an ECC private key from an ASN sequence.
@ -1352,6 +1362,7 @@ if _lib.ECC_ENABLED:
if self.max_signature_size <= 0: # pragma: no cover
raise WolfCryptError(f"Key decode error ({self.max_signature_size})")
@override
def decode_key_raw(self, qx: BytesOrStr, qy: BytesOrStr, d: BytesOrStr, curve_id: int = ECC_SECP256R1) -> None:
"""
Decodes an ECC private key from its raw elements: public (Qx,Qy)
@ -1373,6 +1384,7 @@ if _lib.ECC_ENABLED:
if ret != 0:
raise WolfCryptApiError("Key decode error", ret)
@override
def encode_key(self) -> bytes:
"""
Encodes the ECC private key in an ASN sequence.
@ -1387,6 +1399,7 @@ if _lib.ECC_ENABLED:
return _ffi.buffer(key, ret)[:]
@override
def encode_key_raw(self) -> tuple[bytes, bytes, bytes]:
"""
Encodes the ECC private key in its three raw elements
@ -1620,6 +1633,7 @@ if _lib.ED25519_ENABLED:
return ed25519
@override
def decode_key(self, key: BytesOrStr, pub: bytes | None = None) -> None:
"""
Decodes an ED25519 private + pub key
@ -1656,6 +1670,7 @@ if _lib.ED25519_ENABLED:
if self.max_signature_size <= 0: # pragma: no cover
raise WolfCryptError(f"Key decode error ({self.max_signature_size})")
@override
def encode_key(self) -> tuple[bytes, bytes]:
"""
Encodes the ED25519 private key.
@ -1827,6 +1842,7 @@ if _lib.ED448_ENABLED:
return ed448
@override
def decode_key(self, key: BytesOrStr, pub: bytes | None = None) -> None:
"""
Decodes an ED448 private + pub key
@ -1863,6 +1879,7 @@ if _lib.ED448_ENABLED:
if self.max_signature_size <= 0: # pragma: no cover
raise WolfCryptError(f"Key decode error ({self.max_signature_size})")
@override
def encode_key(self) -> tuple[bytes, bytes]:
"""
Encodes the ED448 private key.
@ -2030,7 +2047,7 @@ if _lib.ML_KEM_ENABLED:
pub_key_bytestype = t2b(pub_key)
ret = _lib.wc_KyberKey_DecodePublicKey(
self.native_object,
_ffi.from_buffer(pub_key_bytestype),
pub_key_bytestype,
len(pub_key_bytestype),
)
@ -2376,23 +2393,22 @@ if _lib.ML_DSA_ENABLED:
return mldsa_priv
@classmethod
def make_key_from_seed(cls, mldsa_type: MlDsaType, seed: bytes) -> MlDsaPrivate:
def make_key_from_seed(cls, mldsa_type: MlDsaType, seed: bytes | list[int] | tuple[int]) -> MlDsaPrivate:
"""
Deterministically generate the key from a seed.
:param mldsa_type: ML-DSA type
:type mldsa_type: MlDsaType
:param seed: the (32 byte) seed from which to deterministically create the key
:type seed: bytes
:type seed: bytes or list/tuple of int
"""
mldsa_priv = cls(mldsa_type)
seed_bytes = t2b(seed)
if len(seed_bytes) != cls.ML_DSA_KEYGEN_SEED_LENGTH:
if len(seed) != cls.ML_DSA_KEYGEN_SEED_LENGTH:
raise ValueError(
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, seed_bytes)
ret = _lib.wc_dilithium_make_key_from_seed(mldsa_priv.native_object, seed)
if ret < 0: # pragma: no cover
raise WolfCryptApiError("wc_dilithium_make_key_from_seed() error", ret)
@ -2525,12 +2541,12 @@ if _lib.ML_DSA_ENABLED:
return _ffi.buffer(signature, out_size[0])[:]
def sign_with_seed(self, message: BytesOrStr, seed: bytes, ctx: BytesOrStr | None = None) -> bytes:
def sign_with_seed(self, message: BytesOrStr, seed: bytes | list[int] | tuple[int], ctx: BytesOrStr | None = None) -> bytes:
"""
:param message: message to be signed
:type message: bytes or str
:param seed: 32-byte seed for deterministic signature generation.
:type seed: bytes
:type seed: bytes or list/tuple of int (value in the range 0-255)
:param ctx: context, maximum 255 bytes (optional by default but that requires support for no-context
signing/verification compiled in; pass empty string "" for FIPS-204 empty-context signing).
:type ctx: bytes or str. None for no-context signing.
@ -2548,8 +2564,7 @@ if _lib.ML_DSA_ENABLED:
if len(seed) != ML_DSA_SIGNATURE_SEED_LENGTH:
raise ValueError(
f"Seed for generating a signature must be {ML_DSA_SIGNATURE_SEED_LENGTH}"
"bytes."
f"Seed for generating a signature must be {ML_DSA_SIGNATURE_SEED_LENGTH} bytes."
)
if ctx is not None:

26
wolfcrypt/hashes.py
View File

@ -25,6 +25,7 @@ from __future__ import annotations
from abc import ABC, abstractmethod
from _cffi_backend import FFI
from typing_extensions import override
from wolfcrypt._ffi import ffi as _ffi
from wolfcrypt._ffi import lib as _lib
from wolfcrypt.exceptions import WolfCryptApiError
@ -69,8 +70,7 @@ class _Hash(ABC):
@classmethod
@abstractmethod
def new(cls, string: BytesOrStr | None) -> _Hash: ...
def new(cls, string: BytesOrStr | None) -> _Hash: ...
def copy(self) -> _Hash:
"""
@ -168,6 +168,7 @@ class _Hash(ABC):
class _Sha(_Hash):
@override
@classmethod
def new(cls, string: BytesOrStr | None = None) -> _Hash:
"""
@ -196,12 +197,15 @@ if _lib.SHA_ENABLED:
if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False):
self._delete(self._native_object)
@override
def _init(self) -> int:
return _lib.wc_InitSha(self._native_object)
@override
def _update(self, data: bytes) -> int:
return _lib.wc_ShaUpdate(self._native_object, data, len(data))
@override
def _final(self, obj: FFI.CData, ret: FFI.CData) -> int:
return _lib.wc_ShaFinal(obj, ret)
@ -224,12 +228,15 @@ if _lib.SHA256_ENABLED:
if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False):
self._delete(self._native_object)
@override
def _init(self) -> int:
return _lib.wc_InitSha256(self._native_object)
@override
def _update(self, data: bytes) -> int:
return _lib.wc_Sha256Update(self._native_object, data, len(data))
@override
def _final(self, obj: FFI.CData, ret: FFI.CData) -> int:
return _lib.wc_Sha256Final(obj, ret)
@ -252,12 +259,15 @@ if _lib.SHA384_ENABLED:
if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False):
self._delete(self._native_object)
@override
def _init(self) -> int:
return _lib.wc_InitSha384(self._native_object)
@override
def _update(self, data: bytes) -> int:
return _lib.wc_Sha384Update(self._native_object, data, len(data))
@override
def _final(self, obj: FFI.CData, ret: FFI.CData) -> int:
return _lib.wc_Sha384Final(obj, ret)
@ -280,12 +290,15 @@ if _lib.SHA512_ENABLED:
if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False):
self._delete(self._native_object)
@override
def _init(self) -> int:
return _lib.wc_InitSha512(self._native_object)
@override
def _update(self, data: bytes) -> int:
return _lib.wc_Sha512Update(self._native_object, data, len(data))
@override
def _final(self, obj: FFI.CData, ret: FFI.CData) -> int:
return _lib.wc_Sha512Final(obj, ret)
@ -345,10 +358,12 @@ if _lib.SHA3_ENABLED:
if string:
self.update(string)
@override
@classmethod
def new(cls, string: BytesOrStr | None = None, size: int = SHA3_384_DIGEST_SIZE) -> Sha3:
return cls(string, size)
@override
def copy(self) -> Sha3:
# Bypass __init__ to avoid calling _init() on a state that _copy
# immediately overwrites (which would leak internal resources in
@ -373,6 +388,7 @@ if _lib.SHA3_ENABLED:
# Keep _shallow_copy = True: memmove shares state with self.
return c
@override
def _init(self) -> int:
if self.digest_size == Sha3.SHA3_224_DIGEST_SIZE:
return _lib.wc_InitSha3_224(self._native_object, _ffi.NULL, 0)
@ -384,6 +400,7 @@ if _lib.SHA3_ENABLED:
return _lib.wc_InitSha3_512(self._native_object, _ffi.NULL, 0)
return -1
@override
def _update(self, data: bytes) -> int:
if self.digest_size == Sha3.SHA3_224_DIGEST_SIZE:
return _lib.wc_Sha3_224_Update(self._native_object, data, len(data))
@ -395,6 +412,7 @@ if _lib.SHA3_ENABLED:
return _lib.wc_Sha3_512_Update(self._native_object, data, len(data))
return -1
@override
def _final(self, obj: FFI.CData, ret: FFI.CData) -> int:
if self.digest_size == Sha3.SHA3_224_DIGEST_SIZE:
return _lib.wc_Sha3_224_Final(obj, ret)
@ -460,9 +478,11 @@ if _lib.HMAC_ENABLED:
if string:
self.update(string)
@override
def _init(self) -> int:
return -1
@override
@classmethod
def new(cls, key: BytesOrStr, string: BytesOrStr | None = None) -> _Hash: # pylint: disable=W0221 # ty: ignore[invalid-method-override]
"""
@ -493,9 +513,11 @@ if _lib.HMAC_ENABLED:
raise WolfCryptApiError("wc_HmacSetKey error", ret)
return ret
@override
def _update(self, data: bytes) -> int:
return _lib.wc_HmacUpdate(self._native_object, data, len(data))
@override
def _final(self, obj: FFI.CData, ret: FFI.CData) -> int:
return _lib.wc_HmacFinal(obj, ret)

18
wolfcrypt/random.py
View File

@ -25,7 +25,7 @@ from __future__ import annotations
from wolfcrypt._ffi import ffi as _ffi
from wolfcrypt._ffi import lib as _lib
from wolfcrypt.exceptions import WolfCryptApiError
from wolfcrypt.exceptions import WolfCryptApiError, WolfCryptError
class Random:
@ -34,23 +34,31 @@ class Random:
"""
def __init__(self, nonce: __builtins__.bytes = b"", device_id: int = -2) -> None:
self.native_object: _lib.RNG = _ffi.new("WC_RNG *")
self._native_object: _lib.RNG | None = None
self._native_object = _ffi.new("WC_RNG *")
ret = _lib.wc_InitRngNonce_ex(self.native_object, nonce, len(nonce), _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 WolfCryptApiError("RNG init error", ret)
# making sure _lib.wc_FreeRng outlives WC_RNG instances
_delete = staticmethod(_lib.wc_FreeRng)
def __del__(self) -> None:
if self.native_object:
if self._native_object is not None:
try:
self._delete(self.native_object)
self._delete(self._native_object)
except AttributeError:
# Can occur during interpreter shutdown
pass
@property
def native_object(self) -> _lib.RNG:
if self._native_object is None:
raise WolfCryptError("RNG not initialized")
return self._native_object
def byte(self) -> __builtins__.bytes:
"""
Generate and return a random byte.