WolfSSL
/
wolfcrypt-py
mirror of https://github.com/wolfSSL/wolfcrypt-py.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Merge 55971eb522 into cb9885ba53

pull/114/merge
Robert de Vries 2026-08-12 16:10:51 +05:30 committed by GitHub
parent cb9885ba53 55971eb522
commit 0df42b70e7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 565 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.github/workflows/python-app.yml vendored
View File

@ -40,10 +40,6 @@ jobs:
uv run --no-sync pytest tests
build-no-pqc:
# Regression coverage for issue #2659: INVALID_DEVID is only declared
# in the CFFI cdef when ML_KEM or ML_DSA is enabled, so a wolfSSL build
# without those features must still import wolfcrypt.random successfully.
runs-on: ubuntu-latest
env:
USE_LOCAL_WOLFSSL: ${{ github.workspace }}/wolfssl-install
@ -93,7 +89,5 @@ jobs:
run: |
uv run --no-sync python -c "from wolfcrypt._ffi import lib as _lib; assert _lib.ML_KEM_ENABLED == 0, 'ML-KEM should be disabled'"
uv run --no-sync python -c "from wolfcrypt._ffi import lib as _lib; assert _lib.ML_DSA_ENABLED == 0, 'ML-DSA should be disabled'"
- name: Import smoke (regression for INVALID_DEVID)
run: uv run --no-sync python -c "from wolfcrypt.random import Random; Random()"
- name: Run tests
run: uv run --no-sync pytest tests

153
scripts/build_ffi.py
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,9 @@ 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;
static const int INVALID_DEVID;
typedef unsigned char byte;
typedef unsigned int word32;
@ -565,6 +575,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);
@ -860,7 +872,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*);
@ -870,7 +882,7 @@ def build_ffi(local_wolfssl, features):
if features["SHA256"]:
cdef += """
typedef struct { ...; } wc_Sha256;
int wc_InitSha256(wc_Sha256*);
int wc_InitSha256_ex(wc_Sha256*, void*, int);
int wc_Sha256Update(wc_Sha256*, const byte*, word32);
int wc_Sha256Final(wc_Sha256*, byte*);
void wc_Sha256Free(wc_Sha256*);
@ -880,7 +892,7 @@ def build_ffi(local_wolfssl, features):
if features["SHA384"]:
cdef += """
typedef struct { ...; } wc_Sha384;
int wc_InitSha384(wc_Sha384*);
int wc_InitSha384_ex(wc_Sha384*, void*, int);
int wc_Sha384Update(wc_Sha384*, const byte*, word32);
int wc_Sha384Final(wc_Sha384*, byte*);
void wc_Sha384Free(wc_Sha384*);
@ -891,7 +903,7 @@ def build_ffi(local_wolfssl, features):
cdef += """
typedef struct { ...; } wc_Sha512;
int wc_InitSha512(wc_Sha512*);
int wc_InitSha512_ex(wc_Sha512*, void*, int);
int wc_Sha512Update(wc_Sha512*, const byte*, word32);
int wc_Sha512Final(wc_Sha512*, byte*);
void wc_Sha512Free(wc_Sha512*);
@ -1304,11 +1316,6 @@ def build_ffi(local_wolfssl, features):
int wolfCrypt_GetPrivateKeyReadEnable_fips(enum wc_KeyType);
"""
if features["ML_KEM"] or features["ML_DSA"]:
cdef += """
static const int INVALID_DEVID;
"""
if features["ML_KEM"]:
cdef += """
static const int WC_ML_KEM_512;
@ -1363,6 +1370,133 @@ 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 {
"""
# The following block is commented out as there are issues with cffi code generation for the
# wc_CryptoInfo structure with two layers of anonymous unions.
# Uncommenting more parts of the cipher struct causes errors regarding conflicting struct sizes of
# other parts of the wc_CryptoInfo struct.
# Cffi cdef and the compiler seem to disagree.
#
# cdef += """
# 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 {
"""
if features["SHA"]:
cdef += """
wc_Sha* sha1;
"""
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;
"""
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 +1534,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.

145
tests/test_cryptocb.py 100644
View File

