pgw#991: a bare-hex blob address is REFUSED, not guessed into a namespace

changelog.d/pgw991.md

@@ -0,0 +1,19 @@
+- **pgw#991: `materialize_blob` tagged bare-hex digests `blake3:`, addressing a
+  namespace the blob was not in.** `_download_blob_by_digest` promoted any
+  digest without a `:` to `blake3:<hex>`, and its docstring asserted the hub
+  indexed all CAS content by blake3. Neither has been true since th#1303: the
+  repo-CAS is sha256, and blake3 survives only as the dataset-CAS contract. So a
+  caller holding a bare sha256 hex — the form the hub's own listings and
+  manifests quote — was silently sent to the wrong namespace and got a miss that
+  read as missing DATA. The guess is **deleted, not re-pointed to `sha256:`**:
+  two live namespaces disagree on the algorithm, so guessing is a coin flip
+  either way. Bare hex is now REFUSED, with the same rule and the same wording
+  the hub uses — `_parse_cas_digest` is a transcription of tensorhub
+  `storage.ParseDigest` (`validateDigestParam`'s implementation, fronting the
+  by-digest route th#1641 added): algorithm-tagged, a supported algorithm
+  (`sha256`/`blake3`), hex of that algorithm's width, lowercased. The refusal
+  happens before any HTTP call and names the fix in the message. It is typed by
+  th#1259 provenance like every other address fault — a caller-supplied one
+  raises `BlobDigestMalformedError` (`blob_digest_malformed`, a `PayloadRefError`
+  that maps to `JOB_STATUS_INVALID`, so a caller's typo is never model-health
+  evidence), while a platform-produced one stays fatal.

src/gen_worker/api/errors.py

@@ -158,6 +158,24 @@ def __init__(self, digest: str) -> None:
         )
 
 
+class BlobDigestMalformedError(PayloadRefError):
+    """A caller-supplied blob address is not an algorithm-tagged digest.
+
+    The hub's `storage.ParseDigest` refuses bare hex outright, so guessing an
+    algorithm here would only turn a local contract error into a remote 400 —
+    or, worse, silently address the wrong one of the two live CAS namespaces
+    (repo-CAS is sha256 since th#1303; dataset-CAS is blake3).
+    """
+
+    def __init__(self, digest: str, detail: str) -> None:
+        super().__init__(
+            f"blob address {digest!r} supplied by the request payload is not a "
+            f"valid content digest ({detail}) — write it algorithm-tagged, as "
+            '"sha256:<64 hex>" or "blake3:<64 hex>"',
+            code="blob_digest_malformed", ref=digest,
+        )
+
+
 class BlobForbiddenError(PayloadRefError):
     """A caller-supplied digest is not readable under this request's grant."""
 

src/gen_worker/request_context/__init__.py

@@ -23,6 +23,7 @@
     List,
     Literal,
     Mapping,
+    NoReturn,
     Optional,
     Sequence,
     Tuple,
