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

271 lines
12 KiB
Python
Raw Permalink Blame History

# 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 objects with a context outlive the registration: the device
stays registered until their contexts are discarded, and another
callback cannot be registered for the device id until then.
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:
owner = CryptoCallback._registered.get(device_id)
if owner is not None and owner._hash_contexts:
raise WolfCryptError(f"CryptoCb device {device_id} still has hash objects in use")
self.device_id = device_id
self._closed = False
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)
if self._closed and key not in self._hash_contexts:
return _lib.CRYPTOCB_UNAVAILABLE
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
self._drop_hash_context(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._drop_hash_context(_address(info.free.obj))
# 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 _drop_hash_context(self, key: int) -> None:
# Unregistering from inside this device's callback is safe: wolfCrypt
# does not touch the device entry after the callback returns.
if self._hash_contexts.pop(key, None) is not None and self._closed:
self._unregister()
def _unregister(self) -> None:
self._closed = True
if self._hash_contexts:
return
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))
Reference in New Issue View Git Blame Copy Permalink