#!/usr/bin/env python3 """ PowerSearch Grid: trust-first release verification helper. This script is intentionally *verification-only*. It does not install services or register nodes. What it can verify: - The signed release manifest served by `GET /api/grid/assets` (Ed25519). - Optional pinned public-key fingerprint (`sha256(pubkey_bytes)`). - Local files against manifest sha256 values. Why it exists: - Users should never feel forced to run an opaque installer. Download + verify is the default. - Teams can pin the Grid public key fingerprint out-of-band for meaningful signature verification. """ from __future__ import annotations import argparse import base64 import hashlib import json import os import sys from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen try: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey # type: ignore except Exception: # pragma: no cover Ed25519PublicKey = None def canonical_json(obj: Any) -> bytes: return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def http_json(url: str, *, timeout_s: float = 15.0) -> dict[str, Any]: req = Request(url, headers={"accept": "application/json"}) try: with urlopen(req, timeout=timeout_s) as resp: raw = resp.read() data = json.loads(raw.decode("utf-8")) return data if isinstance(data, dict) else {"ok": False, "error": "unexpected json"} except HTTPError as exc: try: body = exc.read().decode("utf-8", errors="ignore") except Exception: body = "" raise RuntimeError(f"http {exc.code}: {body[:300]}".strip()) except URLError as exc: raise RuntimeError(f"network error: {exc}") def sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: while True: chunk = f.read(1024 * 1024) if not chunk: break h.update(chunk) return h.hexdigest() def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--base-url", default=os.getenv("POWERSEARCH_GRID_URL", "https://aipowerprogressia.com")) ap.add_argument("--pin-fingerprint-sha256", default=os.getenv("GRID_PUBKEY_FPR_SHA256", "")) ap.add_argument("--timeout-s", type=float, default=15.0) ap.add_argument("--print-fingerprint", action="store_true", help="Print pubkey fingerprint and exit (no verification).") ap.add_argument("--no-signature", action="store_true", help="Skip signature verification (NOT recommended).") ap.add_argument( "--verify-dir", default="", help="Directory containing downloaded Grid assets to verify against the manifest (example: . or ~/.local/share/powersearch-grid-agent).", ) ap.add_argument("--require-all", action="store_true", help="Fail if any manifest asset is missing from --verify-dir.") args = ap.parse_args() base_url = str(args.base_url or "").strip().rstrip("/") if not base_url.startswith(("http://", "https://")): base_url = "https://" + base_url assets_url = base_url + "/api/grid/assets" d = http_json(assets_url, timeout_s=max(1.0, float(args.timeout_s or 15.0))) if d.get("ok") is not True: raise SystemExit(f"error: assets endpoint returned ok=false: {d.get('detail') or d.get('error') or d}") pub_b64 = str(d.get("public_key_b64") or "").strip() if not pub_b64: raise SystemExit("error: missing public_key_b64 in assets response") try: pub = base64.b64decode(pub_b64.encode("utf-8"), validate=True) except Exception: raise SystemExit("error: invalid base64 public_key_b64") fpr = hashlib.sha256(pub).hexdigest() if args.print_fingerprint: sys.stdout.write(fpr + "\n") return 0 expected = str(args.pin_fingerprint_sha256 or "").strip().lower() if expected and expected != fpr: raise SystemExit(f"error: fingerprint mismatch\nexpected: {expected}\nactual: {fpr}") manifest = d.get("release_manifest") if isinstance(d.get("release_manifest"), dict) else None sig_b64 = str(d.get("release_signature_b64") or "").strip() if not args.no_signature: if Ed25519PublicKey is None: raise SystemExit("error: cryptography missing; cannot verify signature (pip install cryptography)") if not manifest or not sig_b64: err = str(d.get("release_signature_error") or "").strip() raise SystemExit(f"error: signed manifest unavailable: {err or 'missing signature/manifest'}") try: sig = base64.b64decode(sig_b64.encode("utf-8"), validate=True) except Exception: raise SystemExit("error: invalid base64 release_signature_b64") try: Ed25519PublicKey.from_public_bytes(pub).verify(sig, canonical_json(manifest)) except Exception: raise SystemExit("error: signature verification failed") assets = manifest.get("assets") if isinstance(manifest, dict) else None if not isinstance(assets, list): if args.no_signature: raise SystemExit("error: release_manifest missing or invalid; cannot verify files") raise SystemExit("error: release_manifest missing or invalid") sha_map: dict[str, str] = {} for a in assets: if not isinstance(a, dict): continue name = str(a.get("name") or "").strip() sha = str(a.get("sha256") or "").strip().lower() if name and len(sha) == 64: sha_map[name] = sha ok_files = True if args.verify_dir: base = Path(str(args.verify_dir)).expanduser() missing: list[str] = [] mismatched: list[tuple[str, str, str]] = [] verified: list[str] = [] for name, want in sha_map.items(): p = base / name if not p.exists(): missing.append(name) continue have = sha256_file(p) if have != want: mismatched.append((name, want, have)) ok_files = False else: verified.append(name) if missing and args.require_all: ok_files = False if verified: sys.stdout.write("ok: files verified: " + ", ".join(sorted(verified)) + "\n") if missing: sys.stdout.write("note: missing files: " + ", ".join(sorted(missing)) + "\n") for name, want, have in mismatched: sys.stdout.write(f"error: checksum mismatch {name}\nexpected: {want}\nactual: {have}\n") if ok_files: sys.stdout.write(f"ok: pubkey_fingerprint_sha256: {fpr}\n") if not args.no_signature: sys.stdout.write("ok: release manifest signature verified\n") return 0 return 2 if __name__ == "__main__": raise SystemExit(main())