Add initial support for crypto callbacks.

roberthdevries · roberthdevries · commit 61356282cb5d · 2026-04-27T15:29:26.000+02:00
diff --git a/scripts/build_ffi.py b/scripts/build_ffi.py
@@ -238,6 +238,10 @@ def make_flags(prefix, fips):
         # ML-DSA
         flags.append("--enable-dilithium")
 
+        # 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")
@@ -378,6 +382,7 @@ def get_features(local_wolfssl, features):
     features["ML_DSA"] = 1 if '#define HAVE_DILITHIUM'  in defines else 0
     features["ML_KEM"] = 1 if '#define WOLFSSL_HAVE_MLKEM'  in defines else 0
     features["HKDF"] = 1 if "#define HAVE_HKDF" in defines else 0
+    features["CRYPTO_CB"] = 1 if "#define WOLF_CRYPTO_CB" in defines else 0
 
     if '#define HAVE_FIPS' in defines:
         if not fips:
@@ -456,6 +461,7 @@ def build_ffi(local_wolfssl, features):
         #include <wolfssl/wolfcrypt/mlkem.h>
         #include <wolfssl/wolfcrypt/wc_mlkem.h>
         #include <wolfssl/wolfcrypt/dilithium.h>
+        #include <wolfssl/wolfcrypt/cryptocb.h>
     """
 
     init_source_string = """
@@ -497,6 +503,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 CRYPTO_CB_ENABLED = """ + str(features["CRYPTO_CB"]) + """;
     """
 
     ffibuilder.set_source( "wolfcrypt._ffi", init_source_string,
@@ -537,13 +544,16 @@ def build_ffi(local_wolfssl, features):
         extern int ML_KEM_ENABLED;
         extern int ML_DSA_ENABLED;
         extern int HKDF_ENABLED;
+        extern int CRYPTO_CB_ENABLED;
 
         typedef unsigned char byte;
         typedef unsigned int word32;
 
         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);
@@ -1331,6 +1341,110 @@ def build_ffi(local_wolfssl, features):
         int wc_MlDsaKey_GetSigLen(MlDsaKey* key, int* len);
         """
 
+    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):
@@ -1365,6 +1479,7 @@ def main(ffibuilder):
         "ML_KEM": 1,
         "ML_DSA": 1,
         "HKDF": 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
@@ -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
@@ -33,7 +33,7 @@
 __all__ = [
     "__title__", "__summary__", "__uri__", "__version__",
     "__author__", "__email__", "__license__", "__copyright__",
-    "ciphers", "hashes", "random", "pwdbased"
+    "ciphers", "hashes", "random", "pwdbased", "cryptocb"
 ]
 
 import os
@@ -46,8 +46,19 @@
 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 WolfCryptError
 
+    _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"))
diff --git a/wolfcrypt/cryptocb.py 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()
