diff --git a/changelog.d/pgw991.md b/changelog.d/pgw991.md new file mode 100644 index 00000000..d3aa3240 --- /dev/null +++ b/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:`, 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. diff --git a/src/gen_worker/api/errors.py b/src/gen_worker/api/errors.py index 6fbdb686..026fe810 100644 --- a/src/gen_worker/api/errors.py +++ b/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.""" diff --git a/src/gen_worker/request_context/__init__.py b/src/gen_worker/request_context/__init__.py index 459e3b86..b868ba1a 100644 --- a/src/gen_worker/request_context/__init__.py +++ b/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 ``:`` 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 ``:`` 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 diff --git a/tests/test_blob_digest_qualified_pgw991.py b/tests/test_blob_digest_qualified_pgw991.py new file mode 100644 index 00000000..e427aec6 --- /dev/null +++ b/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]