diff --git a/scripts/build_ffi.py b/scripts/build_ffi.py index 6807edf..00cc972 100644 --- a/scripts/build_ffi.py +++ b/scripts/build_ffi.py @@ -238,6 +238,10 @@ def make_flags(prefix, fips): # ML-DSA (note: to be able to use the legacy option of signing without context, pass `yes,no-ctx` as argument) flags.append("--enable-mldsa=yes") + # Crypto Callbacks + flags.append("--enable-cryptocb") + # flags.append("EXTRACPPFLAGS=-DDEBUG_CRYPTOCB") + # disabling other configs enabled by default flags.append("--disable-oldtls") flags.append("--disable-oldnames") @@ -394,6 +398,7 @@ def get_features(local_wolfssl, features): # Unlike the other fatures, HASHDRBG is enabled by default in random.h, unless WC_NO_HASHDRBG or # CUSTOM_RAND_GENERATE_BLOCK is defined. features["HASHDRBG"] = 0 if ("#define WC_NO_HASHDRBG" in defines or "#define CUSTOM_RAND_GENERATE_BLOCK" in defines) else 1 + features["CRYPTO_CB"] = 1 if "#define WOLF_CRYPTO_CB" in defines else 0 if '#define HAVE_FIPS' in defines: if not fips: @@ -471,6 +476,7 @@ def build_ffi(local_wolfssl, features): #include #include #include + #include """ init_source_string = f""" @@ -515,6 +521,7 @@ def build_ffi(local_wolfssl, features): int ML_DSA_NO_CTX_ENABLED = {features["ML_DSA_NO_CTX"]}; int HKDF_ENABLED = {features["HKDF"]}; int HASHDRBG_ENABLED = {features["HASHDRBG"]}; + int CRYPTO_CB_ENABLED = {features["CRYPTO_CB"]}; """ ffibuilder.set_source( "wolfcrypt._ffi", init_source_string, @@ -558,6 +565,7 @@ def build_ffi(local_wolfssl, features): extern int ML_DSA_NO_CTX_ENABLED; extern int HKDF_ENABLED; extern int HASHDRBG_ENABLED; + extern int CRYPTO_CB_ENABLED; typedef unsigned char byte; typedef unsigned int word32; @@ -565,6 +573,8 @@ def build_ffi(local_wolfssl, features): typedef struct { ...; } WC_RNG; typedef struct { ...; } OS_Seed; + int wolfCrypt_Init(void); + int wc_InitRng(WC_RNG*); int wc_InitRngNonce(WC_RNG*, byte*, word32); int wc_InitRngNonce_ex(WC_RNG*, byte*, word32, void*, int); @@ -1363,6 +1373,110 @@ def build_ffi(local_wolfssl, features): int wc_dilithium_verify_msg(const byte* sig, word32 sigLen, const byte* msg, word32 msgLen, int* res, dilithium_key* key); """ + if features["CRYPTO_CB"]: + cdef += """ + static const int WC_ALGO_TYPE_NONE; + static const int WC_ALGO_TYPE_HASH; + static const int WC_ALGO_TYPE_CIPHER; + static const int WC_ALGO_TYPE_PK; + static const int WC_ALGO_TYPE_RNG; + static const int WC_ALGO_TYPE_SEED; + static const int WC_ALGO_TYPE_HMAC; + static const int WC_ALGO_TYPE_CMAC; + static const int WC_ALGO_TYPE_CERT; + static const int WC_ALGO_TYPE_KDF; + static const int WC_ALGO_TYPE_COPY; + static const int WC_ALGO_TYPE_FREE; + static const int WC_ALGO_TYPE_MAX; + + static const int WC_HASH_TYPE_SHA; /* SHA-1 (not old SHA-0) */ + static const int WC_HASH_TYPE_SHA256; + static const int WC_HASH_TYPE_SHA384; + static const int WC_HASH_TYPE_SHA512; + static const int WC_HASH_TYPE_SHA3_256; + static const int WC_HASH_TYPE_SHA3_384; + static const int WC_HASH_TYPE_SHA3_512; + """ + + + cdef += """ + typedef struct { + int algo_type; /* enum wc_AlgoType */ + union { + """ + + """ + struct { + int type; /* enum wc_CipherType */ + int enc; + union { + //wc_CryptoCb_AesAuthEnc aesgcm_enc; + //wc_CryptoCb_AesAuthDec aesgcm_dec; + //wc_CryptoCb_AesAuthEnc aesccm_enc; + //wc_CryptoCb_AesAuthDec aesccm_dec; + struct { + Aes* aes; + byte* out; + const byte* in; + word32 sz; + } aescbc; + //struct { + // Aes* aes; + // byte* out; + // const byte* in; + // word32 sz; + //} aesctr; + //struct { + // Aes* aes; + // byte* out; + // const byte* in; + // word32 sz; + //} aesecb; + //struct { + // Des3* des; + // byte* out; + // const byte* in; + // word32 sz; + //} des3; + //void* ctx; + }; + } cipher; + """ + cdef += """ + struct { + int type; /* enum wc_HashType */ + const byte* data; + word32 data_size; + byte* digest; + union { + wc_Sha* sha1; + // wc_Sha224* sha224; + wc_Sha256* sha256; + wc_Sha384* sha384; + wc_Sha512* sha512; + wc_Sha3* sha3; + void* ctx; + } u; + } hash; + """ + cdef += """ + struct { + WC_RNG* rng; + byte* out; + word32 sz; + } rng; + }; + ...; + } wc_CryptoInfo; + + typedef int (*CryptoDevCallbackFunc)(int devId, wc_CryptoInfo* info, void* ctx); + extern "Python" int py_wc_crypto_callback(int devId, wc_CryptoInfo* info, void* ctx); + int wc_CryptoCb_RegisterDevice(int devId, CryptoDevCallbackFunc cb, void* ctx); + void wc_CryptoCb_UnRegisterDevice(int devId); + int wc_CryptoCb_DefaultDevID(); + // void wc_CryptoCb_InfoString(wc_CryptoInfo* info); + """ + ffibuilder.cdef(cdef) def main(ffibuilder): @@ -1400,6 +1514,7 @@ def main(ffibuilder): "ML_DSA_NO_CTX": 0, "HKDF": 1, "HASHDRBG": 1, + "CRYPTO_CB": 1, } # Ed448 requires SHAKE256, which isn't part of the Windows build, yet. diff --git a/tests/test_cryptocb.py b/tests/test_cryptocb.py new file mode 100644 index 0000000..56dbbd9 --- /dev/null +++ b/tests/test_cryptocb.py @@ -0,0 +1,53 @@ +# test_cryptocb.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 + +import pytest + +from wolfcrypt._ffi import lib as _lib +from wolfcrypt.random import Random + + +if not _lib.CRYPTO_CB_ENABLED: + pytest.skip("Crypto Callbacks not supported", allow_module_level=True) + +from wolfcrypt.cryptocb import CryptoCallback + + +def test_default_device_id(): + print(f"Default device ID = {CryptoCallback.default_device_id()}") + +class RngCryptoCallback(CryptoCallback): + def rng_callback(self, _device_id: int, _rng, size: int) -> bytes: + # Generate fake random data for testing purposes. + return bytes(range(1, 1 + size)) + + +def test_rng_callback(): + with RngCryptoCallback(10): + rng = Random(device_id=10) + + random = rng.byte() + assert random == b"\01" + + random = rng.bytes(1) + assert random == b"\01" + + random = rng.bytes(3) + assert random == b"\01\02\03" diff --git a/wolfcrypt/__init__.py b/wolfcrypt/__init__.py index 3578b26..b126b0f 100644 --- a/wolfcrypt/__init__.py +++ b/wolfcrypt/__init__.py @@ -33,7 +33,7 @@ __copyright__ = "Copyright (C) 2006-2026 wolfSSL Inc" __all__ = [ "__title__", "__summary__", "__uri__", "__version__", "__wolfssl_version__", "__author__", "__email__", "__license__", "__copyright__", - "ciphers", "hashes", "random", "pwdbased" + "ciphers", "hashes", "random", "pwdbased", "cryptocb" ] import os @@ -46,8 +46,19 @@ top_level_py = os.path.basename(sys.argv[0]) if top_level_py not in ["setup.py", "build_ffi.py"]: from wolfcrypt._ffi import ffi as _ffi from wolfcrypt._ffi import lib as _lib + from wolfcrypt.cryptocb import CryptoCallback from wolfcrypt.exceptions import WolfCryptApiError + _lib.wolfCrypt_Init() + + if _lib.CRYPTO_CB_ENABLED: + @_ffi.def_extern() + def py_wc_crypto_callback(device_id: int, info: _ffi.CData, ctx: _ffi.CData) -> int: + if ctx == _ffi.NULL: + return _lib.CRYPTOCB_UNAVAILABLE + crypto_cb: CryptoCallback = _ffi.from_handle(ctx) + return crypto_cb.callback(device_id, info) + if hasattr(_lib, 'WC_RNG_SEED_CB_ENABLED'): if _lib.WC_RNG_SEED_CB_ENABLED: ret = _lib.wc_SetSeed_Cb(_ffi.addressof(_lib, "wc_GenerateSeed")) # ty: ignore[no-matching-overload] diff --git a/wolfcrypt/cryptocb.py b/wolfcrypt/cryptocb.py new file mode 100644 index 0000000..97b61d2 --- /dev/null +++ b/wolfcrypt/cryptocb.py @@ -0,0 +1,122 @@ +# cryptocb.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 + +# pylint: disable=no-member,no-name-in-module + +from __future__ import annotations + +from typing import Final +from typing_extensions import Self + +from wolfcrypt._ffi import ffi as _ffi +from wolfcrypt._ffi import lib as _lib + +from wolfcrypt.exceptions import WolfCryptError + +ALGO_TYPE_NAME: Final = { + _lib.WC_ALGO_TYPE_HASH: "hash", + _lib.WC_ALGO_TYPE_CIPHER: "cipher", + _lib.WC_ALGO_TYPE_RNG: "rng", + _lib.WC_ALGO_TYPE_SEED: "seed", +} + +HASH_TYPE_NAME: Final = { + _lib.WC_HASH_TYPE_SHA: "SHA1", + _lib.WC_HASH_TYPE_SHA256: "SHA256", + _lib.WC_HASH_TYPE_SHA384: "SHA384", + _lib.WC_HASH_TYPE_SHA512: "SHA512", + _lib.WC_HASH_TYPE_SHA3_256: "SHA3_256", + _lib.WC_HASH_TYPE_SHA3_384: "SHA3_384", + _lib.WC_HASH_TYPE_SHA3_512: "SHA3_512", +} + +DIGEST_SIZE: Final = { + _lib.WC_HASH_TYPE_SHA: 20, + _lib.WC_HASH_TYPE_SHA256: 32, + _lib.WC_HASH_TYPE_SHA384: 48, + _lib.WC_HASH_TYPE_SHA512: 64, + _lib.WC_HASH_TYPE_SHA3_256: 32, + _lib.WC_HASH_TYPE_SHA3_384: 48, + _lib.WC_HASH_TYPE_SHA3_512: 64, +} + + +if _lib.CRYPTO_CB_ENABLED: + class CryptoCallback: + def __init__(self, device_id: int): + self.device_id = device_id + self.ctx = _ffi.new_handle(self) + ret = _lib.wc_CryptoCb_RegisterDevice(device_id, _lib.py_wc_crypto_callback, self.ctx) + if ret < 0: # pragma: no cover + raise WolfCryptError(f"CryptoCb device registration error ({ret})") + + def __enter__(self) -> Self: + return self + + def __exit__(self, exc_type, exc_value, traceback) -> None: + self._unregister() + + def __del__(self) -> None: + self._unregister() + + def callback(self, device_id: int, info: _ffi.CData) -> int: + print(f"{device_id=} algo = {ALGO_TYPE_NAME[info.algo_type]}") + # _lib.wc_CryptoCb_InfoString(info) + try: + if info.algo_type == _lib.WC_ALGO_TYPE_HASH: + print(f"hash = {HASH_TYPE_NAME[info.hash.type]}") + print(f"{info.hash.data=} {info.hash.data_size=} {info.hash.digest=} {info.hash.u.sha256=}") + if info.hash.digest == _ffi.NULL: + self.hash_update_callback(device_id, info.hash.type, bytes(_ffi.buffer(info.hash.data, info.hash.data_size))) + else: + digest = self.hash_finalize_callback(device_id, info.hash.type) + _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) + _ffi.buffer(info.rng.out, info.rng.sz)[:] = out + return 0 + return _lib.CRYPTOCB_UNAVAILABLE + except NotImplementedError: + return _lib.CRYPTOCB_UNAVAILABLE + + def rng_callback(self, device_id: int, rng, size: int) -> bytes: + raise NotImplementedError + + def hash_update_callback(self, device_id: int, hash_type: int, data: bytes) -> None: + print("hash_update_callback") + raise NotImplementedError + + def hash_finalize_callback(self, device_id: int, hash_type: int) -> bytes: + print("hash_finalize_callback") + raise NotImplementedError + + def cipher_callback(self, device_id: int) -> None: + raise NotImplementedError + + def _unregister(self) -> None: + _lib.wc_CryptoCb_UnRegisterDevice(self.device_id) + + @classmethod + def default_device_id(cls) -> int: + return _lib.wc_CryptoCb_DefaultDevID()