From d1ac3d5f37b15ec825e6a3dc4c2d4396e83058c2 Mon Sep 17 00:00:00 2001 From: Robert de Vries Date: Thu, 21 May 2026 22:00:08 +0200 Subject: [PATCH] Add typing: asn, exceptions, hashes, hkdf, pwdbased, utils. --- pyproject.toml | 17 +++- wolfcrypt/__init__.py | 5 +- wolfcrypt/_ffi/lib.pyi | 88 ++++++++++++++++++++ wolfcrypt/asn.py | 21 ++--- wolfcrypt/exceptions.py | 5 +- wolfcrypt/hashes.py | 174 +++++++++++++++++++++++++--------------- wolfcrypt/hkdf.py | 8 +- wolfcrypt/pwdbased.py | 4 +- wolfcrypt/utils.py | 5 +- 9 files changed, 245 insertions(+), 82 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 35168f4..488d69c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -97,7 +97,7 @@ indent-width = 4 target-version = "py310" [tool.ruff.lint] -select = ["E", "F", "B", "UP", "C4", "DTZ", "EXE", "FA", "INT", "ISC", "ICN", "LOG", "G", "RSE", "SLOT", "TID", "TC", "FLY", "PERF", "W", "FURB"] +select = ["E", "F", "B", "UP", "ANN", "C4", "DTZ", "EXE", "FA", "INT", "ISC", "ICN", "LOG", "G", "RSE", "SLOT", "TID", "TC", "FLY", "PERF", "W", "FURB"] ignore = ["E501"] # Allow fix for all enabled rules (when `--fix`) is provided. @@ -107,6 +107,11 @@ unfixable = [] # Allow unused variables when underscore-prefixed. dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$" +[tool.ruff.lint.per-file-ignores] +"wolfcrypt/ciphers.py" = ["ANN"] +"scripts/*.py" = ["ANN"] +"tests/*.py" = ["ANN"] + [tool.ruff.format] # Like Black, use double quotes for strings. quote-style = "double" @@ -133,3 +138,13 @@ docstring-code-format = false # This only has an effect when the `docstring-code-format` setting is # enabled. docstring-code-line-length = "dynamic" + +[tool.ty.environment] +python-version = "3.10" +root = ["."] + +[tool.ty.src] +exclude = ["./lib", "./wolfcrypt/ciphers.py", "./tests"] + +[tool.ty.rules] +all = "warn" diff --git a/wolfcrypt/__init__.py b/wolfcrypt/__init__.py index 99a27e8..c2514df 100644 --- a/wolfcrypt/__init__.py +++ b/wolfcrypt/__init__.py @@ -17,6 +17,9 @@ # 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 +from typing import cast + +from _cffi_backend import Lib from wolfcrypt._version import __version__, __wolfssl_version__ @@ -50,7 +53,7 @@ if top_level_py not in ["setup.py", "build_ffi.py"]: 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")) + ret = _lib.wc_SetSeed_Cb(_ffi.addressof(cast(Lib, _lib), "wc_GenerateSeed")) if ret < 0: raise WolfCryptApiError("wc_SetSeed_Cb failed", ret) if _lib.FIPS_ENABLED and _lib.FIPS_VERSION >= 5: diff --git a/wolfcrypt/_ffi/lib.pyi b/wolfcrypt/_ffi/lib.pyi index 047e5cf..3421c6a 100644 --- a/wolfcrypt/_ffi/lib.pyi +++ b/wolfcrypt/_ffi/lib.pyi @@ -15,6 +15,7 @@ DES3_ENABLED: int ECC_ENABLED: int ED25519_ENABLED: int ED448_ENABLED: int +ERROR_STRINGS_ENABLED: int FIPS_ENABLED: int HMAC_ENABLED: int KEYGEN_ENABLED: int @@ -68,9 +69,96 @@ WC_ML_DSA_87: int WC_KEYTYPE_ALL: int +PRIVATEKEY_TYPE: int +PUBLICKEY_TYPE: int +CERT_TYPE: int +MAX_DER_DIGEST_SZ: int +SHAh: int +SHA256h: int +SHA384h: int +SHA512h: int + + 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 wc_InitRngNonce_ex(rng: RNG, nonce: bytes, nonce_size: int, heap: FFI.CData, device_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_FreeRng(rng: RNG) -> None: ... + +DerBufferPtr: TypeAlias = FFI.CData +EncryptedInfo: TypeAlias = FFI.CData +IntPtr: TypeAlias = FFI.CData +BytePtr: TypeAlias = FFI.CData + +def wc_PemToDer(buff: bytes, buf_size: int, type: int, der: DerBufferPtr, heap: FFI.CData, info: EncryptedInfo, + key_format: IntPtr) -> int: ... +def wc_DerToPemEx(der: bytes, der_size: int, output: FFI.CData, output_size: int, cipher_info: FFI.CData, + type: int) -> int: ... +def wc_FreeDer(der: DerBufferPtr) -> None: ... +def wc_EncodeSignature(out: BytePtr, digest: bytes, digest_size: int, hash_oid: int) -> int: ... + +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_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_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_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_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: ... +def wc_Sha512Free(obj: FFI.CData) -> None: ... + +def wc_InitSha3_224(obj: FFI.CData, heap: FFI.CData, dev_id: int) -> int: ... +def wc_Sha3_224_Copy(src: FFI.CData, dst: FFI.CData) -> int: ... +def wc_Sha3_224_Update(obj: FFI.CData, data: bytes, size: int) -> int: ... +def wc_Sha3_224_Final(obj: FFI.CData, ret: FFI.CData) -> int: ... +def wc_Sha3_224_Free(obj: FFI.CData) -> None: ... + +def wc_InitSha3_256(obj: FFI.CData, heap: FFI.CData, dev_id: int) -> int: ... +def wc_Sha3_256_Copy(src: FFI.CData, dst: FFI.CData) -> int: ... +def wc_Sha3_256_Update(obj: FFI.CData, data: bytes, size: int) -> int: ... +def wc_Sha3_256_Final(obj: FFI.CData, ret: FFI.CData) -> int: ... +def wc_Sha3_256_Free(obj: FFI.CData) -> None: ... + +def wc_InitSha3_384(obj: FFI.CData, heap: FFI.CData, dev_id: int) -> int: ... +def wc_Sha3_384_Copy(src: FFI.CData, dst: FFI.CData) -> int: ... +def wc_Sha3_384_Update(obj: FFI.CData, data: bytes, size: int) -> int: ... +def wc_Sha3_384_Final(obj: FFI.CData, ret: FFI.CData) -> int: ... +def wc_Sha3_384_Free(obj: FFI.CData) -> None: ... + +def wc_InitSha3_512(obj: FFI.CData, heap: FFI.CData, dev_id: int) -> int: ... +def wc_Sha3_512_Copy(src: FFI.CData, dst: FFI.CData) -> int: ... +def wc_Sha3_512_Update(obj: FFI.CData, data: bytes, size: int) -> int: ... +def wc_Sha3_512_Final(obj: FFI.CData, ret: FFI.CData) -> int: ... +def wc_Sha3_512_Free(obj: FFI.CData) -> None: ... + +def wc_HmacInit(hmac: FFI.CData, heap: FFI.CData, dev_id: int) -> int: ... +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_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: ... +def wc_HKDF_Expand(type: int, in_key: bytes, in_key_size: int, info: bytes, info_size: int, out: FFI.CData, out_size: int) -> int: ... diff --git a/wolfcrypt/asn.py b/wolfcrypt/asn.py index dc702d1..a726b78 100644 --- a/wolfcrypt/asn.py +++ b/wolfcrypt/asn.py @@ -20,23 +20,26 @@ # pylint: disable=no-member,no-name-in-module +from __future__ import annotations + import hmac as _hmac from wolfcrypt._ffi import ffi as _ffi from wolfcrypt._ffi import lib as _lib from wolfcrypt.exceptions import WolfCryptError, WolfCryptApiError +from wolfcrypt.hashes import _Hash if _lib.SHA_ENABLED: - from wolfcrypt.hashes import Sha + from wolfcrypt.hashes import Sha # ty: ignore[possibly-missing-import] if _lib.SHA256_ENABLED: - from wolfcrypt.hashes import Sha256 + from wolfcrypt.hashes import Sha256 # ty: ignore[possibly-missing-import] if _lib.SHA384_ENABLED: - from wolfcrypt.hashes import Sha384 + from wolfcrypt.hashes import Sha384 # ty: ignore[possibly-missing-import] if _lib.SHA512_ENABLED: - from wolfcrypt.hashes import Sha512 + from wolfcrypt.hashes import Sha512 # ty: ignore[possibly-missing-import] if _lib.ASN_ENABLED: - def pem_to_der(pem, pem_type): + def pem_to_der(pem: bytes, pem_type: int) -> bytes: der = _ffi.new("DerBuffer**") ret = _lib.wc_PemToDer(pem, len(pem), pem_type, der, _ffi.NULL, _ffi.NULL, _ffi.NULL) @@ -49,7 +52,7 @@ if _lib.ASN_ENABLED: _lib.wc_FreeDer(der) return result - def der_to_pem(der, pem_type): + def der_to_pem(der: bytes, pem_type: int) -> bytes: pem_length = _lib.wc_DerToPemEx(der, len(der), _ffi.NULL, 0, _ffi.NULL, pem_type) if pem_length <= 0: @@ -63,7 +66,7 @@ if _lib.ASN_ENABLED: return _ffi.buffer(pem, pem_length)[:] - def hash_oid_from_class(hash_cls): + def hash_oid_from_class(hash_cls: type[_Hash]) -> int: if _lib.SHA_ENABLED and hash_cls == Sha: return _lib.SHAh elif _lib.SHA256_ENABLED and hash_cls == Sha256: @@ -75,7 +78,7 @@ if _lib.ASN_ENABLED: else: raise WolfCryptError(f"Unknown hash class {hash_cls.__name__}") - def make_signature(data, hash_cls, key=None): + def make_signature(data: bytes, hash_cls: type[_Hash], key = None) -> bytes: hash_obj = hash_cls() hash_obj.update(data) digest = hash_obj.digest() @@ -93,7 +96,7 @@ if _lib.ASN_ENABLED: else: return plaintext_sig - def check_signature(signature, data, hash_cls, pub_key): + def check_signature(signature: bytes, data: bytes, hash_cls: type[_Hash], pub_key) -> bool: computed_signature = make_signature(data, hash_cls) decrypted_signature = pub_key.verify(signature) return _hmac.compare_digest(computed_signature, decrypted_signature) diff --git a/wolfcrypt/exceptions.py b/wolfcrypt/exceptions.py index 5b41bd7..c691a18 100644 --- a/wolfcrypt/exceptions.py +++ b/wolfcrypt/exceptions.py @@ -18,6 +18,9 @@ # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA +from __future__ import annotations + +from typing import cast from wolfcrypt._ffi import ffi as _ffi from wolfcrypt._ffi import lib as _lib @@ -56,6 +59,6 @@ def error_string(err_code: int) -> str: :return: error string """ if _lib.ERROR_STRINGS_ENABLED: - return _ffi.string(_lib.wc_GetErrorString(err_code)).decode() + return cast(bytes, _ffi.string(_lib.wc_GetErrorString(err_code))).decode() else: return "" diff --git a/wolfcrypt/hashes.py b/wolfcrypt/hashes.py index 59f1bf5..442c698 100644 --- a/wolfcrypt/hashes.py +++ b/wolfcrypt/hashes.py @@ -20,19 +20,23 @@ # pylint: disable=no-member,no-name-in-module, no-self-use +from __future__ import annotations + +from abc import ABC, abstractmethod + +from _cffi_backend import FFI from wolfcrypt._ffi import ffi as _ffi from wolfcrypt._ffi import lib as _lib -from wolfcrypt.utils import t2b, b2h - from wolfcrypt.exceptions import WolfCryptApiError +from wolfcrypt.utils import t2b, b2h, BytesOrStr -class _Hash: +class _Hash(ABC): """ A **PEP 247: Cryptographic Hash Functions** compliant **Hash Function Interface**. """ - def __init__(self, string=None): + def __init__(self, string: BytesOrStr | None = None) -> None: self._native_object = _ffi.new(self._native_type) self._shallow_copy = False ret = self._init() @@ -42,17 +46,33 @@ class _Hash: if string: self.update(string) - @classmethod - def new(cls, string=None): - """ - 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) + @abstractmethod + def _init(self) -> int: ... - def copy(self): + @abstractmethod + def _update(self, data: bytes) -> int: ... + + @abstractmethod + def _final(self, obj: FFI.CData, ret: FFI.CData) -> int: ... + + @property + @abstractmethod + def _native_size(self) -> int: ... + + @property + @abstractmethod + def _native_type(self) -> str: ... + + @property + @abstractmethod + def digest_size(self) -> int: ... + + @classmethod + @abstractmethod + def new(cls, string: BytesOrStr | None) -> _Hash: ... + + + def copy(self) -> _Hash: """ Returns a separate copy of this hashing object. An update to this copy won't affect the original object. @@ -86,7 +106,7 @@ class _Hash: return copy - def update(self, string): + def update(self, string: BytesOrStr) -> None: """ Hashes **string** into the current state of the hashing object. update() can be called any number of times during @@ -98,7 +118,7 @@ class _Hash: if ret < 0: # pragma: no cover raise WolfCryptApiError("Hash update error", ret) - def digest(self): + def digest(self) -> bytes: """ Returns the hash value of this hashing object as a string containing 8-bit data. The object is not altered in any @@ -137,7 +157,7 @@ class _Hash: return _ffi.buffer(result, self.digest_size)[:] - def hexdigest(self): + def hexdigest(self) -> bytes: """ Returns the hash value of this hashing object as a string containing hexadecimal digits. Lowercase letters are used @@ -147,8 +167,20 @@ class _Hash: return b2h(self.digest()) +class _Sha(_Hash): + @classmethod + def new(cls, string: BytesOrStr | None = None) -> _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) + + if _lib.SHA_ENABLED: - class Sha(_Hash): + class Sha(_Sha): """ **SHA-1** is a cryptographic hash function standardized by **NIST**. @@ -160,22 +192,22 @@ if _lib.SHA_ENABLED: _delete = staticmethod(_lib.wc_ShaFree) _copy = staticmethod(_lib.wc_ShaCopy) - def __del__(self): + def __del__(self) -> None: if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False): self._delete(self._native_object) - def _init(self): + def _init(self) -> int: return _lib.wc_InitSha(self._native_object) - def _update(self, data): + def _update(self, data: bytes) -> int: return _lib.wc_ShaUpdate(self._native_object, data, len(data)) - def _final(self, obj, ret): + def _final(self, obj: FFI.CData, ret: FFI.CData) -> int: return _lib.wc_ShaFinal(obj, ret) if _lib.SHA256_ENABLED: - class Sha256(_Hash): + class Sha256(_Sha): """ **SHA-256** is a cryptographic hash function from the **SHA-2 family** and is standardized by **NIST**. @@ -188,22 +220,22 @@ if _lib.SHA256_ENABLED: _delete = staticmethod(_lib.wc_Sha256Free) _copy = staticmethod(_lib.wc_Sha256Copy) - def __del__(self): + def __del__(self) -> None: if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False): self._delete(self._native_object) - def _init(self): + def _init(self) -> int: return _lib.wc_InitSha256(self._native_object) - def _update(self, data): + def _update(self, data: bytes) -> int: return _lib.wc_Sha256Update(self._native_object, data, len(data)) - def _final(self, obj, ret): + def _final(self, obj: FFI.CData, ret: FFI.CData) -> int: return _lib.wc_Sha256Final(obj, ret) if _lib.SHA384_ENABLED: - class Sha384(_Hash): + class Sha384(_Sha): """ **SHA-384** is a cryptographic hash function from the **SHA-2 family** and is standardized by **NIST**. @@ -216,22 +248,22 @@ if _lib.SHA384_ENABLED: _delete = staticmethod(_lib.wc_Sha384Free) _copy = staticmethod(_lib.wc_Sha384Copy) - def __del__(self): + def __del__(self) -> None: if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False): self._delete(self._native_object) - def _init(self): + def _init(self) -> int: return _lib.wc_InitSha384(self._native_object) - def _update(self, data): + def _update(self, data: bytes) -> int: return _lib.wc_Sha384Update(self._native_object, data, len(data)) - def _final(self, obj, ret): + def _final(self, obj: FFI.CData, ret: FFI.CData) -> int: return _lib.wc_Sha384Final(obj, ret) if _lib.SHA512_ENABLED: - class Sha512(_Hash): + class Sha512(_Sha): """ **SHA-512** is a cryptographic hash function from the **SHA-2 family** and is standardized by **NIST**. @@ -244,21 +276,21 @@ if _lib.SHA512_ENABLED: _delete = staticmethod(_lib.wc_Sha512Free) _copy = staticmethod(_lib.wc_Sha512Copy) - def __del__(self): + def __del__(self) -> None: if hasattr(self, '_native_object') and not getattr(self, '_shallow_copy', False): self._delete(self._native_object) - def _init(self): + def _init(self) -> int: return _lib.wc_InitSha512(self._native_object) - def _update(self, data): + def _update(self, data: bytes) -> int: return _lib.wc_Sha512Update(self._native_object, data, len(data)) - def _final(self, obj, ret): + def _final(self, obj: FFI.CData, ret: FFI.CData) -> int: return _lib.wc_Sha512Final(obj, ret) if _lib.SHA3_ENABLED: - class Sha3(_Hash): + class Sha3(_Sha): """ **SHA3 ** is a cryptographic hash function family standardized by **NIST**. @@ -267,6 +299,7 @@ if _lib.SHA3_ENABLED: Using SHA3-384 by default, unless a different digest size is passed through __init__. """ + digest_size = None _native_type = "wc_Sha3 *" _native_size = _ffi.sizeof("wc_Sha3") SHA3_224_DIGEST_SIZE = 28 @@ -288,16 +321,19 @@ if _lib.SHA3_ENABLED: 64: _lib.wc_Sha3_512_Copy, } - def __del__(self): + def __del__(self) -> None: # Unlike the SHA-1/2 classes, Sha3's _delete is set per-instance # from a size->function dict and is None for invalid sizes, so # we need the extra truthiness check. - if (hasattr(self, '_native_object') - and not getattr(self, '_shallow_copy', False) - and getattr(self, '_delete', None)): + if ( + hasattr(self, '_native_object') + and not getattr(self, '_shallow_copy', False) + and getattr(self, '_delete', None) + and self._delete is not None + ): self._delete(self._native_object) - def __init__(self, string=None, size=SHA3_384_DIGEST_SIZE): # pylint: disable=W0231 + def __init__(self, string: BytesOrStr | None = None, size: int = SHA3_384_DIGEST_SIZE) -> None: # pylint: disable=W0231 self._native_object = _ffi.new(self._native_type) self._shallow_copy = False self.digest_size = size @@ -310,10 +346,10 @@ if _lib.SHA3_ENABLED: self.update(string) @classmethod - def new(cls, string=None, size=SHA3_384_DIGEST_SIZE): + def new(cls, string: BytesOrStr | None = None, size: int = SHA3_384_DIGEST_SIZE) -> Sha3: return cls(string, size) - def copy(self): + def copy(self) -> Sha3: # Bypass __init__ to avoid calling _init() on a state that _copy # immediately overwrites (which would leak internal resources in # async/HW-accelerated builds). Mark as shallow up front so @@ -337,12 +373,7 @@ if _lib.SHA3_ENABLED: # Keep _shallow_copy = True: memmove shares state with self. return c - def _init(self): - if (self.digest_size != Sha3.SHA3_224_DIGEST_SIZE and - self.digest_size != Sha3.SHA3_256_DIGEST_SIZE and - self.digest_size != Sha3.SHA3_384_DIGEST_SIZE and - self.digest_size != Sha3.SHA3_512_DIGEST_SIZE): - return -1 + def _init(self) -> int: if self.digest_size == Sha3.SHA3_224_DIGEST_SIZE: return _lib.wc_InitSha3_224(self._native_object, _ffi.NULL, 0) if self.digest_size == Sha3.SHA3_256_DIGEST_SIZE: @@ -351,7 +382,9 @@ if _lib.SHA3_ENABLED: return _lib.wc_InitSha3_384(self._native_object, _ffi.NULL, 0) if self.digest_size == Sha3.SHA3_512_DIGEST_SIZE: return _lib.wc_InitSha3_512(self._native_object, _ffi.NULL, 0) - def _update(self, data): + return -1 + + def _update(self, data: bytes) -> int: if self.digest_size == Sha3.SHA3_224_DIGEST_SIZE: return _lib.wc_Sha3_224_Update(self._native_object, data, len(data)) if self.digest_size == Sha3.SHA3_256_DIGEST_SIZE: @@ -360,7 +393,9 @@ if _lib.SHA3_ENABLED: return _lib.wc_Sha3_384_Update(self._native_object, data, len(data)) if self.digest_size == Sha3.SHA3_512_DIGEST_SIZE: return _lib.wc_Sha3_512_Update(self._native_object, data, len(data)) - def _final(self, obj, ret): + return -1 + + def _final(self, obj: FFI.CData, ret: FFI.CData) -> int: if self.digest_size == Sha3.SHA3_224_DIGEST_SIZE: return _lib.wc_Sha3_224_Final(obj, ret) if self.digest_size == Sha3.SHA3_256_DIGEST_SIZE: @@ -369,6 +404,7 @@ if _lib.SHA3_ENABLED: return _lib.wc_Sha3_384_Final(obj, ret) if self.digest_size == Sha3.SHA3_512_DIGEST_SIZE: return _lib.wc_Sha3_512_Final(obj, ret) + return -1 # Hmac types @@ -408,24 +444,27 @@ if _lib.HMAC_ENABLED: "wc_HmacCopy and byte-copying the state would alias the " "original's internal C resources") - def __del__(self): + 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, string=None): # pylint: disable=W0231 + def __init__(self, key: BytesOrStr, string: BytesOrStr | None = None) -> None: # pylint: disable=W0231 key = t2b(key) self._native_object = _ffi.new(self._native_type) self._shallow_copy = False - ret = self._init(self._type, key) + ret = self._hmac_init(self._type, key) if ret < 0: # pragma: no cover raise WolfCryptApiError("Hmac init error", ret) if string: self.update(string) + def _init(self) -> int: + return -1 + @classmethod - def new(cls, key, string=None): # pylint: disable=W0221 + def new(cls, key: BytesOrStr, string: BytesOrStr | None = None) -> _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 @@ -435,7 +474,12 @@ if _lib.HMAC_ENABLED: """ return cls(key, string) - def _init(self, hmac, key): + + @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) if ret < 0: raise WolfCryptApiError("wc_HmacInit error", ret) @@ -449,10 +493,10 @@ if _lib.HMAC_ENABLED: raise WolfCryptApiError("wc_HmacSetKey error", ret) return ret - def _update(self, data): + def _update(self, data: bytes) -> int: return _lib.wc_HmacUpdate(self._native_object, data, len(data)) - def _final(self, obj, ret): + def _final(self, obj: FFI.CData, ret: FFI.CData) -> int: return _lib.wc_HmacFinal(obj, ret) @@ -465,7 +509,7 @@ if _lib.HMAC_ENABLED: It produces a [ **512-bit | 64 bytes** ] message digest. """ _type = _TYPE_SHA - digest_size = Sha.digest_size + digest_size = Sha.digest_size # ty: ignore[possibly-unresolved-reference] if _lib.SHA256_ENABLED: @@ -477,7 +521,7 @@ if _lib.HMAC_ENABLED: It produces a [ **512-bit | 64 bytes** ] message digest. """ _type = _TYPE_SHA256 - digest_size = Sha256.digest_size + digest_size = Sha256.digest_size # ty: ignore[possibly-unresolved-reference] if _lib.SHA384_ENABLED: @@ -489,7 +533,7 @@ if _lib.HMAC_ENABLED: It produces a [ **512-bit | 64 bytes** ] message digest. """ _type = _TYPE_SHA384 - digest_size = Sha384.digest_size + digest_size = Sha384.digest_size # ty: ignore[possibly-unresolved-reference] if _lib.SHA512_ENABLED: @@ -501,9 +545,9 @@ if _lib.HMAC_ENABLED: It produces a [ **512-bit | 64 bytes** ] message digest. """ _type = _TYPE_SHA512 - digest_size = Sha512.digest_size + digest_size = Sha512.digest_size # ty: ignore[possibly-unresolved-reference] -def hash_type_to_cls(hash_type): +def hash_type_to_cls(hash_type: int) -> type[_Hash] | None: if _lib.SHA_ENABLED and hash_type == _lib.WC_HASH_TYPE_SHA: hash_cls = Sha elif _lib.SHA256_ENABLED and hash_type == _lib.WC_HASH_TYPE_SHA256: diff --git a/wolfcrypt/hkdf.py b/wolfcrypt/hkdf.py index c06ab87..a4ddcdb 100644 --- a/wolfcrypt/hkdf.py +++ b/wolfcrypt/hkdf.py @@ -28,8 +28,9 @@ from wolfcrypt.utils import t2b if _lib.HKDF_ENABLED: + from wolfcrypt.hashes import _Hmac # ty: ignore[possibly-missing-import] - def HKDF(hash_cls, in_key, salt=None, info=None, out_len=None): + def HKDF(hash_cls: _Hmac, in_key: bytes | str, salt: bytes | str | None = None, info: bytes | str | None = None, out_len: int | None = None) -> bytes: """ Perform HKDF Extract-and-Expand in one call (wraps wc_HKDF). @@ -55,6 +56,7 @@ if _lib.HKDF_ENABLED: if out_len is None: out_len = hash_cls.digest_size + assert out_len is not None out = _ffi.new(f"byte[{out_len}]") ret = _lib.wc_HKDF( @@ -73,7 +75,7 @@ if _lib.HKDF_ENABLED: return _ffi.buffer(out, out_len)[:] - def HKDF_Extract(hash_cls, salt, in_key): + def HKDF_Extract(hash_cls: _Hmac, salt: bytes | str | None, in_key: bytes | str) -> bytes: """ HKDF-Extract: PRK = HMAC-Hash(salt, IKM) Wraps wc_HKDF_Extract. @@ -100,7 +102,7 @@ if _lib.HKDF_ENABLED: return _ffi.buffer(out, out_len)[:] - def HKDF_Expand(hash_cls, prk, info, out_len): + def HKDF_Expand(hash_cls: _Hmac, prk: bytes | str, info: bytes | str | None, out_len: int) -> bytes: """ HKDF-Expand: OKM = HKDF-Expand(PRK, info, L) Wraps wc_HKDF_Expand. diff --git a/wolfcrypt/pwdbased.py b/wolfcrypt/pwdbased.py index 56c83b0..ecdbd41 100644 --- a/wolfcrypt/pwdbased.py +++ b/wolfcrypt/pwdbased.py @@ -20,13 +20,15 @@ # pylint: disable=no-member,no-name-in-module +from __future__ import annotations + from wolfcrypt._ffi import ffi as _ffi from wolfcrypt._ffi import lib as _lib from wolfcrypt.exceptions import WolfCryptApiError if _lib.PWDBASED_ENABLED: - def PBKDF2(password, salt, iterations, key_length, hash_type): + def PBKDF2(password: bytes | str, salt: bytes | str, iterations: int, key_length: int, hash_type: int) -> bytes: if isinstance(salt, str): salt = str.encode(salt) diff --git a/wolfcrypt/utils.py b/wolfcrypt/utils.py index 5de6363..6793488 100644 --- a/wolfcrypt/utils.py +++ b/wolfcrypt/utils.py @@ -22,10 +22,13 @@ from __future__ import annotations +from typing import TypeAlias + from binascii import hexlify as b2h, unhexlify as h2b # noqa: F401 +BytesOrStr: TypeAlias = bytes | bytearray | memoryview | str -def t2b(string: bytes | bytearray | memoryview | str) -> bytes: +def t2b(string: BytesOrStr) -> bytes: """ Converts text to bytes.