diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml index 3e0c0f3..0518c7d 100644 --- a/.github/workflows/python-app.yml +++ b/.github/workflows/python-app.yml @@ -40,9 +40,8 @@ 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. + # Regression coverage for issue #2659: a wolfSSL build without ML-KEM + # and ML-DSA must still import wolfcrypt.random successfully. runs-on: ubuntu-latest env: @@ -97,3 +96,61 @@ jobs: run: uv run --no-sync python -c "from wolfcrypt.random import Random; Random()" - name: Run tests run: uv run --no-sync pytest tests + + build-cryptocb-smallstack: + # Crypto callback hash copies must keep wolfCrypt's native resource + # ownership; SMALL_STACK_CACHE allocates per-object buffers that a byte + # copy would share and free twice. + runs-on: ubuntu-latest + env: + USE_LOCAL_WOLFSSL: ${{ github.workspace }}/wolfssl-install + + steps: + - uses: actions/checkout@v7 + with: + submodules: recursive + - name: Set up Python 3.10 + uses: actions/setup-python@v6 + with: + python-version: "3.10" + - name: Install uv + uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b # v8.1.0 + with: + # Install a specific version of uv. + version: "0.11.26" + - name: Install build deps + run: | + sudo apt-get update + sudo apt-get install -y autoconf automake libtool + - name: Build wolfSSL with crypto callbacks and SMALL_STACK_CACHE + run: | + cd lib/wolfssl + ./autogen.sh + ./configure --enable-cryptonly --disable-shared \ + --enable-cryptocb --enable-cryptocbutils=copy,free \ + --enable-smallstack --enable-smallstackcache \ + --enable-aes --enable-aesgcm --enable-aessiv \ + --enable-aesctr --enable-aesgcm-stream \ + --enable-des3 --enable-chacha --enable-poly1305 \ + --enable-sha --enable-sha384 --enable-sha512 --enable-sha3 \ + --enable-hkdf --enable-rsa --enable-rsapss --enable-ecc \ + --enable-ed25519 --enable-ed448 --enable-curve25519 \ + --enable-keygen --enable-pwdbased --enable-pkcs7 \ + --disable-md5 --disable-sha224 --disable-dh \ + --disable-oldtls --disable-oldnames --disable-extended-master \ + CFLAGS=-fPIC \ + --prefix=$GITHUB_WORKSPACE/wolfssl-install + make + make install + - name: Install wolfcrypt-py against the local wolfSSL + run: | + uv venv + uv pip install pytest + uv pip install -e . + - name: Check crypto callback support + run: | + uv run --no-sync python -c "from wolfcrypt._ffi import lib as _lib; assert _lib.CRYPTO_CB_COPY_ENABLED and _lib.CRYPTO_CB_FREE_ENABLED, 'copy and free callbacks should be enabled'" + - name: Run tests + env: + MALLOC_CHECK_: "3" + run: uv run --no-sync pytest tests diff --git a/scripts/build_ffi.py b/scripts/build_ffi.py index 6807edf..7e64853 100644 --- a/scripts/build_ffi.py +++ b/scripts/build_ffi.py @@ -238,6 +238,12 @@ 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") + # Hash callbacks track per-object state across wc_*Copy and wc_*Free + flags.append("--enable-cryptocbutils=copy,free") + # flags.append("EXTRACPPFLAGS=-DDEBUG_CRYPTOCB") + # disabling other configs enabled by default flags.append("--disable-oldtls") flags.append("--disable-oldnames") @@ -394,6 +400,18 @@ 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 + features["CRYPTO_CB_COPY"] = 1 if "#define WOLF_CRYPTO_CB_COPY" in defines else 0 + features["CRYPTO_CB_FREE"] = 1 if "#define WOLF_CRYPTO_CB_FREE" in defines else 0 + # wc_HmacCopy was added in wolfSSL 5.9.0 and has no feature define. + features["HMAC_COPY"] = 0 + if features["HMAC"]: + for d in include_dirs: + hmac_h = os.path.join(d, "wolfssl", "wolfcrypt", "hmac.h") + if os.path.exists(hmac_h): + with open(hmac_h) as f: + if "wc_HmacCopy" in f.read(): + features["HMAC_COPY"] = 1 if '#define HAVE_FIPS' in defines: if not fips: @@ -471,6 +489,7 @@ def build_ffi(local_wolfssl, features): #include #include #include + #include """ init_source_string = f""" @@ -515,6 +534,10 @@ 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"]}; + int CRYPTO_CB_COPY_ENABLED = {features["CRYPTO_CB_COPY"]}; + int CRYPTO_CB_FREE_ENABLED = {features["CRYPTO_CB_FREE"]}; + int HMAC_COPY_ENABLED = {features["HMAC_COPY"]}; """ ffibuilder.set_source( "wolfcrypt._ffi", init_source_string, @@ -558,6 +581,14 @@ 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; + extern int CRYPTO_CB_COPY_ENABLED; + extern int CRYPTO_CB_FREE_ENABLED; + extern int HMAC_COPY_ENABLED; + + static const int INVALID_DEVID; + static const int WC_HW_E; + static const int CRYPTOCB_UNAVAILABLE; typedef unsigned char byte; typedef unsigned int word32; @@ -565,6 +596,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); @@ -759,7 +792,6 @@ def build_ffi(local_wolfssl, features): static const int ECC_PRIVATEONLY_E; static const int EXTKEYUSAGE_E; - static const int WC_HW_E; static const int WC_HW_WAIT_E; static const int PSS_SALTLEN_E; @@ -787,7 +819,6 @@ def build_ffi(local_wolfssl, features): static const int PKCS7_NO_SIGNER_E; static const int WC_PKCS7_WANT_READ_E; - static const int CRYPTOCB_UNAVAILABLE; static const int PKCS7_SIGNEEDS_CHECK; static const int PSS_SALTLEN_RECOVER_E; static const int CHACHA_POLY_OVERFLOW; @@ -860,7 +891,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 +901,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 +911,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 +922,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*); @@ -1018,6 +1049,10 @@ def build_ffi(local_wolfssl, features): int wc_HmacFinal(Hmac*, byte*); void wc_HmacFree(Hmac*); """ + if features["HMAC_COPY"]: + cdef += """ + int wc_HmacCopy(Hmac* src, Hmac* dst); + """ if features["RSA"]: cdef += """ @@ -1304,11 +1339,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 +1393,155 @@ 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; + # """ + + # wolfSSL names the input fields in/inSz; renamed here because "in" is a + # Python keyword. The layout must still match wc_CryptoInfo in cryptocb.h. + 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; + """ + if features["CRYPTO_CB_COPY"]: + cdef += """ + struct { + int algo; /* enum wc_AlgoType */ + int type; /* enum wc_HashType for WC_ALGO_TYPE_HASH */ + void* src; + void* dst; + } copy; + """ + if features["CRYPTO_CB_FREE"]: + cdef += """ + struct { + int algo; /* enum wc_AlgoType */ + int type; /* enum wc_HashType for WC_ALGO_TYPE_HASH */ + int subType; + void* obj; + } free; + """ + cdef += """ + }; + ...; + } 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 +1579,10 @@ def main(ffibuilder): "ML_DSA_NO_CTX": 0, "HKDF": 1, "HASHDRBG": 1, + "CRYPTO_CB": 1, + "CRYPTO_CB_COPY": 1, + "CRYPTO_CB_FREE": 1, + "HMAC_COPY": 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..16873dc --- /dev/null +++ b/tests/test_cryptocb.py @@ -0,0 +1,345 @@ +# 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 gc +import hashlib +import hmac as py_hmac +import struct +from binascii import hexlify as b2h +from typing import TYPE_CHECKING, Any + +import pytest +from typing_extensions import override + +from wolfcrypt._ffi import lib as _lib +from wolfcrypt.exceptions import WolfCryptApiError, WolfCryptError +from wolfcrypt.hashes import _Sha +from wolfcrypt.random import Random + +if TYPE_CHECKING: + from collections.abc import Callable + +if not _lib.CRYPTO_CB_ENABLED: + pytest.skip("Crypto Callbacks not supported", allow_module_level=True) + +from wolfcrypt.cryptocb import CryptoCallback, DIGEST_SIZE + +HASH_CB_ENABLED = _lib.CRYPTO_CB_COPY_ENABLED and _lib.CRYPTO_CB_FREE_ENABLED +needs_hash_cb = pytest.mark.skipif(not HASH_CB_ENABLED, reason="Hash callbacks need copy and free crypto callbacks") + +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): + @override + def hash_update_callback(self, device_id: int, hash_type: int, context: dict[str, Any], data: bytes) -> None: + context.setdefault("data", []).append(data) + + @override + def hash_finalize_callback(self, device_id: int, hash_type: int, context: dict[str, Any]) -> 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(context.get("data", [])))) + + +PY_HASH: dict[int, Callable[[bytes], hashlib._Hash]] = { + _lib.WC_HASH_TYPE_SHA: hashlib.sha1, + _lib.WC_HASH_TYPE_SHA256: hashlib.sha256, + _lib.WC_HASH_TYPE_SHA384: hashlib.sha384, + _lib.WC_HASH_TYPE_SHA512: hashlib.sha512, + _lib.WC_HASH_TYPE_SHA3_256: hashlib.sha3_256, + _lib.WC_HASH_TYPE_SHA3_384: hashlib.sha3_384, + _lib.WC_HASH_TYPE_SHA3_512: hashlib.sha3_512, +} + + +class SoftwareHashCryptoCallback(CryptoCallback): + """Hashes with hashlib, so results must match wolfCrypt without a device.""" + + @override + def hash_update_callback(self, device_id: int, hash_type: int, context: dict[str, Any], data: bytes) -> None: + context["data"] = context.get("data", b"") + data + + @override + def hash_finalize_callback(self, device_id: int, hash_type: int, context: dict[str, Any]) -> bytes: + return PY_HASH[hash_type](context.get("data", b"")).digest() + + +def fake_digest(sha: _Sha, length: int) -> bytes: + assert sha.digest_size is not None + return struct.pack(f"I{sha.digest_size - 4}x", length) + + +@needs_hash_cb +@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)) + assert sha.digest() == fake_digest(sha, 15) + + +@needs_hash_cb +@pytest.mark.parametrize("sha_cls", SHA_CLASSES) +def test_hash_callback_interleaved(sha_cls: type[_Sha]) -> None: + with HashCryptoCallback(20): + first = sha_cls(device_id=20) + second = sha_cls(device_id=20) + first.update(bytes(10)) + second.update(bytes(5)) + assert first.digest() == fake_digest(first, 10) + assert second.digest() == fake_digest(second, 5) + + +@needs_hash_cb +@pytest.mark.parametrize("sha_cls", SHA_CLASSES) +def test_hash_callback_copy(sha_cls: type[_Sha]) -> None: + with HashCryptoCallback(21): + sha = sha_cls(device_id=21) + sha.update(bytes(10)) + dup = sha.copy() + dup.update(bytes(5)) + assert sha.digest() == fake_digest(sha, 10) + assert dup.digest() == fake_digest(dup, 15) + # digest() finalizes a temporary copy, so the original keeps its state + sha.update(bytes(1)) + assert sha.digest() == fake_digest(sha, 11) + assert dup.digest() == fake_digest(dup, 15) + + +@needs_hash_cb +@pytest.mark.parametrize("sha_cls", SHA_CLASSES) +def test_hash_callback_free_drops_context(sha_cls: type[_Sha]) -> None: + with HashCryptoCallback(22) as cb: + sha = sha_cls(device_id=22) + sha.update(bytes(3)) + dup = sha.copy() + assert sha.digest() == fake_digest(sha, 3) + assert len(cb._hash_contexts) == 2 + del sha, dup + gc.collect() + assert not cb._hash_contexts + + +@needs_hash_cb +@pytest.mark.parametrize("sha_cls", SHA_CLASSES) +def test_hash_callback_repeated_copy_free(sha_cls: type[_Sha]) -> None: + # Copies must own their native resources (e.g. the SMALL_STACK_CACHE W + # buffer), otherwise freeing the copies frees the original's. + with SoftwareHashCryptoCallback(24) as cb: + for _ in range(50): + sha = sha_cls(b"abc", device_id=24) + dup = sha.copy() + dup.update(b"def") + assert sha.digest() == sha_cls(b"abc").digest() + assert dup.digest() == sha_cls(b"abcdef").digest() + del sha, dup + gc.collect() + assert not cb._hash_contexts + + +class BadHashCryptoCallback(CryptoCallback): + @override + def hash_update_callback(self, device_id: int, hash_type: int, context: dict[str, Any], data: bytes) -> None: + pass + + @override + def hash_finalize_callback(self, device_id: int, hash_type: int, context: dict[str, Any]) -> bytes: + return bytes(1) # bad hash length + + +if _lib.SHA_ENABLED: + @needs_hash_cb + 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") + + +class BadRngCryptoCallback(CryptoCallback): + @override + def rng_callback(self, device_id: int, rng: _lib.RNG, size: int) -> bytes: + return bytes(1) # bad random length + + +def test_rng_callback_failure(): + with BadRngCryptoCallback(14): + rng = Random(device_id=14) + with pytest.raises(WolfCryptApiError): + rng.bytes(4) + + +def test_reregister_same_device_id(): + gc.disable() + try: + with RngCryptoCallback(15): + pass + # A stale instance collected later must not unregister the new owner. + with RngCryptoCallback(15): + gc.collect() + assert Random(device_id=15).bytes(3) == b"\01\02\03" + # Replacing a live registration hands the device id to the new instance. + first = RngCryptoCallback(15) + with RngCryptoCallback(15): + del first + gc.collect() + assert Random(device_id=15).bytes(3) == b"\01\02\03" + finally: + gc.enable() + + +class FinalizeOnlyCryptoCallback(CryptoCallback): + @override + def hash_finalize_callback(self, device_id: int, hash_type: int, context: dict[str, Any]) -> bytes: + return bytes(DIGEST_SIZE[hash_type]) + + +if _lib.SHA256_ENABLED: + def test_hash_partial_override_falls_back(): + with FinalizeOnlyCryptoCallback(16): + sha = Sha256(b"abc", device_id=16) + assert sha.digest() == hashlib.sha256(b"abc").digest() + + @pytest.mark.skipif(HASH_CB_ENABLED, reason="Copy and free crypto callbacks are available") + def test_hash_without_copy_free_falls_back(): + with HashCryptoCallback(23): + sha = Sha256(b"abc", device_id=23) + assert sha.digest() == hashlib.sha256(b"abc").digest() + + @needs_hash_cb + def test_hash_new_device_id(): + with HashCryptoCallback(17): + sha = Sha256.new(bytes(7), device_id=17) + assert sha.digest() == struct.pack("I28x", 7) + + +if _lib.SHA3_ENABLED: + def test_hash_unsupported_type_falls_back(): + with HashCryptoCallback(18): + sha = Sha3(b"abc", size=Sha3.SHA3_224_DIGEST_SIZE, device_id=18) + assert sha.digest() == hashlib.sha3_224(b"abc").digest() + + +if _lib.HMAC_ENABLED and _lib.SHA256_ENABLED: + from wolfcrypt.hashes import HmacSha256 + + class RecordingCryptoCallback(CryptoCallback): + def __init__(self, device_id: int) -> None: + super().__init__(device_id) + self.algo_types: list[int] = [] + + @override + def callback(self, device_id: int, info: _lib.wc_CryptoInfo) -> int: + self.algo_types.append(info.algo_type) + return super().callback(device_id, info) + + needs_hmac_copy = pytest.mark.skipif(not _lib.HMAC_COPY_ENABLED, reason="HMAC device_id needs wc_HmacCopy") + + @pytest.mark.skipif(bool(HASH_CB_ENABLED and not _lib.HMAC_COPY_ENABLED), reason="HMAC device_id needs wc_HmacCopy") + def test_hmac_device_id(): + with RecordingCryptoCallback(19) as cb: + mac = HmacSha256.new(b"key", b"data", device_id=19).digest() + assert mac == HmacSha256(b"key", b"data").digest() + assert cb.algo_types + + @needs_hash_cb + @needs_hmac_copy + def test_hmac_hash_callback(): + with SoftwareHashCryptoCallback(25) as cb: + mac = HmacSha256(b"key", b"da", device_id=25) + assert mac.digest() == py_hmac.new(b"key", b"da", hashlib.sha256).digest() + mac.update(b"ta") + assert mac.digest() == py_hmac.new(b"key", b"data", hashlib.sha256).digest() + del mac + gc.collect() + assert not cb._hash_contexts + + @pytest.mark.skipif(not HASH_CB_ENABLED or bool(_lib.HMAC_COPY_ENABLED), reason="HMAC device_id is allowed") + def test_hmac_device_id_needs_hmac_copy(): + with pytest.raises(WolfCryptError): + HmacSha256(b"key", device_id=26) diff --git a/wolfcrypt/__init__.py b/wolfcrypt/__init__.py index 3578b26..a257a04 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,52 @@ __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 _cffi_backend import FFI + 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( + exc_type: object, exc_value: BaseException, traceback: TracebackType | None + ) -> int: + if isinstance(exc_value, WolfCryptApiError): + return exc_value.err_code + log.error("crypto callback failed", exc_info=exc_value) + return _lib.WC_HW_E + + @_ffi.def_extern(error=_lib.WC_HW_E, 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..bb700ce 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,9 @@ ASN_ENABLED: int CHACHA_ENABLED: int CHACHA_STREAM_ENABLED: int CHACHA20_POLY1305_ENABLED: int +CRYPTO_CB_ENABLED: int +CRYPTO_CB_COPY_ENABLED: int +CRYPTO_CB_FREE_ENABLED: int DES3_ENABLED: int ECC_ENABLED: int ED25519_ENABLED: int @@ -37,6 +43,7 @@ ED448_ENABLED: int ERROR_STRINGS_ENABLED: int FIPS_ENABLED: int HMAC_ENABLED: int +HMAC_COPY_ENABLED: int KEYGEN_ENABLED: int HKDF_ENABLED: int ML_DSA_ENABLED: int @@ -282,6 +289,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 +349,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 +372,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: ... @@ -403,6 +425,7 @@ def wc_HmacSetKey(hmac: FFI.CData, type: int, key: bytes, length: int) -> int: . def wc_HmacUpdate(hmac: FFI.CData, data: bytes, size: int) -> int: ... def wc_HmacFinal(hmac: FFI.CData, hash: FFI.CData) -> int: ... def wc_HmacFree(hmac: FFI.CData) -> None: ... +def wc_HmacCopy(src: FFI.CData, dst: FFI.CData) -> int: ... def wc_HKDF(type: int, in_key: bytes, in_key_size: int, salt: bytes, salt_size: int, info: bytes, info_size: int, out: FFI.CData, out_size: int) -> int: ... def wc_HKDF_Extract(type: int, salt: bytes, salt_size: int, in_key: bytes, in_key_size: int, out: FFI.CData) -> int: ... @@ -569,3 +592,49 @@ 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 + u: FFI.CData + + +@dataclass +class wc_RngInfo: + rng: RNG + out: FFI.CData + sz: int + + +@dataclass +class wc_CopyInfo: + algo: int + type: int + src: FFI.CData + dst: FFI.CData + + +@dataclass +class wc_FreeInfo: + algo: int + type: int + subType: int + obj: FFI.CData + + +@dataclass +class wc_CryptoInfo: + algo_type: int + hash: wc_HashInfo + rng: wc_RngInfo + copy: wc_CopyInfo + free: wc_FreeInfo + + +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 d66071e..7b31a29 100644 --- a/wolfcrypt/ciphers.py +++ b/wolfcrypt/ciphers.py @@ -437,7 +437,7 @@ class AesGcmStream: # 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 @@ 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, device_id) 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..884af70 --- /dev/null +++ b/wolfcrypt/cryptocb.py @@ -0,0 +1,252 @@ +# 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 copy +import logging +from collections import defaultdict +from typing import TYPE_CHECKING, Any, ClassVar, 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 collections.abc import Callable + from types import TracebackType + + from _cffi_backend import FFI + +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, + } + + # Native type and wolfCrypt copy function for each hash type + NATIVE_HASH_COPY: Final[dict[int, tuple[str, Callable[[FFI.CData, FFI.CData], int]]]] = {} + if _lib.SHA_ENABLED: + NATIVE_HASH_COPY[_lib.WC_HASH_TYPE_SHA] = ("wc_Sha *", _lib.wc_ShaCopy) + if _lib.SHA256_ENABLED: + NATIVE_HASH_COPY[_lib.WC_HASH_TYPE_SHA256] = ("wc_Sha256 *", _lib.wc_Sha256Copy) + if _lib.SHA384_ENABLED: + NATIVE_HASH_COPY[_lib.WC_HASH_TYPE_SHA384] = ("wc_Sha384 *", _lib.wc_Sha384Copy) + if _lib.SHA512_ENABLED: + NATIVE_HASH_COPY[_lib.WC_HASH_TYPE_SHA512] = ("wc_Sha512 *", _lib.wc_Sha512Copy) + if _lib.SHA3_ENABLED: + NATIVE_HASH_COPY[_lib.WC_HASH_TYPE_SHA3_256] = ("wc_Sha3 *", _lib.wc_Sha3_256_Copy) + NATIVE_HASH_COPY[_lib.WC_HASH_TYPE_SHA3_384] = ("wc_Sha3 *", _lib.wc_Sha3_384_Copy) + NATIVE_HASH_COPY[_lib.WC_HASH_TYPE_SHA3_512] = ("wc_Sha3 *", _lib.wc_Sha3_512_Copy) + + + class CryptoCallback: + """ + Registers a wolfCrypt crypto callback device for **device_id**. + Subclasses override rng_callback, or both hash_update_callback + and hash_finalize_callback. Operations that are not overridden + fall back to the software implementation. + + Each hash object has its own context dict, passed to both hash + hooks for keeping per-hash state. It is deep-copied when the hash + object is copied, so its values must support copy.deepcopy, and it + is discarded when the hash is finalized or freed. + Hash callbacks need wolfSSL built with copy and free callback support + (--enable-cryptocbutils=copy,free); without it hashing falls back to + the software implementation. + """ + + # Keeps each registered instance (and its ctx handle) alive until it is + # unregistered, and records which instance owns a device id. + _registered: ClassVar[dict[int, CryptoCallback]] = {} + + def __init__(self, device_id: int) -> None: + self.device_id = device_id + self._hash_contexts: dict[int, dict[str, Any]] = {} + # (src, dst) addresses of native hash copies in progress + self._hash_copies: set[tuple[int, int]] = set() + 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})") + CryptoCallback._registered[device_id] = self + + def __enter__(self) -> Self: + return self + + def __exit__( + self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None + ) -> None: + self._unregister() + + def __del__(self) -> None: + try: + self._unregister() + except AttributeError: + # Can occur during interpreter shutdown + pass + + 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 or not self._handles_hash(): + return _lib.CRYPTOCB_UNAVAILABLE + log.debug("hash_type = %s", HASH_TYPE_NAME[info.hash.type]) + key = _address(info.hash.u.ctx) + context = self._hash_contexts.setdefault(key, {}) + if info.hash.digest == _ffi.NULL: + self.hash_update_callback( + device_id, + info.hash.type, + context, + bytes(_ffi.buffer(info.hash.data, info.hash.data_size)), + ) + else: + digest = self.hash_finalize_callback(device_id, info.hash.type, context) + # wolfCrypt resets a hash object after Final + del self._hash_contexts[key] + 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 + if info.algo_type == _lib.WC_ALGO_TYPE_COPY and info.copy.algo == _lib.WC_ALGO_TYPE_HASH: + return self._hash_copy(info.copy.type, info.copy.src, info.copy.dst) + if info.algo_type == _lib.WC_ALGO_TYPE_FREE and info.free.algo == _lib.WC_ALGO_TYPE_HASH: + self._hash_contexts.pop(_address(info.free.obj), None) + # Let wolfCrypt free the native object + return _lib.CRYPTOCB_UNAVAILABLE + 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, context: dict[str, Any], data: bytes) -> None: + raise NotImplementedError + + def hash_finalize_callback(self, device_id: int, hash_type: int, context: dict[str, Any]) -> bytes: + raise NotImplementedError + + def _hash_copy(self, hash_type: int, src: FFI.CData, dst: FFI.CData) -> int: + src_key = _address(src) + dst_key = _address(dst) + native = NATIVE_HASH_COPY.get(hash_type) + if ( + native is None + or src_key not in self._hash_contexts + or (src_key, dst_key) in self._hash_copies + ): + return _lib.CRYPTOCB_UNAVAILABLE + # Re-enter wolfCrypt's copy so it deep copies the native object + # (e.g. the SMALL_STACK_CACHE W buffer). The nested copy callback + # returns CRYPTOCB_UNAVAILABLE above, and the free of dst it does + # first drops any old context for dst. + ptr_type, copy_fn = native + self._hash_copies.add((src_key, dst_key)) + try: + ret = copy_fn(_ffi.cast(ptr_type, src), _ffi.cast(ptr_type, dst)) + finally: + self._hash_copies.discard((src_key, dst_key)) + if ret == 0: + self._hash_contexts[dst_key] = copy.deepcopy(self._hash_contexts[src_key]) + return ret + + def _handles_hash(self) -> bool: + # Update and finalize must both go to the callback, otherwise wolfCrypt + # would hash the data in software and finalize would ignore it. + # Without copy and free callbacks a hash context could not follow + # its native object. + if not (_lib.CRYPTO_CB_COPY_ENABLED and _lib.CRYPTO_CB_FREE_ENABLED): + return False + cls = type(self) + return ( + cls.hash_update_callback is not CryptoCallback.hash_update_callback + and cls.hash_finalize_callback is not CryptoCallback.hash_finalize_callback + ) + + def _unregister(self) -> None: + if CryptoCallback._registered.get(self.device_id) is self: + del CryptoCallback._registered[self.device_id] + _lib.wc_CryptoCb_UnRegisterDevice(self.device_id) + + @classmethod + def default_device_id(cls) -> int: + return _lib.wc_CryptoCb_DefaultDevID() + + + def _address(ptr: FFI.CData) -> int: + return int(_ffi.cast("uintptr_t", ptr)) 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..5ae2e02 100644 --- a/wolfcrypt/hashes.py +++ b/wolfcrypt/hashes.py @@ -28,7 +28,7 @@ from wolfcrypt._ffi import ffi as _ffi from wolfcrypt._ffi import lib as _lib -from wolfcrypt.exceptions import WolfCryptApiError +from wolfcrypt.exceptions import WolfCryptApiError, WolfCryptError from wolfcrypt.utils import t2b, b2h, BytesOrStr if TYPE_CHECKING: @@ -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: ... @@ -73,7 +73,7 @@ def digest_size(self) -> int | None: ... @classmethod @abstractmethod - def new(cls, string: BytesOrStr | None) -> _Hash: ... + def new(cls, string: BytesOrStr | None, *, device_id: int = _lib.INVALID_DEVID) -> _Hash: ... def copy(self) -> _Hash: """ @@ -173,14 +173,14 @@ def hexdigest(self) -> bytes: class _Sha(_Hash): @override @classmethod - def new(cls, string: BytesOrStr | None = None) -> _Hash: + def new(cls, string: BytesOrStr | None = None, *, device_id: int = _lib.INVALID_DEVID) -> _Hash: """ Creates a new hashing object and returns it. The optional **string** parameter, if supplied, will be immediately hashed into the object's starting state, as if obj.update(string) was called. """ - return cls(string) + return cls(string, device_id) if _lib.SHA_ENABLED: @@ -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 @@ -449,33 +449,32 @@ class _Hmac(_Hash): A **PEP 247: Cryptographic Hash Functions** compliant **Keyed Hash Function Interface**. - Note: wolfCrypt provides no safe public copy for an `Hmac` object. + Note: `copy()` is not supported and raises `NotImplementedError`. Byte-copying the struct would alias the original's internal state - (e.g. async/hardware-accelerated builds keep pointers there), so - `copy()` is not supported and raises `NotImplementedError`. + (e.g. async/hardware-accelerated builds keep pointers there). + `digest()` uses wc_HmacCopy (wolfSSL 5.9.0+) when it is available. """ digest_size = None _native_type = "Hmac *" _native_size = _ffi.sizeof("Hmac") _delete = staticmethod(_lib.wc_HmacFree) + if _lib.HMAC_COPY_ENABLED: + _copy = staticmethod(_lib.wc_HmacCopy) @override def copy(self) -> _Hmac: - raise NotImplementedError( - "HMAC objects cannot be safely copied: wolfCrypt has no " - "wc_HmacCopy and byte-copying the state would alias the " - "original's internal C resources") + raise NotImplementedError("HMAC objects do not support copy()") 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,12 +482,12 @@ 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 @classmethod - def new(cls, key: BytesOrStr, string: BytesOrStr | None = None) -> _Hash: # pylint: disable=W0221 # ty: ignore[invalid-method-override] + def new(cls, key: BytesOrStr, string: BytesOrStr | None = None, *, device_id: int = _lib.INVALID_DEVID) -> _Hash: # pylint: disable=W0221 # ty: ignore[invalid-method-override] """ Creates a new hashing object and returns it. **key** is a required parameter containing a string giving the key @@ -496,15 +495,25 @@ def new(cls, key: BytesOrStr, string: BytesOrStr | None = None) -> _Hash: # pyl will be immediately hashed into the object's starting state, as if obj.update(string) was called. """ - return cls(key, string) + return cls(key, string, device_id) @property @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: + # Without wc_HmacCopy, digest() byte-copies the Hmac, which loses + # the per-hash state a crypto callback keeps for the embedded hash. + # Hash callbacks are only used with copy and free callbacks. + if ( + device_id != _lib.INVALID_DEVID + and not _lib.HMAC_COPY_ENABLED + and _lib.CRYPTO_CB_COPY_ENABLED + and _lib.CRYPTO_CB_FREE_ENABLED + ): + raise WolfCryptError("HMAC device_id with hash crypto callbacks requires wc_HmacCopy") + 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 *")