Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion Lib/test/libregrtest/findtests.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,16 @@
"test_asyncio",
"test_concurrent_futures",
"test_doctests",
"test_free_threading",
"test_future_stmt",
"test_gdb",
"test_inspect",
"test_pydoc",
"test_io",
"test_multiprocessing_fork",
"test_multiprocessing_forkserver",
"test_multiprocessing_spawn",
"test_os",
"test_pydoc",
}


Expand Down
16 changes: 13 additions & 3 deletions Lib/test/support/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -658,6 +658,7 @@ def requires_working_socket(*, module=False):
else:
return unittest.skipUnless(has_socket_support, msg)


@functools.cache
def has_remote_subprocess_debugging():
"""Check if we have permissions to debug subprocesses remotely.
Expand Down Expand Up @@ -1245,7 +1246,6 @@ def __init__(self):
self.started = False

def start(self):
import warnings
try:
f = open(self.procfile, 'r')
except OSError as e:
Expand Down Expand Up @@ -1281,6 +1281,7 @@ def bigmemtest(size, memuse, dry_run=True):
test doesn't support dummy runs when -M is not specified.
"""
def decorator(f):
@functools.wraps(f)
def wrapper(self):
size = wrapper.size
memuse = wrapper.memuse
Expand Down Expand Up @@ -1333,6 +1334,7 @@ def internal(*args, **kwargs):

def bigaddrspacetest(f):
"""Decorator for tests that fill the address space."""
@functools.wraps(f)
def wrapper(self):
if max_memuse < MAX_Py_ssize_t:
if MAX_Py_ssize_t >= 2**63 - 1 and max_memuse >= 2**31:
Expand Down Expand Up @@ -1414,7 +1416,7 @@ def trace_wrapper(*args, **kwargs):
sys.settrace(original_trace)

coverage_wrapper = trace_wrapper
if 'test.cov' in sys.modules: # -Xpresite=test.cov used
if 'test.cov' in sys.modules: # -Xpresite=test.cov:enable used
cov = sys.monitoring.COVERAGE_ID
@functools.wraps(func)
def coverage_wrapper(*args, **kwargs):
Expand All @@ -1438,6 +1440,7 @@ def no_rerun(reason):
def deco(func):
assert not isinstance(func, type), func
_has_run = False
@functools.wraps(func)
def wrapper(self):
nonlocal _has_run
if _has_run:
Expand Down Expand Up @@ -3003,7 +3006,7 @@ def iter_builtin_types():
# Fall back to making a best-effort guess.
if hasattr(object, '__flags__'):
# Look for any type object with the Py_TPFLAGS_STATIC_BUILTIN flag set.
import datetime
import datetime # noqa: F401
seen = set()
for cls, subs in walk_class_hierarchy(object):
if cls in seen:
Expand Down Expand Up @@ -3139,6 +3142,13 @@ def force_color(color: bool):
import _colorize
from .os_helper import EnvironmentVarGuard

if color:
try:
import _pyrepl # noqa: F401
except ModuleNotFoundError:
# Can't force enable color without _pyrepl, so just skip.
raise unittest.SkipTest("_pyrepl is missing")

with (
swap_attr(_colorize, "can_colorize", lambda *, file=None: color),
EnvironmentVarGuard() as env,
Expand Down
247 changes: 245 additions & 2 deletions Lib/test/test_support.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
import contextlib
import errno
import importlib
import itertools
import inspect
import io
import logging
import os
Expand All @@ -18,6 +20,7 @@

from test import support
from test.support import isolation
from test.support import hashlib_helper
from test.support import import_helper
from test.support import os_helper
from test.support import script_helper
Expand Down Expand Up @@ -94,9 +97,15 @@ def test_get_attribute(self):
self.test_get_attribute)
self.assertRaises(unittest.SkipTest, support.get_attribute, self, "foo")

