From e743dc0e08f349c3d1cefb1820799ce1efa6271d Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Sun, 25 Oct 2026 12:24:45 +0000 Subject: [PATCH 1/2] feat(challenge): serve a joined miner its own admission-trial job A public miner could not start a trial in one command: the trial job (IslandJobV1, its input objects, the challenge body) was only reachable through the admin export route, so an operator had to hand the miner a --job folder. GET /v2/runs/{run_id}/join/{admission_id}/job returns the admin export payload (ChallengeStore.export_trial_v2, reused as is) to the currently registered hotkey that owns the admission, never publicly. Unlike the challenge route, the caller is authenticated: admission ids are public through the admission status route, so gating only on the admission's hotkey being registered would let any caller read any miner's trial. The caller sends X-Hotkey and X-Trial-Signature, a hotkey signature of "hypertrain/trial-job/2|{run_id}|{admission_id}" (founder decision, option A, 2026-10-25). Refusals reuse existing codes, in this order: unknown run 404, unknown admission 409 UNKNOWN_ADMISSION and a deregistered or revoked owner 403, all as the challenge route answers; then another hotkey or a missing or bad signature 403 "hotkey is not current accepted identity" (the message of _owner_v2); then the export's own 409/503 refusals. The admin bearer token is not accepted in place of the signature. Rejected: copying the challenge route gate alone (cross-miner reads), and one identical 403 for every caller failure (misleading message for a registered miner using the wrong admission id). Co-Authored-By: Claude Opus 5.5 --- src/hypertrain/challenge/app.py | 26 +++++++++++++++ tests/challenge/test_subnet_gate_v2.py | 30 +++++++++++++++++ tests/challenge/test_trial_export.py | 45 ++++++++++++++++++++++++++ 3 files changed, 101 insertions(+) diff --git a/src/hypertrain/challenge/app.py b/src/hypertrain/challenge/app.py index efc9f86..2bbb23b 100644 --- a/src/hypertrain/challenge/app.py +++ b/src/hypertrain/challenge/app.py @@ -453,6 +453,32 @@ async def challenge_v2(run_id: str, admission_id: str) -> Any: await subnet_member((await run(admission.store.by_id, admission_id)).hotkey) return await run(store.for_run(run_id).admission_v2, run_id, "challenge", b"", admission_id) + @app.get("/v2/runs/{run_id}/join/{admission_id}/job") + async def trial_job_v2( + run_id: str, + admission_id: str, + x_hotkey: Annotated[str | None, Header()] = None, + x_trial_signature: Annotated[str | None, Header()] = None, + ) -> Any: + from hypertrain.protocol.keys import verify + + # The admin export's payload, only to the registered hotkey that owns the admission: + # admission ids are public (admission status), so the caller signs for its hotkey. + _, admission, _ = await run(store.for_run(run_id)._services, run_id) + hotkey = (await run(admission.store.by_id, admission_id)).hotkey + await subnet_member(hotkey) + try: + valid = x_hotkey == hotkey and verify( + decode_hotkey(hotkey), + f"hypertrain/trial-job/2|{run_id}|{admission_id}".encode(), + bytes.fromhex(x_trial_signature or ""), + ) + except (ValueError, KeyError_): + valid = False + if not valid: + raise ChallengeError(403, "hotkey is not current accepted identity") + return await run(store.for_run(run_id).export_trial_v2, run_id, admission_id) + @app.get("/v2/runs/{run_id}/admission/{hotkey}") async def admission_status_v2(run_id: str, hotkey: str) -> Any: return await run(store.for_run(run_id).admission_v2, run_id, "status", b"", hotkey) diff --git a/tests/challenge/test_subnet_gate_v2.py b/tests/challenge/test_subnet_gate_v2.py index b7b5476..38a69eb 100644 --- a/tests/challenge/test_subnet_gate_v2.py +++ b/tests/challenge/test_subnet_gate_v2.py @@ -313,6 +313,36 @@ def test_deregistered_after_join_refused(gated, subnet_master): ) +def test_trial_job_refused_like_challenge_once_owner_deregistered(gated, subnet_master): + # Given: HOT[0] joined while registered and holds an open trial. + n = gated + subnet_master.registered.add(HOT[0].ss58) + identity = n.join(0).json()["admission_id"] + n.push(2) + assert n.client.get(n.url + "/join/" + identity + "/challenge").status_code == 200 + message = f"hypertrain/trial-job/2|{n.manifest.run_id()}|{identity}".encode() + signed = {"x-hotkey": HOT[0].ss58, "x-trial-signature": HOT[0].sign(message).hex()} + url = n.url + "/join/" + identity + "/job" + # When: the hotkey leaves the subnet. + subnet_master.registered.clear() + refresh(n) + # Then: the owner's signature, the admin token, or both get the challenge route's refusal. + challenge = n.client.get(n.url + "/join/" + identity + "/challenge") + assert challenge.status_code == 403 + assert challenge.json() == {"detail": "the hotkey is not registered on the subnet"} + for headers in (signed, service.admin(), {**signed, **service.admin()}): + response = n.client.get(url, headers=headers) + assert (response.status_code, response.json()) == (403, challenge.json()), headers + assert not (n.store.state_dir / "trials-v2").exists() + # And: registered again, the admin token still does not stand in for the signature. + subnet_master.registered.add(HOT[0].ss58) + refresh(n) + response = n.client.get(url, headers=service.admin()) + assert response.status_code == 403 + assert response.json() == {"detail": "hotkey is not current accepted identity"} + assert n.client.get(url, headers=signed).status_code == 200 + + def test_weights_exclude_deregistered_hotkeys(tmp_path, secrets_dir, clock): # Given: two vested entitlements; one hotkey later deregistered. master = Master() diff --git a/tests/challenge/test_trial_export.py b/tests/challenge/test_trial_export.py index a686c88..ff07d79 100644 --- a/tests/challenge/test_trial_export.py +++ b/tests/challenge/test_trial_export.py @@ -40,6 +40,16 @@ def export(n, identity): return n.client.post(n.url + f"/admin/trial/{identity}/export", headers=service.admin()) +def job(n, identity, key=service.HOT[0], hotkey=None, run_id=None): + """Miner-facing job read, signed by `key` (X-Hotkey defaults to the signer).""" + run_id = run_id or n.manifest.run_id() + signature = key.sign(f"hypertrain/trial-job/2|{run_id}|{identity}".encode()).hex() + return n.client.get( + f"/v2/runs/{run_id}/join/{identity}/job", + headers={"x-hotkey": hotkey or key.ss58, "x-trial-signature": signature}, + ) + + def test_export_matches_prepare_reference_inputs(trial): n, identity, envelope = trial response = export(n, identity) @@ -137,6 +147,41 @@ def test_export_requires_admin(trial): assert not (n.store.state_dir / "trials-v2").exists() +def test_owner_reads_the_admin_export_without_admin_token(trial): + n, identity, envelope = trial + response = job(n, identity) + assert response.status_code == 200, response.text + assert response.json() == export(n, identity).json() + assert response.json()["challenge"] == envelope["body"] + + +def test_job_route_refuses_another_miners_admission(trial): + n, identity, _ = trial + other = service.HOT[1] # registered on the fixture subnet, not this admission's hotkey + for response in ( + job(n, identity, key=other), + job(n, identity, key=other, hotkey=service.HOT[0].ss58), + job(n, identity, hotkey=other.ss58), + ): + assert response.status_code == 403, response.text + assert response.json() == {"detail": "hotkey is not current accepted identity"} + assert not (n.store.state_dir / "trials-v2").exists() + assert job(n, identity).status_code == 200 + + +@pytest.mark.parametrize( + "unknown,status,detail", [("run", 404, "unknown run"), ("admission", 409, "UNKNOWN_ADMISSION")] +) +def test_job_route_unknown_ids_answer_like_challenge(trial, unknown, status, detail): + n, identity, _ = trial + run_id = "cd" * 32 if unknown == "run" else n.manifest.run_id() + identity = "ab" * 32 if unknown == "admission" else identity + challenge = n.client.get(f"/v2/runs/{run_id}/join/{identity}/challenge") + response = job(n, identity, run_id=run_id) + assert (response.status_code, response.json()) == (status, {"detail": detail}) + assert (challenge.status_code, challenge.json()) == (status, {"detail": detail}) + + def test_export_refuses_changed_job(trial): n, identity, envelope = trial directory = n.store.state_dir / "trials-v2" / envelope["body"]["nonce"] From 39e2265476c262cdaa3eb5b5e6b897df80b42472 Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Sun, 25 Oct 2026 12:24:45 +0000 Subject: [PATCH 2/2] feat(miner): let probe-v2 fetch its own trial when --job is omitted With the owner-signed trial job route, probe-v2 no longer needs an operator-made folder. Without --job and --challenge it reads its admission id from the admission status, GETs the challenge (which issues the trial when none is open, otherwise returns the open one), GETs the signed job route, and checks the job, objects and signed challenge with the same models and bindings as trial_export.py. Each input is streamed into a private temporary folder under the miner's workdir and verified by sha256, then the probe runs as with --job and the folder is deleted, also on failure (a re-run starts over). The inputs are whole plain objects (the export puts each file as one object), so the 1 MiB part bound of _get_object cannot be used. Like trial_export they are streamed to disk, with no compression added; unlike it, the miner talks to any configured api, so it keeps the _get_object defences: Accept-Encoding identity, refusal of an encoded body, and a bound of MAX_PARTS * PART_BYTES per object (the largest object the round-input path accepts). Measured on the CPU test fixture (824 params): job JSON 5289 bytes, start_state 4256, ef_in 4256, v0 16, samples 20, sample_proofs 271; start_state grows ~4 bytes/param under reset. Rejected: comparing job.manifest with the signed manifest, which IslandJobV1 already forces through manifest.validate_run_id(run_id). --job and --challenge still work as before and must be given together. Co-Authored-By: Claude Opus 5.5 --- docs/network-v2.md | 17 +++ src/hypertrain/miner/cli.py | 11 +- src/hypertrain/miner/core.py | 103 +++++++++++++++++ tests/miner/test_probe_trial_v2.py | 176 +++++++++++++++++++++++++++++ 4 files changed, 304 insertions(+), 3 deletions(-) create mode 100644 tests/miner/test_probe_trial_v2.py diff --git a/docs/network-v2.md b/docs/network-v2.md index 1d124e2..5726474 100644 --- a/docs/network-v2.md +++ b/docs/network-v2.md @@ -143,6 +143,23 @@ uv run --frozen hypertrain-miner probe-v2 --config miner.toml \ uv run --frozen hypertrain-miner proof-v2 --config miner.toml --proof proof.json ``` +A joined miner can skip the export: `probe-v2 --config miner.toml > proof.json` +without `--job`/`--challenge` reads its admission ID (`status-v2`), GETs the +challenge (which issues the trial when none is open), then +`GET /v2/runs/{run_id}/join/{admission_id}/job` with headers `X-Hotkey` and +`X-Trial-Signature` (hex hotkey signature of +`hypertrain/trial-job/2|{run_id}|{admission_id}`). The route returns the admin +export's JSON only to the currently registered hotkey that owns the admission: +unknown run 404, unknown admission 409 `UNKNOWN_ADMISSION`, a deregistered or +revoked owner 403 like the challenge route, any other hotkey or a missing or bad +signature 403 (the admin token is no substitute), then the export's own refusals +(409 `NO_CHALLENGE`, 409 `REFERENCE_TRIAL_CLOSED` once the trial is finalized, +503 on a beacon pause). The miner checks the job, objects and signed challenge as +`trial_export` does, streams each input (one uncompressed, sha256-verified +object; about 4 bytes per parameter for a `reset` start state, more under +`carry`) into a private temporary folder under its workdir, probes, and deletes +the folder, also on failure: a re-run downloads and probes again. + `probe-v2` executes the screen and uploads original rank-0 state/EF/delta/leaves; stdout contains the signed proof/screen pair. The operator independently prepares and finalizes references through authenticated admin routes; miners cannot diff --git a/src/hypertrain/miner/cli.py b/src/hypertrain/miner/cli.py index 340bf04..0c475bc 100644 --- a/src/hypertrain/miner/cli.py +++ b/src/hypertrain/miner/cli.py @@ -385,11 +385,11 @@ def main(argv: list[str] | None = None) -> int: s.add_argument("--request-id", required=True) s.add_argument("--expiry", type=int, required=True) if name in ("probe-v2", "run-v2", "watch"): - s.add_argument("--job", type=Path, required=name == "probe-v2") + s.add_argument("--job", type=Path) if name == "run-v2": s.add_argument("--round", type=int) if name == "probe-v2": - s.add_argument("--challenge", type=Path, required=True) + s.add_argument("--challenge", type=Path) if name == "proof-v2": s.add_argument("--proof", type=Path, required=True) if name == "watch": @@ -465,7 +465,12 @@ def main(argv: list[str] | None = None) -> int: case "status-v2": result = network.status() case "probe-v2": - result = network.probe(args.job, args.challenge) + if (args.job is None) != (args.challenge is None): + raise MinerError("probe-v2 takes --job and --challenge, or neither") + if args.job is None: + result = network.probe_trial() + else: + result = network.probe(args.job, args.challenge) case "proof-v2": result = network.proof(args.proof) case "run-v2": diff --git a/src/hypertrain/miner/core.py b/src/hypertrain/miner/core.py index 98f4e69..70ddb89 100644 --- a/src/hypertrain/miner/core.py +++ b/src/hypertrain/miner/core.py @@ -890,6 +890,109 @@ def traced_launch( raise MinerError("probe deadline/cancellation") return result + def probe_trial(self) -> dict[str, Any]: + """probe-v2 without --job: this hotkey's open trial, fetched into a private temp folder.""" + with tempfile.TemporaryDirectory(prefix="trial-", dir=self.directory) as folder: + job, challenge = self._fetch_trial(Path(folder).resolve()) + return self.probe(job, challenge) + + def _fetch_trial(self, root: Path) -> tuple[Path, Path]: + """The folder trial_export writes, from the miner's own signed job read. + + The challenge GET issues the trial when none is open (else returns the stored one); + the job, its inputs and the signed challenge are checked as trial_export checks them.""" + from pydantic import TypeAdapter + + from hypertrain.protocol import envelope_v2 + from hypertrain.protocol.messages_v2 import Hex64, IslandJobV1, JoinChallenge + + admission_id = self.status()["record"]["admission_id"] + url = self.url + f"/join/{admission_id}" + envelope = self.api.call("GET", url + "/challenge") + signature = self.kp.sign(f"hypertrain/trial-job/2|{self.run_id}|{admission_id}".encode()) + data = self.api.call( + "GET", + url + "/job", + headers={"X-Hotkey": self.kp.ss58, "X-Trial-Signature": signature.hex()}, + ) + try: + job = IslandJobV1.model_validate(data["job"]) + challenge = JoinChallenge.model_validate(data["challenge"]) + objects = TypeAdapter(dict[str, Hex64]).validate_python(data["objects"]) + nonce = data["nonce"] + signed = envelope_v2.parse_envelope(envelope) + except (ValueError, KeyError, TypeError) as error: + raise MinerError("malformed trial job response") from error + if ( + job.run_id != self.run_id + or challenge.manifest_hash != self.run_id + or challenge.admission_id != admission_id + or challenge.nonce != nonce + or set(objects) != set(job.object_paths) + ): + raise MinerError("trial job/challenge/object binding differs") + for name in ("start_state", "ef_in", "v0"): + if objects.get(name) != getattr(job, name + "_sha256"): + raise MinerError(f"trial {name} sha256 differs from job") + if ( + signed.type != "JoinChallenge" + or signed.run_id != self.run_id + or signed.signer != self.manifest.training.coord_pubkey + or signed.body != challenge.body() + or not envelope_v2.verify_envelope(envelope) + ): + raise MinerError("signed challenge differs from the trial job") + destinations = {root / "job.json", root / "challenge.json"} + paths: dict[str, Path] = {} + for name, relative in job.object_paths.items(): + target = (root / relative).resolve() + if ( + target == root + or not target.is_relative_to(root) + or any( + target == p or target in p.parents or p in target.parents for p in destinations + ) + ): + raise MinerError(f"unsafe trial object path: {relative}") + destinations.add(target) + paths[name] = target + for name, digest in objects.items(): + paths[name].parent.mkdir(parents=True, exist_ok=True) + self._get_whole(digest, paths[name]) + (root / "job.json").write_bytes(canonicalize(job.body())) + (root / "challenge.json").write_bytes(canonicalize(envelope)) + return root / "job.json", root / "challenge.json" + + def _get_whole(self, digest: str, target: Path) -> None: + """Stream one whole plain object to `target` and verify it: the trial export stores each + input as a single object, so the 1 MiB part bound of _get_object does not apply; the + bound is the largest object a part index can describe.""" + from hypertrain.protocol.messages_v2 import MAX_PARTS, PART_BYTES + + def get() -> None: + hasher, size = hashlib.sha256(), 0 + with self.api.c.stream( + "GET", + self.api.base + self.url + "/objects/" + digest, + headers={"Accept-Encoding": "identity"}, + ) as r: + if not r.is_success: + raise ChallengeHTTPError(r.status_code, "trial input object unavailable") + # As _get_object: httpx would decode gzip/br before the bound check. + if r.headers.get("Content-Encoding", "identity").lower() != "identity": + raise MinerError("trial input object has a non-identity content encoding") + with target.open("wb") as stream: + for chunk in r.iter_bytes(chunk_size=64 << 10): + size += len(chunk) + if size > MAX_PARTS * PART_BYTES: + raise MinerError("trial input object exceeds the object size bound") + hasher.update(chunk) + stream.write(chunk) + if hasher.hexdigest() != digest: + raise MinerError(f"trial input {target.name} sha256 mismatch") + + _retrying("GET object " + digest, get) + def _bounded_put( self, deadline: float, cancel: threading.Event | None ) -> Callable[[bytes], str]: diff --git a/tests/miner/test_probe_trial_v2.py b/tests/miner/test_probe_trial_v2.py new file mode 100644 index 0000000..34b7095 --- /dev/null +++ b/tests/miner/test_probe_trial_v2.py @@ -0,0 +1,176 @@ +"""`probe-v2` without --job: a joined miner fetches its own open admission trial over HTTP.""" + +from __future__ import annotations + +import gzip +import importlib.util +import json +import sys +from pathlib import Path + +import httpx +import pytest + +from hypertrain.miner import cli + +_spec = importlib.util.spec_from_file_location( + "probe_trial_service_fixture", + Path(__file__).parents[1] / "challenge/test_service_network_v2.py", +) +assert _spec is not None and _spec.loader is not None +service = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = service +_spec.loader.exec_module(service) +network = service.network +HOT = service.HOT[0] + + +def signed_job(n, key, identity): + message = f"hypertrain/trial-job/2|{n.manifest.run_id()}|{identity}".encode() + return n.client.get( + n.url + f"/join/{identity}/job", + headers={"x-hotkey": key.ss58, "x-trial-signature": key.sign(message).hex()}, + ) + + +def tamper(body, fault): + """Apply one named fault to the /job response the miner receives.""" + match fault: + case "nonce": + body["nonce"] = "cd" * 32 + case "challenge": # consistent with itself, but not the signed challenge + body["nonce"] = body["challenge"]["nonce"] = "cd" * 32 + case "job.json" | "ef_in": + body["job"]["object_paths"]["samples"] = fault + + +@pytest.fixture +def miner(network, tmp_path, monkeypatch): + """HOT[0] joined, its trial seed ready; miner.toml whose HTTP goes to the in-process app.""" + n = network + response = n.join() + assert response.status_code == 200, response.text + identity = response.json()["admission_id"] + n.push(n.now + 1) + key = tmp_path / "hot.key" + key.write_bytes(bytes([80]) * 32) + key.chmod(0o600) + work = tmp_path / "work" + config = tmp_path / "miner.toml" + config.write_text( + "\n".join( + f"{k} = {json.dumps(str(v))}" + for k, v in { + "api": "https://coordinator.test", + "keyfile": key, + "workdir": work, + "state_source": "cpu", + "image_digest": n.manifest.training.reference_spec.image_digest, + "run_id": n.manifest.run_id(), + "owner_hotkey": service.OWNER.ss58, + "device": "cpu", + }.items() + ) + ) + requests = [] + fault = {} + + def handler(request): + requests.append(request) + path = request.url.path + for suffix, body in fault.get("serve", {}).items(): + if path.endswith(suffix): + return httpx.Response(200, json=body) + response = n.client.request( + request.method, path, headers=request.headers, content=request.content + ) + name = fault.get("name") + if name and path.endswith("/job") and response.status_code == 200: + body = response.json() + tamper(body, name) + return httpx.Response(200, json=body) + if name == "signature" and path.endswith("/challenge"): + return httpx.Response(200, json={**response.json(), "sig": "00" * 64}) + if "/objects/" in path and request.method == "GET": + if name == "corrupt": + return httpx.Response(200, content=b"corrupt object") + if name == "gzip": + encoded = gzip.compress(response.content) + return httpx.Response(200, content=encoded, headers={"Content-Encoding": "gzip"}) + return httpx.Response(response.status_code, content=response.content) + + client_type = httpx.Client + monkeypatch.setattr( + cli.httpx, + "Client", + lambda **kwargs: client_type(transport=httpx.MockTransport(handler), **kwargs), + ) + return n, identity, config, work / n.manifest.run_id() / HOT.ss58, requests, fault + + +def test_probe_v2_without_job_fetches_own_trial_and_completes(miner, capsys): + n, identity, config, folder, requests, _ = miner + assert cli.main(["probe-v2", "--config", str(config)]) == 0 + signed = json.loads(capsys.readouterr().out) + paths = [r.url.path for r in requests] + assert n.url + f"/join/{identity}/job" in paths + assert not any("/admin/" in p for p in paths) + assert not any("authorization" in r.headers for r in requests) + assert list(folder.iterdir()) == [] + # The service accepts the probe's proof (it replays the reference itself) ... + response = n.client.post(n.url + "/join/proof", json=signed) + assert response.status_code == 200, response.text + assert response.json()["clean_count"] == 1 + # ... and the finalized trial is no longer served. + response = signed_job(n, HOT, identity) + assert response.status_code == 409, response.text + assert response.json() == {"detail": "REFERENCE_TRIAL_CLOSED"} + + +@pytest.mark.parametrize( + "name,error", + [ + ("corrupt", "sha256 mismatch"), + ("gzip", "non-identity content encoding"), + ("nonce", "binding differs"), + ("challenge", "signed challenge differs"), + ("signature", "signed challenge differs"), + ("job.json", "unsafe trial object path"), + ("ef_in", "unsafe trial object path"), + ], +) +def test_probe_v2_refuses_unverified_trial_without_probing(miner, caplog, name, error): + _, _, config, folder, requests, fault = miner + fault["name"] = name + assert cli.main(["probe-v2", "--config", str(config)]) == 2 + assert error in caplog.text + assert not any(r.method == "PUT" for r in requests) + assert list(folder.iterdir()) == [] + + +def test_probe_v2_refuses_another_admissions_genuine_trial(miner, caplog): + n, identity, config, folder, requests, fault = miner + other = n.join(1).json()["admission_id"] + n.push(n.now + 1) + challenge = n.client.get(n.url + f"/join/{other}/challenge") + assert challenge.status_code == 200, challenge.text + job = signed_job(n, service.HOT[1], other) + assert job.status_code == 200, job.text + fault["serve"] = { + f"/join/{identity}/challenge": challenge.json(), + f"/join/{identity}/job": job.json(), + } + assert cli.main(["probe-v2", "--config", str(config)]) == 2 + assert "binding differs" in caplog.text + assert not any(r.method == "PUT" for r in requests) + assert list(folder.iterdir()) == [] + + +@pytest.mark.parametrize("given", ["--job", "--challenge"]) +def test_probe_v2_refuses_job_or_challenge_alone(miner, caplog, tmp_path, given): + _, _, config, _, requests, _ = miner + stray = tmp_path / "stray.json" + stray.write_text("{}") + assert cli.main(["probe-v2", "--config", str(config), given, str(stray)]) == 2 + assert "takes --job and --challenge, or neither" in caplog.text + assert not any(r.url.path.endswith(("/challenge", "/job")) for r in requests)