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/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/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/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"] 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)