+import contextlib
import functools
import hashlib
import importlib
+import inspect
import unittest
+import unittest.mock
+from collections import namedtuple
from test.support.import_helper import import_module
+from types import MappingProxyType
try:
import _hashlib
_hmac = None
+CANONICAL_DIGEST_NAMES = frozenset((
+ 'md5', 'sha1',
+ 'sha224', 'sha256', 'sha384', 'sha512',
+ 'sha3_224', 'sha3_256', 'sha3_384', 'sha3_512',
+ 'shake_128', 'shake_256',
+ 'blake2s', 'blake2b',
+))
+
+NON_HMAC_DIGEST_NAMES = frozenset({
+ 'shake_128', 'shake_256',
+ 'blake2s', 'blake2b',
+})
+
+
+class HashAPI(namedtuple("HashAPI", "builtin openssl hashlib")):
+
+ def fullname(self, typ):
+ match typ:
+ case "builtin":
+ return self.builtin
+ case "openssl":
+ return f"_hashlib.{self.openssl}" if self.openssl else None
+ case "hashlib":
+ return f"hashlib.{self.hashlib}" if self.hashlib else None
+ case _:
+ raise AssertionError(f"unknown type: {typ}")
+
+
+# Mapping from a "canonical" name to a pair (HACL*, _hashlib.*, hashlib.*)
+# constructors. If the constructor name is None, then this means that the
+# algorithm can only be used by the "agile" new() interfaces.
+_EXPLICIT_CONSTRUCTORS = MappingProxyType({
+ "md5": HashAPI("_md5.md5", "openssl_md5", "md5"),
+ "sha1": HashAPI("_sha1.sha1", "openssl_sha1", "sha1"),
+ "sha224": HashAPI("_sha2.sha224", "openssl_sha224", "sha224"),
+ "sha256": HashAPI("_sha2.sha256", "openssl_sha256", "sha256"),
+ "sha384": HashAPI("_sha2.sha384", "openssl_sha384", "sha384"),
+ "sha512": HashAPI("_sha2.sha512", "openssl_sha512", "sha512"),
+ "sha3_224": HashAPI("_sha3.sha3_224", "openssl_sha3_224", "sha3_224"),
+ "sha3_256": HashAPI("_sha3.sha3_256", "openssl_sha3_256", "sha3_256"),
+ "sha3_384": HashAPI("_sha3.sha3_384", "openssl_sha3_384", "sha3_384"),
+ "sha3_512": HashAPI("_sha3.sha3_512", "openssl_sha3_512", "sha3_512"),
+ "shake_128": HashAPI("_sha3.shake_128", "openssl_shake_128", "shake_128"),
+ "shake_256": HashAPI("_sha3.shake_256", "openssl_shake_256", "shake_256"),
+ "blake2s": HashAPI("_blake2.blake2s", None, "blake2s"),
+ "blake2b": HashAPI("_blake2.blake2b", None, "blake2b"),
+})
+assert _EXPLICIT_CONSTRUCTORS.keys() == CANONICAL_DIGEST_NAMES
+
+_EXPLICIT_HMAC_CONSTRUCTORS = {
+ name: f'_hmac.compute_{name}' for name in (
+ 'md5', 'sha1',
+ 'sha224', 'sha256', 'sha384', 'sha512',
+ 'sha3_224', 'sha3_256', 'sha3_384', 'sha3_512',
+ )
+}
+_EXPLICIT_HMAC_CONSTRUCTORS['shake_128'] = None
+_EXPLICIT_HMAC_CONSTRUCTORS['shake_256'] = None
+# Strictly speaking, HMAC-BLAKE is meaningless as BLAKE2 is already a
+# keyed hash function. However, as it's exposed by HACL*, we test it.
+_EXPLICIT_HMAC_CONSTRUCTORS['blake2s'] = '_hmac.compute_blake2s_32'
+_EXPLICIT_HMAC_CONSTRUCTORS['blake2b'] = '_hmac.compute_blake2b_32'
+_EXPLICIT_HMAC_CONSTRUCTORS = MappingProxyType(_EXPLICIT_HMAC_CONSTRUCTORS)
+assert _EXPLICIT_HMAC_CONSTRUCTORS.keys() == CANONICAL_DIGEST_NAMES
+
+
+def _ensure_wrapper_signature(wrapper, wrapped):
+ """Ensure that a wrapper has the same signature as the wrapped function.
+
+ 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).
+ """
+ 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 requires_hashlib():
return unittest.skipIf(_hashlib is None, "requires _hashlib")
def _openssl_availabillity(digestname, *, usedforsecurity):
+ assert isinstance(digestname, str), digestname
try:
_hashlib.new(digestname, usedforsecurity=usedforsecurity)
except AttributeError:
ValueError: [digital envelope routines: EVP_DigestInit_ex] disabled for FIPS
ValueError: unsupported hash type md4
"""
+ assert isinstance(digestname, str), digestname
if openssl and _hashlib is not None:
def test_availability():
_hashlib.new(digestname, usedforsecurity=usedforsecurity)
The hashing algorithm may be missing or blocked by a strict crypto policy.
"""
+ assert isinstance(digestname, str), digestname
def decorator_func(func):
@requires_hashlib() # avoid checking at each call
@functools.wraps(func)
- The *module_name* is the C extension module name based on HACL*.
- The *digestname* is one of its member, e.g., 'md5'.
"""
+ assert isinstance(digestname, str), digestname
def decorator_func(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
- The *module_name* is the C extension module name based on HACL*.
- The *digestname* is one of its member, e.g., 'md5'.
"""
+ assert isinstance(digestname, str), digestname
module = import_module(module_name)
try:
constructor = getattr(module, digestname)
implementation of HMAC).
"""
- ALGORITHMS = [
+ DIGEST_NAMES = [
'md5', 'sha1',
'sha224', 'sha256', 'sha384', 'sha512',
'sha3_224', 'sha3_256', 'sha3_384', 'sha3_512',
# Default 'usedforsecurity' to use when looking up a hash function.
usedforsecurity = True
- def _find_constructor(self, name):
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ assert CANONICAL_DIGEST_NAMES.issuperset(cls.DIGEST_NAMES)
+
+ def is_valid_digest_name(self, digestname):
+ self.assertIn(digestname, self.DIGEST_NAMES)
+
+ def _find_constructor(self, digestname):
# By default, a missing algorithm skips the test that uses it.
- self.assertIn(name, self.ALGORITHMS)
- self.skipTest(f"missing hash function: {name}")
+ self.is_valid_digest_name(digestname)
+ self.skipTest(f"missing hash function: {digestname}")
@property
def md5(self):
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 digestname
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)
+ def _find_constructor(self, digestname):
+ self.is_valid_digest_name(digestname)
return find_openssl_hashdigest_constructor(
- name, usedforsecurity=self.usedforsecurity
+ digestname, usedforsecurity=self.usedforsecurity
)
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_in(self, module, digestname):
+ self.is_valid_digest_name(digestname)
+ return find_builtin_hashdigest_constructor(module, digestname)
@property
def md5(self):
continue
sizes.append(getattr(module, '_GIL_MINSIZE', default))
return max(sizes, default=default)
+
+
+def _block_openssl_hash_new(blocked_name):
+ """Block OpenSSL implementation of _hashlib.new()."""
+ assert isinstance(blocked_name, str), blocked_name
+ if _hashlib is None:
+ return contextlib.nullcontext()
+ @functools.wraps(wrapped := _hashlib.new)
+ def wrapper(name, data=b'', *, usedforsecurity=True, string=None):
+ if name == blocked_name:
+ raise _hashlib.UnsupportedDigestmodError(blocked_name)
+ return wrapped(*args, **kwargs)
+ _ensure_wrapper_signature(wrapper, wrapped)
+ return unittest.mock.patch('_hashlib.new', wrapper)
+
+
+def _block_openssl_hmac_new(blocked_name):
+ """Block OpenSSL HMAC-HASH implementation."""
+ assert isinstance(blocked_name, str), blocked_name
+ if _hashlib is None:
+ return contextlib.nullcontext()
+ @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)
+ _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
+ if _hashlib is None:
+ return contextlib.nullcontext()
+ @functools.wraps(wrapped := _hashlib.hmac_digest)
+ def wrapper(key, msg, digest):
+ if digest == blocked_name:
+ raise _hashlib.UnsupportedDigestmodError(blocked_name)
+ return wrapped(key, msg, digestmod)
+ _ensure_wrapper_signature(wrapper, wrapped)
+ return unittest.mock.patch('_hashlib.hmac_digest', wrapper)
+
+
+@contextlib.contextmanager
+def _block_builtin_hash_new(name):
+ assert isinstance(name, str), name
+ assert name.lower() == name, f"invalid name: {name}"
+
+ builtin_cache = getattr(hashlib, '__builtin_constructor_cache')
+ if name in builtin_cache:
+ f = builtin_cache.pop(name)
+ F = builtin_cache.pop(name.upper(), None)
+ else:
+ f = F = None
+ try:
+ yield
+ finally:
+ if f is not None:
+ builtin_cache[name] = f
+ if F is not None:
+ builtin_cache[name.upper()] = F
+
+
+def _block_builtin_hmac_new(blocked_name):
+ assert isinstance(blocked_name, str), blocked_name
+ if _hmac is None:
+ return contextlib.nullcontext()
+ @functools.wraps(wrapped := _hmac.new)
+ def wrapper(key, msg=None, digestmod=None):
+ if digestmod == blocked_name:
+ raise _hmac.UnknownHashError(blocked_name)
+ return wrapped(key, msg, digestmod)
+ _ensure_wrapper_signature(wrapper, wrapped)
+ return unittest.mock.patch('_hmac.new', wrapper)
+
+
+def _block_builtin_hmac_digest(blocked_name):
+ assert isinstance(blocked_name, str), blocked_name
+ if _hmac is None:
+ return contextlib.nullcontext()
+ @functools.wraps(wrapped := _hmac.compute_digest)
+ def wrapper(key, msg, digest):
+ if digest == blocked_name:
+ raise _hmac.UnknownHashError(blocked_name)
+ return wrapped(key, msg, digest)
+ _ensure_wrapper_signature(wrapper, wrapped)
+ return unittest.mock.patch('_hmac.compute_digest', wrapper)
+
+
+def _make_hash_constructor_blocker(name, dummy, *, interface):
+ assert isinstance(name, str), name
+ assert interface in ('builtin', 'openssl', 'hashlib')
+ assert name in _EXPLICIT_CONSTRUCTORS, f"invalid hash: {name}"
+ fullname = _EXPLICIT_CONSTRUCTORS[name].fullname(interface)
+ if fullname is None:
+ # function shouldn't exist for this implementation
+ return contextlib.nullcontext()
+ assert fullname.count('.') == 1, fullname
+ module_name, method = fullname.split('.', maxsplit=1)
+ try:
+ module = importlib.import_module(module_name)
+ except ImportError:
+ # module is already disabled
+ return contextlib.nullcontext()
+ wrapped = getattr(module, method)
+ wrapper = functools.wraps(wrapped)(dummy)
+ _ensure_wrapper_signature(wrapper, wrapped)
+ return unittest.mock.patch(fullname, wrapper)
+
+
+def _block_hashlib_hash_constructor(name):
+ """Block explicit public constructors."""
+ assert isinstance(name, str), name
+ def dummy(data=b'', *, usedforsecurity=True, string=None):
+ raise ValueError(f"unsupported hash name: {name}")
+ return _make_hash_constructor_blocker(name, dummy, interface='hashlib')
+
+
+def _block_openssl_hash_constructor(name):
+ """Block explicit OpenSSL constructors."""
+ assert isinstance(name, str), name
+ def dummy(data=b'', *, usedforsecurity=True, string=None):
+ raise ValueError(f"unsupported hash name: {name}")
+ return _make_hash_constructor_blocker(name, dummy, interface='openssl')
+
+
+def _block_builtin_hash_constructor(name):
+ """Block explicit HACL* constructors."""
+ assert isinstance(name, str), name
+ def dummy(data=b'', *, usedforsecurity=True, string=b''):
+ raise ValueError(f"unsupported hash name: {name}")
+ return _make_hash_constructor_blocker(name, dummy, interface='builtin')
+
+
+def _block_builtin_hmac_constructor(name):
+ """Block explicit HACL* HMAC constructors."""
+ assert isinstance(name, str), name
+ assert name in _EXPLICIT_HMAC_CONSTRUCTORS, f"invalid hash: {name}"
+ fullname = _EXPLICIT_HMAC_CONSTRUCTORS[name]
+ if fullname is None:
+ # function shouldn't exist for this implementation
+ return contextlib.nullcontext()
+ assert fullname.count('.') == 1, fullname
+ module_name, method = fullname.split('.', maxsplit=1)
+ assert module_name == '_hmac', module_name
+ try:
+ module = importlib.import_module(module_name)
+ except ImportError:
+ # module is already disabled
+ return contextlib.nullcontext()
+ @functools.wraps(wrapped := getattr(module, method))
+ def wrapper(key, obj):
+ raise ValueError(f"unsupported hash name: {name}")
+ _ensure_wrapper_signature(wrapper, wrapped)
+ return unittest.mock.patch(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.
+ """
+ with contextlib.ExitStack() as stack:
+ if not (allow_openssl or allow_builtin):
+ # If one of the private interface is allowed, then the
+ # public interface will fallback to it even though the
+ # comment in hashlib.py says otherwise.
+ #
+ # So we should only block it if the private interfaces
+ # are blocked as well.
+ stack.enter_context(_block_hashlib_hash_constructor(name))
+ if not allow_openssl:
+ stack.enter_context(_block_openssl_hash_new(name))
+ stack.enter_context(_block_openssl_hmac_new(name))
+ stack.enter_context(_block_openssl_hmac_digest(name))
+ stack.enter_context(_block_openssl_hash_constructor(name))
+ if not allow_builtin:
+ stack.enter_context(_block_builtin_hash_new(name))
+ stack.enter_context(_block_builtin_hmac_new(name))
+ stack.enter_context(_block_builtin_hmac_digest(name))
+ stack.enter_context(_block_builtin_hash_constructor(name))
+ stack.enter_context(_block_builtin_hmac_constructor(name))
+ yield
import contextlib
import errno
import importlib
+import itertools
import io
import logging
import os
import warnings
from test import support
+from test.support import hashlib_helper
from test.support import import_helper
from test.support import os_helper
from test.support import script_helper
# SuppressCrashReport
+class TestHashlibSupport(unittest.TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ cls.hashlib = import_helper.import_module("hashlib")
+ cls.hmac = import_helper.import_module("hmac")
+
+ # We required the extension modules to be present since blocking
+ # HACL* implementations while allowing OpenSSL ones would still
+ # result in failures.
+ cls._hashlib = import_helper.import_module("_hashlib")
+ cls._hmac = import_helper.import_module("_hmac")
+
+ def check_context(self, disabled=True):
+ if disabled:
+ return self.assertRaises(ValueError)
+ return contextlib.nullcontext()
+
+ def try_import_attribute(self, fullname, default=None):
+ if fullname is None:
+ return default
+ assert fullname.count('.') == 1, fullname
+ module_name, attribute = fullname.split('.', maxsplit=1)
+ try:
+ module = importlib.import_module(module_name)
+ except ImportError:
+ return default
+ try:
+ return getattr(module, attribute, default)
+ except TypeError:
+ return default
+
+ def validate_modules(self):
+ if hasattr(hashlib_helper, 'hashlib'):
+ self.assertIs(hashlib_helper.hashlib, self.hashlib)
+ if hasattr(hashlib_helper, 'hmac'):
+ self.assertIs(hashlib_helper.hmac, self.hmac)
+
+ def fetch_hash_function(self, name, typ):
+ entry = hashlib_helper._EXPLICIT_CONSTRUCTORS[name]
+ match typ:
+ case "hashlib":
+ assert entry.hashlib is not None, entry
+ return getattr(self.hashlib, entry.hashlib)
+ case "openssl":
+ try:
+ return getattr(self._hashlib, entry.openssl, None)
+ except TypeError:
+ return None
+ case "builtin":
+ return self.try_import_attribute(entry.fullname(typ))
+
+ def fetch_hmac_function(self, name):
+ fullname = hashlib_helper._EXPLICIT_HMAC_CONSTRUCTORS[name]
+ return self.try_import_attribute(fullname)
+
+ def check_openssl_hash(self, name, *, disabled=True):
+ """Check that OpenSSL HASH interface is enabled/disabled."""
+ with self.check_context(disabled):
+ _ = self._hashlib.new(name)
+ if do_hash := self.fetch_hash_function(name, "openssl"):
+ self.assertStartsWith(do_hash.__name__, 'openssl_')
+ with self.check_context(disabled):
+ _ = do_hash(b"")
+
+ def check_openssl_hmac(self, name, *, disabled=True):
+ """Check that OpenSSL HMAC interface is enabled/disabled."""
+ if name in hashlib_helper.NON_HMAC_DIGEST_NAMES:
+ # HMAC-BLAKE and HMAC-SHAKE raise a ValueError as they are not
+ # supported at all (they do not make any sense in practice).
+ with self.assertRaises(ValueError):
+ self._hashlib.hmac_digest(b"", b"", name)
+ else:
+ with self.check_context(disabled):
+ _ = self._hashlib.hmac_digest(b"", b"", name)
+ # OpenSSL does not provide one-shot explicit HMAC functions
+
+ def check_builtin_hash(self, name, *, disabled=True):
+ """Check that HACL* HASH interface is enabled/disabled."""
+ if do_hash := self.fetch_hash_function(name, "builtin"):
+ self.assertEqual(do_hash.__name__, name)
+ with self.check_context(disabled):
+ _ = do_hash(b"")
+
+ def check_builtin_hmac(self, name, *, disabled=True):
+ """Check that HACL* HMAC interface is enabled/disabled."""
+ if name in hashlib_helper.NON_HMAC_DIGEST_NAMES:
+ # HMAC-BLAKE and HMAC-SHAKE raise a ValueError as they are not
+ # supported at all (they do not make any sense in practice).
+ with self.assertRaises(ValueError):
+ self._hmac.compute_digest(b"", b"", name)
+ else:
+ with self.check_context(disabled):
+ _ = self._hmac.compute_digest(b"", b"", name)
+
+ with self.check_context(disabled):
+ _ = self._hmac.new(b"", b"", name)
+
+ if do_hmac := self.fetch_hmac_function(name):
+ self.assertStartsWith(do_hmac.__name__, 'compute_')
+ with self.check_context(disabled):
+ _ = do_hmac(b"", b"")
+ else:
+ self.assertIn(name, hashlib_helper.NON_HMAC_DIGEST_NAMES)
+
+ @support.subTests(
+ ('name', 'allow_openssl', 'allow_builtin'),
+ itertools.product(
+ hashlib_helper.CANONICAL_DIGEST_NAMES,
+ [True, False],
+ [True, False],
+ )
+ )
+ def test_disable_hash(self, name, allow_openssl, allow_builtin):
+ # In FIPS mode, the function may be available but would still need
+ # to raise a ValueError. For simplicity, we don't test the helper
+ # when we're in FIPS mode.
+ if self._hashlib.get_fips_mode():
+ self.skipTest("hash functions may still be blocked in FIPS mode")
+ flags = dict(allow_openssl=allow_openssl, allow_builtin=allow_builtin)
+ is_simple_disabled = not allow_builtin and not allow_openssl
+
+ with hashlib_helper.block_algorithm(name, **flags):
+ self.validate_modules()
+
+ # OpenSSL's blake2s and blake2b are unknown names
+ # when only the OpenSSL interface is available.
+ if allow_openssl and not allow_builtin:
+ aliases = {'blake2s': 'blake2s256', 'blake2b': 'blake2b512'}
+ name_for_hashlib_new = aliases.get(name, name)
+ else:
+ name_for_hashlib_new = name
+
+ with self.check_context(is_simple_disabled):
+ _ = self.hashlib.new(name_for_hashlib_new)
+ with self.check_context(is_simple_disabled):
+ _ = getattr(self.hashlib, name)(b"")
+
+ self.check_openssl_hash(name, disabled=not allow_openssl)
+ self.check_builtin_hash(name, disabled=not allow_builtin)
+
+ if name not in hashlib_helper.NON_HMAC_DIGEST_NAMES:
+ with self.check_context(is_simple_disabled):
+ _ = self.hmac.new(b"", b"", name)
+ with self.check_context(is_simple_disabled):
+ _ = self.hmac.HMAC(b"", b"", name)
+ with self.check_context(is_simple_disabled):
+ _ = self.hmac.digest(b"", b"", name)
+
+ self.check_openssl_hmac(name, disabled=not allow_openssl)
+ self.check_builtin_hmac(name, disabled=not allow_builtin)
+
+
if __name__ == '__main__':
unittest.main()