Skip to content

Add initial support for crypto callbacks. #114

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
roberthdevries wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Add initial support for crypto callbacks. #114

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6135628
Add initial support for crypto callbacks.
roberthdevries Apr 8, 2026
e10a1d3
Addressed review comments
roberthdevries Apr 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 115 additions & 0 deletions scripts/build_ffi.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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 = """
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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;
"""
Comment thread
JeremiahM37 marked this conversation as resolved.
Outdated
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):
Expand Down Expand Up @@ -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.
Expand Down
53 changes: 53 additions & 0 deletions tests/test_cryptocb.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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"
13 changes: 12 additions & 1 deletion wolfcrypt/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
__all__ = [
"__title__", "__summary__", "__uri__", "__version__",
"__author__", "__email__", "__license__", "__copyright__",
"ciphers", "hashes", "random", "pwdbased"
"ciphers", "hashes", "random", "pwdbased", "cryptocb"
]

import os
Expand All @@ -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()
Comment thread
JeremiahM37 marked this conversation as resolved.
Outdated

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"))
Expand Down
122 changes: 122 additions & 0 deletions wolfcrypt/cryptocb.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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:
Copy link
Copy Markdown
Contributor

@JeremiahM37 JeremiahM37 Apr 28, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 [High] Double unregister of device on context-manager exit + GC

Both __exit__ and __del__ call self._unregister(). When used as a context manager (the test does exactly this: with RngCryptoCallback(10): ...), the device is unregistered on __exit__ and again later when the object is garbage-collected. If a second CryptoCallback was registered with the same device_id between exit and GC (a reasonable usage pattern, e.g., in successive tests), the GC-time unregister will rip out the live registration belonging to the new object, causing test pollution / NULL callback dispatch. There is also no flag to record "already unregistered."

def __exit__(self, exc_type, exc_value, traceback) -> None:
    self._unregister()

def __del__(self) -> None:
    self._unregister()

def _unregister(self) -> None:
    _lib.wc_CryptoCb_UnRegisterDevice(self.device_id)

Recommendation: Track registration state with a _registered flag, set it to False after unregistering, and follow the same try/except AttributeError pattern used by Random.__del__ in wolfcrypt/random.py to avoid errors during interpreter shutdown.

Copy link
Copy Markdown
Contributor Author

@roberthdevries roberthdevries Apr 28, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The tracking of the registration is done in the wolfcrypt library itself. Unregistering a device twice is a noop.
__del__ is not certain to be called at any time soon (unless a gc.collect() call is done explicitly) so if more fine-grained control is required to control the lifespan of the callback, a context manager is the typical solution in Python.

So the unregister in __exit__ must not be removed.

Copy link
Copy Markdown
Contributor

@JeremiahM37 JeremiahM37 Apr 29, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

wc_CryptoCb_UnRegisterDevice looks up by devId only — no ownership check (cryptocb.c:428). The double-unregister is only a no-op if the slot is still INVALID_DEVID at the second call. In the cross-instance case (cb1 exits → cb2 registers same devId → cb1 GC'd), cb1's del will find cb2's live slot and ClearDev it. A _registered flag in del avoids this without removing exit's unregister.

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]}")
Comment thread
JeremiahM37 marked this conversation as resolved.
Outdated
# _lib.wc_CryptoCb_InfoString(info)
try:
if info.algo_type == _lib.WC_ALGO_TYPE_HASH:
Comment thread
JeremiahM37 marked this conversation as resolved.
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
Comment thread
JeremiahM37 marked this conversation as resolved.
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
Comment thread
JeremiahM37 marked this conversation as resolved.
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()
Loading