#!/usr/bin/env python3 """Infraveil self-hosted release approver. Run this on a machine YOU control. It is the kill-switch for your own infrastructure: nothing Infraveil ships will run on your servers unless a release is signed by a private key that lives here and never leaves. How the veto works: 1. You generate an approval keypair (--genkey) and register the PUBLIC key in your Infraveil dashboard. Re-deploy your agents once so they pin it. 2. From then on, every payload an agent is told to run must carry a signature made by YOUR private key. No signature -> the agent refuses to run it. 3. This daemon watches the control plane for releases awaiting your approval and signs them according to a policy that YOU set. The private key is used locally to sign a 64-char hash; only the resulting signature is ever sent. Trust properties (read the code -- it is short and stdlib + cryptography only): - The private key never leaves this machine. Only signatures leave. - The approval TOKEN this daemon uses can read pending hashes and submit signatures, but it CANNOT approve anything on its own: a valid approval requires a signature from the private key it does not have. A stolen token cannot push code to your fleet. - Every decision is written to a local append-only log you own (--log). Policies: manual (default) -- show each pending release and ask y/N before signing. allowlist -- auto-approve only hashes you pre-listed in --allowlist-file; hold everything else for manual review. Pre-bless known-good builds, auto-refuse surprises. auto -- approve everything the control plane requests. Convenient, but this hands the veto back; use only if you trust every push. Usage: # one-time: make your keypair, register the printed PUBLIC key in the dashboard python infraveil-approver.py --genkey # generate an approval token in your dashboard (Trust -> Release Approval), # then run the daemon: python infraveil-approver.py \ --base-url https://portal.infraveil.com \ --token-file ./approval-token.txt \ --key-file ./approval-private.key \ --policy allowlist --allowlist-file ./approved-hashes.txt \ --interval 30 # check once and exit (good for cron): python infraveil-approver.py ... --once # intentionally approve every release pending right now, then exit: python infraveil-approver.py ... --approve-all Requires: pip install cryptography """ import sys, os, re, json, time, argparse, getpass, urllib.request, urllib.error from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives import serialization as S _HEX64 = re.compile(r"^[0-9a-f]{64}$") APPROVER_USER_AGENT = "Infraveil-Approver/1.1 (+https://infraveil.com; machine-client)" _HEX_SEED = re.compile(r"^[0-9a-f]{64}$") # 32-byte Ed25519 seed APPROVAL_BATCH_SIZE = 250 APPROVAL_RETRY_ATTEMPTS = 5 def _read_secret_file(path, label): try: st = os.stat(path) if os.name != "nt" and st.st_mode & 0o077: sys.stderr.write(f"warning: {path} is readable by others; chmod 600 it.\n") except FileNotFoundError: sys.exit(f"error: {label} file not found: {path}") with open(path, encoding="utf-8") as f: return f.read().strip() def genkey(out_path): if os.path.exists(out_path): sys.exit(f"Refusing to overwrite existing {out_path} (move it aside first).") priv = Ed25519PrivateKey.generate() seed = priv.private_bytes(S.Encoding.Raw, S.PrivateFormat.Raw, S.NoEncryption()).hex() pub = priv.public_key().public_bytes(S.Encoding.Raw, S.PublicFormat.Raw).hex() fd = os.open(out_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w") as f: f.write(seed + "\n") print(f"PRIVATE KEY written to {out_path} (chmod 600). Keep it offline and safe.") print("There is no recovery if you lose it.\n") print("PUBLIC KEY -- register this in your Infraveil dashboard (Trust -> Release Approval):") print(" " + pub) return 0 def load_private_key(args): if args.key_file: raw = _read_secret_file(args.key_file, "private key").lower() else: raw = getpass.getpass("Approval private key (hex, no echo): ").strip().lower() if not _HEX_SEED.match(raw): sys.exit("Invalid private key: expected a 32-byte Ed25519 seed as 64 hex chars.") return Ed25519PrivateKey.from_private_bytes(bytes.fromhex(raw)) def load_token(args): if args.token_file: return _read_secret_file(args.token_file, "approval token") if args.token: return args.token.strip() env = os.environ.get("INFRAVEIL_APPROVAL_TOKEN", "").strip() if env: return env sys.exit("No approval token. Pass --token-file, --token, or set INFRAVEIL_APPROVAL_TOKEN.") def _api(args, token, method, path, body=None): url = args.base_url.rstrip("/") + path data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request(url, data=data, method=method) req.add_header("User-Agent", APPROVER_USER_AGENT) req.add_header("X-Infraveil-Approval-Token", token) req.add_header("Accept", "application/json") if data is not None: req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=20) as r: return r.status, json.loads(r.read().decode("utf-8") or "{}") except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read().decode("utf-8") or "{}") except Exception: return e.code, {} except Exception as e: return 0, {"error": str(e)} def _load_allowlist(path): if not path or not os.path.exists(path): return set() out = set() for line in open(path, encoding="utf-8"): h = line.strip().lower() if _HEX64.match(h): out.add(h) return out def _log(args, msg): line = time.strftime("%Y-%m-%dT%H:%M:%S") + " " + msg print(line) if args.log: try: with open(args.log, "a", encoding="utf-8") as f: f.write(line + "\n") except Exception: pass def _decide(args, h, agents, allowlist): if args.policy == "auto": return True if args.policy == "allowlist": return h in allowlist # manual who = ", ".join(a.get("name") or a.get("agent_id", "?") for a in agents) or "(no agents reported)" sys.stdout.write(f"\nRelease awaiting approval:\n hash: {h}\n agents: {who}\n Approve and let it run on your infrastructure? [y/N] ") sys.stdout.flush() try: return input().strip().lower() in ("y", "yes") except (EOFError, KeyboardInterrupt): return False def _submit_approval_batch(args, token, approvals): for attempt in range(APPROVAL_RETRY_ATTEMPTS): status, body = _api( args, token, "POST", "/client/api/release-approval/approve-batch", {"approvals": approvals}, ) if status != 429: return status, body retry_after = max(1, min(30, int(body.get("retry_after") or (2 ** attempt)))) if attempt + 1 < APPROVAL_RETRY_ATTEMPTS: _log(args, f"approval API is busy; retrying batch in {retry_after}s") time.sleep(retry_after) return status, body def _submit_approval_fallback(args, token, approvals): approved = [] for approval in approvals: status, body = 0, {} for attempt in range(APPROVAL_RETRY_ATTEMPTS): status, body = _api( args, token, "POST", "/client/api/release-approval/approve", approval, ) if status != 429: break retry_after = max(1, min(30, int(body.get("retry_after") or (2 ** attempt)))) if attempt + 1 < APPROVAL_RETRY_ATTEMPTS: time.sleep(retry_after) if status == 200: approved.append(approval["payload_hash"]) else: return approved, status, body return approved, 200, {"status": "ok"} def run(args): priv = load_private_key(args) pub = priv.public_key().public_bytes(S.Encoding.Raw, S.PublicFormat.Raw).hex() token = load_token(args) _log(args, f"approver started | policy={args.policy} | public_key={pub[:16]}... | base={args.base_url}") while True: allowlist = _load_allowlist(args.allowlist_file) status, body = _api(args, token, "GET", "/client/api/release-approval/pending") if status == 401 or status == 403: edge_error = " ".join(str(body.get(key) or "") for key in ("type", "title", "detail", "error")) if "error-1010" in edge_error.lower() or "error 1010" in edge_error.lower(): _log(args, "ERROR: Cloudflare blocked this approver before it reached Infraveil. Download the current approver and retry.") else: _log(args, "ERROR: approval token rejected or expired (401/403). Generate a fresh token in the dashboard and retry.") if args.once: return 1 time.sleep(args.interval); continue if status != 200: _log(args, f"WARN: could not fetch pending (HTTP {status}): {body.get('error','')}") if args.once: return 1 time.sleep(args.interval); continue registered = (body.get("registered_pubkey") or "").lower() if registered and registered != pub: _log(args, f"REFUSING: dashboard has a DIFFERENT approval key registered ({registered[:16]}...) than this private key ({pub[:16]}...). Not signing.") if args.once: return 1 time.sleep(args.interval); continue pending = [h.lower() for h in (body.get("pending") or []) if _HEX64.match(str(h).lower())] by_hash = {} for a in (body.get("agents") or []): h = str(a.get("payload_hash") or "").lower() if h: by_hash.setdefault(h, []).append(a) approvals_to_submit = [] if not pending: _log(args, "ok: nothing pending; all desired releases already approved.") for h in pending: agents = by_hash.get(h, []) if _decide(args, h, agents, allowlist): sig = priv.sign(h.encode("utf-8")).hex() approvals_to_submit.append({"payload_hash": h, "signature": sig}) else: _log(args, f"HELD {h[:16]}... (policy={args.policy}; not approved)") for offset in range(0, len(approvals_to_submit), APPROVAL_BATCH_SIZE): approval_batch = approvals_to_submit[offset:offset + APPROVAL_BATCH_SIZE] st, resp = _submit_approval_batch(args, token, approval_batch) if st in (404, 405): approved, st, resp = _submit_approval_fallback(args, token, approval_batch) if st == 200: _log(args, f"APPROVED {len(approved)} releases using the legacy approval API") continue if st == 200: approved_count = int(resp.get("approved_count") or len(approval_batch)) _log(args, f"APPROVED {approved_count} releases in one signed batch") else: _log(args, f"FAILED to submit signed batch of {len(approval_batch)} releases (HTTP {st}): {resp.get('error','')}") if args.once: return 1 if args.once: return 0 time.sleep(args.interval) def _here(name): return os.path.join(os.path.dirname(os.path.realpath(__file__)), name) def _enable_ansi(): if os.name == "nt": try: import ctypes h = ctypes.windll.kernel32 h.SetConsoleMode(h.GetStdHandle(-11), 7) except Exception: pass _C = {"d": "\x1b[0m", "b": "\x1b[1m", "dim": "\x1b[90m", "g": "\x1b[92m", "y": "\x1b[93m", "r": "\x1b[91m", "c": "\x1b[96m"} def _col(s, key): return _C.get(key, "") + s + _C["d"] def _cfg_path(): return _here("approver-config.json") def _load_cfg(): cfg = {"base_url": "https://portal.infraveil.com", "policy": "manual", "key_file": _here("approval-private.key"), "token_file": _here("approval-token.txt"), "allowlist_file": _here("approved-hashes.txt"), "interval": 30, "log": _here("approver-decisions.log")} try: with open(_cfg_path(), encoding="utf-8") as f: cfg.update(json.load(f)) except Exception: pass return cfg def _save_cfg(cfg): try: with open(_cfg_path(), "w", encoding="utf-8") as f: json.dump(cfg, f, indent=2) except Exception: pass def _ask(prompt, default=""): suffix = _col(f" [{default}]", "dim") if default else "" try: ans = input(prompt + suffix + ": ").strip() except EOFError: return default return ans or default def _args_from_cfg(cfg, once=False): ns = argparse.Namespace() ns.genkey = False ns.out = cfg["key_file"] ns.base_url = cfg["base_url"] ns.key_file = cfg["key_file"] ns.token_file = cfg["token_file"] ns.token = None ns.policy = cfg["policy"] ns.allowlist_file = cfg["allowlist_file"] ns.interval = int(cfg.get("interval", 30) or 30) ns.log = cfg["log"] ns.once = once return ns def _status_line(label, path, ok_extra=""): if path and os.path.isfile(path): return f" {label:<14}{_col('READY', 'g')} {_col(os.path.basename(path) + ok_extra, 'dim')}" return f" {label:<14}{_col('NOT SET', 'y')}" def _menu_explain(): print(_col("\nHow your release veto works\n", "b")) print(" 1. You generate an approval keypair here. The PRIVATE key never leaves") print(" this machine. You register only the PUBLIC key in your dashboard.") print(" 2. Re-deploy your agents once so they pin that public key.") print(" 3. From then on, every release an agent is told to run must carry a") print(" signature made by your private key. No signature -> the agent refuses.") print(" 4. This tool watches for releases awaiting approval and signs them with") print(" a policy " + _col("you", "b") + " choose:") print(" " + _col("manual", "c") + " - ask you y/N for each release (most control)") print(" " + _col("allowlist", "c") + " - auto-approve only hashes you pre-blessed; hold the rest") print(" " + _col("auto", "c") + " - approve everything (convenient; hands the veto back)") print("\n Even a fully compromised Infraveil cannot run code your key did not sign.") print(" Register your public key under " + _col("Trust -> Release Approval", "c") + " in the dashboard.\n") def interactive_menu(): _enable_ansi() cfg = _load_cfg() while True: print(_col("\n" + "=" * 60, "dim")) print(_col(" INFRAVEIL Release Approver", "b") + _col(" - your offline kill-switch", "dim")) print(_col("=" * 60, "dim")) print(_col(" Nothing Infraveil ships runs on your machines unless YOU sign", "dim")) print(_col(" it with a key that lives only here.\n", "dim")) print(_status_line("Approval key", cfg["key_file"])) print(_status_line("Token", cfg["token_file"])) print(f" {'Policy':<14}{_col(cfg['policy'], 'c')}") print(f" {'Dashboard':<14}{_col(cfg['base_url'], 'dim')}") print() print(" " + _col("1", "b") + ") Generate my approval key " + _col("(one time)", "dim")) print(" " + _col("2", "b") + ") Save my approval token " + _col("(paste from dashboard)", "dim")) print(" " + _col("3", "b") + ") Choose approval policy " + _col("(manual / allowlist / auto)", "dim")) print(" " + _col("4", "b") + ") Set dashboard URL") print(" " + _col("5", "b") + ") " + _col("Start watching for releases", "g") + " " + _col("<- run the approver", "dim")) print(" " + _col("6", "b") + ") Check once and exit " + _col("(good for cron)", "dim")) print(" " + _col("7", "b") + ") How this works") print(" " + _col("0", "b") + ") Quit " + _col("(or q, or Ctrl-C)", "dim")) try: choice = input("\n Choose [1-7, 0 to quit]: ").strip().lower() except (EOFError, KeyboardInterrupt): print(_col("\n Bye. Your key stays here; nothing was uploaded.", "dim")) return 0 if choice in ("0", "q", "quit", "exit", "n", "no"): print(_col(" Bye. Your key stays here; nothing was uploaded.", "dim")) return 0 elif choice == "1": path = _ask(" Where to save the private key", cfg["key_file"]) if os.path.exists(path): if _ask(" " + _col("That file exists. Overwrite? this revokes the old key", "y") + " (y/N)", "N").lower() not in ("y", "yes"): continue try: os.remove(path) except Exception as e: print(_col(f" Could not remove old key: {e}", "r")); continue genkey(path) cfg["key_file"] = path _save_cfg(cfg) print(_col("\n ^ Copy that PUBLIC KEY into the dashboard (Trust -> Release Approval),", "g")) print(_col(" then re-deploy your agents once so they pin it.", "g")) elif choice == "2": print(" Generate a token in the dashboard under " + _col("Trust -> Release Approval", "c") + ".") tok = _ask(" Paste your approval token") if not tok: print(_col(" Nothing pasted; not saved.", "y")); continue path = cfg["token_file"] try: with open(path, "w", encoding="utf-8") as f: f.write(tok + "\n") try: os.chmod(path, 0o600) except Exception: pass _save_cfg(cfg) print(_col(f" Saved to {os.path.basename(path)} (chmod 600).", "g")) except Exception as e: print(_col(f" Could not save token: {e}", "r")) elif choice == "3": pol = _ask(" Policy (manual / allowlist / auto)", cfg["policy"]).lower() if pol not in ("manual", "allowlist", "auto"): print(_col(" Unknown policy; keeping " + cfg["policy"], "y")); continue cfg["policy"] = pol if pol == "allowlist": cfg["allowlist_file"] = _ask(" File of pre-approved 64-hex hashes", cfg["allowlist_file"]) if not os.path.exists(cfg["allowlist_file"]): try: open(cfg["allowlist_file"], "a", encoding="utf-8").close() print(_col(f" Created {os.path.basename(cfg['allowlist_file'])} - add one hash per line.", "dim")) except Exception: pass _save_cfg(cfg) print(_col(f" Policy set to {pol}.", "g")) elif choice == "4": cfg["base_url"] = _ask(" Dashboard base URL", cfg["base_url"]) _save_cfg(cfg) elif choice in ("5", "6"): once = choice == "6" if not os.path.isfile(cfg["key_file"]): print(_col(" No approval key yet - run option 1 first.", "y")); continue if not os.path.isfile(cfg["token_file"]): print(_col(" No approval token yet - run option 2 first.", "y")); continue print(_col(f"\n Watching {cfg['base_url']} | policy={cfg['policy']} | " + ("one pass" if once else f"every {cfg['interval']}s") + ". Ctrl-C to stop.\n", "c")) try: run(_args_from_cfg(cfg, once=once)) except KeyboardInterrupt: print(_col("\n Stopped. Your veto stays in force - agents still refuse unsigned releases.", "dim")) if once: _ask("\n Press Enter to return to the menu") elif choice == "7": _menu_explain() _ask(" Press Enter to return to the menu") elif choice == "": continue else: print(_col(" Not a menu option. Pick 1-7, or 0/q to quit.", "y")) def main(): p = argparse.ArgumentParser(description="Infraveil self-hosted release approver", add_help=True) p.add_argument("--genkey", action="store_true", help="generate an approval keypair and exit") p.add_argument("--out", default="approval-private.key", help="path for --genkey private key file") p.add_argument("--base-url", default="https://portal.infraveil.com") p.add_argument("--key-file", help="path to your Ed25519 private key (64-hex seed); omit to be prompted") p.add_argument("--token-file", help="path to a file containing your approval token") p.add_argument("--token", help="approval token (prefer --token-file; avoids shell history)") p.add_argument("--policy", choices=["manual", "allowlist", "auto"], default="manual") p.add_argument("--allowlist-file", help="file of pre-approved 64-hex hashes (policy=allowlist)") p.add_argument("--interval", type=int, default=30, help="seconds between polls") p.add_argument("--log", help="append decisions to this local file (your own audit trail)") p.add_argument("--once", action="store_true", help="check once and exit (for cron)") p.add_argument("--approve-all", action="store_true", help="sign every release pending in this pass, then exit") p.add_argument("--menu", action="store_true", help="force the interactive menu") args = p.parse_args() if args.approve_all: args.policy = "auto" args.once = True if args.menu or (len(sys.argv) == 1 and sys.stdin.isatty()): try: return interactive_menu() except KeyboardInterrupt: print("\n Bye. Your key stays here; nothing was uploaded.") return 0 if args.genkey: return genkey(args.out) return run(args) if __name__ == "__main__": sys.exit(main())