diff --git a/Lib/hashlib.py b/Lib/hashlib.py index 0e9bd98aa1f..c1e2d07c435 100644 --- a/Lib/hashlib.py +++ b/Lib/hashlib.py @@ -2,7 +2,7 @@ # Licensed to PSF under a Contributor Agreement. # -__doc__ = """hashlib module - A common interface to many hash functions. +__doc__ = r"""hashlib module - A common interface to many hash functions. new(name, data=b'', **kwargs) - returns a new hash object implementing the given hash function; initializing the hash @@ -12,7 +12,7 @@ than using new(name): md5(), sha1(), sha224(), sha256(), sha384(), sha512(), blake2b(), blake2s(), -sha3_224, sha3_256, sha3_384, sha3_512, shake_128, and shake_256. +sha3_224(), sha3_256(), sha3_384(), sha3_512(), shake_128(), and shake_256(). More algorithms may be available on your platform but the above are guaranteed to exist. See the algorithms_guaranteed and algorithms_available attributes @@ -21,8 +21,8 @@ NOTE: If you want the adler32 or crc32 hash functions they are available in the zlib module. -Choose your hash function wisely. Some have known collision weaknesses. -sha384 and sha512 will be slow on 32 bit platforms. +Choose your hash function wisely. Some have known collision weaknesses, +while others may be slower depending on the CPU architecture. Hash objects have these methods: - update(data): Update the hash object with the bytes in data. Repeated calls @@ -36,21 +36,21 @@ efficiently compute the digests of data that share a common initial substring. -For example, to obtain the digest of the byte string 'Nobody inspects the -spammish repetition': +Assuming that Python has been built with SHA-2 support, the SHA-256 digest +of the byte string b'Nobody inspects the spammish repetition' is computed +as follows: >>> import hashlib - >>> m = hashlib.md5() + >>> m = hashlib.sha256() >>> m.update(b"Nobody inspects") >>> m.update(b" the spammish repetition") - >>> m.digest() - b'\\xbbd\\x9c\\x83\\xdd\\x1e\\xa5\\xc9\\xd9\\xde\\xc9\\xa1\\x8d\\xf0\\xff\\xe9' + >>> m.digest() # doctest: +ELLIPSIS + b'\x03\x1e\xdd}Ae\x15\x93\xc5\xfe\\\x00o\xa5u+7...' More condensed: - >>> hashlib.sha224(b"Nobody inspects the spammish repetition").hexdigest() - 'a4337bc45a8fc544c03f52dc550cd6e1e87021bc896588bd79e901e2' - + >>> hashlib.sha256(b"Nobody inspects the spammish repetition").hexdigest() + '031edd7d41651593c5fe5c006fa5752b37fddff7bc4e843aa6af0c950f4b9406' """ # This tuple and __get_builtin_constructor() must be modified if a new @@ -80,6 +80,11 @@ } def __get_builtin_constructor(name): + if not isinstance(name, str): + # Since this function is only used by new(), we use the same + # exception as _hashlib.new() when 'name' is of incorrect type. + err = f"new() argument 'name' must be str, not {type(name).__name__}" + raise TypeError(err) cache = __builtin_constructor_cache constructor = cache.get(name) if constructor is not None: @@ -120,20 +125,33 @@ def __get_builtin_constructor(name): if constructor is not None: return constructor - raise ValueError('unsupported hash type ' + name) + # Keep the message in sync with hashlib.h::HASHLIB_UNSUPPORTED_ALGORITHM. + raise ValueError(f'unsupported hash algorithm {name}') def __get_openssl_constructor(name): + # This function is only used until the module has been initialized. + assert isinstance(name, str), "invalid call to __get_openssl_constructor()" if name in __block_openssl_constructor: # Prefer our builtin blake2 implementation. return __get_builtin_constructor(name) try: - # MD5, SHA1, and SHA2 are in all supported OpenSSL versions - # SHA3/shake are available in OpenSSL 1.1.1+ + # Fetch the OpenSSL hash function if it exists, + # independently of the context security policy. f = getattr(_hashlib, 'openssl_' + name) - # Allow the C module to raise ValueError. The function will be - # defined but the hash not actually available. Don't fall back to - # builtin if the current security policy blocks a digest, bpo#40695. + # Check if the context security policy blocks the digest or not + # by allowing the C module to raise a ValueError. The function + # will be defined but the hash will not be available at runtime. + # + # We use "usedforsecurity=False" to prevent falling back to the + # built-in function in case the security policy does not allow it. + # + # Note that this only affects the explicit named constructors, + # and not the algorithms exposed through hashlib.new() which + # can still be resolved to a built-in function even if the + # current security policy does not allow it. + # + # See https://github.com/python/cpython/issues/84872. f(usedforsecurity=False) # Use the C function directly (very fast) return f @@ -154,6 +172,8 @@ def __hash_new(name, *args, **kwargs): optionally initialized with data (which must be a bytes-like object). """ if name in __block_openssl_constructor: + # __block_openssl_constructor is expected to contain strings only + assert isinstance(name, str), f"unexpected name: {name}" # Prefer our builtin blake2 implementation. return __get_builtin_constructor(name)(*args, **kwargs) try: @@ -187,7 +207,8 @@ def __hash_new(name, *args, **kwargs): try: # OpenSSL's scrypt requires OpenSSL 1.1+ - from _hashlib import scrypt # noqa: F401 + from _hashlib import scrypt + __all__ += ('scrypt',) except ImportError: pass @@ -203,7 +224,7 @@ def file_digest(fileobj, digest, /, *, _bufsize=2**18): *digest* must either be a hash algorithm name as a *str*, a hash constructor, or a callable that returns a hash object. """ - # On Linux we could use AF_ALG sockets and sendfile() to archive zero-copy + # On Linux we could use AF_ALG sockets and sendfile() to achieve zero-copy # hashing with hardware acceleration. if isinstance(digest, str): digestobj = new(digest) @@ -246,9 +267,30 @@ def file_digest(fileobj, digest, /, *, _bufsize=2**18): try: globals()[__func_name] = __get_hash(__func_name) except ValueError: - import logging - logging.exception('code for hash %s was not found.', __func_name) - + # Don't log here: logging at import time has global side effects and + # would tell the wrong audience; code that uses a missing algorithm + # gets a ValueError from the stub installed below. + # The following code can be simplified in Python 3.19 + # once "string" is removed from the signature. + __code = f'''\ +def {__func_name}(data=__UNSET, *, usedforsecurity=True, string=__UNSET): + if data is __UNSET and string is not __UNSET: + import warnings + warnings.warn( + "the 'string' keyword parameter is deprecated since " + "Python 3.15 and slated for removal in Python 3.19; " + "use the 'data' keyword parameter or pass the data " + "to hash as a positional argument instead", + DeprecationWarning, stacklevel=2) + if data is not __UNSET and string is not __UNSET: + raise TypeError("'data' and 'string' are mutually exclusive " + "and support for 'string' keyword parameter " + "is slated for removal in a future version.") + raise ValueError("unsupported hash algorithm {__func_name}") +''' + exec(__code, {"__UNSET": object()}, __locals := {}) + globals()[__func_name] = __locals[__func_name] + del __code, __locals # Cleanup locals() del __always_supported, __func_name, __get_hash diff --git a/Lib/hmac.py b/Lib/hmac.py index 2d6016cda11..e0c040bcd5f 100644 --- a/Lib/hmac.py +++ b/Lib/hmac.py @@ -26,6 +26,16 @@ digest_size = None +def _is_shake_constructor(digest_like): + if isinstance(digest_like, str): + name = digest_like + else: + h = digest_like() if callable(digest_like) else digest_like.new() + if not isinstance(name := getattr(h, "name", None), str): + return False + return name.startswith(("shake", "SHAKE")) + + def _get_digest_constructor(digest_like): if callable(digest_like): return digest_like @@ -109,6 +119,8 @@ def _init_old(self, key, msg, digestmod): import warnings digest_cons = _get_digest_constructor(digestmod) + if _is_shake_constructor(digest_cons): + raise ValueError(f"unsupported hash algorithm {digestmod}") self._hmac = None self._outer = digest_cons() @@ -256,6 +268,8 @@ def digest(key, msg, digest): def _compute_digest_fallback(key, msg, digest): digest_cons = _get_digest_constructor(digest) + if _is_shake_constructor(digest_cons): + raise ValueError(f"unsupported hash algorithm {digest}") inner = digest_cons() outer = digest_cons() blocksize = getattr(inner, 'block_size', 64) diff --git a/Lib/test/support/hashlib_helper.py b/Lib/test/support/hashlib_helper.py index 7032257b068..818f99d0023 100644 --- a/Lib/test/support/hashlib_helper.py +++ b/Lib/test/support/hashlib_helper.py @@ -1,45 +1,515 @@ +import contextlib +import enum import functools -import hashlib import importlib +import inspect import unittest -from test.support.import_helper import import_module +import unittest.mock +from test.support import import_helper -try: - import _hashlib -except ImportError: - _hashlib = None -try: - import _hmac -except ImportError: - _hmac = None +def _parse_fullname(fullname, *, strict=False): + """Parse a fully-qualified name ``.``. + The ``module_name`` component contains one or more dots. + The ``member_name`` component does not contain any dot. -def requires_hashlib(): - return unittest.skipIf(_hashlib is None, "requires _hashlib") + If *strict* is true, *fullname* must be a string. Otherwise, + it can be None, and, ``module_name`` and ``member_name`` will + also be None. + """ + if fullname is None: + assert not strict + return None, None + assert isinstance(fullname, str), fullname + assert fullname.count(".") >= 1, fullname + module_name, member_name = fullname.rsplit(".", maxsplit=1) + return module_name, member_name -def requires_builtin_hmac(): - return unittest.skipIf(_hmac is None, "requires _hmac") +def _import_module(module_name, *, strict=False): + """Import a module from its fully-qualified name. + If *strict* is false, import failures are suppressed and None is returned. + """ + if module_name is None: + # To prevent a TypeError in importlib.import_module + if strict: + raise ImportError("no module to import") + return None + try: + return importlib.import_module(module_name) + except ImportError as exc: + if strict: + raise exc + return None -def _missing_hash(digestname, implementation=None, *, exc=None): - parts = ["missing", implementation, f"hash algorithm: {digestname!r}"] - msg = " ".join(filter(None, parts)) - raise unittest.SkipTest(msg) from exc +def _import_member(module_name, member_name, *, strict=False): + """Import a member from a module. + + If *strict* is false, import failures are suppressed and None is returned. + """ + if member_name is None: + if strict: + raise ImportError(f"no member to import from {module_name}") + return None + module = _import_module(module_name, strict=strict) + if strict: + return getattr(module, member_name) + return getattr(module, member_name, None) + + +class Implementation(enum.StrEnum): + # Indicate that the hash function is implemented by a built-in module. + builtin = enum.auto() + # Indicate that the hash function is implemented by OpenSSL. + openssl = enum.auto() + # Indicate that the hash function is provided through the public API. + hashlib = enum.auto() + + +class _HashId(enum.StrEnum): + """Enumeration containing the canonical digest names. + + Those names should only be used by hashlib.new() or hmac.new(). + Their support by _hashlib.new() is not necessarily guaranteed. + """ + + md5 = enum.auto() + sha1 = enum.auto() + + sha224 = enum.auto() + sha256 = enum.auto() + sha384 = enum.auto() + sha512 = enum.auto() + + sha3_224 = enum.auto() + sha3_256 = enum.auto() + sha3_384 = enum.auto() + sha3_512 = enum.auto() + + shake_128 = enum.auto() + shake_256 = enum.auto() + + blake2s = enum.auto() + blake2b = enum.auto() + + def __repr__(self): + return str(self) + + @property + def is_xof(self): + """Indicate whether the hash is an extendable-output hash function.""" + return self.startswith("shake_") + + @property + def is_keyed(self): + """Indicate whether the hash is a keyed hash function.""" + return self.startswith("blake2") + + +CANONICAL_DIGEST_NAMES = frozenset(map(str, _HashId.__members__)) +NON_HMAC_DIGEST_NAMES = frozenset(( + _HashId.shake_128, _HashId.shake_256, + _HashId.blake2s, _HashId.blake2b, +)) + + +class _HashInfoItem: + """Interface for interacting with a named object. + + The object is entirely described by its fully-qualified *fullname*. + + *fullname* must be None or a string ".". + """ + + def __init__(self, fullname=None, *, strict=False): + module_name, member_name = _parse_fullname(fullname, strict=strict) + self.fullname = fullname + self.module_name = module_name + self.member_name = member_name + + def import_module(self, *, strict=False): + """Import the described module. + + If *strict* is true, an ImportError may be raised if importing fails, + otherwise, None is returned on error. + """ + return _import_module(self.module_name, strict=strict) + + def import_member(self, *, strict=False): + """Import the described member. + + If *strict* is true, an AttributeError or an ImportError may be + raised if importing fails; otherwise, None is returned on error. + """ + return _import_member( + self.module_name, self.member_name, strict=strict + ) -def _openssl_availabillity(digestname, *, usedforsecurity): - try: - _hashlib.new(digestname, usedforsecurity=usedforsecurity) - except AttributeError: - assert _hashlib is None - _missing_hash(digestname, "OpenSSL") - except ValueError as exc: - _missing_hash(digestname, "OpenSSL", exc=exc) +class _HashInfoBase: + """Base dataclass containing "backend" information. + + Subclasses may define an attribute named after one of the known + implementations ("builtin", "openssl" or "hashlib") which stores + an _HashInfoItem object. + + Those attributes can be retrieved through __getitem__(), e.g., + ``info["builtin"]`` returns the _HashInfoItem corresponding to + the builtin implementation. + """ + + def __init__(self, canonical_name): + assert isinstance(canonical_name, _HashId), canonical_name + self.canonical_name = canonical_name + + def __getitem__(self, implementation): + try: + attrname = Implementation(implementation) + except ValueError: + raise self.invalid_implementation_error(implementation) from None + + try: + provider = getattr(self, attrname) + except AttributeError: + raise self.invalid_implementation_error(implementation) from None + + if not isinstance(provider, _HashInfoItem): + raise KeyError(implementation) + return provider + + def invalid_implementation_error(self, implementation): + msg = f"no implementation {implementation} for {self.canonical_name}" + return AssertionError(msg) + + +class _HashTypeInfo(_HashInfoBase): + """Dataclass containing information for hash functions types. + + - *canonical_name* must be a _HashId. + + - *builtin* is the fully-qualified name for the builtin HACL* type, + e.g., "_md5.MD5Type". + + - *openssl* is the fully-qualified name for the OpenSSL wrapper type, + e.g., "_hashlib.HASH". + """ + + def __init__(self, canonical_name, builtin, openssl): + super().__init__(canonical_name) + self.builtin = _HashInfoItem(builtin, strict=True) + self.openssl = _HashInfoItem(openssl, strict=True) + + def fullname(self, implementation): + """Get the fully qualified name of a given implementation. + + This returns a string of the form "MODULE_NAME.OBJECT_NAME" or None + if the hash function does not have a corresponding implementation. + + *implementation* must be "builtin" or "openssl". + """ + return self[implementation].fullname + + def module_name(self, implementation): + """Get the name of the module containing the hash object type.""" + return self[implementation].module_name + + def object_type_name(self, implementation): + """Get the name of the hash object class name.""" + return self[implementation].member_name + + def import_module(self, implementation, *, allow_skip=False): + """Import the module containing the hash object type. + + On error, return None if *allow_skip* is false, or raise SkipNoHash. + """ + target = self[implementation] + module = target.import_module() + if allow_skip and module is None: + reason = f"cannot import module {target.module_name}" + raise SkipNoHash(self.canonical_name, implementation, reason) + return module + + def import_object_type(self, implementation, *, allow_skip=False): + """Get the runtime hash object type. + + On error, return None if *allow_skip* is false, or raise SkipNoHash. + """ + target = self[implementation] + member = target.import_member() + if allow_skip and member is None: + reason = f"cannot import class {target.fullname}" + raise SkipNoHash(self.canonical_name, implementation, reason) + return member + + +class _HashFuncInfo(_HashInfoBase): + """Dataclass containing information for hash functions constructors. + + - *canonical_name* must be a _HashId. + + - *builtin* is the fully-qualified name of the HACL* + hash constructor function, e.g., "_md5.md5". + + - *openssl* is the fully-qualified name of the "_hashlib" method + for the OpenSSL named constructor, e.g., "_hashlib.openssl_md5". + + - *hashlib* is the fully-qualified name of the "hashlib" method + for the explicit named hash constructor, e.g., "hashlib.md5". + """ -def _decorate_func_or_class(func_or_class, decorator_func): + def __init__(self, canonical_name, builtin, openssl=None, hashlib=None): + super().__init__(canonical_name) + self.builtin = _HashInfoItem(builtin, strict=True) + self.openssl = _HashInfoItem(openssl, strict=False) + self.hashlib = _HashInfoItem(hashlib, strict=False) + + def fullname(self, implementation): + """Get the fully qualified name of a given implementation. + + This returns a string of the form "MODULE_NAME.METHOD_NAME" or None + if the hash function does not have a corresponding implementation. + + *implementation* must be "builtin", "openssl" or "hashlib". + """ + return self[implementation].fullname + + def module_name(self, implementation): + """Get the name of the constructor function module. + + The *implementation* must be "builtin", "openssl" or "hashlib". + """ + return self[implementation].module_name + + def method_name(self, implementation): + """Get the name of the constructor function module method. + + Use fullname() to get the constructor function fully-qualified name. + + The *implementation* must be "builtin", "openssl" or "hashlib". + """ + return self[implementation].member_name + + +class _HashInfo: + """Dataclass containing information for supported hash functions. + + Attributes + ---------- + canonical_name : _HashId + The hash function canonical name. + type : _HashTypeInfo + The hash object types information. + func : _HashTypeInfo + The hash object constructors information. + """ + + def __init__( + self, + canonical_name, + builtin_object_type_fullname, + openssl_object_type_fullname, + builtin_method_fullname, + openssl_method_fullname=None, + hashlib_method_fullname=None, + ): + """ + - *canonical_name* must be a _HashId. + + - *builtin_object_type_fullname* is the fully-qualified name + for the builtin HACL* type, e.g., "_md5.MD5Type". + + - *openssl_object_type_fullname* is the fully-qualified name + for the OpenSSL wrapper type, e.g., "_hashlib.HASH". + + - *builtin_method_fullname* is the fully-qualified name + of the HACL* hash constructor function, e.g., "_md5.md5". + + - *openssl_method_fullname* is the fully-qualified name + of the "_hashlib" module method for the explicit OpenSSL + hash constructor function, e.g., "_hashlib.openssl_md5". + + - *hashlib_method_fullname* is the fully-qualified name + of the "hashlib" module method for the explicit hash + constructor function, e.g., "hashlib.md5". + """ + assert isinstance(canonical_name, _HashId), canonical_name + self.canonical_name = canonical_name + self.type = _HashTypeInfo( + canonical_name, + builtin_object_type_fullname, + openssl_object_type_fullname, + ) + self.func = _HashFuncInfo( + canonical_name, + builtin_method_fullname, + openssl_method_fullname, + hashlib_method_fullname, + ) + + +_HASHINFO_DATABASE = frozendict({ + _HashId.md5: _HashInfo( + _HashId.md5, + "_md5.MD5Type", + "_hashlib.HASH", + "_md5.md5", + "_hashlib.openssl_md5", + "hashlib.md5", + ), + _HashId.sha1: _HashInfo( + _HashId.sha1, + "_sha1.SHA1Type", + "_hashlib.HASH", + "_sha1.sha1", + "_hashlib.openssl_sha1", + "hashlib.sha1", + ), + _HashId.sha224: _HashInfo( + _HashId.sha224, + "_sha2.SHA224Type", + "_hashlib.HASH", + "_sha2.sha224", + "_hashlib.openssl_sha224", + "hashlib.sha224", + ), + _HashId.sha256: _HashInfo( + _HashId.sha256, + "_sha2.SHA256Type", + "_hashlib.HASH", + "_sha2.sha256", + "_hashlib.openssl_sha256", + "hashlib.sha256", + ), + _HashId.sha384: _HashInfo( + _HashId.sha384, + "_sha2.SHA384Type", + "_hashlib.HASH", + "_sha2.sha384", + "_hashlib.openssl_sha384", + "hashlib.sha384", + ), + _HashId.sha512: _HashInfo( + _HashId.sha512, + "_sha2.SHA512Type", + "_hashlib.HASH", + "_sha2.sha512", + "_hashlib.openssl_sha512", + "hashlib.sha512", + ), + _HashId.sha3_224: _HashInfo( + _HashId.sha3_224, + "_sha3.sha3_224", + "_hashlib.HASH", + "_sha3.sha3_224", + "_hashlib.openssl_sha3_224", + "hashlib.sha3_224", + ), + _HashId.sha3_256: _HashInfo( + _HashId.sha3_256, + "_sha3.sha3_256", + "_hashlib.HASH", + "_sha3.sha3_256", + "_hashlib.openssl_sha3_256", + "hashlib.sha3_256", + ), + _HashId.sha3_384: _HashInfo( + _HashId.sha3_384, + "_sha3.sha3_384", + "_hashlib.HASH", + "_sha3.sha3_384", + "_hashlib.openssl_sha3_384", + "hashlib.sha3_384", + ), + _HashId.sha3_512: _HashInfo( + _HashId.sha3_512, + "_sha3.sha3_512", + "_hashlib.HASH", + "_sha3.sha3_512", + "_hashlib.openssl_sha3_512", + "hashlib.sha3_512", + ), + _HashId.shake_128: _HashInfo( + _HashId.shake_128, + "_sha3.shake_128", + "_hashlib.HASHXOF", + "_sha3.shake_128", + "_hashlib.openssl_shake_128", + "hashlib.shake_128", + ), + _HashId.shake_256: _HashInfo( + _HashId.shake_256, + "_sha3.shake_256", + "_hashlib.HASHXOF", + "_sha3.shake_256", + "_hashlib.openssl_shake_256", + "hashlib.shake_256", + ), + _HashId.blake2s: _HashInfo( + _HashId.blake2s, + "_blake2.blake2s", + "_hashlib.HASH", + "_blake2.blake2s", + None, + "hashlib.blake2s", + ), + _HashId.blake2b: _HashInfo( + _HashId.blake2b, + "_blake2.blake2b", + "_hashlib.HASH", + "_blake2.blake2b", + None, + "hashlib.blake2b", + ), +}) +assert _HASHINFO_DATABASE.keys() == CANONICAL_DIGEST_NAMES + + +def get_hash_type_info(name): + info = _HASHINFO_DATABASE[name] + assert isinstance(info, _HashInfo), info + return info.type + + +def get_hash_func_info(name): + info = _HASHINFO_DATABASE[name] + assert isinstance(info, _HashInfo), info + return info.func + + +def _iter_hash_func_info(excluded): + for name, info in _HASHINFO_DATABASE.items(): + if name not in excluded: + yield info.func + + +# Mapping from canonical hash names to their explicit HACL* HMAC constructor. +# There is currently no OpenSSL one-shot named function and there will likely +# be none in the future. +_HMACINFO_DATABASE = { + _HashId(canonical_name): _HashInfoItem(f"_hmac.compute_{canonical_name}") + for canonical_name in CANONICAL_DIGEST_NAMES +} +# Neither HACL* nor OpenSSL supports HMAC over XOFs. +_HMACINFO_DATABASE[_HashId.shake_128] = _HashInfoItem() +_HMACINFO_DATABASE[_HashId.shake_256] = _HashInfoItem() +# Strictly speaking, HMAC-BLAKE is meaningless as BLAKE2 is already a +# keyed hash function. However, as it's exposed by HACL*, we test it. +_HMACINFO_DATABASE[_HashId.blake2s] = _HashInfoItem('_hmac.compute_blake2s_32') +_HMACINFO_DATABASE[_HashId.blake2b] = _HashInfoItem('_hmac.compute_blake2b_32') +_HMACINFO_DATABASE = frozendict(_HMACINFO_DATABASE) +assert _HMACINFO_DATABASE.keys() == CANONICAL_DIGEST_NAMES + + +def get_hmac_item_info(name): + info = _HMACINFO_DATABASE[name] + assert isinstance(info, _HashInfoItem), info + return info + + +def _decorate_func_or_class(decorator_func, func_or_class): if not isinstance(func_or_class, type): return decorator_func(func_or_class) @@ -57,114 +527,225 @@ def setUpClass(cls): return decorated_class -def requires_hashdigest(digestname, openssl=None, usedforsecurity=True): - """Decorator raising SkipTest if a hashing algorithm is not available. +def _chain_decorators(decorators): + """Obtain a decorator by chaining multiple decorators. - The hashing algorithm may be missing, blocked by a strict crypto policy, - or Python may be configured with `--with-builtin-hashlib-hashes=no`. + The decorators are applied in the order they are given. + """ + def decorator_func(func): + return functools.reduce(lambda w, deco: deco(w), decorators, func) + return functools.partial(_decorate_func_or_class, decorator_func) - If 'openssl' is True, then the decorator checks that OpenSSL provides - the algorithm. Otherwise the check falls back to (optional) built-in - HACL* implementations. - The usedforsecurity flag is passed to the constructor but has no effect - on HACL* implementations. +def _ensure_wrapper_signature(wrapper, wrapped): + """Ensure that a wrapper has the same signature as the wrapped function. - Examples of exceptions being suppressed: - ValueError: [digital envelope routines: EVP_DigestInit_ex] disabled for FIPS - ValueError: unsupported hash type md4 + This is used to guarantee that a TypeError raised due to a bad API call + is raised consistently (using variadic signatures would hide such errors). """ - if openssl and _hashlib is not None: - def test_availability(): - _hashlib.new(digestname, usedforsecurity=usedforsecurity) - else: - def test_availability(): - hashlib.new(digestname, usedforsecurity=usedforsecurity) + try: + wrapped_sig = inspect.signature(wrapped) + except ValueError: # built-in signature cannot be found + return + + wrapper_sig = inspect.signature(wrapper) + if wrapped_sig != wrapper_sig: + fullname = f"{wrapped.__module__}.{wrapped.__qualname__}" + raise AssertionError( + f"signature for {fullname}() is incorrect:\n" + f" expect: {wrapped_sig}\n" + f" actual: {wrapper_sig}" + ) + +def _make_conditional_decorator(test, /, *test_args, **test_kwargs): def decorator_func(func): @functools.wraps(func) def wrapper(*args, **kwargs): - try: - test_availability() - except ValueError as exc: - _missing_hash(digestname, exc=exc) + test(*test_args, **test_kwargs) return func(*args, **kwargs) return wrapper + return functools.partial(_decorate_func_or_class, decorator_func) - def decorator(func_or_class): - return _decorate_func_or_class(func_or_class, decorator_func) - return decorator +def requires_openssl_hashlib(): + _hashlib = _import_module("_hashlib") + return unittest.skipIf(_hashlib is None, "requires _hashlib") -def requires_openssl_hashdigest(digestname, *, usedforsecurity=True): - """Decorator raising SkipTest if an OpenSSL hashing algorithm is missing. - The hashing algorithm may be missing or blocked by a strict crypto policy. - """ - def decorator_func(func): - @requires_hashlib() # avoid checking at each call - @functools.wraps(func) - def wrapper(*args, **kwargs): - _openssl_availabillity(digestname, usedforsecurity=usedforsecurity) - return func(*args, **kwargs) - return wrapper +def requires_builtin_hmac(): + _hmac = _import_module("_hmac") + return unittest.skipIf(_hmac is None, "requires _hmac") + + +class SkipNoHash(unittest.SkipTest): + """A SkipTest exception raised when a hash is not available.""" + + def __init__(self, digestname, implementation=None, reason=None): + parts = ["missing", implementation, f"hash algorithm {digestname!r}"] + if reason is not None: + parts.insert(0, f"{reason}: ") + super().__init__(" ".join(filter(None, parts))) + - def decorator(func_or_class): - return _decorate_func_or_class(func_or_class, decorator_func) - return decorator +class SkipNoHashInCall(SkipNoHash): + def __init__(self, func, digestname, implementation=None): + super().__init__(digestname, implementation, f"cannot use {func}") -def find_openssl_hashdigest_constructor(digestname, *, usedforsecurity=True): - """Find the OpenSSL hash function constructor by its name.""" + +def _hashlib_new(digestname, openssl, /, **kwargs): + """Check availability of [hashlib|_hashlib].new(digestname, **kwargs). + + If *openssl* is True, module is "_hashlib" (C extension module), + otherwise it is "hashlib" (pure Python interface). + + The constructor function is returned (without binding **kwargs), + or SkipTest is raised if none exists. + """ assert isinstance(digestname, str), digestname - _openssl_availabillity(digestname, usedforsecurity=usedforsecurity) - # This returns a function of the form _hashlib.openssl_ and - # not a lambda function as it is rejected by _hashlib.hmac_new(). - return getattr(_hashlib, f"openssl_{digestname}") + # Re-import 'hashlib' in case it was mocked, but propagate + # exceptions as it should be unconditionally available. + hashlib = importlib.import_module("hashlib") + # re-import '_hashlib' in case it was mocked + _hashlib = _import_module("_hashlib") + module = _hashlib if openssl and _hashlib is not None else hashlib + try: + module.new(digestname, **kwargs) + except ValueError as exc: + raise SkipNoHashInCall(f"{module.__name__}.new", digestname) from exc + return functools.partial(module.new, digestname) -def requires_builtin_hashdigest( - module_name, digestname, *, usedforsecurity=True -): - """Decorator raising SkipTest if a HACL* hashing algorithm is missing. +def _builtin_hash(module_name, digestname, /, **kwargs): + """Check availability of .(**kwargs). - The *module_name* is the C extension module name based on HACL*. - The *digestname* is one of its member, e.g., 'md5'. + + The constructor function is returned, or SkipTest is raised if none exists. """ - def decorator_func(func): - @functools.wraps(func) - def wrapper(*args, **kwargs): - module = import_module(module_name) - try: - getattr(module, digestname) - except AttributeError: - fullname = f'{module_name}.{digestname}' - _missing_hash(fullname, implementation="HACL") - return func(*args, **kwargs) - return wrapper + assert isinstance(module_name, str), module_name + assert isinstance(digestname, str), digestname + fullname = f'{module_name}.{digestname}' + try: + builtin_module = importlib.import_module(module_name) + except ImportError as exc: + raise SkipNoHash(fullname, "builtin") from exc + try: + constructor = getattr(builtin_module, digestname) + except AttributeError as exc: + raise SkipNoHash(fullname, "builtin") from exc + try: + constructor(**kwargs) + except ValueError as exc: + raise SkipNoHash(fullname, "builtin") from exc + return constructor - def decorator(func_or_class): - return _decorate_func_or_class(func_or_class, decorator_func) - return decorator +def _openssl_new(digestname, /, **kwargs): + """Check availability of _hashlib.new(digestname, **kwargs). -def find_builtin_hashdigest_constructor( - module_name, digestname, *, usedforsecurity=True -): - """Find the HACL* hash function constructor. + The constructor function is returned (without binding **kwargs), + or SkipTest is raised if none exists. + """ + assert isinstance(digestname, str), digestname + try: + # re-import '_hashlib' in case it was mocked + _hashlib = importlib.import_module("_hashlib") + except ImportError as exc: + raise SkipNoHash(digestname, "openssl") from exc + try: + _hashlib.new(digestname, **kwargs) + except ValueError as exc: + raise SkipNoHashInCall("_hashlib.new", digestname) from exc + return functools.partial(_hashlib.new, digestname) - - The *module_name* is the C extension module name based on HACL*. - - The *digestname* is one of its member, e.g., 'md5'. + +def _openssl_hash(digestname, /, **kwargs): + """Check availability of _hashlib.openssl_(**kwargs). + + The constructor function is returned (without binding **kwargs), + or SkipTest is raised if none exists. """ - module = import_module(module_name) + assert isinstance(digestname, str), digestname + method_name = f"openssl_{digestname}" + fullname = f"_hashlib.{method_name}" + try: + # re-import '_hashlib' in case it was mocked + _hashlib = importlib.import_module("_hashlib") + except ImportError as exc: + raise SkipNoHash(fullname, "openssl") from exc try: - constructor = getattr(module, digestname) - constructor(b'', usedforsecurity=usedforsecurity) - except (AttributeError, TypeError, ValueError): - _missing_hash(f'{module_name}.{digestname}', implementation="HACL") + constructor = getattr(_hashlib, method_name) + except AttributeError as exc: + raise SkipNoHash(fullname, "openssl") from exc + try: + constructor(**kwargs) + except ValueError as exc: + raise SkipNoHash(fullname, "openssl") from exc return constructor +def requires_hashdigest(digestname, openssl=None, *, usedforsecurity=True): + """Decorator raising SkipTest if a hashing algorithm is not available. + + The hashing algorithm may be missing, blocked by a strict crypto policy, + or Python may be configured with `--with-builtin-hashlib-hashes=no`. + + If 'openssl' is True, then the decorator checks that OpenSSL provides + the algorithm. Otherwise the check falls back to (optional) built-in + HACL* implementations. + + The usedforsecurity flag is passed to the constructor but has no effect + on HACL* implementations. + + Examples of exceptions being suppressed: + ValueError: [digital envelope routines: EVP_DigestInit_ex] disabled for FIPS + ValueError: unsupported hash type md4 + """ + return _make_conditional_decorator( + _hashlib_new, digestname, openssl, usedforsecurity=usedforsecurity + ) + + +def requires_openssl_hashdigest(digestname, *, usedforsecurity=True): + """Decorator raising SkipTest if an OpenSSL hashing algorithm is missing. + + The hashing algorithm may be missing or blocked by a strict crypto policy. + """ + return _make_conditional_decorator( + _openssl_new, digestname, usedforsecurity=usedforsecurity + ) + + +def _make_requires_builtin_hashdigest_decorator(item, *, usedforsecurity=True): + assert isinstance(item, _HashInfoItem), item + return _make_conditional_decorator( + _builtin_hash, + item.module_name, + item.member_name, + usedforsecurity=usedforsecurity, + ) + + +def requires_builtin_hashdigest(canonical_name, *, usedforsecurity=True): + """Decorator raising SkipTest if a HACL* hashing algorithm is missing.""" + info = get_hash_func_info(canonical_name) + return _make_requires_builtin_hashdigest_decorator( + info.builtin, usedforsecurity=usedforsecurity + ) + + +def requires_builtin_hashes(*, exclude=(), usedforsecurity=True): + """Decorator raising SkipTest if one HACL* hashing algorithm is missing.""" + return _chain_decorators(( + _make_requires_builtin_hashdigest_decorator( + info.builtin, usedforsecurity=usedforsecurity + ) for info in _iter_hash_func_info(exclude) + )) + + class HashFunctionsTrait: """Mixin trait class containing hash functions. @@ -178,59 +759,31 @@ class HashFunctionsTrait: implementation of HMAC). """ - ALGORITHMS = [ - 'md5', 'sha1', - 'sha224', 'sha256', 'sha384', 'sha512', - 'sha3_224', 'sha3_256', 'sha3_384', 'sha3_512', - ] - - # Default 'usedforsecurity' to use when looking up a hash function. + # Default 'usedforsecurity' to use when checking a hash function. + # When the trait properties are callables (e.g., _md5.md5) and + # not strings, they must be called with the same 'usedforsecurity'. usedforsecurity = True - def _find_constructor(self, name): - # By default, a missing algorithm skips the test that uses it. - self.assertIn(name, self.ALGORITHMS) - self.skipTest(f"missing hash function: {name}") - - @property - def md5(self): - return self._find_constructor("md5") + def is_valid_digest_name(self, digestname): + self.assertIn(digestname, _HashId) - @property - def sha1(self): - return self._find_constructor("sha1") - - @property - def sha224(self): - return self._find_constructor("sha224") - - @property - def sha256(self): - return self._find_constructor("sha256") - - @property - def sha384(self): - return self._find_constructor("sha384") - - @property - def sha512(self): - return self._find_constructor("sha512") - - @property - def sha3_224(self): - return self._find_constructor("sha3_224") + def _find_constructor(self, digestname): + # By default, a missing algorithm skips the test that uses it. + self.is_valid_digest_name(digestname) + self.skipTest(f"missing hash function: {digestname}") - @property - def sha3_256(self): - return self._find_constructor("sha3_256") + md5 = property(lambda self: self._find_constructor("md5")) + sha1 = property(lambda self: self._find_constructor("sha1")) - @property - def sha3_384(self): - return self._find_constructor("sha3_384") + sha224 = property(lambda self: self._find_constructor("sha224")) + sha256 = property(lambda self: self._find_constructor("sha256")) + sha384 = property(lambda self: self._find_constructor("sha384")) + sha512 = property(lambda self: self._find_constructor("sha512")) - @property - def sha3_512(self): - return self._find_constructor("sha3_512") + sha3_224 = property(lambda self: self._find_constructor("sha3_224")) + sha3_256 = property(lambda self: self._find_constructor("sha3_256")) + sha3_384 = property(lambda self: self._find_constructor("sha3_384")) + sha3_512 = property(lambda self: self._find_constructor("sha3_512")) class NamedHashFunctionsTrait(HashFunctionsTrait): @@ -239,9 +792,9 @@ class NamedHashFunctionsTrait(HashFunctionsTrait): Hash functions are available if and only if they are available in hashlib. """ - def _find_constructor(self, name): - self.assertIn(name, self.ALGORITHMS) - return name + def _find_constructor(self, digestname): + self.is_valid_digest_name(digestname) + return str(digestname) # ensure that we are an exact string class OpenSSLHashFunctionsTrait(HashFunctionsTrait): @@ -250,11 +803,11 @@ class OpenSSLHashFunctionsTrait(HashFunctionsTrait): Hash functions are available if and only if they are available in _hashlib. """ - def _find_constructor(self, name): - self.assertIn(name, self.ALGORITHMS) - return find_openssl_hashdigest_constructor( - name, usedforsecurity=self.usedforsecurity - ) + def _find_constructor(self, digestname): + self.is_valid_digest_name(digestname) + # This returns a function of the form _hashlib.openssl_ and + # not a lambda function as it is rejected by _hashlib.hmac_new(). + return _openssl_hash(digestname, usedforsecurity=self.usedforsecurity) class BuiltinHashFunctionsTrait(HashFunctionsTrait): @@ -265,65 +818,278 @@ class BuiltinHashFunctionsTrait(HashFunctionsTrait): is not since the former is unconditionally built. """ - def _find_constructor_in(self, module, name): - self.assertIn(name, self.ALGORITHMS) - return find_builtin_hashdigest_constructor(module, name) + def _find_constructor(self, digestname): + self.is_valid_digest_name(digestname) + info = get_hash_func_info(digestname) + return _builtin_hash( + info.builtin.module_name, + info.builtin.member_name, + usedforsecurity=self.usedforsecurity, + ) - @property - def md5(self): - return self._find_constructor_in("_md5", "md5") - @property - def sha1(self): - return self._find_constructor_in("_sha1", "sha1") +def find_gil_minsize(modules_names, default=2048): + """Get the largest GIL_MINSIZE value for the given cryptographic modules. - @property - def sha224(self): - return self._find_constructor_in("_sha2", "sha224") + The valid module names are the following: - @property - def sha256(self): - return self._find_constructor_in("_sha2", "sha256") + - _hashlib + - _md5, _sha1, _sha2, _sha3, _blake2 + - _hmac + """ + sizes = [] + for module_name in modules_names: + module = _import_module(module_name) + if module is not None: + sizes.append(getattr(module, '_GIL_MINSIZE', default)) + return max(sizes, default=default) - @property - def sha384(self): - return self._find_constructor_in("_sha2", "sha384") - @property - def sha512(self): - return self._find_constructor_in("_sha2", "sha512") +def _block_openssl_hash_new(blocked_name): + """Block OpenSSL implementation of _hashlib.new().""" + assert isinstance(blocked_name, str), blocked_name - @property - def sha3_224(self): - return self._find_constructor_in("_sha3", "sha3_224") + # re-import '_hashlib' in case it was mocked + if (_hashlib := _import_module("_hashlib")) is None: + return contextlib.nullcontext() - @property - def sha3_256(self): - return self._find_constructor_in("_sha3","sha3_256") + @functools.wraps(wrapped := _hashlib.new) + def _hashlib_new(name, data=b'', *, usedforsecurity=True, string=None): + if name == blocked_name: + raise _hashlib.UnsupportedDigestmodError(blocked_name) + return wrapped(name, data, + usedforsecurity=usedforsecurity, string=string) - @property - def sha3_384(self): - return self._find_constructor_in("_sha3","sha3_384") + _ensure_wrapper_signature(_hashlib_new, wrapped) + return unittest.mock.patch('_hashlib.new', _hashlib_new) - @property - def sha3_512(self): - return self._find_constructor_in("_sha3","sha3_512") +def _block_openssl_hmac_new(blocked_name): + """Block OpenSSL HMAC-HASH implementation.""" + assert isinstance(blocked_name, str), blocked_name -def find_gil_minsize(modules_names, default=2048): - """Get the largest GIL_MINSIZE value for the given cryptographic modules. + # re-import '_hashlib' in case it was mocked + if (_hashlib := _import_module("_hashlib")) is None: + return contextlib.nullcontext() - The valid module names are the following: + @functools.wraps(wrapped := _hashlib.hmac_new) + def wrapper(key, msg=b'', digestmod=None): + if digestmod == blocked_name: + raise _hashlib.UnsupportedDigestmodError(blocked_name) + return wrapped(key, msg, digestmod) - - _hashlib - - _md5, _sha1, _sha2, _sha3, _blake2 - - _hmac + _ensure_wrapper_signature(wrapper, wrapped) + return unittest.mock.patch('_hashlib.hmac_new', wrapper) + + +def _block_openssl_hmac_digest(blocked_name): + """Block OpenSSL HMAC-HASH one-shot digest implementation.""" + assert isinstance(blocked_name, str), blocked_name + + # re-import '_hashlib' in case it was mocked + if (_hashlib := _import_module("_hashlib")) is None: + return contextlib.nullcontext() + + @functools.wraps(wrapped := _hashlib.hmac_digest) + def _hashlib_hmac_digest(key, msg, digest): + if digest == blocked_name: + raise _hashlib.UnsupportedDigestmodError(blocked_name) + return wrapped(key, msg, digest) + + _ensure_wrapper_signature(_hashlib_hmac_digest, wrapped) + return unittest.mock.patch('_hashlib.hmac_digest', _hashlib_hmac_digest) + + +def _block_builtin_hash_new(name): + """Block a buitin-in hash name from the hashlib.new() interface.""" + assert isinstance(name, str), name + assert name.lower() == name, f"invalid name: {name}" + assert name in _HashId, f"invalid hash: {name}" + + # Re-import 'hashlib' in case it was mocked + hashlib = importlib.import_module('hashlib') + builtin_constructor_cache = getattr(hashlib, '__builtin_constructor_cache') + builtin_constructor_cache_mock = builtin_constructor_cache.copy() + builtin_constructor_cache_mock.pop(name, None) + builtin_constructor_cache_mock.pop(name.upper(), None) + + # __get_builtin_constructor() imports the HACL* modules on demand, + # so we need to block the possibility of importing it, but only + # during the call to __get_builtin_constructor(). + get_builtin_constructor = getattr(hashlib, '__get_builtin_constructor') + builtin_module_name = get_hash_func_info(name).builtin.module_name + + @functools.wraps(get_builtin_constructor) + def get_builtin_constructor_mock(name): + with import_helper.isolated_modules(): + sys = importlib.import_module("sys") + sys.modules[builtin_module_name] = None # block module's import + return get_builtin_constructor(name) + + return unittest.mock.patch.multiple( + hashlib, + __get_builtin_constructor=get_builtin_constructor_mock, + __builtin_constructor_cache=builtin_constructor_cache_mock, + ) + + +def _block_builtin_hmac_new(blocked_name): + assert isinstance(blocked_name, str), blocked_name + + # re-import '_hmac' in case it was mocked + if (_hmac := _import_module("_hmac")) is None: + return contextlib.nullcontext() + + @functools.wraps(wrapped := _hmac.new) + def _hmac_new(key, msg=None, digestmod=None): + if digestmod == blocked_name: + raise _hmac.UnknownHashError(blocked_name) + return wrapped(key, msg, digestmod) + + _ensure_wrapper_signature(_hmac_new, wrapped) + return unittest.mock.patch('_hmac.new', _hmac_new) + + +def _block_builtin_hmac_digest(blocked_name): + assert isinstance(blocked_name, str), blocked_name + + # re-import '_hmac' in case it was mocked + if (_hmac := _import_module("_hmac")) is None: + return contextlib.nullcontext() + + @functools.wraps(wrapped := _hmac.compute_digest) + def _hmac_compute_digest(key, msg, digest): + if digest == blocked_name: + raise _hmac.UnknownHashError(blocked_name) + return wrapped(key, msg, digest) + + _ensure_wrapper_signature(_hmac_compute_digest, wrapped) + return unittest.mock.patch('_hmac.compute_digest', _hmac_compute_digest) + + +def _make_hash_constructor_blocker(name, dummy, implementation): + info = get_hash_func_info(name)[implementation] + if (wrapped := info.import_member()) is None: + # function shouldn't exist for this implementation + return contextlib.nullcontext() + wrapper = functools.wraps(wrapped)(dummy) + _ensure_wrapper_signature(wrapper, wrapped) + return unittest.mock.patch(info.fullname, wrapper) + + +def _block_hashlib_hash_constructor(name): + """Block explicit public constructors.""" + def dummy(data=b'', *, usedforsecurity=True, string=None): + raise ValueError(f"blocked explicit public hash name: {name}") + return _make_hash_constructor_blocker(name, dummy, 'hashlib') + + +def _block_openssl_hash_constructor(name): + """Block explicit OpenSSL constructors.""" + def dummy(data=b'', *, usedforsecurity=True, string=None): + raise ValueError(f"blocked explicit OpenSSL hash name: {name}") + return _make_hash_constructor_blocker(name, dummy, 'openssl') + + +def _block_builtin_hash_constructor(name): + """Block explicit HACL* constructors.""" + def dummy(data=b'', *, usedforsecurity=True, string=b''): + raise ValueError(f"blocked explicit builtin hash name: {name}") + return _make_hash_constructor_blocker(name, dummy, 'builtin') + + +def _block_builtin_hmac_constructor(name): + """Block explicit HACL* HMAC constructors.""" + info = get_hmac_item_info(name) + assert info.module_name is None or info.module_name == "_hmac", info + if (wrapped := info.import_member()) is None: + # function shouldn't exist for this implementation + return contextlib.nullcontext() + + @functools.wraps(wrapped) + def wrapper(key, obj): + raise ValueError(f"blocked hash name: {name}") + + _ensure_wrapper_signature(wrapper, wrapped) + return unittest.mock.patch(info.fullname, wrapper) + + +@contextlib.contextmanager +def block_algorithm(name, *, allow_openssl=False, allow_builtin=False): + """Block a hash algorithm for both hashing and HMAC. + + Be careful with this helper as a function may be allowed, but can + still raise a ValueError at runtime if the OpenSSL security policy + disables it, e.g., if allow_openssl=True and FIPS mode is on. """ - sizes = [] - for module_name in modules_names: - try: - module = importlib.import_module(module_name) - except ImportError: - continue - sizes.append(getattr(module, '_GIL_MINSIZE', default)) - return max(sizes, default=default) + with contextlib.ExitStack() as stack: + if not (allow_openssl or allow_builtin): + # Named constructors have a different behavior in the sense + # that they are either built-ins or OpenSSL ones, but not + # "agile" ones (namely once "hashlib" has been imported, + # they are fixed). + # + # If OpenSSL is not available, hashes fall back to built-in ones, + # in which case we don't need to block the explicit public hashes + # as they will call a mocked one. + # + # If OpenSSL is available, hashes fall back to "openssl_*" ones, + # except for BLAKE2b and BLAKE2s. + stack.enter_context(_block_hashlib_hash_constructor(name)) + elif ( + # In FIPS mode, hashlib.() functions may raise if they use + # the OpenSSL implementation, except with usedforsecurity=False. + # However, blocking such functions also means blocking them + # so we again need to block them if we want to. + (_hashlib := _import_module("_hashlib")) + and _hashlib.get_fips_mode() + and not allow_openssl + ) or ( + # Without OpenSSL, hashlib.() functions are aliases + # to built-in functions, so both of them must be blocked + # as the module may have been imported before the HACL ones. + not (_hashlib := _import_module("_hashlib")) + and not allow_builtin + ): + stack.enter_context(_block_hashlib_hash_constructor(name)) + + if not allow_openssl: + # _hashlib.new() + stack.enter_context(_block_openssl_hash_new(name)) + # _hashlib.openssl_*() + stack.enter_context(_block_openssl_hash_constructor(name)) + # _hashlib.hmac_new() + stack.enter_context(_block_openssl_hmac_new(name)) + # _hashlib.hmac_digest() + stack.enter_context(_block_openssl_hmac_digest(name)) + + if not allow_builtin: + # __get_builtin_constructor(name) + stack.enter_context(_block_builtin_hash_new(name)) + # .() + stack.enter_context(_block_builtin_hash_constructor(name)) + # _hmac.new(..., name) + stack.enter_context(_block_builtin_hmac_new(name)) + # _hmac.compute_() + stack.enter_context(_block_builtin_hmac_constructor(name)) + # _hmac.compute_digest(..., name) + stack.enter_context(_block_builtin_hmac_digest(name)) + yield + + +@contextlib.contextmanager +def block_openssl_algorithms(*, exclude=()): + """Block OpenSSL implementations, except those given in *exclude*.""" + with contextlib.ExitStack() as stack: + for name in CANONICAL_DIGEST_NAMES.difference(exclude): + stack.enter_context(block_algorithm(name, allow_builtin=True)) + yield + + +@contextlib.contextmanager +def block_builtin_algorithms(*, exclude=()): + """Block HACL* implementations, except those given in *exclude*.""" + with contextlib.ExitStack() as stack: + for name in CANONICAL_DIGEST_NAMES.difference(exclude): + stack.enter_context(block_algorithm(name, allow_openssl=True)) + yield diff --git a/Lib/test/test_hashlib.py b/Lib/test/test_hashlib.py index c7d7b801ac7..253a8f45585 100644 --- a/Lib/test/test_hashlib.py +++ b/Lib/test/test_hashlib.py @@ -18,7 +18,6 @@ import tempfile import threading import unittest -import warnings from test import support from test.support import _4G, bigmemtest from test.support import hashlib_helper @@ -95,6 +94,14 @@ def read_vectors(hash_name): yield parts +DEPRECATED_STRING_PARAMETER = re.escape( + "the 'string' keyword parameter is deprecated since " + "Python 3.15 and slated for removal in Python 3.19; " + "use the 'data' keyword parameter or pass the data " + "to hash as a positional argument instead" +) + + class HashLibTestCase(unittest.TestCase): supported_hash_names = ( 'md5', 'MD5', 'sha1', 'SHA1', 'sha224', 'SHA224', 'sha256', 'SHA256', @@ -199,6 +206,11 @@ def hash_constructors(self): constructors = self.constructors_to_test.values() return itertools.chain.from_iterable(constructors) + @property + def shake_constructors(self): + for shake_name in self.shakes: + yield from self.constructors_to_test.get(shake_name, ()) + @property def is_fips_mode(self): return get_fips_mode() @@ -260,20 +272,26 @@ def test_clinic_signature(self): with self.subTest(constructor.__name__): constructor(b'') constructor(data=b'') - constructor(string=b'') # should be deprecated in the future + with self.assertWarnsRegex(DeprecationWarning, + DEPRECATED_STRING_PARAMETER): + constructor(string=b'') digest_name = constructor(b'').name with self.subTest(digest_name): hashlib.new(digest_name, b'') hashlib.new(digest_name, data=b'') - hashlib.new(digest_name, string=b'') + with self.assertWarnsRegex(DeprecationWarning, + DEPRECATED_STRING_PARAMETER): + hashlib.new(digest_name, string=b'') # Make sure that _hashlib contains the constructor # to test when using a combination of libcrypto and # interned hash implementations. if self._hashlib and digest_name in self._hashlib._constructors: self._hashlib.new(digest_name, b'') self._hashlib.new(digest_name, data=b'') - self._hashlib.new(digest_name, string=b'') + with self.assertWarnsRegex(DeprecationWarning, + DEPRECATED_STRING_PARAMETER): + self._hashlib.new(digest_name, string=b'') @unittest.skipIf(get_fips_mode(), "skip in FIPS mode") def test_clinic_signature_errors(self): @@ -329,7 +347,9 @@ def test_clinic_signature_errors(self): def test_unknown_hash(self): self.assertRaises(ValueError, hashlib.new, 'spam spam spam spam spam') - self.assertRaises(TypeError, hashlib.new, 1) + # ensure that the exception message remains consistent + err = re.escape("new() argument 'name' must be str, not int") + self.assertRaisesRegex(TypeError, err, hashlib.new, 1) def test_new_upper_to_lower(self): self.assertEqual(hashlib.new("SHA256").name, "sha256") @@ -356,11 +376,23 @@ def test_get_builtin_constructor(self): sys.modules['_md5'] = _md5 else: del sys.modules['_md5'] - self.assertRaises(TypeError, get_builtin_constructor, 3) + # ensure that the exception message remains consistent + err = re.escape("new() argument 'name' must be str, not int") + self.assertRaises(TypeError, err, get_builtin_constructor, 3) constructor = get_builtin_constructor('md5') self.assertIs(constructor, _md5.md5) self.assertEqual(sorted(builtin_constructor_cache), ['MD5', 'md5']) + def test_copy(self): + for cons in self.hash_constructors: + h1 = cons(os.urandom(16), usedforsecurity=False) + h2 = h1.copy() + self.assertIs(type(h1), type(h2)) + self.assertEqual(h1.name, h2.name) + size = (16,) if h1.name in self.shakes else () + self.assertEqual(h1.digest(*size), h2.digest(*size)) + self.assertEqual(h1.hexdigest(*size), h2.hexdigest(*size)) + def test_hexdigest(self): for cons in self.hash_constructors: h = cons(usedforsecurity=False) @@ -371,21 +403,50 @@ def test_hexdigest(self): self.assertIsInstance(h.digest(), bytes) self.assertEqual(hexstr(h.digest()), h.hexdigest()) - def test_digest_length_overflow(self): - # See issue #34922 - large_sizes = (2**29, 2**32-10, 2**32+10, 2**61, 2**64-10, 2**64+10) - for cons in self.hash_constructors: - h = cons(usedforsecurity=False) - if h.name not in self.shakes: - continue - if HASH is not None and isinstance(h, HASH): - # _hashopenssl's take a size_t - continue - for digest in h.digest, h.hexdigest: - self.assertRaises(ValueError, digest, -10) - for length in large_sizes: - with self.assertRaises((ValueError, OverflowError)): - digest(length) + def test_shakes_zero_digest_length(self): + for constructor in self.shake_constructors: + with self.subTest(constructor=constructor): + h = constructor(b'abcdef', usedforsecurity=False) + self.assertEqual(h.digest(0), b'') + self.assertEqual(h.hexdigest(0), '') + + def test_shakes_invalid_digest_length(self): + # See https://github.com/python/cpython/issues/79103. + for constructor in self.shake_constructors: + with self.subTest(constructor=constructor): + h = constructor(usedforsecurity=False) + # Note: digest() and hexdigest() take a signed input and + # raise if it is negative; the rationale is that we use + # internally PyBytes_FromStringAndSize() and _Py_strhex() + # which both take a Py_ssize_t. + for negative_size in (-1, -10, -(1 << 31), -sys.maxsize): + self.assertRaises(ValueError, h.digest, negative_size) + self.assertRaises(ValueError, h.hexdigest, negative_size) + + def test_shakes_overflow_digest_length(self): + # See https://github.com/python/cpython/issues/135759. + + exc_types = (OverflowError, ValueError) + # HACL* accepts an 'uint32_t' while OpenSSL accepts a 'size_t'. + openssl_overflown_sizes = (sys.maxsize + 1, 2 * sys.maxsize) + # https://github.com/python/cpython/issues/79103 restricts + # the accepted built-in lengths to 2 ** 29, even if OpenSSL + # accepts such lengths. + builtin_overflown_sizes = openssl_overflown_sizes + ( + 2 ** 29, 2 ** 32 - 10, 2 ** 32, 2 ** 32 + 10, + 2 ** 61, 2 ** 64 - 10, 2 ** 64, 2 ** 64 + 10, + ) + + for constructor in self.shake_constructors: + with self.subTest(constructor=constructor): + h = constructor(usedforsecurity=False) + if HASH is not None and isinstance(h, HASH): + overflown_sizes = openssl_overflown_sizes + else: + overflown_sizes = builtin_overflown_sizes + for invalid_size in overflown_sizes: + self.assertRaises(exc_types, h.digest, invalid_size) + self.assertRaises(exc_types, h.hexdigest, invalid_size) def test_name_attribute(self): for cons in self.hash_constructors: @@ -489,13 +550,17 @@ def check(self, name, data, hexdigest, shake=False, **kwargs): def check_file_digest(self, name, data, hexdigest): hexdigest = hexdigest.lower() - try: - hashlib.new(name) - except ValueError: - # skip, algorithm is blocked by security policy. - return - digests = [name] - digests.extend(self.constructors_to_test[name]) + digests = [] + for digest in [name, *self.constructors_to_test[name]]: + try: + if callable(digest): + digest(b"") + else: + hashlib.new(digest) + except ValueError: + # skip, algorithm is blocked by security policy. + continue + digests.append(digest) with tempfile.TemporaryFile() as f: f.write(data) @@ -626,12 +691,12 @@ def test_case_md5_2(self): ) @unittest.skipIf(sys.maxsize < _4G + 5, 'test cannot run on 32-bit systems') - @bigmemtest(size=_4G + 5, memuse=1, dry_run=False) + @bigmemtest(size=_4G + 5, memuse=2, dry_run=False) def test_case_md5_huge(self, size): self.check('md5', b'A'*size, 'c9af2dff37468ce5dfee8f2cfc0a9c6d') @unittest.skipIf(sys.maxsize < _4G - 1, 'test cannot run on 32-bit systems') - @bigmemtest(size=_4G - 1, memuse=1, dry_run=False) + @bigmemtest(size=_4G - 1, memuse=2, dry_run=False) def test_case_md5_uintmax(self, size): self.check('md5', b'A'*size, '28138d306ff1b8281f1a9067e1a1a2b3') @@ -1016,49 +1081,73 @@ def test_gil(self): def test_sha256_gil(self): gil_minsize = hashlib_helper.find_gil_minsize(['_sha2', '_hashlib']) + data = b'1' + b'#' * gil_minsize + b'1' + expected = hashlib.sha256(data).hexdigest() + m = hashlib.sha256() m.update(b'1') m.update(b'#' * gil_minsize) m.update(b'1') - self.assertEqual( - m.hexdigest(), - '1cfceca95989f51f658e3f3ffe7f1cd43726c9e088c13ee10b46f57cef135b94' - ) + self.assertEqual(m.hexdigest(), expected) - m = hashlib.sha256(b'1' + b'#' * gil_minsize + b'1') - self.assertEqual( - m.hexdigest(), - '1cfceca95989f51f658e3f3ffe7f1cd43726c9e088c13ee10b46f57cef135b94' - ) + @threading_helper.reap_threads + @threading_helper.requires_working_threading() + def test_threaded_hashing_fast(self): + # Same as test_threaded_hashing_slow() but only tests some functions + # since otherwise test_hashlib.py becomes too slow during development. + algos = ['md5', 'sha1', 'sha256', 'sha3_256', 'blake2s'] + for name in algos: + if constructor := getattr(hashlib, name, None): + with self.subTest(name): + try: + self.do_test_threaded_hashing(constructor, is_shake=False) + except ValueError as exc: + self.skip_if_blake2_not_builtin(name, exc) + raise + if shake_128 := getattr(hashlib, 'shake_128', None): + self.do_test_threaded_hashing(shake_128, is_shake=True) + + @requires_resource('cpu') @threading_helper.reap_threads @threading_helper.requires_working_threading() - def test_threaded_hashing(self): + def test_threaded_hashing_slow(self): + for algorithm, constructors in self.constructors_to_test.items(): + is_shake = algorithm in self.shakes + for constructor in constructors: + with self.subTest(constructor.__name__, is_shake=is_shake): + self.do_test_threaded_hashing(constructor, is_shake) + + def do_test_threaded_hashing(self, constructor, is_shake): # Updating the same hash object from several threads at once # using data chunk sizes containing the same byte sequences. # # If the internal locks are working to prevent multiple # updates on the same object from running at once, the resulting # hash will be the same as doing it single threaded upfront. - hasher = hashlib.sha1() - num_threads = 5 - smallest_data = b'swineflu' - data = smallest_data * 200000 - expected_hash = hashlib.sha1(data*num_threads).hexdigest() - - def hash_in_chunks(chunk_size): - index = 0 - while index < len(data): - hasher.update(data[index:index + chunk_size]) - index += chunk_size + + # The data to hash has length s|M|q^N and the chunk size for the i-th + # thread is s|M|q^(N-i), where N is the number of threads, M is a fixed + # message of small length, and s >= 1 and q >= 2 are small integers. + smallest_size, num_threads, s, q = 8, 5, 2, 10 + + smallest_data = os.urandom(smallest_size) + data = s * smallest_data * (q ** num_threads) + + h1 = constructor(usedforsecurity=False) + h2 = constructor(data * num_threads, usedforsecurity=False) + + def update(chunk_size): + for index in range(0, len(data), chunk_size): + h1.update(data[index:index + chunk_size]) threads = [] - for threadnum in range(num_threads): - chunk_size = len(data) // (10 ** threadnum) + for thread_num in range(num_threads): + # chunk_size = len(data) // (q ** thread_num) + chunk_size = s * smallest_size * q ** (num_threads - thread_num) self.assertGreater(chunk_size, 0) - self.assertEqual(chunk_size % len(smallest_data), 0) - thread = threading.Thread(target=hash_in_chunks, - args=(chunk_size,)) + self.assertEqual(chunk_size % smallest_size, 0) + thread = threading.Thread(target=update, args=(chunk_size,)) threads.append(thread) for thread in threads: @@ -1066,7 +1155,10 @@ def hash_in_chunks(chunk_size): for thread in threads: thread.join() - self.assertEqual(expected_hash, hasher.hexdigest()) + if is_shake: + self.assertEqual(h1.hexdigest(16), h2.hexdigest(16)) + else: + self.assertEqual(h1.hexdigest(), h2.hexdigest()) def test_get_fips_mode(self): fips_mode = self.is_fips_mode @@ -1123,12 +1215,6 @@ class KDFTests(unittest.TestCase): (b'pass\0word', b'sa\0lt', 4096, 16), ] - scrypt_test_vectors = [ - (b'', b'', 16, 1, 1, unhexlify('77d6576238657b203b19ca42c18a0497f16b4844e3074ae8dfdffa3fede21442fcd0069ded0948f8326a753a0fc81f17e8d3e0fb2e0d3628cf35e20c38d18906')), - (b'password', b'NaCl', 1024, 8, 16, unhexlify('fdbabe1c9d3472007856e7190d01e9fe7c6ad7cbc8237830e77376634b3731622eaf30d92e22a3886ff109279d9830dac727afb94a83ee6d8360cbdfa2cc0640')), - (b'pleaseletmein', b'SodiumChloride', 16384, 8, 1, unhexlify('7023bdcb3afd7348461c06cd81fd38ebfda8fbba904f8e3ea9b543f6545da1f2d5432955613f0fcf62d49705242a9af9e61e85dc0d651e40dfcf017b45575887')), - ] - pbkdf2_results = { "sha1": [ # official test vectors from RFC 6070 @@ -1220,46 +1306,6 @@ def _test_pbkdf2_hmac(self, pbkdf2, supported): def test_pbkdf2_hmac_c(self): self._test_pbkdf2_hmac(openssl_hashlib.pbkdf2_hmac, openssl_md_meth_names) - @unittest.skipUnless(hasattr(hashlib, 'scrypt'), - ' test requires OpenSSL > 1.1') - @unittest.skipIf(get_fips_mode(), reason="scrypt is blocked in FIPS mode") - def test_scrypt(self): - for password, salt, n, r, p, expected in self.scrypt_test_vectors: - result = hashlib.scrypt(password, salt=salt, n=n, r=r, p=p) - self.assertEqual(result, expected) - - # this values should work - hashlib.scrypt(b'password', salt=b'salt', n=2, r=8, p=1) - # password and salt must be bytes-like - with self.assertRaises(TypeError): - hashlib.scrypt('password', salt=b'salt', n=2, r=8, p=1) - with self.assertRaises(TypeError): - hashlib.scrypt(b'password', salt='salt', n=2, r=8, p=1) - # require keyword args - with self.assertRaises(TypeError): - hashlib.scrypt(b'password') - with self.assertRaises(TypeError): - hashlib.scrypt(b'password', b'salt') - with self.assertRaises(TypeError): - hashlib.scrypt(b'password', 2, 8, 1, salt=b'salt') - for n in [-1, 0, 1, None]: - with self.assertRaises((ValueError, OverflowError, TypeError)): - hashlib.scrypt(b'password', salt=b'salt', n=n, r=8, p=1) - for r in [-1, 0, None]: - with self.assertRaises((ValueError, OverflowError, TypeError)): - hashlib.scrypt(b'password', salt=b'salt', n=2, r=r, p=1) - for p in [-1, 0, None]: - with self.assertRaises((ValueError, OverflowError, TypeError)): - hashlib.scrypt(b'password', salt=b'salt', n=2, r=8, p=p) - for maxmem in [-1, None]: - with self.assertRaises((ValueError, OverflowError, TypeError)): - hashlib.scrypt(b'password', salt=b'salt', n=2, r=8, p=1, - maxmem=maxmem) - for dklen in [-1, None]: - with self.assertRaises((ValueError, OverflowError, TypeError)): - hashlib.scrypt(b'password', salt=b'salt', n=2, r=8, p=1, - dklen=dklen) - def test_normalized_name(self): self.assertNotIn("blake2b512", hashlib.algorithms_available) self.assertNotIn("sha3-512", hashlib.algorithms_available) @@ -1301,5 +1347,76 @@ def readable(self): hashlib.file_digest(NonBlocking(), hashlib.sha256) +@unittest.skipUnless(hasattr(hashlib, 'scrypt'), 'requires OpenSSL 1.1+') +@unittest.skipIf(get_fips_mode(), reason="scrypt is blocked in FIPS mode") +class TestScrypt(unittest.TestCase): + + scrypt_test_vectors = [ + (b'', b'', 16, 1, 1, unhexlify('77d6576238657b203b19ca42c18a0497f16b4844e3074ae8dfdffa3fede21442fcd0069ded0948f8326a753a0fc81f17e8d3e0fb2e0d3628cf35e20c38d18906')), + (b'password', b'NaCl', 1024, 8, 16, unhexlify('fdbabe1c9d3472007856e7190d01e9fe7c6ad7cbc8237830e77376634b3731622eaf30d92e22a3886ff109279d9830dac727afb94a83ee6d8360cbdfa2cc0640')), + (b'pleaseletmein', b'SodiumChloride', 16384, 8, 1, unhexlify('7023bdcb3afd7348461c06cd81fd38ebfda8fbba904f8e3ea9b543f6545da1f2d5432955613f0fcf62d49705242a9af9e61e85dc0d651e40dfcf017b45575887')), + ] + + def test_scrypt(self): + for password, salt, n, r, p, expected in self.scrypt_test_vectors: + result = hashlib.scrypt(password, salt=salt, n=n, r=r, p=p) + self.assertEqual(result, expected) + + # these parameters must be valid + hashlib.scrypt(b'password', salt=b'salt', n=2, r=8, p=1) + hashlib.scrypt(b'password', salt=b'salt', n=2, r=8, p=1, maxmem=0) + hashlib.scrypt(b'password', salt=b'salt', n=2, r=8, p=1, dklen=1) + + def test_scrypt_types(self): + # password and salt must be bytes-like + with self.assertRaises(TypeError): + hashlib.scrypt('password', salt=b'salt', n=2, r=8, p=1) + with self.assertRaises(TypeError): + hashlib.scrypt(b'password', salt='salt', n=2, r=8, p=1) + # require keyword args + with self.assertRaises(TypeError): + hashlib.scrypt(b'password') + with self.assertRaises(TypeError): + hashlib.scrypt(b'password', b'salt') + with self.assertRaises(TypeError): + hashlib.scrypt(b'password', 2, 8, 1, salt=b'salt') + + def test_scrypt_validate(self): + def scrypt(password=b"password", /, **kwargs): + # overwrite well-defined parameters with bad ones + kwargs = dict(salt=b'salt', n=2, r=8, p=1) | kwargs + return hashlib.scrypt(password, **kwargs) + + for param_name in ('n', 'r', 'p', 'maxmem', 'dklen'): + param = {param_name: None} + with self.subTest(**param): + self.assertRaises(TypeError, scrypt, **param) + + self.assertRaises(ValueError, scrypt, n=0) + self.assertRaises(ValueError, scrypt, n=-1) + self.assertRaises(ValueError, scrypt, n=1) + + self.assertRaises(ValueError, scrypt, r=0) + self.assertRaises(ValueError, scrypt, r=-1) + + self.assertRaises(ValueError, scrypt, p=-1) + self.assertRaises(ValueError, scrypt, p=0) + + self.assertRaises(ValueError, scrypt, maxmem=-1) + # OpenSSL hard limit for 'maxmem' is an 'uint64_t' but for now, + # we do not use the 'uint64' Clinic converter but the 'long' one. + self.assertRaises(OverflowError, scrypt, maxmem=(1 << 64)) + # Historically, Python allowed 'maxmem' to be at most INT_MAX, + # which is at most 2**32-1 (on Windows, sizeof(long) == 4, so + # an OverflowError will be raised instead of a ValueError). + numeric_exc_types = (OverflowError, ValueError) + self.assertRaises(numeric_exc_types, scrypt, maxmem=(1 << 32)) + + self.assertRaises(ValueError, scrypt, dklen=-1) + self.assertRaises(ValueError, scrypt, dklen=0) + MAX_DKLEN = ((1 << 32) - 1) * 32 # see RFC 7914 + self.assertRaises(numeric_exc_types, scrypt, dklen=MAX_DKLEN + 1) + + if __name__ == "__main__": unittest.main() diff --git a/Lib/test/test_hmac.py b/Lib/test/test_hmac.py index 1d270b2a43c..c43188cc333 100644 --- a/Lib/test/test_hmac.py +++ b/Lib/test/test_hmac.py @@ -21,10 +21,10 @@ import hmac import hashlib import random -import types import unittest import warnings from _operator import _compare_digest as operator_compare_digest +from test import support from test.support import _4G, bigmemtest from test.support import check_disallow_instantiation from test.support import hashlib_helper, import_helper @@ -161,7 +161,7 @@ def hmac_digest(self, key, msg=None, digestmod=DIGESTMOD_SENTINEL): return _call_digest_func(self.hmac.digest, key, msg, digestmod) -@hashlib_helper.requires_hashlib() +@hashlib_helper.requires_openssl_hashlib() class ThroughOpenSSLAPIMixin(CreatorMixin, DigestMixin): """Mixin delegating to _hashlib.hmac_new() and _hashlib.hmac_digest().""" @@ -303,7 +303,7 @@ def assert_hmac_new_by_name( def check_hmac_new( self, key, msg, hexdigest, hashname, digest_size, block_size, - hmac_new_func, hmac_new_kwds=types.MappingProxyType({}), + hmac_new_func, hmac_new_kwds=frozendict(), ): """Check that HMAC(key, msg) == digest. @@ -349,7 +349,7 @@ def assert_hmac_hexdigest_by_name( def check_hmac_hexdigest( self, key, msg, hexdigest, digest_size, - hmac_digest_func, hmac_digest_kwds=types.MappingProxyType({}), + hmac_digest_func, hmac_digest_kwds=frozendict(), ): """Check and return a HMAC digest computed by hmac_digest_func(). @@ -965,7 +965,7 @@ def raiser(): with self.assertRaisesRegex(RuntimeError, "custom exception"): func(b'key', b'msg', raiser) - with self.assertRaisesRegex(ValueError, 'hash type'): + with self.assertRaisesRegex(ValueError, 'unsupported hash algorithm'): func(b'key', b'msg', 'unknown') with self.assertRaisesRegex(AttributeError, 'new'): @@ -1025,6 +1025,13 @@ def test_hmac_digest_digestmod_parameter(self): ): self.hmac_digest(b'key', b'msg', value) + @support.subTests("xof_name", ("shake_128", "shake_256")) + def test_hmac_new_xof_digestmod(self, xof_name): + # gh-145200: XOF digests (SHAKE) are not supported by HMAC. + # Verify that the error path does not leak the EVP_MAC_CTX. + with self.assertRaises(_hashlib.UnsupportedDigestmodError): + self.hmac_new(b'key', digestmod=xof_name) + class BuiltinConstructorTestCase(ThroughBuiltinAPIMixin, ExtensionConstructorTestCaseMixin, @@ -1440,7 +1447,7 @@ def test_compare_digest_func(self): self.assertIs(self.compare_digest, operator_compare_digest) -@hashlib_helper.requires_hashlib() +@hashlib_helper.requires_openssl_hashlib() class OpenSSLCompareDigestTestCase(CompareDigestMixin, unittest.TestCase): compare_digest = openssl_compare_digest @@ -1452,19 +1459,25 @@ class OperatorCompareDigestTestCase(CompareDigestMixin, unittest.TestCase): class PyMiscellaneousTests(unittest.TestCase): """Miscellaneous tests for the pure Python HMAC module.""" + @staticmethod + def mock__compute_digest_fallback(hmac): + fn = getattr(hmac, meth := "_compute_digest_fallback") + return patch.object(hmac, meth, wraps=fn) + + @staticmethod + def mock_HMAC_method(hmac, method): + fn = getattr(hmac.HMAC, method) + return patch.object(hmac.HMAC, method, autospec=True, wraps=fn) + @hashlib_helper.requires_builtin_hmac() def test_hmac_constructor_uses_builtin(self): # Block the OpenSSL implementation and check that # HMAC() uses the built-in implementation instead. hmac = import_fresh_module("hmac", blocked=["_hashlib"]) - def watch_method(cls, name): - wraps = getattr(cls, name) - return patch.object(cls, name, autospec=True, wraps=wraps) - with ( - watch_method(hmac.HMAC, '_init_openssl_hmac') as f, - watch_method(hmac.HMAC, '_init_builtin_hmac') as g, + self.mock_HMAC_method(hmac, '_init_openssl_hmac') as f, + self.mock_HMAC_method(hmac, '_init_builtin_hmac') as g, ): _ = hmac.HMAC(b'key', b'msg', digestmod="sha256") f.assert_not_called() @@ -1518,7 +1531,7 @@ def test_hmac_digest_overflow_error_openssl_only(self, size): hmac = import_fresh_module("hmac", blocked=["_hmac"]) self.do_test_hmac_digest_overflow_error_switch_to_slow(hmac, size) - @hashlib_helper.requires_builtin_hashdigest("_md5", "md5") + @hashlib_helper.requires_builtin_hashdigest("md5") @bigmemtest(size=_4G + 5, memuse=2, dry_run=False) def test_hmac_digest_overflow_error_builtin_only(self, size): hmac = import_fresh_module("hmac", blocked=["_hashlib"]) @@ -1535,11 +1548,11 @@ def do_test_hmac_digest_overflow_error_switch_to_slow(self, hmac, size): bigkey = b'K' * size bigmsg = b'M' * size - with patch.object(hmac, "_compute_digest_fallback") as slow: + with self.mock__compute_digest_fallback(hmac) as slow: hmac.digest(bigkey, b'm', "md5") slow.assert_called_once() - with patch.object(hmac, "_compute_digest_fallback") as slow: + with self.mock__compute_digest_fallback(hmac) as slow: hmac.digest(b'k', bigmsg, "md5") slow.assert_called_once() @@ -1550,7 +1563,7 @@ def test_hmac_digest_no_overflow_error_in_fallback(self, size): for key, msg in [(b'K' * size, b'm'), (b'k', b'M' * size)]: with self.subTest(keysize=len(key), msgsize=len(msg)): - with patch.object(hmac, "_compute_digest_fallback") as slow: + with self.mock__compute_digest_fallback(hmac) as slow: hmac.digest(key, msg, "md5") slow.assert_called_once() diff --git a/crates/stdlib/src/hashlib.rs b/crates/stdlib/src/hashlib.rs index ceca00d0bd9..76b17c999ca 100644 --- a/crates/stdlib/src/hashlib.rs +++ b/crates/stdlib/src/hashlib.rs @@ -244,7 +244,18 @@ pub(crate) mod _hashlib { ) -> PyResult> { match (data.into_option(), string) { (Some(d), None) => Ok(OptionalArg::Present(d)), - (None, Some(s)) => Ok(OptionalArg::Present(s)), + (None, Some(s)) => { + crate::vm::stdlib::_warnings::warn( + vm.ctx.exceptions.deprecation_warning, + "the 'string' keyword parameter is deprecated since Python 3.15 \ + and slated for removal in Python 3.19; use the 'data' keyword \ + parameter or pass the data to hash as a positional argument instead" + .to_owned(), + 1, + vm, + )?; + Ok(OptionalArg::Present(s)) + } (None, None) => Ok(OptionalArg::Missing), (Some(_), Some(_)) => Err(vm.new_type_error( "'data' and 'string' are mutually exclusive \