@unittest.skip("failing buildbots")
@unittest.skipIf(support.is_android or support.is_apple_mobile,
'Mobile platforms redirect stdout to system log')
def test_get_original_stdout(self):
self.assertEqual(support.get_original_stdout(), sys.stdout)
if isinstance(sys.stdout, io.StringIO):
# gh-55258: When --junit-xml is used, stdout is a StringIO:
# use sys.__stdout__ in this case.
self.assertEqual(support.get_original_stdout(), sys.__stdout__)
else:
self.assertEqual(support.get_original_stdout(), sys.stdout)

def test_unload(self):
import sched # noqa: F401
Expand Down Expand Up @@ -483,6 +492,7 @@ def test_check__all__(self):

self.assertRaises(AssertionError, support.check__all__, self, unittest)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()
@unittest.skipUnless(hasattr(os, 'waitpid') and hasattr(os, 'WNOHANG'),
'need os.waitpid() and os.WNOHANG')
@support.requires_fork()
Expand Down Expand Up @@ -834,6 +844,239 @@ def test_linked_to_musl(self):
# SuppressCrashReport


@hashlib_helper.requires_builtin_hashes()
class TestHashlibSupport(unittest.TestCase):

@classmethod
def setUpClass(cls):
super().setUpClass()
cls.hashlib = import_helper.import_module("hashlib")
cls.hmac = import_helper.import_module("hmac")

# All C extension modules must be present since blocking
# the built-in implementation while allowing OpenSSL or vice-versa
# may result in failures depending on the exposed built-in hashes.
cls._hashlib = import_helper.import_module("_hashlib")
cls._hmac = import_helper.import_module("_hmac")
cls._md5 = import_helper.import_module("_md5")

def skip_if_fips_mode(self):
if self._hashlib.get_fips_mode():
self.skipTest("disabled in FIPS mode")

def skip_if_not_fips_mode(self):
if not self._hashlib.get_fips_mode():
self.skipTest("requires FIPS mode")

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 fetch_hash_function(self, name, implementation):
info = hashlib_helper.get_hash_func_info(name)
match hashlib_helper.Implementation(implementation):
case hashlib_helper.Implementation.hashlib:
method_name = info.hashlib.member_name
assert isinstance(method_name, str), method_name
return getattr(self.hashlib, method_name)
case hashlib_helper.Implementation.openssl:
method_name = info.openssl.member_name
assert isinstance(method_name, str | None), method_name
return getattr(self._hashlib, method_name or "", None)
fullname = info[implementation].fullname
return self.try_import_attribute(fullname)

def fetch_hmac_function(self, name):
target = hashlib_helper.get_hmac_item_info(name)
return target.import_member()

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, so we will test the helper separately.
self.skip_if_fips_mode()
flags = dict(allow_openssl=allow_openssl, allow_builtin=allow_builtin)
is_fully_disabled = not allow_builtin and not allow_openssl

with hashlib_helper.block_algorithm(name, **flags):
# 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_fully_disabled):
_ = self.hashlib.new(name_for_hashlib_new)

# Since _hashlib is present, explicit blake2b/blake2s constructors
# use the built-in implementation, while others (since we are not
# in FIPS mode and since _hashlib exists) use the OpenSSL function.
with self.check_context(is_fully_disabled):
_ = getattr(self.hashlib, name)()

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_fully_disabled):
_ = self.hmac.new(b"", b"", name)
with self.check_context(is_fully_disabled):
_ = self.hmac.HMAC(b"", b"", name)
with self.check_context(is_fully_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)

@hashlib_helper.block_algorithm("md5")
def test_disable_hash_md5_in_fips_mode(self):
self.skip_if_not_fips_mode()

self.assertRaises(ValueError, self.hashlib.new, "md5")
self.assertRaises(ValueError, self._hashlib.new, "md5")
self.assertRaises(ValueError, self.hashlib.md5)
self.assertRaises(ValueError, self._hashlib.openssl_md5)

