Add archive contents_sha256 and ARCHIVE_FORMATS

README.md

@@ -202,6 +202,13 @@ for generating proquint identifiers for content channels. These are short string
 that are easy to enter on devices without a full keyboard, e.g. `sutul-hakuh`.
 
 
+Archive contents hash
+---------------------
+[le_utils/archive.py](./le_utils/archive.py) provides `contents_sha256`, a hash of a zip
+archive's member paths and bytes that ignores compression and metadata. It identifies files
+in `file_formats.ARCHIVE_FORMATS`. Its definition is frozen: ricecooker and Studio index archives by it.
+
+
 Roles
 -----
 The `role` constants are used for Role-based access control (RBAC) within the

le_utils/archive.py

@@ -0,0 +1,38 @@
+import hashlib
+import zipfile
+
+_CHUNK_SIZE = 64 * 1024
+
+
+def _member_sha256(zf, info):
+    digest = hashlib.sha256()
+    with zf.open(info) as member:
+        for chunk in iter(lambda: member.read(_CHUNK_SIZE), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def contents_sha256(fileobj):
+    """
+    Identity hash of a zip archive's member paths and bytes, independent of compression and metadata.
+    FROZEN: ricecooker and Studio's upload Cloud Function index archives by this digest.
+    :param fileobj: A seekable binary file object of a zip archive
+    :return: A hex SHA-256 digest
+    """
+    with zipfile.ZipFile(fileobj) as zf:
+        members = {}
+        for info in zf.infolist():
+            # Not .filename: Windows rewrites "\\" in it and Python >= 3.12 overrides it from 0x7075 extra fields.
+            path = info.orig_filename
+            # Before the directory skip: zipfile reads "x\0/" as file "x".
+            if "\0" in path:
+                raise ValueError(f"NUL in member path {path!r}")
+            if path.endswith("/"):
+                continue
+            if path in members:
+                raise ValueError(f"Duplicate member path {path!r}")
+            members[path] = info
+        digest = hashlib.sha256()
+        for path in sorted(members):
+            digest.update(f"{path}\0{_member_sha256(zf, members[path])}\n".encode("utf-8"))
+    return digest.hexdigest()

le_utils/constants/file_formats.py

@@ -85,6 +85,10 @@
 BLOOMD = "bloomd"
 BLOOMPUB_MIMETYPE = "application/bloompub+zip"
 
+# Zip-based formats identified by le_utils.archive.contents_sha256.
+# FROZEN: ricecooker and Studio's upload Cloud Function pin this; removing a format orphans its index entries.
+ARCHIVE_FORMATS = (HTML5, H5P, EPUB, HTML5_ARTICLE, BLOOMPUB, BLOOMD)
+
 choices = (
     (MP4, "MP4 Video"),
     (WEBM, "WEBM Video"),

tests/test_archive.py

@@ -0,0 +1,87 @@
+import io
+import struct
+import zipfile
+import zlib
+
+import pytest
+
+from le_utils.archive import contents_sha256
+
+MEMBERS = [
+    ("README.txt", b"Kolibri\n"),
+    ("index.html", b"<html><body><p>Kolibri</p></body></html>\n"),
+    ("css/site/style.css", b"p { color: #333; }\n"),
+    ("assets/data.bin", bytes(range(256)) * 400),
+    ("données/été.txt", "café\n".encode("utf-8")),
+    ("empty.txt", b""),
+]
+EXPECTED_DIGEST = "9ccc8c88bedb2106e59879ba70e7c84f38f098bf8fa4a7a2476dad63eab0ae15"
+
+
+def _zip(members, compression=zipfile.ZIP_DEFLATED):
+    archive = io.BytesIO()
+    with zipfile.ZipFile(archive, "w", compression) as zf:
+        for name, data in members:
+            zf.writestr(name, data)
+    archive.seek(0)
+    return archive
+
+
+@pytest.mark.parametrize("compression", [zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED], ids=["stored", "deflated"])
+def test_contents_sha256_is_pinned(compression):
+    assert contents_sha256(_zip(MEMBERS, compression)) == EXPECTED_DIGEST
+
+
+def test_contents_sha256_ignores_member_order():
+    assert contents_sha256(_zip(reversed(MEMBERS))) == EXPECTED_DIGEST
+
+
+def test_contents_sha256_ignores_directory_entries():
+    directories = [("css/", b""), ("css/site/", b""), ("données/", b"")]
+    assert contents_sha256(_zip(directories + MEMBERS)) == EXPECTED_DIGEST
+
+
+def test_contents_sha256_depends_on_member_paths():
+    moved = [("moved/" + name if name == "index.html" else name, data) for name, data in MEMBERS]
+    assert contents_sha256(_zip(moved)) != EXPECTED_DIGEST
+
+
+def _with_metadata(name):
+    info = zipfile.ZipInfo(name, date_time=(2001, 2, 3, 4, 5, 6))
+    info.external_attr = 0o100755 << 16
+    info.comment = b"comment"
+    # Python >= 3.12 replaces ZipInfo.filename with this Unicode Path field's name.
+    renamed = ("renamed/" + name).encode("utf-8")
+    info.extra = struct.pack("<HHBL", 0x7075, 5 + len(renamed), 1, zlib.crc32(name.encode("utf-8"))) + renamed
+    return info
+
+
+def test_contents_sha256_ignores_metadata():
+    archive = io.BytesIO()
+    with zipfile.ZipFile(archive, "w") as zf:
+        zf.comment = b"archive comment"
+        for name, data in MEMBERS:
+            zf.writestr(_with_metadata(name), data)
+    assert contents_sha256(archive) == EXPECTED_DIGEST
+
+
+def test_contents_sha256_is_independent_of_file_position():
+    archive = _zip(MEMBERS)
+    archive.read()
+    assert contents_sha256(archive) == EXPECTED_DIGEST
+    assert not archive.closed
+
+
+def test_contents_sha256_rejects_duplicate_paths():
+    with pytest.warns(UserWarning):
+        archive = _zip(MEMBERS + [("index.html", b"other")])
+    with pytest.raises(ValueError):
+        contents_sha256(archive)
+
+
+def test_contents_sha256_rejects_nul_in_path():
+    # zipfile truncates names at NUL when writing, so patch the stored bytes.
+    # Trailing "/": NUL must be rejected before directory entries are skipped.
+    archive = io.BytesIO(_zip([("nul#name/", b"x")]).getvalue().replace(b"nul#name", b"nul\0name"))
+    with pytest.raises(ValueError):
+        contents_sha256(archive)

tests/test_formats.py

@@ -18,3 +18,7 @@ def test_file_format_extensions_are_synced():
 
 def test_FORMATLIST_exists():
     assert file_formats.FORMATLIST, "FORMATLIST did not genereate properly"
+
+
+def test_ARCHIVE_FORMATS_does_not_change():
+    assert set(file_formats.ARCHIVE_FORMATS) == {"zip", "h5p", "epub", "kpub", "bloompub", "bloomd"}