pgw#1138 (th#1722 §C): the worker stops reconstructing an org the hub derives itself

changelog.d/pgw1138.md

@@ -0,0 +1 @@
+- **pgw#1138 (th#1722 §C): media uploads stop reconstructing the org.** tensorhub `de30113d` mounts the media-upload family org-less (`POST /api/v1/media/uploads`) because an upload addresses the CALLER'S OWN namespace — the org comes from the credential, never from the path — and keeps the org-addressed shape only as a transitional alias for wheels ≤0.106.0 (th#1799 deletes it once the fleet relocks onto a wheel carrying this change). So the client stops rebuilding a segment the hub derives itself: `RequestContext._media_upload_owner()` (which decoded the capability JWT's `tenant` claim for no other purpose) is **deleted**, the `X-Cozy-Owner` header is **deleted** (nothing in tensorhub reads it, in any version — the only `Cozy-Owner`-ish header in that tree is the orchestrator-internal `X-Cozy-Owner-Forwarded`), and with them goes the `RuntimeError("file save failed (missing owner)")` refusal and the whole J19-run34 403 class it was written against: a client that does not supply the org cannot supply the wrong one. `_decode_unverified_jwt_claims` stays — ten other call sites (transport, lifecycle, worker_identity, fleet_cells, procsplit, …) use it. The `upload.create/put/complete` phase split (th#1795) and the sha256-declare direct PUT are untouched; `complete_url` is derived from the create URL, so it follows automatically. **Test:** `tests/harness/upload_sink.py` now ROUTES like the hub — a pattern table mirroring `registerMediaUploadRoutes(v1, "/media")`, 404 for anything else including the alias — so `tests/test_media_upload_orgless_pgw1138.py` proves the URL by driving the real `ctx.save_bytes` codepath against a server that serves only tensorhub's canonical routes, rather than asserting a path string a stub pinned to itself. Seven tests across four files go RED on the pre-change source (four new, plus pgw#767, P9 and th#1111, which fail at the create leg with `status=404`).

src/gen_worker/presigned_upload.py

@@ -19,7 +19,8 @@
 used at different route prefixes for datasets
 (/api/v1/datasets/:dataset_id/upload-sessions/:session_id/uploads),
 endpoint source (/api/v1/endpoints/:owner/:endpoint/releases/uploads),
