multihash.py
compute multiple hashes over file sets
Create or verify multi-hash checksums of files. Supported hash types are all that Python hashlib supports (cryptographic hashes like sha256) and crc32.
The program takes care to read each file just one time. If a free-threaded Python interpreter is detected, then the work of hashing is distributed across multiple threads, one per hash type. The amount of buffered data is carefully controlled to be no more than 3 chunks of 256KiB. (Assuming prompt collection by the Python garbage collector; there are some caveats about collection in multi-threaded circumstances) This can produce a healthy speed-up in elapsed time, 3.6x measured when computing sha512+sha384+sha256+md5 on a 8-core Ryzen 3700X. (comparing 3.15 to 3.15t on linux)
This program produces checksum files that look something like this:
$ python multihash.py -h crc32 -h md5 /dev/null > /tmp/digest
crc32 md5 filename
00000000 d41d8cd98f00b204e9800998ecf8427e /dev/null
Note that the format is self-describing but is NOT compatible with traditional hash verification programs like sha256sum.
The hashes can then be checked:
$ python multihash.py -c /tmp/digest
/tmp/digest:2: /dev/null: All hashes match
Available hash types can be listed:
$ python multihash.py --list-hashes
Hashes guaranteed available: blake2b blake2s crc32 md5 sha1 sha224 sha256 sha384 sha3_224 sha3_256 sha3_384 sha3_512 sha512 shake_128 shake_256
Additional hashes available: md5-sha1 ripemd160 sha512_224 sha512_256 sm3
multihash.py
multihash.py (5484 bytes)
# SPDX-FileCopyrightText: 2026 Jeff Epler
#
# SPDX-License-Identifier: GPL-3.0-only
import io
import binascii
import getopt
import hashlib
import queue
import sys
import typing
import collections.abc
import shlex
import threading
try:
use_concurrency: bool = not sys._is_gil_enabled()
except AttributeError:
use_concurrency = False
chunksize = 262144
class Hashlike(typing.Protocol):
@property
def name(self) -> str: ...
def __init__(self) -> None: ...
def update(self, data: collections.abc.Buffer, /) -> None: ...
def hexdigest(self) -> str: ...
class Crc32:
name = "crc32"
def __init__(self) -> None:
self.value = binascii.crc32(b"")
def update(self, b: collections.abc.Buffer) -> None:
self.value = binascii.crc32(b, self.value)
def hexdigest(self) -> str:
return f"{self.value:08x}"
def chunked(f: io.RawIOBase, chunksize: int) -> typing.Iterator[bytes]:
while chunk := f.read(chunksize):
yield chunk
def hash_new(name: str) -> Hashlike:
if name == "crc32":
return Crc32()
return hashlib.new(name)
def hash_list():
algorithms_guaranteed = hashlib.algorithms_guaranteed | {"crc32"}
algorithms_available = hashlib.algorithms_available | {"crc32"}
print("Hashes guaranteed available:", " ".join(sorted(algorithms_guaranteed)))
print(
"Additional hashes available:",
" ".join(sorted(algorithms_available - algorithms_guaranteed)),
)
def opener(filename):
if filename == "-":
return sys.stdin.buffer
return open(filename, "rb")
def hash_do(h: Hashlike, q: queue.Queue) -> None:
while chunk := q.get():
h.update(chunk)
def multihash(filename, hashes):
hashers = [hash_new(h) for h in hashes]
with opener(filename) as f:
if use_concurrency:
queues = [queue.Queue(3) for _ in hashers]
threads = [
threading.Thread(target=hash_do, args=args)
for args in zip(hashers, queues)
]
for t in threads:
t.start()
for chunk in chunked(f, chunksize):
for q in queues:
q.put(chunk)
for q in queues:
q.put(None)
for t in threads:
t.join()
else:
for chunk in chunked(f, chunksize):
for h in hashers:
h.update(chunk)
return [h.hexdigest() for h in hashers]
def print_header(hashes):
for h in hashes:
hashlen = len(hash_new(h).hexdigest())
print(f"{h:{hashlen}} ", end="")
print("", "filename")
def check(fn: str, quiet: bool) -> bool:
success = True
with open(fn, "r") as f:
header = f.readline()
header_words = header.split()
if not header_words or header_words[-1] != "filename":
print(f"{fn}:1: Bad header", file=sys.stderr)
return False
hashes = header_words[:-1]
for i, row in enumerate(f, start=2):
words = row.split(None, maxsplit=1 + len(hashes))
if len(words) != len(header_words):
print("{fn}:{i}: Bad syntax", file=sys.stderr)
success = False
continue
parsed = shlex.split(words[-1])
if len(parsed) != 1:
print("{fn}:{i}: Bad syntax (filename)", file=sys.stderr)
success = False
continue
filename = parsed[0]
expected = words[:-1]
try:
hashed = multihash(filename, hashes)
except Exception as e:
print(f"{fn}:{i}: Failed to hash: {e}", file=sys.stderr)
success = False
continue
file_success = True
for hashname, want, got in zip(hashes, expected, hashed):
if want != got:
print(
f"{fn}:{i}: {hashname}: Expected {want}, got {got}",
file=sys.stderr,
)
success = False
file_success = False
if file_success and not quiet:
print(f"{fn}:{i}: {filename}: All hashes match")
return success
def main():
opts, files = getopt.getopt(sys.argv[1:], "ch:q", ["list-hashes"])
hashes = []
do_check = False
quiet = False
files = files or ["-"]
for o in opts:
if o[0] == "--list-hashes":
hash_list()
raise SystemExit(0)
if o[0] == "-h":
hashes.extend(o[1].split(","))
if o[0] == "-c":
do_check = True
if o[0] == "-q":
quiet = True
if not hashes:
hashes = ["sha256"]
if not quiet:
print(f"Note: {use_concurrency=}", file=sys.stderr)
if do_check:
status = 0
for fn in files:
if not check(fn, quiet):
status = 1
raise SystemExit(status)
else:
print_header(hashes)
for fn in files:
hashers = multihash(fn, hashes)
for h in hashers:
print(h, end=" ")
print("", shlex.quote(fn))
def hash(fn):
hashers = [hash_new(h) for h in hashes]
with opener(fn) as f:
while chunk := f.read(chunksize):
for h in hashers:
h.update(chunk)
hashes = []
if __name__ == "__main__":
main()