@ -0,0 +1,145 @@
# 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
# ty: ignore[possibly-missing-import]
from __future__ import annotations
import struct
from binascii import hexlify as b2h
import pytest
from typing_extensions import override
from wolfcrypt._ffi import lib as _lib
from wolfcrypt.exceptions import WolfCryptApiError
from wolfcrypt.hashes import _Sha
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, DIGEST_SIZE
SHA_CLASSES: list[type[_Sha]] = []
if _lib.SHA_ENABLED:
from wolfcrypt.hashes import Sha
SHA_CLASSES.append(Sha)
if _lib.SHA256_ENABLED:
from wolfcrypt.hashes import Sha256
SHA_CLASSES.append(Sha256)
if _lib.SHA384_ENABLED:
from wolfcrypt.hashes import Sha384
SHA_CLASSES.append(Sha384)
if _lib.SHA512_ENABLED:
from wolfcrypt.hashes import Sha512
SHA_CLASSES.append(Sha512)
if _lib.SHA3_ENABLED:
from wolfcrypt.hashes import Sha3
SHA_CLASSES.append(Sha3)
def test_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):
@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))
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"
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(f"I{DIGEST_SIZE[hash_type] - 4}x", len(b"".join(self.data)))
@pytest.mark.parametrize("sha_cls", SHA_CLASSES)
def test_hash_callback(sha_cls: type[_Sha]) -> None:
with HashCryptoCallback(11):
sha = sha_cls(device_id=11)
sha.update(bytes(10))
sha.update(bytes(5))
digest = sha.digest()
digest_size = sha.digest_size
assert digest_size is not None
assert digest == struct.pack(f"I{digest_size - 4}x", 15)
class BadHashCryptoCallback(CryptoCallback):
@override
def hash_finalize_callback(self, device_id: int, hash_type: int) -> bytes:
return bytes(1) # bad hash length
if _lib.SHA_ENABLED:
def test_hash_callback_failure():
with BadHashCryptoCallback(12):
sha = Sha(device_id=12)
sha.update(bytes(10))
with pytest.raises(WolfCryptApiError):
sha.digest()
if _lib.AESGCM_STREAM_ENABLED:
from wolfcrypt.ciphers import AesGcmStream
def test_crypto_callback_aes_gcm_stream_encrypt():
"""Verify that the crypto callback that does not support AesGcmStream works as if not callback was installed."""
with CryptoCallback(13):
"""Known answer encrypt-decrypt test with default authentication tag size of 16 bytes"""
key = "fedcba9876543210"
iv = "0123456789abcdef"
gcm = AesGcmStream(key, iv, device_id=13)
buf = gcm.encrypt("hello world")
authTag = gcm.final()
assert authTag is not None
assert b2h(authTag) == bytes('ac8fcee96dc6ef8e5236da19b6197d2e', 'utf-8')
assert b2h(buf) == bytes('5ba7d42e1bf01d7998e932', "utf-8")

40
wolfcrypt/__init__.py
View File

@ -18,6 +18,11 @@
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1335, USA
from __future__ import annotations
import os
import sys
from wolfcrypt._version import __version__, __wolfssl_version__
__title__ = "wolfcrypt"
@ -33,21 +38,48 @@ __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
import sys
top_level_py = os.path.basename(sys.argv[0])
# The code below is intended to only be used after the CFFI is built, so we
# don't want it invoked whilst building the CFFI with build_ffi.py or setup.py.
if top_level_py not in ["setup.py", "build_ffi.py"]:
import logging
from typing import TYPE_CHECKING
from wolfcrypt._ffi import ffi as _ffi
from wolfcrypt._ffi import lib as _lib
if TYPE_CHECKING:
if _lib.CRYPTO_CB_ENABLED:
from types import TracebackType
from wolfcrypt.cryptocb import CryptoCallback # ty: ignore[possibly-missing-import]
from wolfcrypt.exceptions import WolfCryptApiError
log = logging.getLogger("wolfcrypt")
# 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)
if _lib.CRYPTO_CB_ENABLED:
def crypto_cb_error_handler(exception: Exception, exc_value: Exception, traceback: TracebackType | None) -> int: # noqa: ANN401
if isinstance(exc_value, WolfCryptApiError):
return exc_value.err_code
log.error(exc_value)
return -1
@_ffi.def_extern(onerror=crypto_cb_error_handler)
def py_wc_crypto_callback(device_id: int, info: _lib.wc_CryptoInfo, 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]

64
wolfcrypt/_ffi/lib.pyi
View File

@ -18,9 +18,12 @@
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
from _cffi_backend import FFI
from collections.abc import Callable
from dataclasses import dataclass
from typing import TypeAlias
from _cffi_backend import FFI
INVALID_DEVID: int
AES_ENABLED: int
@ -30,6 +33,7 @@ ASN_ENABLED: int
CHACHA_ENABLED: int
CHACHA_STREAM_ENABLED: int
CHACHA20_POLY1305_ENABLED: int
CRYPTO_CB_ENABLED: int
DES3_ENABLED: int
ECC_ENABLED: int
ED25519_ENABLED: int
@ -282,6 +286,20 @@ WC_MGF1SHA256: int
WC_MGF1SHA384: int
WC_MGF1SHA512: int
WC_ALGO_TYPE_NONE: int
WC_ALGO_TYPE_HASH: int
WC_ALGO_TYPE_CIPHER: int
WC_ALGO_TYPE_PK: int
WC_ALGO_TYPE_RNG: int
WC_ALGO_TYPE_SEED: int
WC_ALGO_TYPE_HMAC: int
WC_ALGO_TYPE_CMAC: int
WC_ALGO_TYPE_CERT: int
WC_ALGO_TYPE_KDF: int
WC_ALGO_TYPE_COPY: int
WC_ALGO_TYPE_FREE: int
WC_ALGO_TYPE_MAX: int
WC_HASH_TYPE_NONE: int
WC_HASH_TYPE_MD2: int
WC_HASH_TYPE_MD4: int
@ -328,8 +346,9 @@ RNG: TypeAlias = FFI.CData
def wc_SetSeed_Cb(cb: FFI.CData) -> int: ...
def wolfCrypt_SetPrivateKeyReadEnable_fips(enable: int, key_type: int) -> int: ...
def wc_GetErrorString(error: int) -> FFI.CData: ...
def wolfCrypt_Init() -> int: ...
def wc_InitRngNonce_ex(rng: RNG, nonce: bytes, nonce_size: int, heap: FFI.CData, device_id: int) -> int: ...
def wc_InitRngNonce_ex(rng: RNG, nonce: bytes, nonce_size: int, heap: FFI.CData, dev_id: int) -> int: ...
def wc_RNG_GenerateByte(rng: RNG, buffer: FFI.CData) -> int: ...
def wc_RNG_GenerateBlock(rng: RNG, buffer: FFI.CData, len: int) -> int: ...
def wc_RNG_DRBG_Reseed(rng: RNG, seed: bytes, seed_size: int) -> int: ...
@ -350,25 +369,25 @@ 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, dev_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: ...
def wc_ShaFree(obj: FFI.CData) -> None: ...
def wc_InitSha256(obj: FFI.CData) -> int: ...
def wc_InitSha256_ex(obj: FFI.CData, heap: FFI.CData, dev_id: int) -> int: ...
def wc_Sha256Copy(src: FFI.CData, dst: FFI.CData) -> int: ...
def wc_Sha256Update(obj: FFI.CData, data: bytes, size: int) -> int: ...
def wc_Sha256Final(obj: FFI.CData, ret: FFI.CData) -> int: ...
def wc_Sha256Free(obj: FFI.CData) -> None: ...
def wc_InitSha384(obj: FFI.CData) -> int: ...
def wc_InitSha384_ex(obj: FFI.CData, heap: FFI.CData, dev_id: int) -> int: ...
def wc_Sha384Copy(src: FFI.CData, dst: FFI.CData) -> int: ...
def wc_Sha384Update(obj: FFI.CData, data: bytes, size: int) -> int: ...
def wc_Sha384Final(obj: FFI.CData, ret: FFI.CData) -> int: ...
def wc_Sha384Free(obj: FFI.CData) -> None: ...
def wc_InitSha512(obj: FFI.CData) -> int: ...
def wc_InitSha512_ex(obj: FFI.CData, heap: FFI.CData, dev_id: int) -> int: ...
def wc_Sha512Copy(src: FFI.CData, dst: FFI.CData) -> int: ...
def wc_Sha512Update(obj: FFI.CData, data: bytes, size: int) -> int: ...
def wc_Sha512Final(obj: FFI.CData, ret: FFI.CData) -> int: ...
@ -569,3 +588,36 @@ def wc_dilithium_sign_msg_with_seed(msg: bytes, msg_len: int, sig: BytePtr, sig_
def wc_MlDsaKey_GetPrivLen(key: DilithiumKey, len: IntPtr) -> int: ...
def wc_MlDsaKey_GetPubLen(key: DilithiumKey, len: IntPtr) -> int: ...
def wc_MlDsaKey_GetSigLen(key: DilithiumKey, len: IntPtr) -> int: ...
@dataclass
class wc_HashInfo:
type: int
data: FFI.CData
data_size: int
digest: FFI.CData
sha1: FFI.CData
sha256: FFI.CData
sha384: FFI.CData
sha512: FFI.CData
sha3: FFI.CData
ctx: FFI.CData
@dataclass
class wc_RngInfo:
rng: RNG
out: FFI.CData
sz: int
@dataclass
class wc_CryptoInfo:
algo_type: int
hash: wc_HashInfo
rng: wc_RngInfo
def wc_CryptoCb_RegisterDevice(dev_id: int, cb: Callable[[int, wc_CryptoInfo, FFI.CData], int], ctx: FFI.CData) -> int: ...
def wc_CryptoCb_UnRegisterDevice(dev_id: int) -> None: ...
def wc_CryptoCb_DefaultDevID() -> int: ...
def py_wc_crypto_callback(dev_id: int, info: wc_CryptoInfo, ctx: FFI.CData) -> int: ...

4
wolfcrypt/ciphers.py
View File

@ -437,7 +437,7 @@ if _lib.AESGCM_STREAM_ENABLED:
# making sure _lib.wc_AesFree outlives Aes instances
_delete = staticmethod(_lib.wc_AesFree)
def __init__(self, key: BytesOrStr, IV: BytesOrStr, tag_bytes: int = 16) -> None:
def __init__(self, key: BytesOrStr, IV: BytesOrStr, tag_bytes: int = 16, device_id: int = _lib.INVALID_DEVID) -> None:
"""
tag_bytes is the number of bytes to use for the authentication tag during encryption
"""
@ -460,7 +460,7 @@ if _lib.AESGCM_STREAM_ENABLED:
raise ValueError(f"key must be {self._key_sizes} in length, not {len(key)}")
self._init_done = False
self._native_object = _ffi.new(self._native_type)
ret = _lib.wc_AesInit(self._native_object, _ffi.NULL, -2)
ret = _lib.wc_AesInit(self._native_object, _ffi.NULL, device_id)
if ret < 0:
raise WolfCryptApiError("AES init error", ret)
self._init_done = True

153
wolfcrypt/cryptocb.py 100644
View File

@ -0,0 +1,153 @@
# 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
import logging
from collections import defaultdict
from typing import TYPE_CHECKING, 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
if TYPE_CHECKING:
from types import TracebackType
log = logging.getLogger(__name__)
if _lib.CRYPTO_CB_ENABLED:
ALGO_TYPE_NAME: Final = defaultdict(
lambda: "unknown",
{
_lib.WC_ALGO_TYPE_NONE: "none",
_lib.WC_ALGO_TYPE_HASH: "hash",
_lib.WC_ALGO_TYPE_CIPHER: "cipher",
_lib.WC_ALGO_TYPE_PK: "pk",
_lib.WC_ALGO_TYPE_RNG: "rng",
_lib.WC_ALGO_TYPE_SEED: "seed",
_lib.WC_ALGO_TYPE_HMAC: "hmac",
_lib.WC_ALGO_TYPE_CMAC: "cmac",
_lib.WC_ALGO_TYPE_CERT: "cert",
_lib.WC_ALGO_TYPE_KDF: "kdf",
_lib.WC_ALGO_TYPE_COPY: "copy",
_lib.WC_ALGO_TYPE_FREE: "free",
_lib.WC_ALGO_TYPE_MAX: "max",
},
)
HASH_TYPE_NAME: Final = defaultdict(
lambda: "unknown",
{
_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,
}
class CryptoCallback:
def __init__(self, device_id: int) -> None:
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: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None
) -> bool:
self._unregister()
return False
def __del__(self) -> None:
self._unregister()
def callback(self, device_id: int, info: _lib.wc_CryptoInfo) -> int:
log.debug("device_id=%d algo_type = %s", device_id, ALGO_TYPE_NAME[info.algo_type])
try:
if info.algo_type == _lib.WC_ALGO_TYPE_HASH:
if info.hash.type not in DIGEST_SIZE:
return _lib.CRYPTOCB_UNAVAILABLE
log.debug("hash_type = %s", HASH_TYPE_NAME[info.hash.type])
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)
if len(digest) != DIGEST_SIZE[info.hash.type]:
raise ValueError(
f"Generated digest is expected to be {DIGEST_SIZE[info.hash.type]} bytes long, "
f"but is {len(digest)} bytes long"
)
_ffi.buffer(info.hash.digest, DIGEST_SIZE[info.hash.type])[:] = digest
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:
raise ValueError(
f"Generated random is expected to be {info.rng.sz} bytes long, but is {len(out)} bytes long"
)
_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: _lib.RNG, size: int) -> bytes:
raise NotImplementedError
def hash_update_callback(self, device_id: int, hash_type: int, data: bytes) -> None:
raise NotImplementedError
def hash_finalize_callback(self, device_id: int, hash_type: int) -> bytes:
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()