-and user media (/api/v1/media/:owner/uploads). Repo checkpoints do NOT use
+and user media (/api/v1/media/uploads — org-less, the hub derives the org
+from the credential; th#1722 §C). Repo checkpoints do NOT use
 this client anymore — they publish via the /commits API (gw#471,
 gen_worker.convert.hub).
 
@@ -263,8 +264,8 @@ def presigned_upload_file(
     Args:
         file_path: Local path to the file.
         base_url: TensorHub base URL.
-        endpoint_path: e.g. "/api/v1/media/:owner/uploads" or "/api/v1/repos/.../uploads".
-        headers: Auth headers (Authorization, X-Cozy-Owner).
+        endpoint_path: e.g. "/api/v1/media/uploads" or "/api/v1/repos/.../uploads".
+        headers: Auth headers (Authorization).
         create_payload: Additional fields for the create POST (ref, path, request_id, etc.).
         blake3_hex: Pre-computed BLAKE3 hash of the file.
         size_bytes: File size in bytes.

src/gen_worker/request_context/__init__.py

@@ -645,21 +645,6 @@ def _get_worker_capability_token(self) -> str:
             return self._worker_capability_token
         return _require_worker_capability_token()
 
-    def _media_upload_owner(self) -> str:
-        """Owner segment for /api/v1/media/:owner/uploads calls.
-
-        The capability token's `tenant` claim is the canonical org uuid its
-        upload_media grant is bound to — the only owner tensorhub authorizes
-        media writes for. Falls back to ctx.owner for dev/local paths where
-        the token is absent or not a JWT.
-        """
-        token = self._worker_capability_token or ""
-        if token:
-            claim = str(_decode_unverified_jwt_claims(token).get("tenant") or "").strip()
-            if claim:
-                return claim
-        return (self._owner or "").strip()
-
     def _resolve_local_output_path(self, ref: str) -> Optional[str]:
         """
         Dev-only local output backend.

src/gen_worker/request_context/_stream.py

@@ -14,7 +14,6 @@
 import threading
 import time
 import tempfile
-import urllib.parse
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Dict, Optional
 
@@ -425,25 +424,17 @@ def _finalize_presigned_upload(self) -> Any:
         # /complete measured at 1060 ms server-side per image.
         create_payload["sha256"] = self._sha.hexdigest()
 
-        # Media upload. The URL owner segment MUST be the owner the
-        # capability token's upload_media grant is bound to (the token's
-        # `tenant` claim: the canonical invoking-org uuid). The
-        # dispatch-stamped ctx.owner can be a slug or a destination-repo
-        # owner resolving to a DIFFERENT org — tensorhub then finds no
-        # matching grant and 403s (J19 run34 sample images). Inference
-        # outputs work exactly because URL owner == token-bound owner.
+        # Media upload. th#1722 §C / pgw#1138: an upload addresses the
+        # CALLER'S OWN namespace, so the hub derives the org from the
+        # credential and the path never names one. The client cannot get the
+        # org wrong because it no longer supplies it — which retires the J19
+        # run34 403 class (dispatch-stamped ctx.owner was a slug resolving to
+        # a different org than the capability grant's).
         create_payload["ref"] = self._ref
         job_id = str(self._ctx._job_id or "").strip()
         if job_id:
             create_payload["job_id"] = job_id
-        owner = self._ctx._media_upload_owner()
-        if not owner:
-            raise RuntimeError(
-                "file save failed (missing owner): media uploads require ctx.owner"
-            )
-        headers["X-Cozy-Owner"] = owner
-        owner_seg = urllib.parse.quote(owner, safe="")
-        endpoint_path = f"/api/v1/media/{owner_seg}/uploads"
+        endpoint_path = "/api/v1/media/uploads"
 
         def _progress_cb(parts_done: int, total_parts: int, bytes_up: int) -> None:
             with self._progress_lock:

tests/harness/upload_sink.py

@@ -1,11 +1,20 @@
-"""A real local stand-in for tensorhub's `/api/v1/media/:owner/uploads`.
+"""A real local stand-in for tensorhub's media-upload route family.
 
 Answers a dedup create, so a test needs no S3 part-PUT scripting to prove that
 an upload really happened. `requests_seen` being non-empty is the observable:
 it distinguishes "uploaded" from "returned a ref for bytes that never left the
 process", which is the pgw#767 defect class and cannot be told apart from the
 result envelope alone.
 
+**The sink ROUTES; it does not accept everything.** `_ORG_LESS_ROUTES` mirrors
+`registerMediaUploadRoutes(v1, "/media")` in tensorhub `internal/api/files.go`
+(th#1722 §C, `de30113d`): the org is derived from the credential and is never a
+path segment. Anything else 404s exactly as gin does — including the
+transitional org-addressed alias `/api/v1/media/<org>/uploads`, which th#1799
+deletes. That is what makes a client's URL construction testable end to end: a
+test that let the server answer any path could not tell the two shapes apart,
+which is the defect class pgw#1138 was filed against.
+
 The v2 suite has its own richer `UploadSink` fixture (`tests_v2/conftest.py`,
 with a `status=` knob for refusal rows). This is the v1 suite's minimal twin —
 kept separate deliberately, because a cross-suite import would make the v2
@@ -15,37 +24,82 @@
 from __future__ import annotations
 
 import json
+import re
 import threading
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import Any, ClassVar, Dict, List, Tuple
 
+#: The canonical create route. Clients build this and nothing else.
+MEDIA_UPLOADS_PATH = "/api/v1/media/uploads"
+
+#: The org-less media-upload family, one entry per tensorhub route. Kept as
+#: patterns rather than prefixes so `/api/v1/media/<org>/uploads` — one extra
+#: segment — cannot match by accident.
+_ORG_LESS_ROUTES: Tuple[re.Pattern[str], ...] = tuple(
+    re.compile(p)
+    for p in (
+        r"^/api/v1/media/uploads$",
+        r"^/api/v1/media/uploads/batch$",
+        r"^/api/v1/media/uploads/batch/complete$",
+        r"^/api/v1/media/uploads/[^/]+$",
+        r"^/api/v1/media/uploads/[^/]+/parts$",
+        r"^/api/v1/media/uploads/[^/]+/complete$",
+    )
+)
+
+
+def is_media_upload_route(path: str) -> bool:
+    """True when `path` is served by tensorhub's org-less upload family."""
+    bare = str(path or "").split("?", 1)[0]
+    return any(rx.match(bare) for rx in _ORG_LESS_ROUTES)
+
 
 class DedupUploadSink(BaseHTTPRequestHandler):
     requests_seen: ClassVar[List[Tuple[str, Dict[str, Any]]]] = []
+    headers_seen: ClassVar[List[Dict[str, str]]] = []
+    rejected: ClassVar[List[str]] = []
 
     def log_message(self, *_args: Any) -> None:
         pass
 
     def do_POST(self) -> None:  # noqa: N802
         length = int(self.headers.get("Content-Length", "0"))
-        body = json.loads(self.rfile.read(length) or b"{}")
+        raw = self.rfile.read(length)
+        if not is_media_upload_route(self.path):
+            type(self).rejected.append(self.path)
+            self._send(404, {
+                "error": "not_found",
+                "message": f"no route for POST {self.path}",
+            })
+            return
+        body = json.loads(raw or b"{}")
         type(self).requests_seen.append((self.path, body))
-        resp = json.dumps({
+        type(self).headers_seen.append({k: v for k, v in self.headers.items()})
+        self._send(200, {
             "dedup": True, "ref": body.get("ref") or "", "filename": "out.bin",
             "blake3": body.get("blake3") or "", "size_bytes": body.get("size_bytes") or 0,
             "mime_type": "application/octet-stream", "media_id": "m1",
-        }).encode("utf-8")
-        self.send_response(200)
+        })
+
+    def _send(self, code: int, payload: Dict[str, Any]) -> None:
+        resp = json.dumps(payload).encode("utf-8")
+        self.send_response(code)
         self.send_header("Content-Type", "application/json")
         self.send_header("Content-Length", str(len(resp)))
         self.end_headers()
         self.wfile.write(resp)
 
 
-def serve_upload_sink() -> Tuple[ThreadingHTTPServer, str]:
-    """Start the sink on an ephemeral port. Caller shuts it down and resets
-    `DedupUploadSink.requests_seen`."""
+def reset_upload_sink() -> None:
     DedupUploadSink.requests_seen = []
+    DedupUploadSink.headers_seen = []
+    DedupUploadSink.rejected = []
+
+
+def serve_upload_sink() -> Tuple[ThreadingHTTPServer, str]:
+    """Start the sink on an ephemeral port. Caller shuts it down and calls
+    `reset_upload_sink()`."""
+    reset_upload_sink()
     httpd = ThreadingHTTPServer(("127.0.0.1", 0), DedupUploadSink)
     threading.Thread(target=httpd.serve_forever, daemon=True).start()
     return httpd, f"http://127.0.0.1:{httpd.server_address[1]}"

tests/test_inline_envelope_pgw767.py

@@ -22,7 +22,12 @@
 
 from harness.hub_double import hub_double, is_ready, is_result_for
 from harness.toy_endpoints import EchoIn
-from harness.upload_sink import DedupUploadSink, serve_upload_sink
+from harness.upload_sink import (
+    MEDIA_UPLOADS_PATH,
+    DedupUploadSink,
+    reset_upload_sink,
+    serve_upload_sink,
+)
 
 
 def test_inline_dispatch_over_the_envelope_ceiling_still_really_uploads() -> None:
@@ -49,11 +54,11 @@ def test_inline_dispatch_over_the_envelope_ceiling_still_really_uploads() -> Non
                 "the inline media hint must not reach the result envelope"
             )
             path, body = DedupUploadSink.requests_seen[-1]
-            assert path.startswith(f"/api/v1/media/{org_id}/uploads")
+            assert path == MEDIA_UPLOADS_PATH
             assert body["size_bytes"] > 64 * 1024
     finally:
         httpd.shutdown()
-        DedupUploadSink.requests_seen = []
+        reset_upload_sink()
 
 
 def test_save_bytes_still_inlines_media_for_the_client() -> None:

tests/test_media_upload_orgless_pgw1138.py

@@ -0,0 +1,167 @@
+"""pgw#1138 / th#1722 §C: a media upload addresses the CALLER'S OWN namespace.
+
+tensorhub `de30113d` mounts the upload family org-less
+(`registerMediaUploadRoutes(v1, "/media")`, `internal/api/files.go`) and derives
+the org from the credential; the org-addressed shape survives only as a
+transitional alias for wheels <=0.106.0, which th#1799 deletes. So the client
+stops reconstructing the org: no `tenant`-claim decode, no `:owner` segment, no
+`X-Cozy-Owner` header (nothing in tensorhub reads it, in any version).
+
+These run against `harness.upload_sink`, which ROUTES like the hub does —
+404 for anything outside the org-less family, including the alias. The
+assertion that matters is therefore "the upload SUCCEEDED against a server
+that serves only the hub's canonical routes", not a path string the test
+pinned to itself. Against master every test here fails at the create leg with
+`ArtifactTransferError`, because the client posts `/api/v1/media/<org>/uploads`.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+from typing import Any, Dict
+
+import pytest
+
+from gen_worker import RequestContext
+from gen_worker.api.errors import ArtifactTransferError
+
+from harness.upload_sink import (
+    MEDIA_UPLOADS_PATH,
+    DedupUploadSink,
+    is_media_upload_route,
+    reset_upload_sink,
+    serve_upload_sink,
+)
+
+
+def _unsigned_jwt(claims: Dict[str, Any]) -> str:
+    def seg(obj: Dict[str, Any]) -> str:
+        raw = json.dumps(obj).encode("utf-8")
+        return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
+
+    return f"{seg({'alg': 'none', 'typ': 'JWT'})}.{seg(claims)}.sig"
+
+
+def test_save_bytes_uploads_against_the_org_less_route_family() -> None:
+    """The real `ctx.save_bytes` codepath, against a server that serves only
+    tensorhub's org-less routes. RED on master: the create POST goes to
+    `/api/v1/media/<tenant>/uploads`, the sink 404s it, and the save raises."""
+    httpd, base_url = serve_upload_sink()
+    try:
+        token = _unsigned_jwt(
+            {"tenant": "019f4c33-f3a5-705b-9848-0b3b0863c416", "request_id": "req-1138"}
+        )
+        ctx = RequestContext(
+            request_id="req-1138",
+            owner="tensorhub",  # the dispatch-stamped SLUG (the J19 run34 shape)
+            file_api_base_url=base_url,
+            worker_capability_token=token,
+        )
+        asset = ctx.save_bytes("samples/pair-000.bin", b"payload")
+
+        assert asset.ref == "samples/pair-000.bin"
+        assert not DedupUploadSink.rejected, (
+            "the client addressed a route tensorhub does not serve: "
+            f"{DedupUploadSink.rejected}"
+        )
+        assert DedupUploadSink.requests_seen, "the real upload sink must have been hit"
+        path, body = DedupUploadSink.requests_seen[-1]
+        assert path == MEDIA_UPLOADS_PATH, (
+            "the org comes from the credential, never from the path"
+        )
+        assert body["ref"] == "samples/pair-000.bin"
+    finally:
+        httpd.shutdown()
+        reset_upload_sink()
+
+
+def test_the_upload_carries_no_owner_header() -> None:
+    """`X-Cozy-Owner` was write-only: no tensorhub version reads it."""
+    httpd, base_url = serve_upload_sink()
+    try:
+        ctx = RequestContext(
+            request_id="req-1138-hdr",
+            owner="org",
+            file_api_base_url=base_url,
+            worker_capability_token=_unsigned_jwt({"tenant": "org-uuid"}),
+        )
+        ctx.save_bytes("samples/hdr.bin", b"payload")
+        assert DedupUploadSink.headers_seen
+        sent = {k.lower() for k in DedupUploadSink.headers_seen[-1]}
+        assert "authorization" in sent, "the credential is what carries the org"
+        assert "x-cozy-owner" not in sent
+    finally:
+        httpd.shutdown()
+        reset_upload_sink()
+
+
+def test_an_upload_with_no_derivable_owner_is_no_longer_a_failure_mode() -> None:
+    """`RuntimeError("file save failed (missing owner)")` is gone with the
+    reconstruction: a capability token without a `tenant` claim and an empty
+    `ctx.owner` used to refuse client-side. The hub knows the org."""
+    httpd, base_url = serve_upload_sink()
+    try:
+        ctx = RequestContext(
+            request_id="req-1138-noowner",
+            owner="",
+            file_api_base_url=base_url,
+            worker_capability_token=_unsigned_jwt({"request_id": "req-1138-noowner"}),
+        )
+        asset = ctx.save_bytes("samples/no-owner.bin", b"payload")
+        assert asset.ref == "samples/no-owner.bin"
+        assert DedupUploadSink.requests_seen
+        assert DedupUploadSink.requests_seen[-1][0] == MEDIA_UPLOADS_PATH
+    finally:
+        httpd.shutdown()
+        reset_upload_sink()
+
+
+def test_the_org_reconstruction_helper_is_gone() -> None:
+    assert not hasattr(RequestContext, "_media_upload_owner")
+
+
+def test_the_sink_would_have_caught_the_old_shape() -> None:
+    """The harness is only evidence if it can go red. The alias th#1799
+    deletes must not match the org-less family."""
+    assert is_media_upload_route(MEDIA_UPLOADS_PATH)
+    assert is_media_upload_route(MEDIA_UPLOADS_PATH + "/u1/complete")
+    assert not is_media_upload_route("/api/v1/media/some-org-uuid/uploads")
+    assert not is_media_upload_route("/api/v1/media/some-org-uuid/uploads/u1/complete")
+
+
+def test_the_sink_404s_a_route_tensorhub_does_not_serve() -> None:
+    """The 404 the client would hit against a post-th#1799 hub surfaces as a
+    typed transfer error, not a silent success."""
+    import requests
+
+    httpd, base_url = serve_upload_sink()
+    try:
+        resp = requests.post(
+            f"{base_url}/api/v1/media/some-org/uploads", json={"ref": "x"}, timeout=5
+        )
+        assert resp.status_code == 404
+        assert DedupUploadSink.rejected == ["/api/v1/media/some-org/uploads"]
+        assert not DedupUploadSink.requests_seen
+    finally:
+        httpd.shutdown()
+        reset_upload_sink()
+
+
+def test_a_create_404_raises_a_typed_transfer_error() -> None:
+    """Guards the RED signal itself: had the sink answered every path, the
+    tests above would pass on master. Against a base URL with no upload
+    routes at all the client must fail loudly."""
+    httpd, base_url = serve_upload_sink()
+    try:
+        ctx = RequestContext(
+            request_id="req-1138-404",
+            owner="org",
+            file_api_base_url=base_url + "/nope",
+            worker_capability_token=_unsigned_jwt({"tenant": "org-uuid"}),
+        )
+        with pytest.raises(ArtifactTransferError):
+            ctx.save_bytes("samples/dead.bin", b"payload")
+    finally:
+        httpd.shutdown()
+        reset_upload_sink()