kwargs = dict(usedforsecurity=True)
self.assertRaises(ValueError, self.hashlib.new, "md5", **kwargs)
self.assertRaises(ValueError, self._hashlib.new, "md5", **kwargs)
self.assertRaises(ValueError, self.hashlib.md5, **kwargs)
self.assertRaises(ValueError, self._hashlib.openssl_md5, **kwargs)

@hashlib_helper.block_algorithm("md5", allow_openssl=True)
def test_disable_hash_md5_in_fips_mode_allow_openssl(self):
self.skip_if_not_fips_mode()
# Allow the OpenSSL interface to be used but not the HACL* one.
# hashlib.new("md5") is dispatched to hashlib.openssl_md5()
self.assertRaises(ValueError, self.hashlib.new, "md5")
# dispatched to hashlib.openssl_md5() in FIPS mode
h2 = self.hashlib.new("md5", usedforsecurity=False)
self.assertIsInstance(h2, self._hashlib.HASH)

# block_algorithm() does not mock hashlib.md5 and _hashlib.openssl_md5
self.assertNotHasAttr(self.hashlib.md5, "__wrapped__")
self.assertNotHasAttr(self._hashlib.openssl_md5, "__wrapped__")

hashlib_md5 = inspect.unwrap(self.hashlib.md5)
self.assertIs(hashlib_md5, self._hashlib.openssl_md5)
self.assertRaises(ValueError, self.hashlib.md5)
# allow MD5 to be used in FIPS mode if usedforsecurity=False
h3 = self.hashlib.md5(usedforsecurity=False)
self.assertIsInstance(h3, self._hashlib.HASH)

@hashlib_helper.block_algorithm("md5", allow_builtin=True)
def test_disable_hash_md5_in_fips_mode_allow_builtin(self):
self.skip_if_not_fips_mode()
# Allow the HACL* interface to be used but not the OpenSSL one.
h1 = self.hashlib.new("md5") # dispatched to _md5.md5()
self.assertNotIsInstance(h1, self._hashlib.HASH)
h2 = self.hashlib.new("md5", usedforsecurity=False)
self.assertIsInstance(h2, type(h1))

# block_algorithm() mocks hashlib.md5 and _hashlib.openssl_md5
self.assertHasAttr(self.hashlib.md5, "__wrapped__")
self.assertHasAttr(self._hashlib.openssl_md5, "__wrapped__")

hashlib_md5 = inspect.unwrap(self.hashlib.md5)
openssl_md5 = inspect.unwrap(self._hashlib.openssl_md5)
self.assertIs(hashlib_md5, openssl_md5)
self.assertRaises(ValueError, self.hashlib.md5)
self.assertRaises(ValueError, self.hashlib.md5,
usedforsecurity=False)

@hashlib_helper.block_algorithm("md5",
allow_openssl=True,
allow_builtin=True)
def test_disable_hash_md5_in_fips_mode_allow_all(self):
self.skip_if_not_fips_mode()
# hashlib.new() isn't blocked as it falls back to _md5.md5
self.assertIsInstance(self.hashlib.new("md5"), self._md5.MD5Type)
self.assertRaises(ValueError, self._hashlib.new, "md5")
h = self._hashlib.new("md5", usedforsecurity=False)
self.assertIsInstance(h, self._hashlib.HASH)

self.assertNotHasAttr(self.hashlib.md5, "__wrapped__")
self.assertNotHasAttr(self._hashlib.openssl_md5, "__wrapped__")

self.assertIs(self.hashlib.md5, self._hashlib.openssl_md5)
self.assertRaises(ValueError, self.hashlib.md5)
h = self.hashlib.md5(usedforsecurity=False)
self.assertIsInstance(h, self._hashlib.HASH)


class TestIsolated(unittest.TestCase):
# Drive the sample tests in test._isolated_sample (which really spawn
# subprocesses through @isolation.runInSubprocess()) under a private
Expand Down
Loading