Use hashing from antsibull-fileutils.

src/antsibull_core/ansible_core.py

@@ -16,12 +16,12 @@
 from urllib.parse import urljoin
 
 import aiofiles
+from antsibull_fileutils.hashing import verify_a_hash
 from packaging.version import Version as PypiVer
 
 from . import app_context
 from .logging import log
 from .subprocess_util import async_log_run
-from .utils.hashing import verify_a_hash
 from .utils.http import retry_get
 from .utils.io import copy_file
 
@@ -168,7 +168,9 @@ async def retrieve(  # noqa C901
         if lib_ctx.ansible_core_cache and "sha256" in digests:
             cached_path = os.path.join(lib_ctx.ansible_core_cache, tar_filename)
             if os.path.isfile(cached_path):
-                if await verify_a_hash(cached_path, digests):
+                if await verify_a_hash(
+                    cached_path, digests, chunksize=lib_ctx.chunksize
+                ):
                     await copy_file(cached_path, tar_path, check_content=False)
                     return tar_path
 

src/antsibull_core/galaxy.py

@@ -15,9 +15,9 @@
 
 import aiofiles
 import semantic_version as semver
+from antsibull_fileutils.hashing import verify_hash
 
 from . import app_context
-from .utils.hashing import verify_hash
 from .utils.http import retry_get
 from .utils.io import copy_file
 
@@ -392,7 +392,10 @@ async def download(
         if self.collection_cache:
             cached_copy = os.path.join(self.collection_cache, filename)
             if os.path.isfile(cached_copy):
-                if await verify_hash(cached_copy, sha256sum):
+                lib_ctx = app_context.lib_ctx.get()
+                if await verify_hash(
+                    cached_copy, sha256sum, chunksize=lib_ctx.chunksize
+                ):
                     await copy_file(cached_copy, download_filename, check_content=False)
                     return download_filename
 
@@ -408,7 +411,9 @@ async def download(
                     await f.write(chunk)
 
         # Verify the download
-        if not await verify_hash(download_filename, sha256sum):
+        if not await verify_hash(
+            download_filename, sha256sum, chunksize=lib_ctx.chunksize
+        ):
             raise DownloadFailure(
                 f"{release_url} failed to download correctly."
                 f" Expected checksum: {sha256sum}"

src/antsibull_core/utils/hashing.py

@@ -7,33 +7,18 @@
 
 from __future__ import annotations
 
-import dataclasses
-import hashlib
 import typing as t
 from collections.abc import Mapping
 
-import aiofiles
+from antsibull_fileutils.hashing import verify_a_hash as _verify_a_hash
+from antsibull_fileutils.hashing import verify_hash as _verify_hash
 
 from .. import app_context
 
 if t.TYPE_CHECKING:
     from _typeshed import StrOrBytesPath
 
 
-@dataclasses.dataclass(frozen=True)
-class _AlgorithmData:
-    name: str
-    algorithm: str
-    kwargs: dict[str, t.Any]
-
-
-_PREFERRED_HASHES: tuple[_AlgorithmData, ...] = (
-    # https://pypi.org/help/#verify-hashes, https://github.com/pypi/warehouse/issues/9628
-    _AlgorithmData(name="sha256", algorithm="sha256", kwargs={}),
-    _AlgorithmData(name="blake2b_256", algorithm="blake2b", kwargs={"digest_size": 32}),
-)
-
-
 async def verify_hash(
     filename: StrOrBytesPath,
     hash_digest: str,
@@ -50,15 +35,14 @@ async def verify_hash(
     :kwarg algorithm_kwargs: Parameters to provide to the hash algorithm's constructor.
     :returns: True if the hash matches, otherwise False.
     """
-    hasher = getattr(hashlib, algorithm)(**(algorithm_kwargs or {}))
-    async with aiofiles.open(filename, "rb") as f:
-        ctx = app_context.lib_ctx.get()
-        while chunk := await f.read(ctx.chunksize):
-            hasher.update(chunk)
-    if hasher.hexdigest() != hash_digest:
-        return False
-
-    return True
+    ctx = app_context.lib_ctx.get()
+    return await _verify_hash(
+        filename,
+        hash_digest,
+        algorithm=algorithm,
+        algorithm_kwargs=algorithm_kwargs,
+        chunksize=ctx.chunksize,
+    )
 
 
 async def verify_a_hash(
@@ -72,12 +56,5 @@ async def verify_a_hash(
     :arg hash_digest: A mapping of hash types to digests.
     :returns: True if the hash matches, otherwise False.
     """
-    for algorithm_data in _PREFERRED_HASHES:
-        if algorithm_data.name in hash_digests:
-            return await verify_hash(
-                filename,
-                hash_digests[algorithm_data.name],
-                algorithm=algorithm_data.algorithm,
-                algorithm_kwargs=algorithm_data.kwargs,
-            )
-    return False
+    ctx = app_context.lib_ctx.get()
+    return await _verify_a_hash(filename, hash_digests, chunksize=ctx.chunksize)