1
wolfcrypt/exceptions.py
View File

@ -41,6 +41,7 @@ class WolfCryptApiError(WolfCryptError):
:param message: error message
:param err_code: WolfCrypt error code
"""
self.err_code = err_code
err_string = error_string(err_code)
if err_string:

50
wolfcrypt/hashes.py
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,8 +232,8 @@ if _lib.SHA256_ENABLED:
self._delete(self._native_object)
@override
def _init(self) -> int:
return _lib.wc_InitSha256(self._native_object)
def _init(self, device_id: int) -> int:
return _lib.wc_InitSha256_ex(self._native_object, _ffi.NULL, device_id)
@override
def _update(self, data: bytes) -> int:
@ -263,8 +263,8 @@ if _lib.SHA384_ENABLED:
self._delete(self._native_object)
@override
def _init(self) -> int:
return _lib.wc_InitSha384(self._native_object)
def _init(self, device_id: int) -> int:
return _lib.wc_InitSha384_ex(self._native_object, _ffi.NULL, device_id)
@override
def _update(self, data: bytes) -> int:
@ -294,8 +294,8 @@ if _lib.SHA512_ENABLED:
self._delete(self._native_object)
@override
def _init(self) -> int:
return _lib.wc_InitSha512(self._native_object)
def _init(self, device_id: int) -> int:
return _lib.wc_InitSha512_ex(self._native_object, _ffi.NULL, device_id)
@override
def _update(self, data: bytes) -> int:
@ -349,13 +349,13 @@ if _lib.SHA3_ENABLED:
):
self._delete(self._native_object)
def __init__(self, string: BytesOrStr | None = None, size: int = SHA3_384_DIGEST_SIZE) -> None: # pylint: disable=W0231
def __init__(self, string: BytesOrStr | None = None, size: int = SHA3_384_DIGEST_SIZE, device_id:int = _lib.INVALID_DEVID) -> None: # pylint: disable=W0231
self._native_object = _ffi.new(self._native_type)
self._shallow_copy = False
self.digest_size = size
self._delete = self._SHA3_FREE.get(size)
self._copy = self._SHA3_COPY.get(size)
ret = self._init()
ret = self._init(device_id)
if ret < 0: # pragma: no cover
raise WolfCryptApiError("Sha3 init error", ret)
if string:
@ -363,8 +363,8 @@ if _lib.SHA3_ENABLED:
@override
@classmethod
def new(cls, string: BytesOrStr | None = None, size: int = SHA3_384_DIGEST_SIZE) -> Sha3:
return cls(string, size)
def new(cls, string: BytesOrStr | None = None, size: int = SHA3_384_DIGEST_SIZE, device_id: int = _lib.INVALID_DEVID) -> Sha3:
return cls(string, size, device_id)
@override
def copy(self) -> Sha3:
@ -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
@ -470,12 +470,12 @@ if _lib.HMAC_ENABLED:
if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False):
self._delete(self._native_object)
def __init__(self, key: BytesOrStr, string: BytesOrStr | None = None) -> None: # pylint: disable=W0231
def __init__(self, key: BytesOrStr, string: BytesOrStr | None = None, device_id: int = _lib.INVALID_DEVID) -> None: # pylint: disable=W0231
key = t2b(key)
self._native_object = _ffi.new(self._native_type)
self._shallow_copy = False
ret = self._hmac_init(self._type, key)
ret = self._hmac_init(self._type, key, device_id)
if ret < 0: # pragma: no cover
raise WolfCryptApiError("Hmac init error", ret)
@ -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
@ -503,8 +503,8 @@ if _lib.HMAC_ENABLED:
@abstractmethod
def _type(self) -> int: ...
def _hmac_init(self, hmac: int, key: bytes) -> int:
ret = _lib.wc_HmacInit(self._native_object, _ffi.NULL, -2)
def _hmac_init(self, hmac: int, key: bytes, device_id: int) -> int:
ret = _lib.wc_HmacInit(self._native_object, _ffi.NULL, device_id)
if ret < 0:
raise WolfCryptApiError("wc_HmacInit error", ret)
# If the key isn't set, don't call wc_HmacSetKey. This can happen,

2
wolfcrypt/random.py
View File

@ -33,7 +33,7 @@ class Random:
A Cryptographically Secure Pseudo Random Number Generator - CSPRNG
"""
def __init__(self, nonce: __builtins__.bytes = b"", device_id: int = -2) -> None:
def __init__(self, nonce: __builtins__.bytes = b"", device_id: int = _lib.INVALID_DEVID) -> None:
self._native_object: _lib.RNG | None = None
self._native_object = _ffi.new("WC_RNG *")