Add initial support for crypto callbacks.

pull/114/head
Robert de Vries 2026-04-08 15:52:05 +02:00 committed by David Garske
parent cb9885ba53
commit e57bcfe097
4 changed files with 302 additions and 1 deletions

View File

@ -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 <wolfssl/wolfcrypt/chacha20_poly1305.h>
#include <wolfssl/wolfcrypt/wc_mlkem.h>
#include <wolfssl/wolfcrypt/wc_mldsa.h>
#include <wolfssl/wolfcrypt/cryptocb.h>
"""
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.

View File

@ -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"

View File

@ -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]

View File

@ -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()