diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml index 3e0c0f3..c46f061 100644 --- a/.github/workflows/python-app.yml +++ b/.github/workflows/python-app.yml @@ -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 diff --git a/scripts/build_ffi.py b/scripts/build_ffi.py index 82087b3..3b4adb3 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,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. diff --git a/tests/test_cryptocb.py b/tests/test_cryptocb.py new file mode 100644 index 0000000..21ecfae --- /dev/null +++ b/tests/test_cryptocb.py @@ -0,0 +1,101 @@ +# 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 + +import pytest +from typing_extensions import override + +from wolfcrypt.exceptions import WolfCryptApiError +from wolfcrypt._ffi import lib as _lib +from wolfcrypt.random import Random + +if _lib.SHA_ENABLED: + from wolfcrypt.hashes import Sha + +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(): + # 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("I16x", len(b"".join(self.data))) + + +if _lib.SHA_ENABLED: + def test_hash_callback(): + with HashCryptoCallback(11): + sha = Sha(device_id=11) + sha.update(bytes(10)) + sha.update(bytes(5)) + digest = sha.digest() + assert digest == struct.pack("I16x", 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() diff --git a/wolfcrypt/__init__.py b/wolfcrypt/__init__.py index 3578b26..10893fb 100644 --- a/wolfcrypt/__init__.py +++ b/wolfcrypt/__init__.py @@ -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 @@ __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] diff --git a/wolfcrypt/_ffi/lib.pyi b/wolfcrypt/_ffi/lib.pyi index ca50324..f429d00 100644 --- a/wolfcrypt/_ffi/lib.pyi +++ b/wolfcrypt/_ffi/lib.pyi @@ -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: ... diff --git a/wolfcrypt/ciphers.py b/wolfcrypt/ciphers.py index b7b8c05..87a1fac 100644 --- a/wolfcrypt/ciphers.py +++ b/wolfcrypt/ciphers.py @@ -460,7 +460,7 @@ def __init__(self, key: BytesOrStr, IV: BytesOrStr, tag_bytes: int = 16) -> None 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, _lib.INVALID_DEVID) if ret < 0: raise WolfCryptApiError("AES init error", ret) self._init_done = True diff --git a/wolfcrypt/cryptocb.py b/wolfcrypt/cryptocb.py new file mode 100644 index 0000000..b1c18d8 --- /dev/null +++ b/wolfcrypt/cryptocb.py @@ -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() diff --git a/wolfcrypt/exceptions.py b/wolfcrypt/exceptions.py index ddebc7a..4493499 100644 --- a/wolfcrypt/exceptions.py +++ b/wolfcrypt/exceptions.py @@ -41,6 +41,7 @@ def __init__(self, message: str, err_code: int) -> None: :param message: error message :param err_code: WolfCrypt error code """ + self.err_code = err_code err_string = error_string(err_code) if err_string: diff --git a/wolfcrypt/hashes.py b/wolfcrypt/hashes.py index 76180af..3b952e7 100644 --- a/wolfcrypt/hashes.py +++ b/wolfcrypt/hashes.py @@ -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 @@ def __init__(self, string: BytesOrStr | None = None) -> None: 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 @@ def __del__(self) -> None: 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 @@ def __del__(self) -> None: 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 @@ def __del__(self) -> None: 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 @@ def __del__(self) -> None: 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 @@ def __del__(self) -> None: ): 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 @@ def __init__(self, string: BytesOrStr | None = None, size: int = SHA3_384_DIGEST @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 @@ def copy(self) -> Sha3: 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 @@ def __del__(self) -> None: 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 @@ def __init__(self, key: BytesOrStr, string: BytesOrStr | None = None) -> None: self.update(string) @override - def _init(self) -> int: + def _init(self, device_id: int) -> int: return -1 @override @@ -503,8 +503,8 @@ def new(cls, key: BytesOrStr, string: BytesOrStr | None = None) -> _Hash: # pyl @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, diff --git a/wolfcrypt/random.py b/wolfcrypt/random.py index c29844c..d259507 100644 --- a/wolfcrypt/random.py +++ b/wolfcrypt/random.py @@ -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 *")