Normalize invalid digest inputs

CHANGELOG.md

@@ -3,6 +3,14 @@
 All notable changes are recorded here. The format follows Keep a Changelog and the
 project uses Semantic Versioning.
 
+## [Unreleased]
+
+### Fixed
+
+- Normalize invalid digest input types and count ranges to stable `SS012` errors.
+- Reject whitespace and every other non-exact spelling of 64-character SHA-256 hex.
+- Preserve one-pass digest iterables required by bounded-memory release processing.
+
 ## [0.2.3] - 2026-08-24
 
 ### Added

docs/api.md

@@ -16,6 +16,13 @@ digest = record_digest(record)
 It rejects non-string object keys, non-finite floats, unsupported objects, and integers
 outside the exactly interoperable range.
 
+`dataset_digest` accepts string split names paired with a non-negative 64-bit record
+count and exactly 64 lowercase or uppercase hexadecimal SHA-256 characters.
+`sequence_digest` applies the same exact grammar to each record digest from a one-pass
+iterable. Text and byte containers are not treated as iterables of digests. Whitespace is
+not accepted. Invalid runtime types, encodings, lengths, and count ranges fail with
+`SS012`.
+
 ## Release operations
 
 ```python

src/splitseal/canonical.py

@@ -4,7 +4,8 @@
 
 import hashlib
 import math
-from collections.abc import Mapping, Sequence
+import re
+from collections.abc import Iterable, Mapping
 from typing import TypeAlias, cast
 
 import rfc8785
@@ -19,6 +20,9 @@
 _SEQUENCE_DOMAIN = b"splitseal-sequence-v1\x00"
 _DATASET_DOMAIN = b"splitseal-dataset-v1\x00"
 _MAX_INTEROPERABLE_INTEGER = 9_007_199_254_740_991
+_MAX_RECORD_COUNT = 2**64 - 1
+_SPLIT_DIGEST_ENTRY_SIZE = 2
+_SHA256_HEX = re.compile(r"[0-9A-Fa-f]{64}")
 
 
 def _validate_json(value: object, location: str = "$") -> None:
@@ -70,36 +74,59 @@ def record_digest(record: Record) -> str:
     return hashlib.sha256(_RECORD_DOMAIN + _framed(payload)).hexdigest()
 
 
-def sequence_digest(record_digests: Sequence[str]) -> str:
+def sequence_digest(record_digests: Iterable[str]) -> str:
     """Hash an ordered sequence of hexadecimal record digests."""
 
+    if isinstance(record_digests, (str, bytes, bytearray)) or not isinstance(
+        record_digests, Iterable
+    ):
+        raise fail("SS012", "record digests must be an iterable of strings")
     digest = hashlib.sha256(_SEQUENCE_DOMAIN)
     for item in record_digests:
-        try:
-            raw = bytes.fromhex(item)
-        except ValueError as exc:
-            raise fail("SS012", "record digest is not hexadecimal") from exc
-        if len(raw) != hashlib.sha256().digest_size:
-            raise fail("SS012", "record digest has an invalid length")
+        if not isinstance(item, str):
+            raise fail("SS012", "record digest must be a string")
+        if not _SHA256_HEX.fullmatch(item):
+            raise fail("SS012", "record digest must contain exactly 64 hexadecimal characters")
+        raw = bytes.fromhex(item)
         digest.update(_framed(raw))
     return digest.hexdigest()
 
 
 def dataset_digest(splits: Mapping[str, tuple[int, str]]) -> str:
     """Hash named split roots and counts in split-name order."""
 
+    if not isinstance(splits, Mapping):
+        raise fail("SS012", "dataset splits must be a mapping")
+    validated: list[tuple[str, int, str]] = []
+    for name, value in splits.items():
+        if not isinstance(name, str):
+            raise fail("SS012", "split names must be strings")
+        if not isinstance(value, tuple) or len(value) != _SPLIT_DIGEST_ENTRY_SIZE:
+            raise fail("SS012", "split digest entry must be a count and digest pair", split=name)
+        count, split_digest = value
+        if type(count) is not int or count < 0 or count > _MAX_RECORD_COUNT:
+            raise fail(
+                "SS012",
+                "record count must be an unsigned 64-bit integer",
+                split=name,
+            )
+        if not isinstance(split_digest, str):
+            raise fail("SS012", "split digest must be a string", split=name)
+        validated.append((name, count, split_digest))
+
     digest = hashlib.sha256(_DATASET_DOMAIN)
-    for name in sorted(splits):
-        count, split_digest = splits[name]
-        if count < 0:
-            raise fail("SS012", "record count cannot be negative", split=name)
+    for name, count, split_digest in sorted(validated, key=lambda item: item[0]):
+        if not _SHA256_HEX.fullmatch(split_digest):
+            raise fail(
+                "SS012",
+                "split digest must contain exactly 64 hexadecimal characters",
+                split=name,
+            )
+        raw_digest = bytes.fromhex(split_digest)
         try:
-            raw_digest = bytes.fromhex(split_digest)
-        except ValueError as exc:
-            raise fail("SS012", "split digest is not hexadecimal", split=name) from exc
-        if len(raw_digest) != hashlib.sha256().digest_size:
-            raise fail("SS012", "split digest has an invalid length", split=name)
-        encoded_name = name.encode("utf-8")
+            encoded_name = name.encode("utf-8")
+        except UnicodeEncodeError as exc:
+            raise fail("SS012", "split name is not valid UTF-8") from exc
         digest.update(_framed(encoded_name))
         digest.update(count.to_bytes(8, "big"))
         digest.update(_framed(raw_digest))

tests/test_canonical.py

@@ -62,13 +62,46 @@ def test_sequence_digest_is_order_sensitive() -> None:
     assert sequence_digest([first, second]) != sequence_digest([second, first])
 
 
+def test_sequence_digest_accepts_one_pass_iterables() -> None:
+    items = [record_digest({"id": "one"}), record_digest({"id": "two"})]
+    assert sequence_digest(item for item in items) == sequence_digest(items)
+
+
 @pytest.mark.parametrize("digest", ["not-hex", "ab"])
 def test_sequence_digest_rejects_malformed_digest(digest: str) -> None:
     with pytest.raises(SplitSealError) as caught:
         sequence_digest([digest])
     assert caught.value.code == "SS012"
 
 
+@pytest.mark.parametrize(
+    "digests",
+    [None, "00" * 32, b"00", [None], [1]],
+)
+def test_sequence_digest_rejects_wrong_runtime_types(digests: object) -> None:
+    with pytest.raises(SplitSealError) as caught:
+        sequence_digest(digests)  # type: ignore[arg-type]
+    assert caught.value.code == "SS012"
+
+
+@pytest.mark.parametrize(
+    "digest",
+    [
+        "00 " * 32,
+        "00" * 16 + "\n" + "00" * 16,
+        "00" * 16 + "\t" + "00" * 16,
+    ],
+)
+def test_digest_functions_reject_ascii_whitespace(digest: str) -> None:
+    with pytest.raises(SplitSealError) as sequence_error:
+        sequence_digest([digest])
+    assert sequence_error.value.code == "SS012"
+
+    with pytest.raises(SplitSealError) as dataset_error:
+        dataset_digest({"split": (1, digest)})
+    assert dataset_error.value.code == "SS012"
+
+
 def test_dataset_digest_sorts_split_names_but_includes_counts() -> None:
     root = record_digest({"id": "one"})
     left = dataset_digest({"b": (1, root), "a": (1, root)})
@@ -86,6 +119,25 @@ def test_dataset_digest_rejects_invalid_inputs() -> None:
         dataset_digest({"a": (1, "00")})
 
 
+@pytest.mark.parametrize(
+    "splits",
+    [
+        None,
+        {1: (1, "00" * 32)},
+        {"a": [1, "00" * 32]},
+        {"a": ("1", "00" * 32)},
+        {"a": (True, "00" * 32)},
+        {"a": (2**64, "00" * 32)},
+        {"a": (1, None)},
+        {"\ud800": (1, "00" * 32)},
+    ],
+)
+def test_dataset_digest_rejects_wrong_runtime_types(splits: object) -> None:
+    with pytest.raises(SplitSealError) as caught:
+        dataset_digest(splits)  # type: ignore[arg-type]
+    assert caught.value.code == "SS012"
+
+
 def test_ensure_record_rejects_non_object() -> None:
     with pytest.raises(SplitSealError) as caught:
         ensure_record(["not", "an", "object"], location="test")