| from __future__ import annotations |
|
|
| import base64 |
| import hashlib |
| import json |
| import os |
| import shutil |
| import subprocess |
| import tempfile |
| import time |
| import zipfile |
| from pathlib import Path |
| from typing import Iterator |
|
|
| import gradio as gr |
|
|
| DEFAULT_REPOSITORY = "toxzak-svg/starfire" |
| DEFAULT_COMMIT = "902c88e533a75076542726d5975880ef0234d54c" |
| DEFAULT_FEATURE = "omega-g2-shadow" |
| SCIENTIFIC_PATHS = ( |
| "lib/omega_g2_shadow.rs", |
| "lib/prediction/mod.rs", |
| "lib/examples/omega_g2_s0_real_trace_shadow.rs", |
| ) |
| MAX_LOG_CHARS = 180_000 |
| ARTIFACT_ROOT = Path("/tmp/starfire-verifier-artifacts") |
| ARTIFACT_ROOT.mkdir(parents=True, exist_ok=True) |
|
|
|
|
| def _append_log(log: str, line: str) -> str: |
| combined = f"{log}{line}\n" |
| if len(combined) > MAX_LOG_CHARS: |
| combined = "[earlier output truncated]\n" + combined[-MAX_LOG_CHARS:] |
| return combined |
|
|
|
|
| def _run_streaming( |
| command: list[str], |
| cwd: Path, |
| env: dict[str, str], |
| display: str, |
| ) -> Iterator[tuple[str, int | None]]: |
| yield f"\n$ {display}\n", None |
| process = subprocess.Popen( |
| command, |
| cwd=cwd, |
| env=env, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT, |
| text=True, |
| encoding="utf-8", |
| errors="replace", |
| bufsize=1, |
| ) |
| assert process.stdout is not None |
| for line in process.stdout: |
| yield line, None |
| code = process.wait() |
| yield f"[exit {code}]\n", code |
|
|
|
|
| def _sha256(path: Path) -> str: |
| digest = hashlib.sha256() |
| with path.open("rb") as handle: |
| for chunk in iter(lambda: handle.read(1024 * 1024), b""): |
| digest.update(chunk) |
| return digest.hexdigest() |
|
|
|
|
| def _source_hashes(repo: Path) -> dict[str, str]: |
| hashes: dict[str, str] = {} |
| for relative in SCIENTIFIC_PATHS: |
| target = repo / relative |
| if not target.is_file(): |
| raise FileNotFoundError(f"required scientific source missing: {relative}") |
| hashes[relative] = _sha256(target) |
| return hashes |
|
|
|
|
| def _ensure_rust(env: dict[str, str], work: Path) -> Iterator[tuple[str, int | None]]: |
| cargo = shutil.which("cargo", path=env.get("PATH")) |
| if cargo: |
| yield f"Using existing Cargo: {cargo}\n", 0 |
| return |
|
|
| installer = work / "rustup-init.sh" |
| download = [ |
| "curl", |
| "--proto", |
| "=https", |
| "--tlsv1.2", |
| "-sSf", |
| "https://sh.rustup.rs", |
| "-o", |
| str(installer), |
| ] |
| yield from _run_streaming(download, work, env, "download rustup installer") |
| if not installer.is_file(): |
| yield "rustup installer was not downloaded\n", 1 |
| return |
| install = ["sh", str(installer), "-y", "--profile", "minimal", "--default-toolchain", "stable"] |
| yield from _run_streaming(install, work, env, "install stable Rust") |
|
|
|
|
| def _clone_private_repo( |
| repository: str, |
| commit: str, |
| token: str, |
| work: Path, |
| env: dict[str, str], |
| ) -> Iterator[tuple[str, int | None]]: |
| askpass = work / "git-askpass.sh" |
| askpass.write_text( |
| "#!/bin/sh\n" |
| "case \"$1\" in\n" |
| " *Username*) printf '%s\\n' 'x-access-token' ;;\n" |
| " *) printf '%s\\n' \"$GITHUB_TOKEN\" ;;\n" |
| "esac\n", |
| encoding="utf-8", |
| ) |
| askpass.chmod(0o700) |
| env["GIT_ASKPASS"] = str(askpass) |
| env["GIT_TERMINAL_PROMPT"] = "0" |
| env["GITHUB_TOKEN"] = token |
|
|
| repo_dir = work / "starfire" |
| clone = ["git", "clone", "--no-checkout", "--filter=blob:none", f"https://github.com/{repository}.git", str(repo_dir)] |
| yield from _run_streaming(clone, work, env, f"git clone --no-checkout https://github.com/{repository}.git starfire") |
| if not repo_dir.is_dir(): |
| return |
| fetch = ["git", "fetch", "--depth", "1", "origin", commit] |
| yield from _run_streaming(fetch, repo_dir, env, f"git fetch --depth 1 origin {commit}") |
| checkout = ["git", "checkout", "--detach", commit] |
| yield from _run_streaming(checkout, repo_dir, env, f"git checkout --detach {commit}") |
|
|
|
|
| def _write_evidence( |
| artifact_dir: Path, |
| report: dict[str, object], |
| log: str, |
| before: dict[str, str], |
| after: dict[str, str], |
| ) -> Path: |
| artifact_dir.mkdir(parents=True, exist_ok=True) |
| (artifact_dir / "report.json").write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") |
| (artifact_dir / "validation.log").write_text(log, encoding="utf-8") |
| (artifact_dir / "source-before.sha256.json").write_text(json.dumps(before, indent=2, sort_keys=True) + "\n", encoding="utf-8") |
| (artifact_dir / "source-after.sha256.json").write_text(json.dumps(after, indent=2, sort_keys=True) + "\n", encoding="utf-8") |
| archive = artifact_dir.with_suffix(".zip") |
| with zipfile.ZipFile(archive, "w", compression=zipfile.ZIP_DEFLATED) as bundle: |
| for path in sorted(artifact_dir.iterdir()): |
| bundle.write(path, arcname=path.name) |
| return archive |
|
|
|
|
| def validate(repository: str, commit: str) -> Iterator[tuple[str, str, str | None]]: |
| repository = repository.strip() |
| commit = commit.strip() |
| token = os.environ.get("GITHUB_TOKEN", "").strip() |
| if not token: |
| message = ( |
| "BLOCKED: Space secret GITHUB_TOKEN is missing. Add a fine-grained, read-only token " |
| "for toxzak-svg/starfire under Settings → Variables and secrets, then run again." |
| ) |
| yield message, "BLOCKED", None |
| return |
| if repository != DEFAULT_REPOSITORY: |
| yield "BLOCKED: repository must remain toxzak-svg/starfire for this frozen verifier.", "BLOCKED", None |
| return |
| if len(commit) != 40 or any(ch not in "0123456789abcdefABCDEF" for ch in commit): |
| yield "BLOCKED: commit must be a full 40-character hexadecimal Git SHA.", "BLOCKED", None |
| return |
|
|
| started = int(time.time()) |
| run_id = f"omega-g2-s0-{started}" |
| artifact_dir = ARTIFACT_ROOT / run_id |
| log = "" |
| status = "RUNNING" |
| archive: Path | None = None |
| before: dict[str, str] = {} |
| after: dict[str, str] = {} |
| command_results: list[dict[str, object]] = [] |
| terminal_probe_pass = False |
|
|
| with tempfile.TemporaryDirectory(prefix="starfire-hf-") as temp: |
| work = Path(temp) |
| env = os.environ.copy() |
| env["CARGO_HOME"] = str(work / ".cargo") |
| env["RUSTUP_HOME"] = str(work / ".rustup") |
| env["CARGO_TERM_COLOR"] = "never" |
| env["RUST_BACKTRACE"] = "1" |
| env["PATH"] = f"{env['CARGO_HOME']}/bin:{env.get('PATH', '')}" |
|
|
| try: |
| for chunk, code in _ensure_rust(env, work): |
| log = _append_log(log, chunk.rstrip("\n")) |
| yield log, status, None |
| if code not in (None, 0): |
| raise RuntimeError("stable Rust installation failed") |
|
|
| for chunk, code in _clone_private_repo(repository, commit, token, work, env): |
| log = _append_log(log, chunk.rstrip("\n")) |
| yield log, status, None |
| if code not in (None, 0): |
| raise RuntimeError("private repository checkout failed") |
|
|
| repo = work / "starfire" |
| resolved = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=repo, env=env, text=True).strip() |
| if resolved.lower() != commit.lower(): |
| raise RuntimeError(f"checkout mismatch: expected {commit}, got {resolved}") |
| before = _source_hashes(repo) |
| log = _append_log(log, f"Pinned source confirmed: {resolved}") |
| log = _append_log(log, "Recorded pre-execution scientific source hashes") |
| yield log, status, None |
|
|
| commands = [ |
| ( |
| ["cargo", "check", "-p", "star", "--all-targets", "--locked"], |
| "cargo check -p star --all-targets --locked", |
| ), |
| ( |
| ["cargo", "check", "-p", "star", "--all-targets", "--features", DEFAULT_FEATURE, "--locked"], |
| f"cargo check -p star --all-targets --features {DEFAULT_FEATURE} --locked", |
| ), |
| ( |
| ["cargo", "test", "-p", "star", "omega_g2_shadow", "--features", DEFAULT_FEATURE, "--locked"], |
| f"cargo test -p star omega_g2_shadow --features {DEFAULT_FEATURE} --locked", |
| ), |
| ( |
| ["cargo", "run", "-p", "star", "--example", "omega_g2_s0_real_trace_shadow", "--features", DEFAULT_FEATURE, "--locked"], |
| f"cargo run -p star --example omega_g2_s0_real_trace_shadow --features {DEFAULT_FEATURE} --locked", |
| ), |
| ] |
|
|
| for index, (command, display) in enumerate(commands): |
| output_for_command = "" |
| final_code: int | None = None |
| for chunk, code in _run_streaming(command, repo, env, display): |
| output_for_command += chunk |
| log = _append_log(log, chunk.rstrip("\n")) |
| yield log, status, None |
| if code is not None: |
| final_code = code |
| command_results.append({"command": display, "exit_code": final_code}) |
| if final_code != 0: |
| raise RuntimeError(f"validation command failed: {display}") |
| if index == 3: |
| terminal_probe_pass = '"terminal_classification": "PASS"' in output_for_command |
| if not terminal_probe_pass: |
| raise RuntimeError("probe completed but did not emit terminal PASS") |
|
|
| after = _source_hashes(repo) |
| if before != after: |
| raise RuntimeError("scientific source hashes changed during execution") |
| diff = subprocess.run( |
| ["git", "diff", "--exit-code", "--", *SCIENTIFIC_PATHS], |
| cwd=repo, |
| env=env, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT, |
| text=True, |
| ) |
| log = _append_log(log, diff.stdout.rstrip("\n")) if diff.stdout else log |
| if diff.returncode != 0: |
| raise RuntimeError("git source-immutability check failed") |
|
|
| status = "PASS" |
| log = _append_log(log, "terminal_classification: PASS") |
| except Exception as exc: |
| status = "FAIL" |
| log = _append_log(log, f"terminal_classification: FAIL") |
| log = _append_log(log, f"error: {type(exc).__name__}: {exc}") |
| finally: |
| env.pop("GITHUB_TOKEN", None) |
| report: dict[str, object] = { |
| "terminal_classification": status, |
| "repository": repository, |
| "requested_commit": commit, |
| "started_at_unix": started, |
| "finished_at_unix": int(time.time()), |
| "feature": DEFAULT_FEATURE, |
| "commands": command_results, |
| "probe_emitted_pass": terminal_probe_pass, |
| "source_immutable": bool(before) and before == after, |
| "scientific_paths": list(SCIENTIFIC_PATHS), |
| "authority": { |
| "response_influence": False, |
| "routing": False, |
| "persistence_mutation": False, |
| "belief_or_ontology_promotion": False, |
| "tool_selection": False, |
| "external_action": False, |
| "source_modification": False, |
| }, |
| } |
| archive = _write_evidence(artifact_dir, report, log, before, after) |
|
|
| yield log, status, str(archive) if archive else None |
|
|
|
|
| with gr.Blocks(title="Starfire ΩG2-S0 Verifier") as demo: |
| gr.Markdown( |
| """ |
| # Starfire ΩG2-S0 Exact-Source Verifier |
| |
| Runs the frozen, default-off shadow integration checks against one exact private Git commit. |
| The verifier **observes and reports only**. It has no live response, routing, persistence, ontology, tool, or action authority. |
| """ |
| ) |
| with gr.Row(): |
| repository = gr.Textbox(label="Repository", value=DEFAULT_REPOSITORY, interactive=False) |
| commit = gr.Textbox(label="Exact commit SHA", value=DEFAULT_COMMIT) |
| run = gr.Button("Run exact-source validation", variant="primary") |
| classification = gr.Textbox(label="Terminal classification", value="NOT RUN", interactive=False) |
| logs = gr.Textbox(label="Streaming validation log", lines=28, max_lines=40, interactive=False) |
| evidence = gr.File(label="Evidence bundle") |
| run.click(validate, inputs=[repository, commit], outputs=[logs, classification, evidence], concurrency_limit=1) |
|
|
| demo.queue(default_concurrency_limit=1, max_size=2).launch( |
| server_name="0.0.0.0", |
| server_port=7860, |
| show_error=True, |
| allowed_paths=[str(ARTIFACT_ROOT)], |
| ) |
|
|