@@ -50,6 +51,7 @@ class LoraOverlay(TypedDict):
 
 from ..api.errors import (
     AuthError,
+    BlobDigestMalformedError,
     BlobForbiddenError,
     BlobNotFoundError,
     DatasetNotFoundError,
@@ -132,6 +134,48 @@ def _copy_context_metadata(value: Any) -> Any:
     return value
 
 
+_CAS_DIGEST_WIDTHS = {"blake3": 64, "sha256": 64}
+"""The algorithms the hub's CAS keys on, and their hex widths — the exact
+content of tensorhub `storage.casSupportedAlgos`. Two live namespaces with
+different algorithms is why a bare hex string cannot be promoted to a digest
+by guessing."""
+
+
+def _parse_cas_digest(digest: str, *, origin: str) -> str:
+    """Return the canonical ``<algo>:<hex>`` form, or refuse.
+
+    Byte-for-byte the same contract as the hub's `storage.ParseDigest`
+    (`internal/storage/cas_paths.go`), which `validateDigestParam` fronts on
+    every by-digest route: algorithm-tagged, a supported algorithm, hex of
+    that algorithm's width, lowercased. Bare hex is refused rather than
+    tagged, because the two CAS namespaces disagree on the algorithm and
+    guessing addresses the wrong one silently.
+    """
+    raw = (digest or "").strip()
+    caller_supplied = origin == REF_ORIGIN_PAYLOAD
+
+    def _refuse(detail: str) -> NoReturn:
+        if caller_supplied:
+            raise BlobDigestMalformedError(raw, detail)
+        raise RuntimeError(f"malformed platform blob digest {raw!r}: {detail}")
+
+    if not raw:
+        _refuse("empty")
+    algo, sep, hexpart = raw.partition(":")
+    if not sep:
+        _refuse("not algorithm-tagged")
+    algo = algo.strip().lower()
+    hexpart = hexpart.strip()
+    width = _CAS_DIGEST_WIDTHS.get(algo)
+    if width is None:
+        _refuse(f"unsupported algorithm {algo!r}")
+    if not re.fullmatch(r"[0-9a-fA-F]*", hexpart):
+        _refuse("non-hex character")
+    if len(hexpart) != width:
+        _refuse(f"{algo} hex must be {width} chars")
+    return f"{algo}:{hexpart.lower()}"
+
+
 def _as_asset(asset: Asset, cls: type) -> Any:
     """Re-type a plain Asset as a media Asset subclass (same fields)."""
     kw = {f: getattr(asset, f) for f in asset.__struct_fields__}
@@ -1592,11 +1636,12 @@ def _download_blob_by_digest(
     ) -> None:
         """Fetch a blob by ``<algo>:<hex>`` digest to ``dest``.
 
-        Uses the repo-CAS by-digest read endpoint — works for any blob
-        uploaded via ``save_checkpoint`` regardless of whether it's a
-        checkpoint file or a dataset file. The server indexes all CAS
-        content by blake3 digest; callers that know the digest can fetch
-        without needing to know which subsystem the blob belongs to.
+        Uses the by-digest CAS read endpoint — works for any blob uploaded
+        via ``save_checkpoint`` regardless of whether it is a checkpoint file
+        or a dataset file. The digest must be ALGORITHM-TAGGED: the hub keys
+        two CAS namespaces on different algorithms (repo-CAS is sha256 since
+        th#1303; dataset-CAS is blake3), so a bare hex string does not name a
+        blob and is refused here for the same reason the hub refuses it.
 
         ``origin`` is the th#1259 provenance of the ADDRESS, and it is the
         only thing that decides how a terminal miss classifies. See
@@ -1608,8 +1653,7 @@ def _download_blob_by_digest(
 
         base = (self._file_api_base_url or "").strip().rstrip("/")
         token = self._get_worker_capability_token()
-        # Normalize digest format for URL.
-        digest_norm = digest if ":" in digest else f"blake3:{digest}"
+        digest_norm = _parse_cas_digest(digest, origin=origin)
         url = f"{base}/api/v1/blobs/{urllib.parse.quote(digest_norm, safe=':')}/content"
         headers = {"Authorization": f"Bearer {token}"}
         caller_supplied = origin == REF_ORIGIN_PAYLOAD

tests/test_blob_digest_qualified_pgw991.py

@@ -0,0 +1,207 @@
+"""pgw#991: a bare-hex blob address is refused, not tagged `blake3:`.
+
+`_download_blob_by_digest` used to promote a bare hex string to a digest by
+prefixing `blake3:`. That was wrong twice over: the repo-CAS has been sha256
+since th#1303, so the guess addressed a namespace the blob was not in; and the
+hub's own `storage.ParseDigest` (fronted by `validateDigestParam` on the
+by-digest route th#1641 added) REFUSES bare hex outright, so even a correct
+guess is not the contract.
+
+The hub in this rig answers the by-digest route exactly like the real one —
+`400 invalid_digest` for anything `ParseDigest` rejects — so a regression that
+re-introduces the guess fails here on the hub's own rule rather than on a
+worker-side assertion about it.
+"""
+from __future__ import annotations
+
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+from typing import Iterator
+from urllib.parse import unquote
+
+import pytest
+
+from gen_worker.pb import worker_scheduler_pb2 as pb
+from gen_worker.api.errors import (
+    BlobDigestMalformedError,
+    PayloadRefError,
+    ValidationError,
+)
+from gen_worker.executor import _map_exception
+from gen_worker.request_context import (
+    REF_ORIGIN_PLATFORM,
+    ConversionContext,
+)
+
+BARE_HEX = "6f3306ab3b849905dd21c6e3073a2f88a4ae34ac4ee5f3af4bda597f559e9d17"
+SHA256_DIGEST = f"sha256:{BARE_HEX}"
+BLAKE3_DIGEST = "blake3:" + "aa" * 32
+BLOB_BYTES = b"real blob bytes"
+
+# Every algorithm tensorhub `storage.casSupportedAlgos` keys on, and its width.
+_SUPPORTED = {"blake3": 64, "sha256": 64}
+
+_REQUESTED: list[str] = []
+
+
+def _hub_parse_digest(ref: str) -> str | None:
+    """tensorhub `internal/storage/cas_paths.go` ParseDigest, transcribed."""
+    ref = ref.strip()
+    if not ref:
+        return None
+    algo, sep, hexpart = ref.partition(":")
+    if not sep:
+        return None  # "bare hex is refused"
+    algo = algo.strip().lower()
+    hexpart = hexpart.strip()
+    width = _SUPPORTED.get(algo)
+    if width is None:
+        return None
+    if any(c not in "0123456789abcdefABCDEF" for c in hexpart):
+        return None
+    if len(hexpart) != width:
+        return None
+    return f"{algo}:{hexpart.lower()}"
+
+
+class _Hub(BaseHTTPRequestHandler):
+    def log_message(self, *_a: object) -> None:
+        pass
+
+    def _send(self, code: int, body: bytes = b"") -> None:
+        self.send_response(code)
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        if body:
+            self.wfile.write(body)
+
+    def do_GET(self) -> None:  # noqa: N802
+        path = self.path
+        if "/blobs/" not in path or not path.endswith("/content"):
+            return self._send(404, b"")
+        raw = unquote(path.split("/blobs/", 1)[1][: -len("/content")])
+        _REQUESTED.append(raw)
+        if _hub_parse_digest(raw) is None:
+            # api.WriteError(c, http.StatusBadRequest, "invalid_digest", "")
+            return self._send(400, b'{"error":{"code":"invalid_digest"}}')
+        return self._send(200, BLOB_BYTES)
+
+
+@pytest.fixture()
+def hub() -> Iterator[str]:
+    _REQUESTED.clear()
+    srv = HTTPServer(("127.0.0.1", 0), _Hub)
+    threading.Thread(target=srv.serve_forever, daemon=True).start()
+    try:
+        yield f"http://127.0.0.1:{srv.server_port}"
+    finally:
+        srv.shutdown()
+        srv.server_close()
+
+
+def _ctx(hub_url: str) -> ConversionContext:
+    ctx = ConversionContext(request_id="r-pgw991")
+    ctx._file_api_base_url = hub_url
+    ctx._worker_capability_token = "test-token"
+    return ctx
+
+
+def test_bare_hex_is_refused_before_any_request(hub: str, tmp_path: Path) -> None:
+    """The filed defect. Bare hex is a malformed ADDRESS, and the caller owns
+    it — so it fails the REQUEST typed, and no HTTP call is made at all."""
+    ctx = _ctx(hub)
+
+    with pytest.raises(BlobDigestMalformedError) as ei:
+        ctx.materialize_blob(BARE_HEX, tmp_path / "out.bin")
+
+    exc = ei.value
+    assert exc.code == "blob_digest_malformed"
+    assert exc.ref == BARE_HEX
+    assert isinstance(exc, PayloadRefError)
+    assert isinstance(exc, ValidationError)
+    # The refusal names the fix, in the hub's own words.
+    assert "sha256:<64 hex>" in str(exc)
+    assert _REQUESTED == [], "a malformed address must not reach the network"
+
+
+def test_bare_hex_maps_to_INVALID_not_FATAL(hub: str, tmp_path: Path) -> None:
+    """th#1259's rule holds for this class too: a caller's bad address is
+    never model-health evidence."""
+    ctx = _ctx(hub)
+    with pytest.raises(BlobDigestMalformedError) as ei:
+        ctx.materialize_blob(BARE_HEX, tmp_path / "out.bin")
+    status, message = _map_exception(ei.value)
+    assert status == pb.JOB_STATUS_INVALID
+    assert message.startswith("blob_digest_malformed: ")
+
+
+def test_the_old_blake3_guess_is_what_the_hub_rejects(hub: str, tmp_path: Path) -> None:
+    """Guard on the regression itself. Had the code kept tagging bare hex
+    `blake3:`, this sha256 blob would have been fetched from the blake3
+    namespace — a wrong address that the route answers, not an error. The
+    qualified form is the ONLY thing that reaches the right one."""
+    ctx = _ctx(hub)
+    out = ctx.materialize_blob(SHA256_DIGEST, tmp_path / "ok.bin")
+    assert out.read_bytes() == BLOB_BYTES
+    assert _REQUESTED == [SHA256_DIGEST], (
+        "the digest must reach the hub verbatim and algorithm-tagged; "
+        f"a blake3: guess would have sent blake3:{BARE_HEX}"
+    )
+
+
+def test_blake3_still_works_for_the_dataset_cas(hub: str, tmp_path: Path) -> None:
+    """blake3 is not dead — it is the dataset-CAS contract. A caller that
+    declares it explicitly is still served."""
+    ctx = _ctx(hub)
+    out = ctx.materialize_blob(BLAKE3_DIGEST, tmp_path / "ds.bin")
+    assert out.read_bytes() == BLOB_BYTES
+    assert _REQUESTED == [BLAKE3_DIGEST]
+
+
+def test_platform_origin_malformed_stays_fatal(hub: str, tmp_path: Path) -> None:
+    """A malformed address the PLATFORM produced is a platform fault, so it
+    keeps the fatal classification instead of blaming the caller."""
+    ctx = _ctx(hub)
+    with pytest.raises(RuntimeError) as ei:
+        ctx.materialize_blob(BARE_HEX, tmp_path / "p.bin", origin=REF_ORIGIN_PLATFORM)
+    assert not isinstance(ei.value, PayloadRefError)
+    assert "malformed platform blob digest" in str(ei.value)
+    assert _REQUESTED == []
+
+
+@pytest.mark.parametrize(
+    "bad",
+    [
+        "",
+        "   ",
+        BARE_HEX,                       # not algorithm-tagged
+        f"md5:{BARE_HEX}",              # unsupported algorithm
+        "sha256:" + "zz" * 32,          # non-hex character
+        "sha256:" + "ab" * 16,          # wrong width
+        "sha256:",                      # empty hex
+    ],
+)
+def test_worker_refusal_matches_the_hub_exactly(hub: str, tmp_path: Path, bad: str) -> None:
+    """Parity, asserted rather than asserted-about: every address the worker
+    refuses is one the hub's ParseDigest also refuses. A worker that refused
+    MORE would strand a legal address; one that refused LESS would ship a 400
+    to a pod that could have named the fault locally."""
+    assert _hub_parse_digest(bad) is None, "rig disagrees with the hub's rule"
+    ctx = _ctx(hub)
+    with pytest.raises(BlobDigestMalformedError):
+        ctx.materialize_blob(bad, tmp_path / "x.bin")
+    assert _REQUESTED == []
+
+
+@pytest.mark.parametrize(
+    "good",
+    [SHA256_DIGEST, SHA256_DIGEST.upper().replace("SHA256", "sha256"), BLAKE3_DIGEST],
+)
+def test_worker_accepts_everything_the_hub_accepts(hub: str, tmp_path: Path, good: str) -> None:
+    canonical = _hub_parse_digest(good)
+    assert canonical is not None
+    ctx = _ctx(hub)
+    ctx.materialize_blob(good, tmp_path / "y.bin")
+    # Canonicalised the same way the hub canonicalises it (lowercased hex).
+    assert _REQUESTED == [canonical]