diff --git a/tests/test_integrity_trained.py b/tests/test_integrity_trained.py index 5343ea4..931a692 100644 --- a/tests/test_integrity_trained.py +++ b/tests/test_integrity_trained.py @@ -8,12 +8,78 @@ from validator.integrity import ( check_canonical_data_source, + check_checkpoint_not_blocklisted, check_checkpoint_trained, check_compute_plausibility, check_recipe_config_matches_proof, + check_training_timing, + compare_loss_trajectory, nats_per_token_from_bpb, ) + +# --- training-timing gate (anti off-protocol) -------------------------------- +def test_timing_rejects_run_longer_than_canonical_code_age(): + # aa427cd1/6a25bdc8: ~6.95h wall_clock but the canonical code was ~2h old. + fs = {"wall_clock_s": 25011.0} + ok, reason = check_training_timing(fs, canonical_code_epoch=1_000_000.0, now_epoch=1_000_000.0 + 7380, slack_s=7200) + assert not ok and "off-protocol" in reason + + +def test_timing_accepts_run_within_code_age(): + fs = {"wall_clock_s": 21600.0} # 6h, code existed 8h + assert check_training_timing(fs, canonical_code_epoch=1e6, now_epoch=1e6 + 8 * 3600, slack_s=7200)[0] + + +def test_timing_skips_without_wall_clock_or_epoch(): + assert check_training_timing({"wall_clock_s": 0}, canonical_code_epoch=1e6, now_epoch=1e6 + 10, slack_s=7200)[0] + assert check_training_timing({"wall_clock_s": 9e9}, canonical_code_epoch=None, now_epoch=1e6, slack_s=7200)[0] + + +# --- fraud-checkpoint blocklist ---------------------------------------------- +def test_blocklist_rejects_known_fraud_checkpoint(): + fraud = "f06a9090548978a1987c2b3ada48746348d8134e68f598e0be982e4d5f26f7ab" + ok, reason = check_checkpoint_not_blocklisted(fraud, {fraud}) + assert not ok and "blocklisted" in reason + + +def test_blocklist_passes_unknown_and_none(): + assert check_checkpoint_not_blocklisted("d" * 64, {"a" * 64})[0] + assert check_checkpoint_not_blocklisted(None, {"a" * 64})[0] + assert check_checkpoint_not_blocklisted("a" * 64, set())[0] + + +# --- pre-crown re-derivation trajectory compare ------------------------------ +_HONEST = [(0, 11.02), (50, 6.77), (100, 5.78), (150, 5.20)] + + +def test_rederive_accepts_matching_trajectory(): + assert compare_loss_trajectory(_HONEST, [(0, 11.03), (50, 6.80), (100, 5.75), (150, 5.24)])[0] + + +def test_rederive_rejects_fabricated_trajectory(): + ok, reason = compare_loss_trajectory(_HONEST, [(0, 11.0), (50, 8.9), (100, 8.1), (150, 7.6)]) + assert not ok and "trajectory mismatch" in reason + + +def test_rederive_rejects_offprotocol_faster_drop(): + # A different (e.g. bigger) off-protocol model / different data drops differently. + ok, reason = compare_loss_trajectory(_HONEST, [(0, 11.01), (50, 5.1), (100, 4.0), (150, 3.4)]) + assert not ok + + +def test_rederive_rejects_step0_fingerprint_mismatch(): + ok, reason = compare_loss_trajectory(_HONEST, [(0, 9.8), (50, 6.8), (100, 5.8), (150, 5.2)]) + assert not ok and "step 0" in reason + + +def test_rederive_tolerates_one_noisy_point(): + assert compare_loss_trajectory(_HONEST, [(0, 11.02), (50, 6.77), (100, 6.30), (150, 5.20)])[0] + + +def test_rederive_needs_min_points(): + assert not compare_loss_trajectory(_HONEST, [(0, 11.0)])[0] + VOCAB = 50257 RANDOM_NATS = math.log(VOCAB) # ~10.82 diff --git a/validator/integrity.py b/validator/integrity.py index e4d6819..f89c5d1 100644 --- a/validator/integrity.py +++ b/validator/integrity.py @@ -151,6 +151,144 @@ def check_compute_plausibility( return True, f"compute plausible: {tokens / wall:,.0f} tok/s, {mfu * 100:.0f}% MFU" +# --- Training-timing plausibility (anti off-protocol-training) ----------------- +# +# op2 attestation proves the canonical recipe/runner CODE was present in the +# enclave (container_measurement) and that the bundle is BOUND to it (report_data), +# but NOT that the code EXECUTED to produce the checkpoint. A miner with real CC +# hardware can train a model OFF-PROTOCOL (own box, any data/compute, no data-lock, +# no step/compute gate), then spin up the canonical container and mint an +# attestation over the pre-trained checkpoint + a fabricated final_state. +# +# The physical tell: a checkpoint that attests to the canonical code cannot have +# been trained for longer than that code has EXISTED. If the declared wall_clock_s +# exceeds the wall-clock time elapsed since the canonical code was committed, the +# run necessarily started before this code existed -> it was produced off-protocol +# and the enclave only attested a pre-trained model. Pairs with +# check_compute_plausibility: too-LONG wall_clock trips THIS gate; too-SHORT trips +# the MFU gate. Together they box in the off-protocol class (a real model needs +# real FLOPs => a minimum wall_clock the window cannot contain). +def check_training_timing( + final_state: dict, + *, + canonical_code_epoch: float | None, + now_epoch: float, + slack_s: float = 7200.0, +) -> tuple[bool, str]: + """Reject a checkpoint whose declared training duration exceeds the lifetime of + the canonical code it attests to. Best-effort: skipped (ok) when the canonical + code epoch is unknown or no wall_clock_s is declared. Returns (ok, reason); + ok=False -> reject as off-protocol.""" + fs = final_state or {} + if canonical_code_epoch is None: + return True, "timing: unknown canonical code epoch (skipped)" + try: + wall = float(fs.get("wall_clock_s", 0) or 0) + except (TypeError, ValueError): + return True, "timing: non-numeric wall_clock_s (skipped)" + if wall <= 0: + return True, "timing: no declared wall_clock_s (skipped)" + code_age = float(now_epoch) - float(canonical_code_epoch) + if wall > code_age + slack_s: + return False, ( + f"off-protocol training: declared wall_clock_s={wall:,.0f}s ({wall / 3600:.1f}h) " + f"exceeds canonical code age {code_age:,.0f}s ({code_age / 3600:.1f}h, " + f"+{slack_s / 3600:.1f}h slack) — the attested canonical recipe is younger than " + f"the claimed run, so the checkpoint was trained before this code existed" + ) + return True, f"timing plausible: wall {wall / 3600:.1f}h <= code age {code_age / 3600:.1f}h" + + +def check_checkpoint_not_blocklisted(checkpoint_sha256: str | None, blocked: set) -> tuple[bool, str]: + """Reject a checkpoint whose SHA-256 was previously dethroned as fraud/off-protocol. + + Stopgap against re-submitting the IDENTICAL off-protocol model under a fresh + bundle hash + adjusted final_state metadata. The timing gate weakens as the + canonical code ages (a 7h claim becomes "possible" 7h after the cutover), so an + unchanged fraud checkpoint can otherwise be re-crowned by simply waiting. The + caller passes manifest['checkpoint_sha256'], which is authenticated against the + on-disk checkpoint by the artifact-integrity loop. Returns (ok, reason).""" + if isinstance(checkpoint_sha256, str) and checkpoint_sha256 in blocked: + return False, ( + f"blocklisted checkpoint {checkpoint_sha256[:16]}… — this exact model was " + f"previously dethroned as off-protocol/fabricated; re-derive on the canonical " + f"code to resubmit" + ) + return True, "checkpoint not blocklisted" + + +# --- Pre-crown re-derivation (proof of EXECUTION, not just presence) ----------- +# +# op2 attests that the canonical code was PRESENT; nothing proves it EXECUTED to +# produce the checkpoint. The timing gate + fraud blocklist raise the cost but a +# patient attacker retrains a fresh off-protocol checkpoint and waits out the +# timing window. The only check that proves the declared run actually happened is +# to RE-RUN a slice of it: apply the miner's patch to the canonical recipe, run the +# real train.py for the first N steps on CANONICAL data with the miner's config + +# seed, and compare the re-derived per-step loss trajectory against the declared +# training_log.jsonl. +# +# Why it works: the step-0 loss (init-seed weights forward on the first canonical +# batch) is a near-deterministic FINGERPRINT of (arch, seed, data) — an off-protocol +# run on different data/arch, or a fabricated log, misses it. The next few logged +# points can't be reproduced without actually running the canonical optimizer on +# canonical data, so faking them == honestly training (the attacker gains nothing). +# Coverage limit: partial re-derivation proves the run STARTED honestly; a +# "run N canonical steps then swap the final checkpoint" attack needs the attacker +# to actually run N canonical steps AND the swapped checkpoint still faces op4 — it +# raises cost sharply but only full re-derivation closes it completely. +def compare_loss_trajectory( + declared, + rederived, + *, + step0_tol: float = 0.10, + abs_tol: float = 0.40, + rel_tol: float = 0.10, + min_points: int = 2, +) -> tuple[bool, str]: + """Compare a declared vs a re-derived training-loss trajectory. + + Args: + declared/rederived: iterables of (step, loss), matched by step number. + step0_tol: tight band for the step-0 fingerprint (init forward, deterministic). + abs_tol/rel_tol: looser band for later steps (benign GPU/compile nondeterminism); + a step passes if |declared - rederived| <= max(abs_tol, rel_tol*|declared|). + min_points: minimum matched steps required to render a verdict. + + Returns (ok, reason). ok=False => the declared run was not reproduced on canonical + data (off-protocol / fabricated log).""" + dd = {int(s): float(v) for s, v in declared if v == v} # drop NaN + rr = {int(s): float(v) for s, v in rederived if v == v} + common = sorted(set(dd) & set(rr)) + if len(common) < min_points: + return False, ( + f"re-derivation produced too few comparable points ({len(common)} < {min_points}) " + f"— cannot confirm the declared training ran on canonical code/data" + ) + # Step-0 fingerprint: init-seed weights forwarded on the first canonical batch. + # A miss here means different arch/seed/data than the canonical recipe. + if 0 in common and abs(dd[0] - rr[0]) > step0_tol: + return False, ( + f"re-derivation mismatch at step 0: declared loss {dd[0]:.3f} vs re-derived " + f"{rr[0]:.3f} (tol {step0_tol:.2f}) — different init/data/arch than the canonical " + f"recipe: the checkpoint was trained off-protocol or the log is fabricated" + ) + fails = [] + for s in common: + tol = step0_tol if s == 0 else max(abs_tol, rel_tol * abs(dd[s])) + if abs(dd[s] - rr[s]) > tol: + fails.append((s, dd[s], rr[s], tol)) + # Tolerate a single noisy point; a majority outside band = systematic divergence. + if len(fails) > max(0, (len(common) - 1) // 2): + s, d, r, t = fails[0] + return False, ( + f"re-derivation trajectory mismatch: {len(fails)}/{len(common)} steps outside band " + f"(e.g. step {s}: declared {d:.3f} vs re-derived {r:.3f}, tol {t:.2f}) — the declared " + f"training was not reproduced on canonical data (off-protocol)" + ) + return True, f"re-derivation reproduced {len(common) - len(fails)}/{len(common)} trajectory points" + + def _added_config_jsons(patch_text: str) -> list[dict]: """Parse every NEW/whole configs/*.json the patch adds (best-effort).""" import json diff --git a/validator/validator.py b/validator/validator.py index 3d26f8d..d4c5cf3 100644 --- a/validator/validator.py +++ b/validator/validator.py @@ -17,7 +17,10 @@ import hashlib import json import math +import os +import subprocess import sys +import time from dataclasses import asdict, dataclass, field from pathlib import Path @@ -43,8 +46,10 @@ from proof.sources import compute_container_measurement from validator.integrity import ( check_canonical_data_source, + check_checkpoint_not_blocklisted, check_compute_plausibility, check_recipe_config_matches_proof, + check_training_timing, ) # Hard-coded sanity bounds for the miner-submitted model config. The validator @@ -173,6 +178,61 @@ def _safe_load_checkpoint_weights(ckpt_path: Path, expected_keys: set[str] | Non return state_dict +def _git_commit_epoch(repo_dir: str) -> float | None: + """Unix commit time (UTC seconds) of HEAD in repo_dir, or None if unavailable.""" + try: + out = subprocess.run( + ["git", "-C", repo_dir, "show", "-s", "--format=%ct", "HEAD"], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + if out.returncode == 0 and out.stdout.strip(): + return float(out.stdout.strip()) + except (OSError, ValueError, subprocess.SubprocessError): + return None + return None + + +def _fraud_checkpoints(ralph_root: Path) -> set: + """Load the fraud/off-protocol checkpoint blocklist (chain/fraud_checkpoints.json). + Entries may be bare SHA-256 strings or {"checkpoint_sha256": ...} dicts. A missing + or malformed file yields an empty set (fail-open).""" + p = ralph_root / "chain" / "fraud_checkpoints.json" + try: + data = json.loads(p.read_text(encoding="utf-8")) + except (OSError, ValueError): + return set() + out: set = set() + for e in data if isinstance(data, list) else []: + if isinstance(e, str): + out.add(e) + elif isinstance(e, dict) and isinstance(e.get("checkpoint_sha256"), str): + out.add(e["checkpoint_sha256"]) + return out + + +def _canonical_code_epoch() -> float | None: + """Wall-clock epoch the attested canonical code was 'born' — the commit time of + the PINNED canonical recipe (RECIPE_DIR HEAD), or RALPH_CANONICAL_CODE_EPOCH if + set. Anchored to the pinned recipe (not ralph HEAD) so validator-only deploys + don't reset the clock and retro-invalidate honest in-flight runs; only a + measurement cutover re-pins RECIPE_DIR. Used by the training-timing gate.""" + env = os.environ.get("RALPH_CANONICAL_CODE_EPOCH", "").strip() + if env: + try: + return float(env) + except ValueError: + pass + try: + from ralph_bootstrap import RECIPE_DIR + + return _git_commit_epoch(str(RECIPE_DIR)) + except Exception: # noqa: BLE001 — bootstrap/git failure -> gate skips (fail-open) + return None + + def op1_diff_and_integrity( ralph_root: Path, submission_payload: dict, @@ -215,6 +275,13 @@ def op1_diff_and_integrity( ("checkpoint", proof_dir / "training" / "checkpoint.pt", manifest.get("checkpoint_sha256")), ("training_log", proof_dir / "training" / "training_log.jsonl", manifest.get("training_log_sha256")), ("calibration", proof_dir / "calibration.json", manifest.get("calibration_sha256")), + # final_state is REQUIRED. The anti-gaming gates (compute/timing/data/config) + # all read final_state.config, and the runner folds final_state into bundle_hash + # UNCONDITIONALLY (proof.runner). If a bundle could omit it, the validator would + # (a) skip every gate (the `if fs_path.exists()` block below) and (b) recompute a + # 4-component hash that still self-consistently matches a tampered submission — + # a one-file bypass of the entire anti-gaming surface. Requiring it here closes that. + ("final_state", proof_dir / "training" / "final_state.json", manifest.get("final_state_sha256")), ] # Attestation is now required (single attested-execution tier). if manifest.get("attestation_sha256"): @@ -233,6 +300,15 @@ def op1_diff_and_integrity( if actual != expected: return False, f"{name} hash mismatch (expected {expected[:8]}, got {actual[:8]})" + # Fraud-checkpoint blocklist: reject re-submission of a checkpoint previously + # dethroned as off-protocol/fabricated (identical model, new bundle hash + + # adjusted metadata). manifest['checkpoint_sha256'] is authenticated by the loop above. + ok_bl, detail_bl = check_checkpoint_not_blocklisted( + manifest.get("checkpoint_sha256"), _fraud_checkpoints(ralph_root) + ) + if not ok_bl: + return False, detail_bl + # Recompute bundle_hash from disk and require it match BOTH the # submission's signed-over hash AND the manifest's declared bundle hash. # The recipe for bundle_hash is the same as in proof.runner.run_proof_test. @@ -245,9 +321,9 @@ def op1_diff_and_integrity( bundle_components.append(_file_sha256(proof_dir / "training" / "checkpoint.pt").encode()) bundle_components.append(_file_sha256(proof_dir / "training" / "training_log.jsonl").encode()) bundle_components.append(_file_sha256(proof_dir / "calibration.json").encode()) - fs_path = proof_dir / "training" / "final_state.json" - if fs_path.exists(): - bundle_components.append(_file_sha256(fs_path).encode()) + # final_state is guaranteed present by the required-artifact loop above, so append + # it unconditionally — matching proof.runner's unconditional 5-component hash. + bundle_components.append(_file_sha256(proof_dir / "training" / "final_state.json").encode()) recomputed = hashlib.sha256(b"".join(bundle_components)).hexdigest() if submission_payload.get("bundle_hash") != recomputed: return False, ( @@ -325,6 +401,22 @@ def op1_diff_and_integrity( ok_c, detail_c = check_compute_plausibility(final_state, calibration) if not ok_c: return False, detail_c + # Off-protocol training: a checkpoint attesting to the canonical code cannot + # have trained longer than that code has existed. Pairs with the MFU gate + # above (too-long wall_clock here, too-short there). Disable: RALPH_TIMING_GATE_OFF=1. + if os.environ.get("RALPH_TIMING_GATE_OFF") != "1": + try: + _slack = float(os.environ.get("RALPH_TIMING_SLACK_S") or 7200.0) + except ValueError: + _slack = 7200.0 + ok_t, detail_t = check_training_timing( + final_state, + canonical_code_epoch=_canonical_code_epoch(), + now_epoch=time.time(), + slack_s=_slack, + ) + if not ok_t: + return False, detail_t ok_d, detail_d = check_canonical_data_source(final_state) if not ok_d: return False, detail_d @@ -1204,6 +1296,139 @@ def _legacy_hidden_eval( return _run_eval_subprocess(RECIPE_DIR, ckpt_path, ralph_root, "canonical-eval") +def _read_training_log_points(log_path: Path, max_step: int | None = None) -> list: + """Parse (step, loss) points from a training_log.jsonl, up to max_step (inclusive).""" + out: list = [] + try: + for ln in log_path.read_text().splitlines(): + ln = ln.strip() + if not ln: + continue + e = json.loads(ln) + s = int(e["step"]) + if max_step is not None and s > max_step: + break + out.append((s, float(e["loss"]))) + except (OSError, ValueError, KeyError, TypeError): + pass + return out + + +def op_rederive_trajectory(ralph_root: Path, proof_dir: Path) -> tuple[bool, str]: + """Pre-crown proof-of-EXECUTION: re-run the FIRST few steps of the declared + training on CANONICAL data and require the loss trajectory to reproduce. + + op2 attests code PRESENCE, not EXECUTION; the timing gate + fraud blocklist only + raise the cost of off-protocol training. This is the check that actually proves the + declared run happened: apply the miner's patch to the canonical recipe, run the real + train.py with the miner's exact config+seed but pinned to CANONICAL data, collect the + first few logged points, and compare the trajectory (validator.integrity.compare_loss_trajectory). + + OFF by default (RALPH_REDERIVE=1 to enable). Fail-OPEN (skip) when disabled or when the + canonical training data is not materialized on the validator — enable only AFTER + materializing data/data_manifest.json + shards and calibrating the tolerance band on a + known-honest bundle. Expensive (GPU-minutes); the caller should run it only on a bundle + that would actually beat the king. NOTE: written but UNVALIDATED until canonical data + exists on a validator to run it against. + """ + import os + import shutil + import subprocess + import tempfile + import time + + if os.environ.get("RALPH_REDERIVE") != "1": + return True, "re-derivation disabled (RALPH_REDERIVE!=1)" + from ralph_bootstrap import RECIPE_DIR + + canon_manifest = RECIPE_DIR / "data" / "data_manifest.json" + if not canon_manifest.exists(): + return True, "re-derivation skipped: canonical data_manifest.json not materialized on validator" + + fs_path = proof_dir / "training" / "final_state.json" + log_path = proof_dir / "training" / "training_log.jsonl" + if not (fs_path.exists() and log_path.exists()): + return True, "re-derivation skipped: missing final_state/training_log" + try: + cfg = (json.loads(fs_path.read_text()).get("config")) or {} + except (OSError, ValueError): + return True, "re-derivation skipped: unreadable final_state" + + n_points = int(os.environ.get("RALPH_REDERIVE_POINTS", "3")) + log_every = int(cfg.get("log_every", 50) or 50) + max_step = log_every * (n_points - 1) + declared = _read_training_log_points(log_path, max_step=max_step) + if len(declared) < 2: + return True, "re-derivation skipped: declared log too short to compare" + + from proof.runner import _sanitized_env, apply_patch + + patch_path = proof_dir / "patch.diff" + with tempfile.TemporaryDirectory() as tmp: + tmp_p = Path(tmp) + workdir = tmp_p / "workdir" + for sub in ("model", "recipe", "data", "configs"): + src = RECIPE_DIR / sub + if src.exists(): + shutil.copytree(src, workdir / sub, dirs_exist_ok=True) + try: + if patch_path.exists(): + apply_patch(workdir, patch_path) + except Exception as e: # noqa: BLE001 + return False, f"re-derivation: patch apply failed: {str(e)[:150]}" + + # Reproduce the miner's exact config (hence LR schedule) but pin data to canonical. + cfg_file = tmp_p / "rederive_config.json" + cfg_file.write_text(json.dumps(cfg)) + out_dir = tmp_p / "out" + out_dir.mkdir(parents=True, exist_ok=True) + rlog = out_dir / "training_log.jsonl" + train_py = workdir / "recipe" / "train.py" + if not train_py.exists(): + return True, "re-derivation skipped: recipe/train.py not found in workdir" + cmd = [ + sys.executable, str(train_py), + "--config", str(cfg_file), + "--manifest", str(canon_manifest.resolve()), + "--data-base-dir", str((RECIPE_DIR / "data").resolve()), + "--out-dir", str(out_dir), + ] + seed = cfg.get("init_seed", cfg.get("data_seed")) + if seed is not None: + cmd += ["--seed", str(int(seed))] + + env = _sanitized_env(extra={"PYTHONPATH": str(workdir)}) + timeout_s = int(os.environ.get("RALPH_REDERIVE_TIMEOUT_S", "1200")) + proc = subprocess.Popen(cmd, cwd=str(workdir), env=env, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + # Watch the log until the target step is written (schedule preserved, we just + # stop early), then terminate — a faithful partial re-run, not a full retrain. + start = time.time() + rederived: list = [] + try: + while time.time() - start < timeout_s: + if rlog.exists(): + rederived = _read_training_log_points(rlog, max_step=max_step) + if rederived and max(s for s, _ in rederived) >= max_step: + break + if proc.poll() is not None: # train.py finished (ran fewer than max_step) + rederived = _read_training_log_points(rlog, max_step=max_step) + break + time.sleep(2) + finally: + if proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=30) + except subprocess.TimeoutExpired: + proc.kill() + if len(rederived) < 2: + return True, f"re-derivation inconclusive: only {len(rederived)} point(s) produced in {timeout_s}s (skip)" + + from validator.integrity import compare_loss_trajectory + return compare_loss_trajectory(declared, rederived) + + def op4_hidden_eval( ralph_root: Path, proof_dir: Path, @@ -1297,6 +1522,19 @@ def judge_submission( result.rejected = ValidatorReject("op3_log_plausibility", detail) return result + # Pre-crown proof-of-EXECUTION (OFF by default; RALPH_REDERIVE=1). Re-runs the + # first training steps on CANONICAL data and requires the loss trajectory to + # reproduce — the only check that off-protocol training can't satisfy without + # actually running the canonical recipe. Skips cleanly when disabled or when the + # canonical data isn't materialized. Expensive: for efficiency the crown path + # should ideally invoke this only for a bundle that would beat the king (it still + # short-circuits behind op1-op3 here). + ok, detail = op_rederive_trajectory(ralph_root, proof_dir) + result.operations["op_rederive"] = {"ok": ok, "detail": detail} + if not ok: + result.rejected = ValidatorReject("op_rederive_trajectory", detail) + return result + ok, detail, hidden_eval = op4_hidden_eval(ralph_root, proof_dir, chain=chain) result.operations["op4_hidden_eval"] = {"ok": ok, "detail": detail} if not ok: