Addressed review comments

* added various tests
* adapted hash interface to allow callbacks
pull/114/head
Robert de Vries 2026-07-09 22:46:20 +02:00
parent c7306ed9ef
commit d2debf4c8d
6 changed files with 74 additions and 27 deletions

View File

@ -870,7 +870,7 @@ def build_ffi(local_wolfssl, features):
if features["SHA"]:
cdef += """
typedef struct { ...; } wc_Sha;
int wc_InitSha(wc_Sha*);
int wc_InitSha_ex(wc_Sha*, void*, int);
int wc_ShaUpdate(wc_Sha*, const byte*, word32);
int wc_ShaFinal(wc_Sha*, byte*);
void wc_ShaFree(wc_Sha*);
@ -1456,12 +1456,28 @@ def build_ffi(local_wolfssl, features):
word32 data_size;
byte* digest;
union {
"""
if features["SHA"]:
cdef += """
wc_Sha* sha1;
// wc_Sha224* sha224;
"""
if features["SHA256"]:
cdef += """
wc_Sha256* sha256;
"""
if features["SHA384"]:
cdef += """
wc_Sha384* sha384;
"""
if features["SHA512"]:
cdef += """
wc_Sha512* sha512;
"""
if features["SHA3"]:
cdef += """
wc_Sha3* sha3;
"""
cdef += """
void* ctx;
} u;
} hash;

View File

@ -17,12 +17,20 @@
# 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
# ty: ignore[possibly-missing-import]
from __future__ import annotations
import struct
import pytest
from typing_extensions import override
from wolfcrypt._ffi import lib as _lib
from wolfcrypt.random import Random
if _lib.SHA_ENABLED:
from wolfcrypt.hashes import Sha
if not _lib.CRYPTO_CB_ENABLED:
pytest.skip("Crypto Callbacks not supported", allow_module_level=True)
@ -31,10 +39,12 @@ from wolfcrypt.cryptocb import CryptoCallback
def test_default_device_id():
print(f"Default device ID = {CryptoCallback.default_device_id()}")
# In the python implementation the default device ID is the invalid device ID.
assert CryptoCallback.default_device_id() == _lib.INVALID_DEVID
class RngCryptoCallback(CryptoCallback):
def rng_callback(self, _device_id: int, _rng, size: int) -> bytes:
@override
def rng_callback(self, device_id: int, rng: _lib.RNG, size: int) -> bytes:
# Generate fake random data for testing purposes.
return bytes(range(1, 1 + size))
@ -51,3 +61,27 @@ def test_rng_callback():
random = rng.bytes(3)
assert random == b"\01\02\03"
class HashCryptoCallback(CryptoCallback):
def __init__(self, device_id):
super().__init__(device_id)
self.data: list[bytes] = []
@override
def hash_update_callback(self, device_id: int, hash_type: int, data: bytes) -> None:
self.data.append(data)
@override
def hash_finalize_callback(self, device_id: int, hash_type: int) -> bytes:
# quite lame hash function, just returns the length of the data as an integer (padded to match the expected hash length).
return struct.pack("I16x", len(b"".join(self.data)))
if _lib.SHA_ENABLED:
def test_hash_callback():
with HashCryptoCallback(11):
sha = Sha(device_id=11)
sha.update(bytes(10))
sha.update(bytes(5))
digest = sha.digest()
assert digest == struct.pack("I16x", 15)

View File

@ -50,9 +50,12 @@ if top_level_py not in ["setup.py", "build_ffi.py"]:
if TYPE_CHECKING:
if _lib.CRYPTO_CB_ENABLED:
from wolfcrypt.cryptocb import CryptoCallback
from wolfcrypt.cryptocb import CryptoCallback # ty: ignore[possibly-missing-import]
from wolfcrypt.exceptions import WolfCryptApiError
# Only wolfCrypt_Init() is called here.
# Calling wolfCrypt_Cleanup() is not needed as the application exit() will clean up the entire process
# including any wolfcrypt data in any case.
ret = _lib.wolfCrypt_Init()
if ret < 0:
raise WolfCryptApiError("WolfCrypt_Init failed", ret)

View File

@ -369,7 +369,7 @@ def wc_EncodeSignature(out: BytePtr, digest: bytes, digest_size: int, hash_oid:
def wc_PBKDF2(out: BytePtr, passwd: bytes, pass_len: int, salt: bytes, salt_len: int, iterations: int, keylen: int,
hash_type: int) -> int: ...
def wc_InitSha(obj: FFI.CData) -> int: ...
def wc_InitSha_ex(obj: FFI.CData, heap: FFI.CData, device_id: int) -> int: ...
def wc_ShaCopy(src: FFI.CData, dst: FFI.CData) -> int: ...
def wc_ShaUpdate(obj: FFI.CData, data: bytes, size: int) -> int: ...
def wc_ShaFinal(obj: FFI.CData, ret: FFI.CData) -> int: ...

View File

@ -124,9 +124,6 @@ if _lib.CRYPTO_CB_ENABLED:
)
_ffi.buffer(info.hash.digest, DIGEST_SIZE[info.hash.type])[:] = digest
return 0
if info.algo_type == _lib.WC_ALGO_TYPE_CIPHER:
self.cipher_callback(device_id)
return 0
if info.algo_type == _lib.WC_ALGO_TYPE_RNG:
out = self.rng_callback(device_id, info.rng.rng, info.rng.sz)
if len(out) != info.rng.sz:
@ -148,9 +145,6 @@ if _lib.CRYPTO_CB_ENABLED:
def hash_finalize_callback(self, device_id: int, hash_type: int) -> bytes:
raise NotImplementedError
def cipher_callback(self, device_id: int) -> None:
raise NotImplementedError
def _unregister(self) -> None:
_lib.wc_CryptoCb_UnRegisterDevice(self.device_id)

View File

@ -40,10 +40,10 @@ class _Hash(ABC):
A **PEP 247: Cryptographic Hash Functions** compliant
**Hash Function Interface**.
"""
def __init__(self, string: BytesOrStr | None = None) -> None:
def __init__(self, string: BytesOrStr | None = None, device_id: int = _lib.INVALID_DEVID) -> None:
self._native_object = _ffi.new(self._native_type)
self._shallow_copy = False
ret = self._init()
ret = self._init(device_id)
if ret < 0: # pragma: no cover
raise WolfCryptApiError("Hash init error", ret)
@ -51,7 +51,7 @@ class _Hash(ABC):
self.update(string)
@abstractmethod
def _init(self) -> int: ...
def _init(self, device_id: int) -> int: ...
@abstractmethod
def _update(self, data: bytes) -> int: ...
@ -201,8 +201,8 @@ if _lib.SHA_ENABLED:
self._delete(self._native_object)
@override
def _init(self) -> int:
return _lib.wc_InitSha(self._native_object)
def _init(self, device_id: int) -> int:
return _lib.wc_InitSha_ex(self._native_object, _ffi.NULL, device_id)
@override
def _update(self, data: bytes) -> int:
@ -232,7 +232,7 @@ if _lib.SHA256_ENABLED:
self._delete(self._native_object)
@override
def _init(self) -> int:
def _init(self, device_id: int) -> int:
return _lib.wc_InitSha256(self._native_object)
@override
@ -263,7 +263,7 @@ if _lib.SHA384_ENABLED:
self._delete(self._native_object)
@override
def _init(self) -> int:
def _init(self, device_id: int) -> int:
return _lib.wc_InitSha384(self._native_object)
@override
@ -294,7 +294,7 @@ if _lib.SHA512_ENABLED:
self._delete(self._native_object)
@override
def _init(self) -> int:
def _init(self, device_id: int) -> int:
return _lib.wc_InitSha512(self._native_object)
@override
@ -355,7 +355,7 @@ if _lib.SHA3_ENABLED:
self.digest_size = size
self._delete = self._SHA3_FREE.get(size)
self._copy = self._SHA3_COPY.get(size)
ret = self._init()
ret = self._init(_lib.INVALID_DEVID)
if ret < 0: # pragma: no cover
raise WolfCryptApiError("Sha3 init error", ret)
if string:
@ -392,15 +392,15 @@ if _lib.SHA3_ENABLED:
return c
@override
def _init(self) -> int:
def _init(self, device_id: int) -> int:
if self.digest_size == Sha3.SHA3_224_DIGEST_SIZE:
return _lib.wc_InitSha3_224(self._native_object, _ffi.NULL, 0)
return _lib.wc_InitSha3_224(self._native_object, _ffi.NULL, device_id)
if self.digest_size == Sha3.SHA3_256_DIGEST_SIZE:
return _lib.wc_InitSha3_256(self._native_object, _ffi.NULL, 0)
return _lib.wc_InitSha3_256(self._native_object, _ffi.NULL, device_id)
if self.digest_size == Sha3.SHA3_384_DIGEST_SIZE:
return _lib.wc_InitSha3_384(self._native_object, _ffi.NULL, 0)
return _lib.wc_InitSha3_384(self._native_object, _ffi.NULL, device_id)
if self.digest_size == Sha3.SHA3_512_DIGEST_SIZE:
return _lib.wc_InitSha3_512(self._native_object, _ffi.NULL, 0)
return _lib.wc_InitSha3_512(self._native_object, _ffi.NULL, device_id)
return -1
@override
@ -483,7 +483,7 @@ if _lib.HMAC_ENABLED:
self.update(string)
@override
def _init(self) -> int:
def _init(self, device_id: int) -> int:
return -1
@override