diff --git a/Lib/hashlib.py b/Lib/hashlib.py index c1e2d07c4357ff..43175d11f87046 100644 --- a/Lib/hashlib.py +++ b/Lib/hashlib.py @@ -233,7 +233,8 @@ def file_digest(fileobj, digest, /, *, _bufsize=2**18): if hasattr(fileobj, "getbuffer"): # io.BytesIO object, use zero-copy buffer - digestobj.update(fileobj.getbuffer()) + with fileobj.getbuffer() as buf: + digestobj.update(buf[fileobj.tell():]) return digestobj # Only binary files implement readinto(). diff --git a/Lib/test/test_hashlib.py b/Lib/test/test_hashlib.py index 253a8f455853f5..2b67572b412b4e 100644 --- a/Lib/test/test_hashlib.py +++ b/Lib/test/test_hashlib.py @@ -12,6 +12,7 @@ import itertools import logging import os +import random import re import sys import sysconfig @@ -102,6 +103,20 @@ def read_vectors(hash_name): ) +def make_hash_objects(*digestmods, buf=b"", **kwargs): + objects = [] + for digestmod in digestmods: + try: + if callable(digestmod): + obj = digestmod(buf) + else: + obj = hashlib.new(digestmod, buf, **kwargs) + except ValueError: + continue + objects.append((digestmod, obj)) + return objects + + class HashLibTestCase(unittest.TestCase): supported_hash_names = ( 'md5', 'MD5', 'sha1', 'SHA1', 'sha224', 'SHA224', 'sha256', 'SHA256', @@ -524,7 +539,8 @@ def test_blake2_update_over_4gb(self): self.assertEqual(h.hexdigest(), "8a268e83dd30528bc0907fa2008c91de8f090a0b6e0e60a5ff0d999d8485526f") def check(self, name, data, hexdigest, shake=False, **kwargs): - length = len(hexdigest)//2 + n = len(data) + length = len(hexdigest) // 2 hexdigest = hexdigest.lower() constructors = self.constructors_to_test[name] # 2 is for hashlib.name(...) and hashlib.new(name, ...) @@ -533,47 +549,47 @@ def check(self, name, data, hexdigest, shake=False, **kwargs): m = hash_object_constructor(data, **kwargs) computed = m.hexdigest() if not shake else m.hexdigest(length) self.assertEqual( - computed, hexdigest, - "Hash algorithm %s constructed using %s returned hexdigest" - " %r for %d byte input data that should have hashed to %r." - % (name, hash_object_constructor, - computed, len(data), hexdigest)) + computed, hexdigest, + "Hash algorithm %s constructed using %s returned hexdigest" + " %r for %d byte input data that should have hashed to %r." + % (name, hash_object_constructor, computed, n, hexdigest) + ) computed = m.digest() if not shake else m.digest(length) digest = bytes.fromhex(hexdigest) self.assertEqual(computed, digest) if not shake: self.assertEqual(len(digest), m.digest_size) + def generate_sub_hexdigest(pos=-1): + if pos < 0: + pos = 0 if n == 0 else random.randrange(0, n) + pos += 1 # ensure pos is at least 1 (allowed to be out of range) + buf = data[pos:] + objects = make_hash_objects(name, *constructors, buf=buf, **kwargs) + hexdigests = {obj.hexdigest() for _, obj in objects} + self.assertEqual(len(hexdigests), 1, f"bad digests: {objects}") + return pos, hexdigests.pop() + if not shake and kwargs.get("key") is None: # skip shake and blake2 extended parameter tests self.check_file_digest(name, data, hexdigest) + for pos in sorted({-1, 1, n - 1, n, n + 1}): + with self.subTest(pos=pos): + pos, hexdigest2 = generate_sub_hexdigest(pos=pos) + self.check_file_digest(name, data, hexdigest2, pos) - def check_file_digest(self, name, data, hexdigest): + def check_file_digest(self, name, data, hexdigest, pos=0): hexdigest = hexdigest.lower() - 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) - + digests = make_hash_objects(name, *self.constructors_to_test[name]) with tempfile.TemporaryFile() as f: f.write(data) + buf = io.BytesIO(data) - for digest in digests: - buf = io.BytesIO(data) - buf.seek(0) - self.assertEqual( - hashlib.file_digest(buf, digest).hexdigest(), hexdigest - ) - f.seek(0) - digestobj = hashlib.file_digest(f, digest) - self.assertEqual(digestobj.hexdigest(), hexdigest) + for digest, _ in digests: + for fobj in [buf, f]: + fobj.seek(pos) + digestobj = hashlib.file_digest(fobj, digest) + self.assertEqual(digestobj.hexdigest(), hexdigest) def check_no_unicode(self, algorithm_name): # Unicode objects are not allowed as input. diff --git a/Misc/NEWS.d/next/Library/2026-08-09-13-23-12.gh-issue-155419.J6dSFJ.rst b/Misc/NEWS.d/next/Library/2026-08-09-13-23-12.gh-issue-155419.J6dSFJ.rst new file mode 100644 index 00000000000000..2117f189d64871 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-08-09-13-23-12.gh-issue-155419.J6dSFJ.rst @@ -0,0 +1,3 @@ +Ensure that :func:`hashlib.file_digest` honors the current position of the +file object when given an :class:`io.BytesIO` object. Patch by Bénédikt +Tran.