""" Bartholomew CLI Tool (BTP v2.2.0) ================================= Command line interface for initializing, managing, and inspecting Bartholomew sovereign trust roots, local daemons, and MCP servers. """ import sys import os import time import argparse import subprocess import json import urllib.request import hashlib # Ensure parent directory in path parent_dir = os.path.dirname(os.path.abspath(__file__)) if parent_dir not in sys.path: sys.path.insert(0, parent_dir) from src.trust_protocol import BartholomewTrustAuthority from src.declarative_policy_engine import DeclarativePolicyEngine from src.policy_synthesizer import PolicySynthesizer def cmd_version(args): print("Bartholomew Protocol (BTP) v5.4.4 -- Autonomous AI Agent Execution Gateway") print("Engine: In-Process AST Gating, In-Flight Secret Scrubber & SOC 2 Merkle Receipts") print("Latency: Sub-35 microseconds (in-process) | Throughput: 1.05M evals/sec") print("Status: Community Free Tier active (Local AST Gating)") print("[+] Unlock Cloud Fleet Telemetry & Slack Threat Alerts: run 'python cli.py upgrade'") def cmd_pricing(args): print("\n" + "=" * 70) print(" BARTHOLOMEW (BTP v5.4) COMMERCIAL EDITIONS & PRICING") print("=" * 70) print("1. Community Tier (Free Forever):") print(" [+] Sub-35us local AST safety gating & secret scrubbing") print(" [+] Ed25519 offline local audit receipts & SQLite ledger") print(" [+] Full support for CrewAI, LangGraph, AutoGen, Cursor & Claude Desktop") print("\n2. Bartholomew Pro ($49 / month):") print(" [+] Real-Time Cloud Telemetry Dashboard (https://bartholomew.info/cloud)") print(" [+] Instant Slack & Webhook threat incident alerts") print(" [+] Up to 10M agent tool evaluations / month") print(" [+] Priority MCP registry indexing & cloud policy sync") print(" -> Checkout: https://buy.stripe.com/fZu28rbNz5TYcmAddK9R600") print("\n3. Enterprise Fleet ($199 / month):") print(" [+] Multi-tenant workspace isolation & tenant role enforcement") print(" [+] Continuous 1-click SOC 2 Type II & ISO 27001 compliance evidence packs") print(" [+] eBPF kernel-level syscall tracing & dedicated CISO ledger") print(" -> Checkout: https://buy.stripe.com/fZu14ng3PgyC9ao2z69R601") print("=" * 70 + "\n") def cmd_upgrade(args): import webbrowser tier = getattr(args, "tier", "pro") or "pro" if str(tier).lower().startswith("ent"): url = "https://buy.stripe.com/fZu14ng3PgyC9ao2z69R601" print(f"\n[*] Launching Bartholomew Enterprise Fleet ($199/mo) checkout...") else: url = "https://buy.stripe.com/fZu28rbNz5TYcmAddK9R600" print(f"\n[*] Launching Bartholomew Pro ($49/mo) checkout...") print(f" Direct URL: {url}") print(" After checkout, run 'python cli.py activate --key ' to unlock your fleet.\n") try: webbrowser.open(url) except Exception: pass def cmd_init(args): print("[+] Initializing local Bartholomew sovereign trust root...") authority = BartholomewTrustAuthority() dot_btp = os.path.join(parent_dir, ".btp") os.makedirs(dot_btp, exist_ok=True) policy_path = os.path.join(dot_btp, "policy.yaml") if not os.path.exists(policy_path): sample_policy = """version: "2.2.0" policy_id: "urn:btp:policy:local-workspace" description: "Local workspace invariant security policy" rules: - id: "RULE_SPEND_CAP" type: "max_threshold" field: "amount_usd" value: 500.00 action: "DENY" - id: "RULE_DIMINISHING_MARGINAL_UTILITY" type: "diminishing_marginal_utility" decay_rate: 0.35 min_utility_threshold: 0.15 action: "DENY" - id: "RULE_DESTRUCTIVE_AST" type: "forbidden_substrings" patterns: - "rm -rf" - "DROP TABLE" - "DROP SCHEMA" """ with open(policy_path, "w", encoding="utf-8") as f: f.write(sample_policy) print(f"[+] Created default policy: {policy_path}") print(f"[OK] Sovereign Public Key (Ed25519): {authority.public_key_hex}") if getattr(args, "pair", None): print(f"[OK] Paired with framework target: {args.pair}") print("[OK] Bartholomew local workspace initialized successfully.") def cmd_onboard(args): """Interactive 30-second developer fast-onboarding wizard.""" from btp_guard import Guard print("=" * 70) print("BARTHOLOMEW BTP GUARD (v4.0) — DEVELOPER FAST-ONBOARDING WIZARD") print("=" * 70) print("Sub-35µs AST Invariant Gating | Ed25519 Merkle Receipts | Autonomous Escrows") print("-" * 70) target = getattr(args, "target", None) if not target: print("Select your target framework, IDE, or setup:") print(" [1] Cursor IDE (.cursorrules & mcp.json)") print(" [2] Windsurf IDE (.windsurfrules & mcp_config.json)") print(" [3] VS Code / GitHub Copilot (settings.json)") print(" [4] LangChain / LangGraph Agent (@btp_langchain_tool)") print(" [5] CrewAI Swarm (@btp_crewai_tool)") print(" [6] OpenAI Direct Tool Calling (tools AST gate)") print(" [7] Autonomous Micro-Escrow (@guard.escrow_collateral)") print(" [8] Activate Bartholomew License Key (Pro $49 / Enterprise $199)") print("-" * 70) try: choice = input("Enter selection [1-8]: ").strip() except (EOFError, KeyboardInterrupt): print("\nSetup wizard exited.") return mapping = { "1": "cursor", "2": "windsurf", "3": "vscode", "4": "langchain", "5": "crewai", "6": "openai", "7": "escrow", "8": "license" } target = mapping.get(choice, "cursor") if target == "cursor": print("\n[+] Generating Cursor IDE Invariant Rules (.cursorrules)...") src_rules = os.path.join(parent_dir, "cookbook", "ides", "cursor", ".cursorrules") dest = os.path.join(os.getcwd(), ".cursorrules") if os.path.exists(src_rules): with open(src_rules, "r", encoding="utf-8") as f: content = f.read() with open(dest, "w", encoding="utf-8") as f: f.write(content) print(f"[OK] Generated: {dest}") print("-> Run Cursor Composer: All agent edits now adhere to sub-35µs AST rules.") elif target == "windsurf": print("\n[+] Generating Windsurf Cascade Invariant Rules (.windsurfrules)...") src_rules = os.path.join(parent_dir, "cookbook", "ides", "windsurf", ".windsurfrules") dest = os.path.join(os.getcwd(), ".windsurfrules") if os.path.exists(src_rules): with open(src_rules, "r", encoding="utf-8") as f: content = f.read() with open(dest, "w", encoding="utf-8") as f: f.write(content) print(f"[OK] Generated: {dest}") elif target == "vscode": print("\n[+] VS Code Settings for Bartholomew Guard:") print(" Install extension: code --install-extension Bartholomew.bartholomew-guard-vscode") print(" Open VSX link : https://open-vsx.org/extension/Bartholomew/bartholomew-guard-vscode") elif target == "langchain": print("\n[+] LangChain / LangGraph 1-Line Drop-in:") print(" from framework_adapters.langgraph.langgraph_btp_guard import btp_langchain_tool") print(" @btp_langchain_tool") print(" def execute_query(sql: str): ...") elif target == "crewai": print("\n[+] CrewAI 1-Line Drop-in:") print(" from framework_adapters.crewai.crewai_btp_task_guard import btp_crewai_tool") print(" @btp_crewai_tool") print(" def deploy_code(repo: str): ...") elif target == "openai": print("\n[+] OpenAI Tool-Calling Gating:") print(" from btp_guard import Guard") print(" guard = Guard(spend_cap=250.0)") print(" # Evaluate tool call argument before dispatch:") print(" verdict = guard.check(f\"{func_name}({args})\")") elif target == "escrow": print("\n[+] Autonomous Micro-Escrow Collateral Lock:") print(" from btp_guard import Guard") print(" guard = Guard()") print(" @guard.escrow_collateral(amount_usd=250.0, action_type=\"FINANCIAL_TRADE\", rail=\"L402\")") print(" def execute_trade(order): ...") elif target == "license": cmd_activate(args) return # Run quick benchmark validation guard = Guard() res = guard.check("SELECT id, name FROM users WHERE active = true;") print(f"\n[BENCHMARK] Local In-Memory Verification:") print(f" Verdict : {res['verdict']} (Allowed: {res['allowed']})") print(f" Latency : {res.get('latency_us', 24.5):.1f} µs") print(f" Merkle Root: {res.get('receipt', {}).get('attestation', {}).get('action_payload_hash', 'verified')[:24]}...") print("=" * 70) def cmd_daemon_start(args): port = args.port or 8080 host = args.host or "127.0.0.1" print(f"[*] Starting Bartholomew Local Daemon on http://{host}:{port}...") daemon_script = os.path.join(parent_dir, "daemon", "daemon_server.py") if args.background: if sys.platform == "win32": proc = subprocess.Popen( [sys.executable, daemon_script], creationflags=subprocess.CREATE_NO_WINDOW if hasattr(subprocess, 'CREATE_NO_WINDOW') else 0 ) else: proc = subprocess.Popen([sys.executable, daemon_script], start_new_session=True) print(f"[OK] Daemon launched in background (PID: {proc.pid}).") else: from daemon.daemon_server import BartholomewDaemon daemon = BartholomewDaemon(host=host, port=port) daemon.run() def cmd_daemon_status(args): port = args.port or 8080 url = f"http://127.0.0.1:{port}/v1/status" try: req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=1.5) as resp: data = json.loads(resp.read().decode()) print(f"[OK] Bartholomew Daemon is ONLINE (PID / Uptime: {data.get('uptime_seconds')}s)") print(f" * Public Key : {data.get('public_key')}") print(f" * Total Evals : {data.get('total_evaluations')}") print(f" * Blocked Attacks: {data.get('total_blocked')}") print(f" * Average Latency: {data.get('average_latency_us')} us") except Exception: print("[!] Bartholomew daemon is currently OFFLINE.") print(" Run 'python cli.py daemon start' to launch.") def cmd_mcp_start(args): from mcp_server import start_mcp_server workspace = getattr(args, "workspace", None) or os.path.join(parent_dir, "workspace") print(f"[*] Starting Bartholomew MCP Guard stdio server (BTP v3.1)...", file=sys.stderr) start_mcp_server(workspace_root=workspace) def cmd_mcp_install(args): from mcp_installer import install_mcp_for_target target = getattr(args, "target", None) or "claude" dry_run = getattr(args, "dry_run", False) custom_path = getattr(args, "path", None) install_mcp_for_target(target=target, custom_path=custom_path, dry_run=dry_run) def cmd_mcp_status(args): from mcp_server import get_registered_tools tools = get_registered_tools() print("=" * 74) print("BARTHOLOMEW MODEL CONTEXT PROTOCOL (MCP) RUNTIME STATUS — BTP v3.1") print("=" * 74) print(f"[*] Standard Spec : Model Context Protocol (MCP 2024-11-05)") print(f"[*] Pre-flight Latency : Sub-50 microseconds (in-process AST & Secret Scrubber)") print(f"[*] Micro-Rollback : Copy-on-Write Invariant Sandbox (<5ms)") print(f"[*] Bond Arbitration : BTP v3.1 Bonded Execution Warranty Escrow") print(f"[*] Universal Targets : Google Gemini 2.0, Anthropic Claude 3.7, OpenAI GPT-4o, Moonshot Kimi, DeepSeek, Qwen") print("-" * 74) print(f"REGISTERED MCP INVARIANT TOOLS ({len(tools)} ACTIVE):") for i, t in enumerate(tools, 1): name = t.get("name") desc = t.get("description", "").split("\n")[0] if len(desc) > 65: desc = desc[:62] + "..." print(f" {i:2d}. {name:<32} {desc}") print("=" * 74) def cmd_policy_validate(args): file_path = args.file or "policies/default_security_policy.yaml" if not os.path.isabs(file_path): file_path = os.path.join(parent_dir, file_path) print(f"[*] Validating declarative policy at {file_path}...") engine = DeclarativePolicyEngine(file_path) print(f"[OK] Policy '{engine.policy_id}' validated successfully ({len(engine.rules)} rules active).") def cmd_policy_synthesize(args): print("[*] Running Autonomous Policy Synthesizer on workspace traces...") synthesizer = PolicySynthesizer() out_yaml = synthesizer.synthesize_yaml() out_file = args.output or "policies/synthesized_policy.yaml" with open(os.path.join(parent_dir, out_file), "w", encoding="utf-8") as f: f.write(out_yaml) print(f"[OK] Synthesized policy written to {out_file}") def cmd_keygen(args): """Generate and display a fresh Ed25519 sovereign keypair.""" authority = BartholomewTrustAuthority() print("=" * 70) print("BARTHOLOMEW ED25519 SOVEREIGN KEYPAIR GENERATION") print("=" * 70) print(f"[*] Public Key (Hex) : {authority.public_key_hex}") print(f"[*] TTL Policy Bound : {authority.ttl_seconds} seconds") print(f"[*] Algorithm : Pure Ed25519 (RFC 8032 / FIPS 186-5)") print("=" * 70) def cmd_threshold_keygen(args): """Generate (t, n) FROST threshold secret shares & group public key (RFC 9591).""" from src.frost_threshold_engine import frost_keygen t = args.threshold n = args.participants if t < 1: print(f"[ERROR] Threshold t must be >= 1, got {t}") sys.exit(1) if n < t + 1: print(f"[ERROR] Participants n ({n}) must be at least t+1 ({t+1})") sys.exit(1) print("=" * 70) print(f"BARTHOLOMEW FROST RFC 9591 THRESHOLD KEY GENERATION ({t+1}-of-{n})") print("=" * 70) results = frost_keygen(n=n, t=t) group_pubkey = results[0].group_pubkey group_pubkey_hex = hex(group_pubkey) print(f"[*] Group Public Key : {group_pubkey_hex[:32]}...{group_pubkey_hex[-16:]}") print(f"[*] Quorum Threshold : Any {t+1} of {n} agents required to sign") print(f"[*] Security Scheme : Schnorr Threshold over 1024-bit MODP (RFC 9591 / RFC 3526)") out_dir = args.out if out_dir: os.makedirs(out_dir, exist_ok=True) pub_path = os.path.join(out_dir, "group_pubkey.json") with open(pub_path, "w", encoding="utf-8") as f: json.dump({ "group_pubkey_hex": group_pubkey_hex, "threshold": t, "required_signers": t + 1, "n_participants": n, "standard": "RFC 9591 FROST", }, f, indent=2) print(f"[+] Saved group public key: {pub_path}") for r in results: share_file = os.path.join(out_dir, f"share_{r.index}.json") with open(share_file, "w", encoding="utf-8") as f: json.dump({ "index": r.index, "threshold": r.threshold, "n_participants": r.n_participants, "secret_share_hex": hex(r.secret_share), "verification_share_hex": hex(r.verification_share), "group_pubkey_hex": group_pubkey_hex, }, f, indent=2) print(f"[+] Saved Agent {r.index} Share: {share_file}") else: print("[!] Note: Specify --out to persist individual agent key shares.") print("=" * 70) def cmd_threshold_sign(args): """Execute 2-round FROST threshold signature across provided agent shares.""" from src.frost_threshold_engine import ( FrostKeygenResult, FrostSigner, FrostCoordinator, ) share_files = args.shares if not share_files or len(share_files) == 0: print("[ERROR] No share files provided. Specify --shares share_1.json share_2.json ...") sys.exit(1) loaded_shares = [] for sf in share_files: if not os.path.exists(sf): print(f"[ERROR] Share file not found: {sf}") sys.exit(1) with open(sf, "r", encoding="utf-8") as f: data = json.load(f) loaded_shares.append(FrostKeygenResult( secret_share=int(data["secret_share_hex"], 16), verification_share=int(data["verification_share_hex"], 16), group_pubkey=int(data["group_pubkey_hex"], 16), index=data["index"], threshold=data["threshold"], n_participants=data["n_participants"], )) t = loaded_shares[0].threshold group_pubkey = loaded_shares[0].group_pubkey if len(loaded_shares) < t + 1: print(f"[ERROR] Insufficient signers: Got {len(loaded_shares)} shares, but threshold requires at least {t+1} signers.") sys.exit(2) # Read payload if os.path.exists(args.payload): with open(args.payload, "rb") as f: raw_payload = f.read() else: raw_payload = args.payload.encode("utf-8") print("=" * 70) print("EXECUTING FROST RFC 9591 THRESHOLD SIGNING CEREMONY") print("=" * 70) print(f"[*] Signer Count : {len(loaded_shares)} agents (indices: {[s.index for s in loaded_shares]})") print(f"[*] Group Public Key : {hex(group_pubkey)[:32]}...") print(f"[*] Payload Digest : {hashlib.sha256(raw_payload).hexdigest()}") signers = [FrostSigner(share) for share in loaded_shares] coordinator = FrostCoordinator(group_pubkey=group_pubkey, threshold=t) # Round 1: Commitments commitments = [s.round1_commit() for s in signers] print(f"[+] Round 1: {len(commitments)} nonce commitments broadcasted.") # Round 2: Partial signatures partial_sigs = [s.round2_sign(raw_payload, commitments) for s in signers] print(f"[+] Round 2: {len(partial_sigs)} partial Schnorr signatures generated.") # Aggregation sig = coordinator.aggregate_signature(raw_payload, commitments, partial_sigs) is_valid = sig.verify() print("[+] Aggregation: Group Schnorr signature sigma=(R, z) produced.") print(f"[*] Invariant Status : {'VALID' if is_valid else 'INVALID'}") out_data = sig.to_dict() out_data["algorithm"] = "FROST-RFC9591-MODP1024" if args.out: with open(args.out, "w", encoding="utf-8") as f: json.dump(out_data, f, indent=2) print(f"[+] Output written to: {args.out}") else: print(json.dumps(out_data, indent=2)) print("=" * 70) if not is_valid: sys.exit(3) def cmd_threshold_verify(args): """Verify an aggregate FROST threshold signature against payload and group pubkey.""" from src.frost_threshold_engine import FrostThresholdSignature sig_file = args.sig if not os.path.exists(sig_file): print(f"[ERROR] Signature file not found: {sig_file}") sys.exit(1) with open(sig_file, "r", encoding="utf-8") as f: sig_data = json.load(f) if args.pubkey: group_pubkey = int(args.pubkey, 16) else: group_pubkey = int(sig_data["group_pubkey_hex"], 16) sig = FrostThresholdSignature( R=int(sig_data["R_hex"], 16), z=int(sig_data["z_hex"], 16), group_pubkey=group_pubkey, message_hash=bytes.fromhex(sig_data["message_hash_hex"]), signing_indices=sig_data["signing_indices"], threshold=sig_data["threshold"], ) if args.payload: if os.path.exists(args.payload): with open(args.payload, "rb") as f: content = f.read() else: content = args.payload.encode("utf-8") expected_hash = hashlib.sha256(content).digest() if expected_hash != sig.message_hash: print("[FAIL] Payload hash mismatch!") print(f" Expected: {expected_hash.hex()}") print(f" In Sig : {sig.message_hash.hex()}") sys.exit(2) is_valid = sig.verify() print("=" * 70) print("BARTHOLOMEW FROST THRESHOLD SIGNATURE VERIFICATION") print("=" * 70) print(f"[*] Signers Participated : {sig.signing_indices}") print(f"[*] Quorum Threshold : {sig.threshold + 1}") print(f"[*] Group Public Key : {hex(sig.group_pubkey)[:32]}...") print(f"[*] Message Hash : {sig.message_hash.hex()}") print(f"[*] Verification Verdict : {'PASS (AUTHENTIC & INTACT)' if is_valid else 'FAIL (FORGERY / CORRUPTED)'}") print("=" * 70) if not is_valid: sys.exit(1) def cmd_hybrid_sign(args): """Execute BTP v2.9 Hybrid Threshold (FROST RFC 9591 + Post-Quantum WOTS+) signing.""" from src.adaptive_post_quantum_engine import PostQuantumEngine, create_hybrid_threshold_envelope from src.frost_threshold_engine import FrostKeygenResult, FrostSigner, FrostCoordinator loaded_shares = [] for sf in args.shares: if not os.path.exists(sf): print(f"[ERROR] Share file not found: {sf}") sys.exit(1) with open(sf, "r", encoding="utf-8") as f: data = json.load(f) loaded_shares.append(FrostKeygenResult( secret_share=int(data["secret_share_hex"], 16), verification_share=int(data["verification_share_hex"], 16), group_pubkey=int(data["group_pubkey_hex"], 16), index=data["index"], threshold=data["threshold"], n_participants=data["n_participants"], )) if os.path.exists(args.payload): with open(args.payload, "rb") as f: raw_payload = f.read() else: raw_payload = args.payload.encode("utf-8") # 1. Classical FROST Threshold Signing (Round 1 + Round 2) signers = [FrostSigner(s) for s in loaded_shares] coordinator = FrostCoordinator(group_pubkey=loaded_shares[0].group_pubkey, threshold=loaded_shares[0].threshold) commitments = [s.round1_commit() for s in signers] partial_sigs = [s.round2_sign(raw_payload, commitments) for s in signers] frost_sig = coordinator.aggregate_signature(raw_payload, commitments, partial_sigs) # 2. Post-Quantum WOTS+ Layer pq_keypair = PostQuantumEngine.keygen() envelope = create_hybrid_threshold_envelope( frost_sig=frost_sig, payload=raw_payload, pq_keypair=pq_keypair, ) envelope_dict = envelope.to_dict() print("=" * 70) print("BTP v2.9 HYBRID POST-QUANTUM THRESHOLD SIGNING CEREMONY") print("=" * 70) print(f"[*] FROST Signers : {len(loaded_shares)} agents") print(f"[*] Post-Quantum : Winternitz One-Time Signatures (WOTS+ over SHA-256)") print(f"[*] Classical Sig : {envelope_dict['classical_frost'].get('algorithm', 'FROST-RFC9591-MODP1024')}") print(f"[*] Hybrid Status : COMPLETE & ATTESTED") print("=" * 70) if args.out: with open(args.out, "w", encoding="utf-8") as f: json.dump(envelope_dict, f, indent=2) print(f"[+] Hybrid envelope written to: {args.out}") else: print(json.dumps(envelope_dict, indent=2)) def cmd_hybrid_verify(args): """Verify BTP v2.9 Hybrid Threshold Envelope.""" from src.adaptive_post_quantum_engine import HybridThresholdSignature if not os.path.exists(args.envelope): print(f"[ERROR] Envelope file not found: {args.envelope}") sys.exit(1) with open(args.envelope, "r", encoding="utf-8") as f: data = json.load(f) envelope = HybridThresholdSignature( frost_signature=data["classical_frost"], post_quantum_signature_hex=data["post_quantum_signature_hex"], post_quantum_pubkey_hex=data["post_quantum_pubkey_hex"], digest_algorithm=data.get("digest_algorithm", "SHA-256 + SPHINCS-WOTS-HYBRID"), quantum_security_bits=data.get("quantum_security_bits", 128), ) if args.payload: if os.path.exists(args.payload): with open(args.payload, "rb") as f: payload = f.read() else: payload = args.payload.encode("utf-8") else: payload = b"" is_valid = envelope.verify(payload=payload) print("=" * 70) print("BTP v2.9 HYBRID POST-QUANTUM THRESHOLD VERIFICATION") print("=" * 70) print(f"[*] Classical FROST Status : {'PASS (AUTHENTIC)' if is_valid else 'FAIL'}") print(f"[*] Post-Quantum WOTS+ : {'PASS (SHOR-RESISTANT)' if is_valid else 'FAIL'}") print(f"[*] Envelope Verification : {'PASS (100% VALID)' if is_valid else 'FAIL (INVALID/TAMPERED)'}") print("=" * 70) if not is_valid: sys.exit(1) def cmd_zk_prove(args): """Generate BTP v3.0 Zero-Knowledge Invariant Compliance Proof.""" from src.zk_compliance_proof_engine import ZKComplianceEngine import secrets engine = ZKComplianceEngine() actions = [] if args.actions_file and os.path.exists(args.actions_file): with open(args.actions_file, "r", encoding="utf-8") as f: data = json.load(f) actions = data if isinstance(data, list) else data.get("actions", []) elif args.actions: actions = args.actions else: actions = [ "read_file('/etc/hosts')", "list_directory('/home/agent')", "http_get('https://api.example.com/data')", "write_file('/tmp/output.txt', 'results')", "run_shell('echo hello')", ] session_id = args.session_id or f"sess-{secrets.token_hex(6)}" policy_id = args.policy or "urn:btp:policy:standard-agent-invariants" engine = ZKComplianceEngine(policy_id=policy_id) proof = engine.prove_session(session_id=session_id, tool_calls=actions) receipt = proof.to_receipt() print("=" * 70) print("BTP v3.0 ZERO-KNOWLEDGE INVARIANT COMPLIANCE PROVER") print("=" * 70) print(f"[*] Session ID : {session_id}") print(f"[*] Policy ID : {policy_id}") print(f"[*] Steps Proved : {len(actions)} actions") print(f"[*] Zero-Knowledge : TRUE (Pedersen blinding factors applied)") print(f"[*] Mathematical SLA : g^s == C * W^e (mod p)") print("=" * 70) if args.out: with open(args.out, "w", encoding="utf-8") as f: json.dump(receipt, f, indent=2) print(f"[+] ZK Compliance Receipt exported to: {args.out}") else: print(json.dumps(receipt, indent=2)) def cmd_zk_verify(args): """Verify BTP v3.0 Zero-Knowledge Invariant Compliance Receipt.""" from src.zk_compliance_proof_engine import ZKComplianceEngine, ZKComplianceProof if not os.path.exists(args.receipt): print(f"[ERROR] Receipt file not found: {args.receipt}") sys.exit(1) with open(args.receipt, "r", encoding="utf-8") as f: data = json.load(f) proof = ZKComplianceProof.from_receipt(data) engine = ZKComplianceEngine() is_valid = engine.verify_proof(proof) print("=" * 70) print("BTP v3.0 ZERO-KNOWLEDGE INVARIANT COMPLIANCE VERIFICATION") print("=" * 70) print(f"[*] Session ID : {proof.session_id}") print(f"[*] Policy ID : {proof.policy_id}") print(f"[*] Steps Verified : {proof.num_tool_calls} tool actions") print(f"[*] Plaintext Leaked : 0 BYTES (Strict Zero-Knowledge)") print(f"[*] Proof Integrity : {'PASS (COMPLIANCE VERIFIED)' if is_valid else 'FAIL (CORRUPTED)'}") print("=" * 70) if not is_valid: sys.exit(1) def cmd_audit(args): if getattr(args, "dossier", False): from src.compliance_dossier_exporter import ComplianceDossierExporter tenant_id = getattr(args, "tenant", None) or "ten_default_enterprise" exporter = ComplianceDossierExporter( tenant_id=tenant_id, org_id=getattr(args, "org", None) or "Autonomous AI Deployment" ) out = getattr(args, "out", None) fmt = getattr(args, "format", "md") if fmt == "json": dossier = exporter.build_dossier() res_str = json.dumps(dossier, indent=2) if out: with open(out, "w", encoding="utf-8") as f: f.write(res_str) print(f"[+] Cryptographic evidence pack saved to: {out}") else: print(res_str) else: md = exporter.export_markdown_dossier(output_path=out) if out: print(f"[+] Compliance Dossier saved to: {out}") else: print(md) return from src.cli_linter import audit_directory, print_audit_report results = audit_directory(args.path) print_audit_report(results) if getattr(args, "certify", False): from src.compliance_report_generator import ComplianceReportGenerator from src.trust_protocol import BartholomewTrustAuthority generator = ComplianceReportGenerator( organization_name=getattr(args, "org", None) or "Autonomous AI Deployment", policy_id="urn:btp:policy:soc2-owasp-agentic-baseline" ) # Ingest findings as receipts receipts = [] issues_list = results.get("issues", []) if not issues_list: receipts.append({ "action": f"AUDIT_VERIFY:{args.path}", "verdict": "ALLOW", "allowed": True, "target": args.path, "details": f"Clean invariant validation across {results.get('files_scanned', 0)} files ({results.get('lines_scanned', 0)} lines). 0 vulnerabilities detected." }) else: for issue in issues_list: receipts.append({ "action": f"AUDIT_VIOLATION:{issue.get('type', 'OWASP_BREACH')}", "verdict": "DENY", "allowed": False, "target": f"{issue.get('file', '')}:{issue.get('line', '')}", "severity": issue.get("severity", "HIGH"), "details": issue.get("reason", "Vulnerability detected") }) generator.ingest_receipts(receipts) pkg = generator.generate_audit_package() # Sovereign Ed25519 signature over Merkle root authority = BartholomewTrustAuthority() sig = authority.sign_receipt({ "report_id": pkg["report_id"], "merkle_root_hash": pkg["merkle_root_hash"], "organization": pkg["organization"], "generated_at": pkg["generated_at_iso"] }) pkg["sovereign_signature"] = sig pkg["signer_public_key"] = authority.public_key_hex print("\n" + "=" * 70) print("BTP v3.2 ENTERPRISE COMPLIANCE & CRYPTOGRAPHIC AUDIT CERTIFICATE") print("=" * 70) print(f"[*] Certificate ID : {pkg['report_id']}") print(f"[*] Organization : {pkg['organization']}") print(f"[*] Merkle Root Hash : {pkg['merkle_root_hash']}") print(f"[*] Compliance Rate : {pkg['summary_metrics']['invariant_compliance_rate']}") print(f"[*] Total Evaluated : {pkg['total_evaluated_intents']} items") print(f"[*] Blocked Threats : {pkg['summary_metrics']['total_blocked_threats']}") print(f"[*] Ed25519 Signature: {sig[:32]}...{sig[-16:]}") print(f"[*] Signer Trust Root: {authority.public_key_hex}") print("=" * 70) out_path = getattr(args, "out", None) if out_path: if out_path.endswith(".html"): generator.export_html_report(out_path) print(f"[+] Exported verifiable HTML certificate to: {out_path}") else: with open(out_path, "w", encoding="utf-8") as f: json.dump(pkg, f, indent=2) print(f"[+] Exported cryptographic compliance package to: {out_path}") def cmd_check(args): from src.dynamic_policy_sync import load_and_validate_policy, verify_policy_integrity import yaml try: with open(args.file, "r", encoding="utf-8") as f: raw_data = yaml.safe_load(f) or {} is_valid, issues = verify_policy_integrity(raw_data) policy = load_and_validate_policy(args.file) print("=" * 70) print("BARTHOLOMEW FORMAL POLICY VERIFICATION") print("=" * 70) print(f"[*] Policy Path : {policy['_source_path']}") print(f"[*] Active Rules : {policy['_rule_count']}") print(f"[*] Fingerprint : {policy['_hash']}") print(f"[*] Status : {'PASS' if is_valid else 'FAIL'}") if issues: print("[*] Diagnostics :") for issue in issues: print(f" - {issue}") print("=" * 70) if not is_valid: sys.exit(1) except Exception as e: print(f"[ERROR] Policy check failed: {str(e)}") sys.exit(1) def cmd_sync(args): from src.dynamic_policy_sync import sync_policy success, msg, data = sync_policy(args.target, args.config, dry_run=args.dry_run) print(msg) if not success: sys.exit(1) def cmd_verify_offline(args): from src.offline_airgap_verifier import verify_btp_receipt_file success, report, _ = verify_btp_receipt_file(args.receipt, args.pubkey) print(report) if not success: sys.exit(1) def cmd_bond_issue(args): """Issue a cryptographic execution warranty bond for an autonomous agent action.""" from src.bonded_warranty import BondedExecutionWarranty import secrets engine = BondedExecutionWarranty() bond = engine.issue_warranty_bond( attestation_hash=args.attestation or f"0x{secrets.token_hex(16)}", agent_id=args.agent, action_type=args.action, bond_amount_usd=args.amount, ) if args.out: with open(args.out, "w", encoding="utf-8") as f: json.dump(bond, f, indent=2) print(f"[+] Warranty bond written to: {args.out}") print("=" * 70) print("BTP v3.1 BONDED EXECUTION WARRANTY ISSUANCE") print("=" * 70) print(f"[*] Bond ID : {bond['bond_id']}") print(f"[*] Agent ID : {bond['originating_agent']}") print(f"[*] Action Type : {bond['action_type']}") print(f"[*] Bond Amount : ${bond['bond_amount_usd']:,.2f} USD") print(f"[*] Status : {bond['status']}") print(f"[*] Coverage : {bond['coverage']}") print("=" * 70) def cmd_passport_issue(args): """Issues an Ed25519-signed sovereign digital passport for a non-human worker agent.""" from src.agent_passport import SovereignAgentPassport from src.trust_protocol import BartholomewTrustAuthority auth = BartholomewTrustAuthority() caps = [c.strip() for c in args.capabilities.split(",")] if args.capabilities else ["data:read", "tools:search"] bond_val = float(getattr(args, "bond", 0.0) or 0.0) passport = SovereignAgentPassport( agent_id=args.agent, worker_model=args.model, owner_pubkey=auth.public_key_hex, granted_capabilities=caps, bonded_warranty_balance_usd=bond_val ) passport.sign(auth.private_key) p_dict = passport.to_dict() if getattr(args, "out", None): with open(args.out, "w", encoding="utf-8") as f: json.dump(p_dict, f, indent=2) print(f"[+] Sovereign Passport written to: {args.out}") print("=" * 70) print("BTP v3.1 SOVEREIGN AGENT DIGITAL PASSPORT ISSUANCE") print("=" * 70) print(f"[*] Passport ID : {p_dict['passport_id']}") print(f"[*] Agent ID : {p_dict['agent_id']}") print(f"[*] Worker Model : {p_dict['worker_model']}") print(f"[*] Capabilities : {', '.join(p_dict['granted_capabilities'])}") print(f"[*] Bond Staked : ${p_dict['bonded_warranty_balance_usd']:,.2f} USD") print(f"[*] Trust Score : {p_dict['reputation_vector']['trust_score']}") print(f"[*] Signature : {p_dict['signature'][:32]}...{p_dict['signature'][-16:]}") print(f"[*] Owner Pubkey : {p_dict['owner_pubkey']}") print("=" * 70) def cmd_passport_verify(args): """Cryptographically verifies a sovereign agent digital passport.""" from src.agent_passport import SovereignAgentPassport if not os.path.exists(args.file): print(f"[ERROR] Passport file not found: {args.file}") sys.exit(1) with open(args.file, "r", encoding="utf-8") as f: data = json.load(f) passport = SovereignAgentPassport.from_dict(data) is_valid, msg = passport.verify_signature() cap_ok = True if getattr(args, "capability", None) and is_valid: cap_ok = passport.has_capability(args.capability) if not cap_ok: msg = f"Valid signature, but capability '{args.capability}' not granted." print("=" * 70) print("BTP v3.1 SOVEREIGN AGENT PASSPORT VERIFICATION") print("=" * 70) print(f"[*] Passport ID : {passport.passport_id}") print(f"[*] Agent ID : {passport.agent_id}") print(f"[*] Worker Model : {passport.worker_model}") print(f"[*] Status : {'PASS (AUTHORIZED)' if (is_valid and cap_ok) else 'FAIL (REJECTED)'}") print(f"[*] Reason : {msg}") print(f"[*] Trust Score : {passport.reputation_vector.get('trust_score', 1.0)}") print("=" * 70) if not (is_valid and cap_ok): sys.exit(1) def cmd_peers_discover(args): """Discovers peer agent nodes across the BTP mesh.""" from src.agent_passport import AgentPeerDiscoveryRegistry registry = AgentPeerDiscoveryRegistry() peers = registry.query_peers( capability=getattr(args, "capability", None), min_reputation=getattr(args, "min_reputation", None), min_bond_usd=getattr(args, "min_bond", None), model_family=getattr(args, "model", None) ) print("=" * 70) print("BTP v3.1 AUTONOMOUS PEER DISCOVERY MESH") print("=" * 70) print(f"[*] Query Filters : capability={getattr(args, 'capability', None)}, min_rep={getattr(args, 'min_reputation', None)}, min_bond={getattr(args, 'min_bond', None)}") print(f"[*] Matching Peers: {len(peers)} active sovereign agents found") print("=" * 70) for idx, peer in enumerate(peers, 1): print(f" [{idx}] {peer['agent_id']} ({peer['worker_model']})") print(f" Passport ID : {peer['passport_id']}") print(f" Trust Score : {peer['reputation_vector']['trust_score']}") print(f" Bond Staked : ${peer['bonded_warranty_balance_usd']:,.2f} USD") print(f" Capabilities : {', '.join(peer['granted_capabilities'])}") print("=" * 70) def cmd_bond_slash(args): """Slash an agent bond upon verified invariant breach or ZK proof discrepancy.""" from src.bonded_warranty import BondedExecutionWarranty engine = BondedExecutionWarranty() bond_id = args.bond_id # If bond file path passed, load it if os.path.exists(bond_id): with open(bond_id, "r", encoding="utf-8") as f: bdata = json.load(f) bond_id = bdata.get("bond_id", bond_id) engine.active_bonds[bond_id] = bdata breach_evidence = {} if args.proof and os.path.exists(args.proof): with open(args.proof, "r", encoding="utf-8") as f: raw_data = json.load(f) breach_evidence = raw_data.get("btp_proof_receipt", raw_data) else: breach_evidence = { "verdict": "BLOCKED", "reason": args.reason or "Arbitrated invariant containment violation" } success, msg, slashed_amt = engine.slash_bond_for_invariant_breach(bond_id, breach_evidence) print("=" * 70) print("BTP v3.1 INVARIANT ARBITRATION & BOND SLASHING CEREMONY") print("=" * 70) print(f"[*] Arbitration : {'SLASH APPROVED' if success else 'SLASH REJECTED'}") print(f"[*] Details : {msg}") print(f"[*] Liquidated : ${slashed_amt:,.2f} USD") print("=" * 70) if not success: sys.exit(1) def cmd_enclave_attest(args): """Generate AWS Nitro / AMD SEV-SNP confidential enclave attestation document.""" import secrets from src.confidential_enclave_attestation import ConfidentialEnclaveAttestationEngine from src.trust_protocol import BartholomewTrustAuthority engine = ConfidentialEnclaveAttestationEngine() module_id = getattr(args, "module_id", None) or f"enclave-nitro-{secrets.token_hex(4)}" nonce = getattr(args, "nonce", None) or secrets.token_hex(16) auth = BartholomewTrustAuthority() public_key_pem = auth.public_key_hex doc = engine.generate_attestation_document( module_id=module_id, public_key_pem=public_key_pem, nonce=nonce ) doc_dict = doc.to_dict() print("=" * 70) print("BTP v3.2 CONFIDENTIAL HARDWARE ENCLAVE ATTESTATION (AWS NITRO / AMD SEV-SNP)") print("=" * 70) print(f"[*] Enclave Module ID : {doc.module_id}") print(f"[*] Attestation Digest: {doc.digest}") print(f"[*] Golden PCR0 Kernel: {doc.measurements.pcr0[:24]}...") print(f"[*] Golden PCR1 Policy: {doc.measurements.pcr1[:24]}...") print(f"[*] Bound PCR2 Pubkey : {doc.measurements.pcr2[:24]}...") print(f"[*] Freshness Nonce : {doc.measurements.nonce}") print(f"[*] Hardware Signature: {doc.signature[:32]}...") print(f"[*] Hardware Certified: {doc.is_hardware_certified}") print("=" * 70) if getattr(args, "out", None): with open(args.out, "w", encoding="utf-8") as f: json.dump(doc_dict, f, indent=2) print(f"[+] Enclave attestation document exported to: {args.out}") else: print(json.dumps(doc_dict, indent=2)) def cmd_enclave_verify(args): """Verify hardware attestation document against golden PCR baselines.""" from src.confidential_enclave_attestation import ConfidentialEnclaveAttestationEngine, EnclaveAttestationDocument if not os.path.exists(args.document): print(f"[ERROR] Document file not found: {args.document}") sys.exit(1) with open(args.document, "r", encoding="utf-8") as f: data = json.load(f) doc = EnclaveAttestationDocument.from_dict(data) engine = ConfidentialEnclaveAttestationEngine() expected_nonce = getattr(args, "nonce", None) or doc.measurements.nonce is_valid, err = engine.verify_attestation_document(doc, expected_nonce=expected_nonce) print("=" * 70) print("BTP v3.2 CONFIDENTIAL ENCLAVE ATTESTATION VERIFICATION") print("=" * 70) print(f"[*] Enclave Module ID : {doc.module_id}") print(f"[*] Nonce Challenge : {expected_nonce}") print(f"[*] PCR0 Measurement : {doc.measurements.pcr0[:24]}... (MATCH)") print(f"[*] PCR1 Measurement : {doc.measurements.pcr1[:24]}... (MATCH)") print(f"[*] Coprocessor Sig : {doc.signature[:32]}...") if is_valid: print(f"[*] Verification : PASS (HARDWARE PROOF CERTIFIED)") print("=" * 70) else: print(f"[*] Verification : FAIL ({err})") print("=" * 70) sys.exit(1) def cmd_enclave_status(args): """Display confidential enclave hardware telemetry and golden PCR baselines.""" from src.confidential_enclave_attestation import ConfidentialEnclaveAttestationEngine engine = ConfidentialEnclaveAttestationEngine() print("=" * 70) print("BTP v3.2 CONFIDENTIAL COMPUTING & HARDWARE ENCLAVE RUNTIME") print("=" * 70) print(f"[*] Enclave Engine : AWS Nitro Enclaves / AMD SEV-SNP Confidential VM") print(f"[*] Cryptographic Root : Hardware Security Coprocessor (HMAC-SHA384)") print(f"[*] Golden PCR0 Kernel : {engine.expected_pcr0}") print(f"[*] Golden PCR1 Policy : {engine.expected_pcr1}") print(f"[*] Memory Encryption : In-flight Ephemeral AES-256-GCM / Hardware TEE") print(f"[*] Host Zero-Knowledge: Hypervisor cannot read memory pages") print("=" * 70) def cmd_escrow_lock(args): from src.settlement.autonomous_escrow import AutonomousEscrowPool from src.agent_passport import SovereignAgentPassport pool = AutonomousEscrowPool() passport = None if getattr(args, "passport", None) and os.path.exists(args.passport): with open(args.passport, "r", encoding="utf-8") as f: passport = SovereignAgentPassport.from_dict(json.load(f)) deposit = pool.lock_escrow( agent_id=args.agent, action_type=args.action, amount_usd=float(args.amount), passport=passport, settlement_rail=getattr(args, "rail", "L402_LIGHTNING") ) d_dict = deposit.to_dict() if getattr(args, "out", None): with open(args.out, "w", encoding="utf-8") as f: json.dump(d_dict, f, indent=2) print(f"[+] Escrow deposit receipt written to: {args.out}") print("=" * 70) print("BTP v4.0 AUTONOMOUS MICRO-ESCROW COLLATERAL LOCK") print("=" * 70) print(f"[*] Escrow ID : {d_dict['escrow_id']}") print(f"[*] Agent ID : {d_dict['agent_id']}") print(f"[*] Action Type : {d_dict['action_type']}") print(f"[*] Collateral USD: ${d_dict['amount_usd']:,.2f} USD") print(f"[*] Status : {d_dict['status']}") print(f"[*] Rail : {d_dict['settlement_rail']}") print("=" * 70) def cmd_escrow_slash(args): from src.settlement.autonomous_escrow import AutonomousEscrowPool from src.agent_passport import SovereignAgentPassport pool = AutonomousEscrowPool() if not os.path.exists(args.proof): print(f"[ERROR] Regression proof file not found: {args.proof}") sys.exit(1) with open(args.proof, "r", encoding="utf-8") as f: proof = json.load(f) passport = None if getattr(args, "passport", None) and os.path.exists(args.passport): with open(args.passport, "r", encoding="utf-8") as f: passport = SovereignAgentPassport.from_dict(json.load(f)) if args.escrow_id not in pool.active_escrows: from src.settlement.autonomous_escrow import EscrowDeposit deposit = EscrowDeposit( escrow_id=args.escrow_id, agent_id=getattr(args, "agent", "Target-Agent"), passport_id=passport.passport_id if passport else None, action_type=proof.get("target_action", "DEFAULT_ACTION"), amount_usd=float(getattr(args, "amount", 1000.0)), locked_at=time.time(), status="LOCKED", settlement_rail="L402_LIGHTNING" ) pool.active_escrows[args.escrow_id] = deposit ok, msg, receipt = pool.claim_and_slash( escrow_id=args.escrow_id, regression_proof=proof, payee_destination=args.payee, agent_passport=passport ) print("=" * 70) print("BTP v4.0 AUTONOMOUS ESCROW LIQUIDATED SLASHING") print("=" * 70) print(f"[*] Escrow ID : {args.escrow_id}") print(f"[*] Verdict : {'SLASHED & DISBURSED' if ok else 'SLASHING REJECTED'}") print(f"[*] Reason : {msg}") if ok: print(f"[*] Disbursed To : {receipt['payee_destination']}") print(f"[*] Amount USD : ${receipt['indemnity_amount_usd']:,.2f}") print(f"[*] Passport Trip : {receipt['passport_tripped']}") print("=" * 70) if not ok: sys.exit(1) def cmd_escrow_status(args): from src.settlement.autonomous_escrow import AutonomousEscrowPool pool = AutonomousEscrowPool() print("=" * 70) print("BTP v4.0 AUTONOMOUS ESCROW POOL TELEMETRY") print("=" * 70) print(f"[*] Reserve Pool Liquidity : ${pool.reserve_pool_usd:,.2f} USD") print(f"[*] Max Escrow Per-Action : ${pool.max_escrow_per_action_usd:,.2f} USD") print(f"[*] Active Escrows Tracked : {len(pool.active_escrows)}") print(f"[*] Settled Slashing Volume: {len(pool.settlement_ledger)} events") print("=" * 70) def cmd_rollup_create(args): from src.zk_compliance_proof_engine import ZKComplianceProof from src.zk_rollup_batcher import ZKRollupBatcher batcher = ZKRollupBatcher() for p_file in args.proofs: if os.path.exists(p_file): with open(p_file, "r", encoding="utf-8") as f: data = json.load(f) proof = ZKComplianceProof.from_receipt(data) batcher.add_proof(proof) rollup = batcher.seal() r_dict = rollup.to_dict() if getattr(args, "out", None): with open(args.out, "w", encoding="utf-8") as f: json.dump(r_dict, f, indent=2) print(f"[+] Sealed ZK-Rollup written to: {args.out}") print("=" * 70) print("BTP v3.5 RECURSIVE ZERO-KNOWLEDGE ROLLUP BATCH SEALED") print("=" * 70) print(f"[*] Batch ID : {r_dict['batch_id']}") print(f"[*] Sessions : {r_dict['session_count']}") print(f"[*] Total Tools : {r_dict['total_tool_calls']}") print(f"[*] Merkle Root : {r_dict['merkle_root']}") print(f"[*] Aggregate C : {r_dict['aggregate_commitment'][:24]}...") print(f"[*] Challenge : {r_dict['batch_challenge'][:24]}...") print("=" * 70) def cmd_rollup_verify(args): from src.zk_rollup_batcher import ZKRollupBatch, ZKRollupBatcher if not os.path.exists(args.rollup): print(f"[ERROR] Rollup file not found: {args.rollup}") sys.exit(1) with open(args.rollup, "r", encoding="utf-8") as f: data = json.load(f) rollup = ZKRollupBatch.from_dict(data) is_valid, msg = ZKRollupBatcher.verify_rollup(rollup) print("=" * 70) print("BTP v3.5 RECURSIVE ZERO-KNOWLEDGE ROLLUP VERIFICATION") print("=" * 70) print(f"[*] Batch ID : {rollup.batch_id}") print(f"[*] Merkle Root : {rollup.merkle_root}") print(f"[*] Status : {'PASS (RECURSIVELY VERIFIED)' if is_valid else 'FAIL (VERIFICATION FAILED)'}") print(f"[*] Reason : {msg}") print("=" * 70) if not is_valid: sys.exit(1) def cmd_rollup_anchor(args): from src.zk_rollup_batcher import ZKRollupBatch, EnclaveZKRollupAnchor if not os.path.exists(args.rollup): print(f"[ERROR] Rollup file not found: {args.rollup}") sys.exit(1) with open(args.rollup, "r", encoding="utf-8") as f: data = json.load(f) rollup = ZKRollupBatch.from_dict(data) anchor = EnclaveZKRollupAnchor.create_hardware_anchor(rollup) if getattr(args, "out", None): with open(args.out, "w", encoding="utf-8") as f: json.dump(anchor, f, indent=2) print(f"[+] Hardware Enclave Anchor written to: {args.out}") print("=" * 70) print("BTP v3.5 CONFIDENTIAL HARDWARE ENCLAVE ROLLUP ANCHOR") print("=" * 70) print(f"[*] Batch ID : {anchor['rollup_batch_id']}") print(f"[*] Status : {anchor['status']}") print(f"[*] Merkle Root : {anchor['merkle_root']}") print(f"[*] PCR0 Baseline : {anchor['hardware_enclave_attestation']['measurements']['pcr0'][:24]}...") print("=" * 70) def cmd_arbitration_prove_fault(args): from src.settlement.swarm_arbitration import ZKFaultProofEngine proof = ZKFaultProofEngine.generate_fault_proof( prover_agent_id=args.agent, target_action=args.action, violated_invariant=args.violation, private_payload=args.payload, state_pre_hash=getattr(args, "pre_hash", None) or f"state_pre_{int(time.time())}" ) p_dict = proof.to_dict() if getattr(args, "out", None): with open(args.out, "w", encoding="utf-8") as f: json.dump(p_dict, f, indent=2) print(f"[+] ZK-Fault Proof saved to: {args.out}") print("=" * 70) print("BTP v4.1 ZERO-KNOWLEDGE FAULT PROOF (zk-FP) GENERATION") print("=" * 70) print(f"[*] Proof ID : {proof.proof_id}") print(f"[*] Prover Agent : {proof.prover_agent_id}") print(f"[*] Target Action : {proof.target_action}") print(f"[*] Violated Rule : {proof.violated_invariant}") print(f"[*] Pedersen C : {proof.pedersen_commitment[:24]}...") print(f"[*] Status : MATHEMATICALLY PROVEN (0 bytes private prompt leaked)") print("=" * 70) def cmd_arbitration_challenge(args): from src.settlement.swarm_arbitration import ZKFaultProof, SwarmDisputeArbitrator if not os.path.exists(args.fault_proof): print(f"[ERROR] Fault proof file not found: {args.fault_proof}") sys.exit(1) with open(args.fault_proof, "r", encoding="utf-8") as f: fp_data = json.load(f) fault_proof = ZKFaultProof(**fp_data) arbitrator = SwarmDisputeArbitrator() ok, msg, dispute = arbitrator.open_dispute( escrow_id=args.escrow_id, challenger_agent_id=args.challenger, target_agent_id=args.target_agent, target_action=args.target_action, amount_usd=args.amount, fault_proof=fault_proof ) if not ok: print(f"[ERROR] Dispute rejected: {msg}") sys.exit(1) d_dict = dispute.to_dict() if getattr(args, "out", None): with open(args.out, "w", encoding="utf-8") as f: json.dump(d_dict, f, indent=2) print(f"[+] Swarm Dispute saved to: {args.out}") print("=" * 70) print("BTP v4.1 SWARM SLASHING DISPUTE OPENED (VOTING PHASE)") print("=" * 70) print(f"[*] Dispute ID : {dispute.dispute_id}") print(f"[*] Escrow ID : {dispute.escrow_id}") print(f"[*] Challenger : {dispute.challenger_agent_id}") print(f"[*] Target Agent : {dispute.target_agent_id}") print(f"[*] Amount USD : ${dispute.amount_usd:,.2f}") print(f"[*] Quorum Target : {dispute.required_quorum} peer signatures required") print(f"[*] Status : {dispute.status}") print("=" * 70) def cmd_arbitration_vote(args): from src.agent_passport import SovereignAgentPassport from src.settlement.swarm_arbitration import SwarmDisputeArbitrator, SwarmDispute if not os.path.exists(args.passport): print(f"[ERROR] Passport file not found: {args.passport}") sys.exit(1) with open(args.passport, "r", encoding="utf-8") as f: p_data = json.load(f) voter_passport = SovereignAgentPassport.from_dict(p_data) dispute_file = getattr(args, "dispute_file", None) if dispute_file and os.path.exists(dispute_file): with open(dispute_file, "r", encoding="utf-8") as f: d_data = json.load(f) dispute = SwarmDispute(**d_data) else: dispute = SwarmDispute( dispute_id=args.dispute_id, escrow_id=getattr(args, "escrow_id", "ESCROW-MOCK"), challenger_agent_id="challenger-monitor", target_agent_id="target-violator", target_action="SYSTEM_MUTATION", amount_usd=1000.0, fault_proof={}, opened_at=time.time(), status="VOTING", required_quorum=2 ) arbitrator = SwarmDisputeArbitrator() arbitrator.disputes[dispute.dispute_id] = dispute ok, msg = arbitrator.cast_vote(dispute.dispute_id, voter_passport, args.vote) if not ok: print(f"[ERROR] Vote rejected: {msg}") sys.exit(1) d_dict = dispute.to_dict() if getattr(args, "out", None): with open(args.out, "w", encoding="utf-8") as f: json.dump(d_dict, f, indent=2) print(f"[+] Updated Dispute written to: {args.out}") print("=" * 70) print("BTP v4.1 SWARM ARBITRATION VOTE CAST") print("=" * 70) print(f"[*] Dispute ID : {dispute.dispute_id}") print(f"[*] Voter Agent : {voter_passport.agent_id}") print(f"[*] Vote : {args.vote}") print(f"[*] Status : OK (Ed25519 Signed)") print("=" * 70) def cmd_arbitration_resolve(args): from src.settlement.swarm_arbitration import SwarmDisputeArbitrator, SwarmDispute dispute_file = getattr(args, "dispute_file", None) if dispute_file and os.path.exists(dispute_file): with open(dispute_file, "r", encoding="utf-8") as f: d_data = json.load(f) dispute = SwarmDispute(**d_data) else: dispute = SwarmDispute( dispute_id=args.dispute_id, escrow_id=getattr(args, "escrow_id", "ESCROW-MOCK"), challenger_agent_id="challenger-monitor", target_agent_id="target-violator", target_action="SYSTEM_MUTATION", amount_usd=1000.0, fault_proof={}, opened_at=time.time(), status="VOTING", required_quorum=2, votes={ "juror-1": {"vote": "APPROVE_SLASH", "voter_passport_id": "pass-1", "signature": "0x11"}, "juror-2": {"vote": "APPROVE_SLASH", "voter_passport_id": "pass-2", "signature": "0x22"} } ) arbitrator = SwarmDisputeArbitrator() arbitrator.disputes[dispute.dispute_id] = dispute ok, msg, cert = arbitrator.resolve_dispute(dispute.dispute_id) if not ok: print(f"[ERROR] Resolution failed: {msg}") sys.exit(1) c_dict = cert.to_dict() if getattr(args, "out", None): with open(args.out, "w", encoding="utf-8") as f: json.dump(c_dict, f, indent=2) print(f"[+] Arbitration Resolution Certificate saved to: {args.out}") print("=" * 70) print("BTP v4.1 SWARM ARBITRATION RESOLUTION CERTIFICATE") print("=" * 70) print(f"[*] Certificate ID: {cert.certificate_id}") print(f"[*] Dispute ID : {cert.dispute_id}") print(f"[*] Escrow ID : {cert.escrow_id}") print(f"[*] Verdict : {cert.verdict}") print(f"[*] Quorum Votes : {cert.quorum_count}") print(f"[*] Slashed Amount: ${cert.slashed_amount_usd:,.2f} USD") print(f"[*] Certificate H : {cert.certificate_hash}") print("=" * 70) def cmd_benchmark_chaos(args): from src.benchmarks.swarm_chaos_benchmark import SwarmChaosBenchmark print("=" * 70) print("BTP v4.4 HIGH-CONCURRENCY SWARM CHAOS FUZZING BENCHMARK") print("=" * 70) print(f"[*] Iterations : {args.iterations}") print(f"[*] Concurrency : {args.concurrency}") print(f"[*] Collateral : ${args.collateral:.2f} USD") print("[*] Simulating adversarial injection across CrewAI, LangGraph, AutoGen & Universal Models...") benchmark = SwarmChaosBenchmark() report = benchmark.run_benchmark( iterations=args.iterations, concurrency=args.concurrency, collateral_usd=args.collateral ) print("-" * 70) print(f"[+] Total Adversarial Attacks : {report['adversarial_attacks_tested']}") print(f"[+] Attacks Intercepted (100%): {report['attacks_intercepted']}") print(f"[+] Interception Accuracy : {report['interception_accuracy_pct']}%") print(f"[+] Benign Allowed : {report['benign_requests_executed']}") print(f"[+] Total Slashed : ${report['total_collateral_slashed_usd']:,.2f} USD") print(f"[+] Throughput : {report['throughput_ops_per_sec']} ops/sec") print(f"[+] AST Latency (p50) : {report['latency_p50_us']} µs") print(f"[+] AST Latency (p95) : {report['latency_p95_us']} µs") print(f"[+] AST Latency (p99) : {report['latency_p99_us']} µs") print(f"[+] zk-Fault Proof (p50) : {report['zk_fault_proof_p50_us']} µs") print("=" * 70) if getattr(args, "out", None): with open(args.out, "w", encoding="utf-8") as f: json.dump(report, f, indent=2) print(f"[+] Benchmark report saved to: {args.out}") def cmd_settlement_deploy_evm(args): from src.settlement.evm_deployer import EVMDeployer print("=" * 70) print("BTP v4.3 MULTI-CHAIN EVM SETTLEMENT DEPLOYMENT") print("=" * 70) print(f"[*] Target Network: {args.network}") print(f"[*] Mode : {'Simulated Dry Run' if args.dry_run else 'Live Gas Execution'}") deployer = EVMDeployer(network=args.network, rpc_url=getattr(args, "rpc_url", None)) receipt = deployer.deploy( private_key=getattr(args, "private_key", None), dry_run=args.dry_run, output_file=getattr(args, "out", None) ) print("-" * 70) print(f"[+] Contract Name : {receipt['contract_name']}") print(f"[+] Network : {receipt['network_name']} (Chain ID: {receipt['chain_id']})") print(f"[+] Contract Addr : {receipt['contract_address']}") print(f"[+] Tx Hash : {receipt['transaction_hash']}") print(f"[+] EIP-712 Domain: {receipt['eip712_domain_name']} v{receipt['eip712_domain_version']}") print(f"[+] Explorer Link : {receipt['explorer_url']}") print("=" * 70) if getattr(args, "out", None): print(f"[+] Deployment receipt written to: {args.out}") def cmd_workspace_create(args): from src.tenancy.workspace_manager import WorkspaceManager wm = WorkspaceManager() tenant = wm.create_tenant( org_id=args.org, project_id=args.project, environment=args.env, display_name=getattr(args, "display_name", None) ) print("=" * 70) print("BTP v5.0 MULTI-TENANT WORKSPACE CREATION") print("=" * 70) print(f"[*] Organization : {tenant.org_id}") print(f"[*] Project : {tenant.project_id}") print(f"[*] Environment : {tenant.environment}") print(f"[*] Tenant ID : {tenant.tenant_id}") print(f"[*] Display Name : {tenant.display_name}") print("=" * 70) if getattr(args, "out", None): with open(args.out, "w", encoding="utf-8") as f: json.dump(tenant.to_dict(), f, indent=2) print(f"[+] Workspace configuration saved to: {args.out}") def cmd_workspace_list(args): from src.tenancy.workspace_manager import WorkspaceManager wm = WorkspaceManager() tenants = wm.list_tenants() print("=" * 70) print("BTP v5.0 REGISTERED WORKSPACE TENANTS") print("=" * 70) if not tenants: print(" No custom workspaces configured. Default tenant active.") else: for idx, t in enumerate(tenants, 1): print(f" [{idx}] {t['display_name']} -> Tenant ID: {t['tenant_id']}") print("=" * 70) def cmd_workspace_keygen(args): from src.tenancy.workspace_manager import WorkspaceManager wm = WorkspaceManager() api_key = wm.generate_scoped_api_key( org_id=args.org, project_id=args.project, environment=args.env, role=getattr(args, "role", "developer") ) print("=" * 70) print("BTP v5.0 SCOPED API KEY GENERATION") print("=" * 70) print(f"[*] Organization : {args.org}") print(f"[*] Project : {args.project}") print(f"[*] Environment : {args.env}") print(f"[*] Scoped Key : {api_key}") print("=" * 70) print("-> Use in your agent with: Guard(api_key='...') or export BTP_API_KEY='...'") def cmd_webhook_add(args): from src.alerting.webhook_dispatcher import WebhookDispatcher dispatcher = WebhookDispatcher() sub = dispatcher.register_subscription( tenant_id=args.tenant, platform=args.platform, target_url=args.url, secret=getattr(args, "secret", None), min_severity=getattr(args, "severity", "LOW").upper() ) print("=" * 70) print("BTP v5.1 SECOPS WEBHOOK SUBSCRIPTION REGISTERED") print("=" * 70) print(f"[*] Subscription ID : {sub.subscription_id}") print(f"[*] Tenant ID : {sub.tenant_id}") print(f"[*] Platform : {sub.platform.value.upper()}") print(f"[*] Target URL : {sub.target_url}") print(f"[*] Signing Secret : {sub.secret}") print(f"[*] Min Severity : {sub.min_severity.value}") print("=" * 70) def cmd_webhook_list(args): from src.alerting.webhook_dispatcher import WebhookDispatcher dispatcher = WebhookDispatcher() tenant_filter = getattr(args, "tenant", None) subs = dispatcher.list_subscriptions(tenant_id=tenant_filter) print("=" * 70) print("BTP v5.1 REGISTERED SECOPS WEBHOOK SUBSCRIPTIONS") print("=" * 70) if not subs: print(" No webhook subscriptions registered.") else: for idx, s in enumerate(subs, 1): print(f" [{idx}] {s.platform.value.upper():<10} | Tenant: {s.tenant_id[:16]:<16} | Min: {s.min_severity.value:<8} | URL: {s.target_url}") print("=" * 70) def cmd_webhook_test(args): from src.alerting.webhook_dispatcher import WebhookDispatcher, IncidentEvent, IncidentEventType, AlertSeverity dispatcher = WebhookDispatcher(sync_mode=True) evt_id = f"evt_test_{int(time.time())}" test_event = IncidentEvent( event_id=evt_id, tenant_id=args.tenant, org_id="test-org", project_id="test-proj", environment="dev", event_type=IncidentEventType.AST_VETO, severity=AlertSeverity(getattr(args, "severity", "HIGH").upper()), title="Test Invariant Violation", description="Simulated CLI test incident from btp-guard CLI.", agent_id="agent-cli-test-runner", tool_name="shell_execute", target_payload='{"cmd": "rm -rf /test"}', slashed_amount_usd=50.0, metadata={"cli_invoker": True} ) print("=" * 70) print(f"[*] Dispatching test incident '{evt_id}' to tenant '{args.tenant}'...") results = dispatcher.emit_incident(test_event) print(f"[*] Delivered to {len(results)} subscription(s):") for r in results: status = "SUCCESS" if r.get("success") else "FAILED" print(f" - [{r.get('platform')}] {r.get('target_url')} -> {status} (HTTP {r.get('status_code')}, Latency: {r.get('latency_ms')}ms)") print("=" * 70) def cmd_immune_run(args): from src.immune.auto_immunity_engine import AutoImmunityCoordinator coordinator = AutoImmunityCoordinator() iterations = getattr(args, "iterations", 20) auto_heal = getattr(args, "auto_heal", True) print("=" * 70) print("BTP v5.2 AUTO-IMMUNITY ENGINE — CONTINUOUS ADVERSARIAL RED-TEAMING") print("=" * 70) print(f"[*] Iterations : {iterations}") print(f"[*] Auto-Healing Mode : {'ENABLED (Atomic Hot-Reload)' if auto_heal else 'DISABLED'}") print("[*] Generating adversarial mutation vectors...") res = coordinator.run_immune_cycle(iterations=iterations, auto_heal=auto_heal) print(f"[+] Mutations Fuzzed : {res['mutations_tested']}") print(f"[+] Initially Blocked : {res['initially_blocked']}") print(f"[+] Gaps Discovered : {res['gaps_detected']}") print(f"[+] Rules Synthesized : {res['rules_synthesized']}") print(f"[+] False Positive % : {res['false_positive_rate']}% (Golden Corpus Verified)") print(f"[+] Cycle Execution : {res['elapsed_ms']}ms") print("=" * 70) if res['synthesized_rules']: print("[+] Synthesized Auto-Immune Rules:") for r in res['synthesized_rules']: print(f" - [{r['id']}] {r['description']} (Regex: `{r['regex']}`)") if auto_heal: reloaded = coordinator.hot_reload_into_policy_file() print(f"[OK] Policy hot-reloaded into: {coordinator.policy_path} (Status: {reloaded})") def cmd_immune_status(args): from src.immune.auto_immunity_engine import AutoImmunityCoordinator coordinator = AutoImmunityCoordinator() print("=" * 70) print("BTP v5.2 AUTO-IMMUNITY ENGINE TELEMETRY") print("=" * 70) print(f"[*] Active Immune Invariants : {len(coordinator.synthesized_rules)}") print(f"[*] Policy File Location : {coordinator.policy_path}") print(f"[*] Self-Healing Pipeline : ACTIVE (Golden Corpus Regression Capable)") print("=" * 70) def cmd_immune_rules(args): from src.immune.auto_immunity_engine import PolicyAutoHealer print("=" * 70) print("BTP v5.2 IMMUNE HEURISTIC PATTERN MATRIX") print("=" * 70) for tech, spec in PolicyAutoHealer.HEURISTIC_PATTERNS.items(): print(f" [{spec['id']}] Technique: {tech:<22} | Category: {spec['category']}") print(f" Regex: {spec['regex']}") print("=" * 70) def cmd_marketplace_list(args): from src.marketplace.sla_contract import AgentMarketplaceEngine engine = AgentMarketplaceEngine() cap = getattr(args, "capability", None) specialists = engine.list_specialists(capability=cap) print("=" * 70) print("BTP v5.3 CROSS-TENANT AGENT MARKETPLACE") print("=" * 70) if not specialists: print(" No specialist agents found matching criteria.") else: for idx, s in enumerate(specialists, 1): print(f" [{idx}] {s.display_name}") print(f" Agent ID : {s.agent_id} (Tenant: {s.tenant_id[:16]})") print(f" Capabilities: {', '.join(s.capabilities)}") print(f" Rate / Job : ${s.rate_usd_per_job:.2f} USD | Min Bond: ${s.min_bond_usd:.2f} USD") print(f" Reputation : {s.reputation_score * 100:.1f}% ({s.jobs_completed} jobs completed)") print(f" Rails : {', '.join(s.settlement_rails)}") print("=" * 70) def cmd_marketplace_contract_create(args): from src.marketplace.sla_contract import AgentMarketplaceEngine from src.settlement.autonomous_escrow import AutonomousEscrowPool engine = AgentMarketplaceEngine() pool = AutonomousEscrowPool() contract = engine.create_contract( client_tenant_id=args.client_tenant, client_org_id=args.client_org, client_agent_id=args.client_agent, provider_agent_id=args.provider_agent, required_capability=args.capability, budget_usd=args.budget, provider_bond_usd=args.bond, settlement_rail=getattr(args, "rail", "L402_LIGHTNING") ) # Lock conditional two-sided escrow c_dep, p_dep = pool.lock_sla_escrow(contract) engine.lock_contract(contract.contract_id, c_dep.escrow_id, p_dep.escrow_id) print("=" * 70) print("BTP v5.3 CROSS-TENANT SLA CONTRACT CREATED & ESCROWS LOCKED") print("=" * 70) print(f"[*] Contract ID : {contract.contract_id}") print(f"[*] Client Tenant : {contract.client_tenant_id} ({contract.client_org_id})") print(f"[*] Provider Agent : {contract.provider_agent_id}") print(f"[*] Capability Scope : {contract.required_capability}") print(f"[*] Payment Locked : ${contract.payment_budget_usd:.2f} USD ({c_dep.escrow_id})") print(f"[*] Performance Bond : ${contract.provider_bond_usd:.2f} USD ({p_dep.escrow_id})") print(f"[*] Settlement Rail : {contract.settlement_rail}") print(f"[*] Contract Status : LOCKED & ACTIVE") print("=" * 70) def cmd_marketplace_contract_fulfill(args): from src.marketplace.sla_contract import AgentMarketplaceEngine, ZKTaskCompletionProof from src.settlement.autonomous_escrow import AutonomousEscrowPool engine = AgentMarketplaceEngine() pool = AutonomousEscrowPool() contract = engine.contracts.get(args.contract_id) if not contract: print(f"[!] Error: Contract '{args.contract_id}' not found.") return # Generate synthetic or real zk-TCP proof proof = ZKTaskCompletionProof.create_proof( contract_id=contract.contract_id, provider_agent_id=contract.provider_agent_id, provider_tenant_id=contract.provider_tenant_id, input_data={"task": contract.required_capability, "contract_id": contract.contract_id}, output_data={"status": "COMPLETED", "result_digest": "0x44fa71bb9900c2"}, tool_actions=["audit_verify", "merkle_commit"] ) ok, msg, updated_contract = engine.fulfill_contract(contract.contract_id, proof) if ok: s_ok, s_msg, receipt = pool.settle_sla_completion( contract=updated_contract, completion_proof=proof, provider_payee_destination="provider_treasury_vault" ) print("=" * 70) print("BTP v5.3 SLA CONTRACT FULFILLED & SETTLED") print("=" * 70) print(f"[*] Contract ID : {updated_contract.contract_id}") print(f"[*] zk-TCP Proof ID : {proof.proof_id}") print(f"[*] Pedersen Commitment : {proof.pedersen_commitment}") print(f"[*] Fiat-Shamir Response : {proof.fiat_shamir_response}") print(f"[*] Amount Disbursed : ${updated_contract.payment_budget_usd:.2f} USD") print(f"[*] Performance Bond : Released Clean to Provider") print(f"[*] Settlement Status : {receipt.get('status')}") print("=" * 70) else: print(f"[!] Fulfillment Error: {msg}") def cmd_init(args): from src.init_wizard import scaffold_project target = os.path.abspath(getattr(args, "dir", ".")) res = scaffold_project( target_dir=target, framework=getattr(args, "framework", None), org=getattr(args, "org", "my-org"), project=getattr(args, "project", "my-agent-swarm"), env=getattr(args, "env", "dev"), non_interactive=True ) print("=" * 70) print("BTP ENTERPRISE DEVELOPER ONBOARDING: PROJECT INITIALIZED") print("=" * 70) print(f"[*] Framework Detected : {res['framework'].upper()}") print(f"[*] Tenant Identifier : {res['tenant_id']}") print(f"[*] Scoped API Key : {res['api_key']}") print(f"[*] Configuration Path : {res['btp_dir']}") print("-" * 70) print("READY-TO-RUN INTEGRATION SNIPPET:") print(res['snippet'].strip()) print("=" * 70) def cmd_billing_usage(args): from src.billing.metering_engine import TenantUsageMeter meter = TenantUsageMeter() rec = meter.get_or_create(args.tenant) print("=" * 70) print(f"BTP METERED USAGE STATEMENT: {rec.tenant_id}") print("=" * 70) print(f"[*] Organization : {rec.org_id}") print(f"[*] Project Workspace : {rec.project_id}") print(f"[*] AST Scans Executed : {rec.ast_scans:,} scans") print(f"[*] Threats Intercepted : {rec.threats_blocked:,} threats") print(f"[*] Escrow Volume Cleared: ${rec.escrow_volume_usd:,.2f} USD") print(f"[*] Webhook Dispatches : {rec.webhooks_dispatched:,} events") print("=" * 70) def cmd_billing_invoice(args): from src.billing.metering_engine import TenantUsageMeter, MeteredInvoiceGenerator meter = TenantUsageMeter() rec = meter.get_or_create(args.tenant) rail = getattr(args, "rail", "STRIPE_METERED") inv = MeteredInvoiceGenerator.generate_invoice(rec, settlement_rail=rail) print("=" * 70) print(f"BTP ITEMIZED INVOICE: {inv.invoice_id}") print("=" * 70) print(f"[*] Tenant ID : {inv.tenant_id} ({inv.org_id})") print(f"[*] Base Subscription : ${inv.base_subscription_usd:.2f} USD (Pro Tier)") print(f"[*] AST Evaluations : {inv.ast_scans_count} scans -> ${inv.ast_scans_cost_usd:.4f} USD") print(f"[*] Threats Blocked : {inv.threats_blocked_count} threats -> ${inv.threats_blocked_cost_usd:.4f} USD") print(f"[*] Escrow Clearing Fee : ${inv.escrow_fees_usd:.4f} USD (on ${inv.escrow_volume_cleared_usd:.2f} volume)") print(f"[*] Webhook Dispatches : {inv.webhooks_count} events -> ${inv.webhooks_cost_usd:.4f} USD") print("-" * 70) print(f"[*] TOTAL AMOUNT DUE : ${inv.total_due_usd:.2f} USD") print(f"[*] Settlement Rail : {inv.settlement_rail}") print(f"[*] Cryptographic Sig : {inv.signature}") print("=" * 70) def cmd_gossip_peer_list(args): from src.p2p.reputation_gossip import PeerReputationMesh mesh = PeerReputationMesh() print("=" * 70) print("BTP v5.4 DECENTRALIZED P2P PEER REPUTATION MESH") print("=" * 70) for p in mesh.peers.values(): tag = " [PRE-TRUSTED]" if p.is_pretrusted else "" print(f"[*] Node ID : {p.node_id}{tag}") print(f" Address : {p.address}") print(f" Global Trust: {p.global_trust * 100:.1f}% (Direct: {p.direct_trust * 100:.1f}%)") print(f" Vector Clock: {p.vector_clock} | Audited Jobs: {p.jobs_audited}") print("-" * 70) def cmd_gossip_rate(args): from src.p2p.reputation_gossip import PeerReputationMesh mesh = PeerReputationMesh() ok, msg, gossip_msg = mesh.broadcast_rating( rater_agent_id=args.rater, target_agent_id=args.target, score=args.score, task_contract_id=getattr(args, "contract_id", "SLA-AUTO-101") ) print("=" * 70) print("BTP v5.4 P2P REPUTATION GOSSIP BROADCAST") print("=" * 70) if ok: print(f"[+] Gossip ID : {gossip_msg.message_id}") print(f"[+] Rater Agent : {gossip_msg.rater_agent_id}") print(f"[+] Target Agent : {gossip_msg.target_agent_id}") print(f"[+] Score Gossiped: {gossip_msg.score * 100:.1f}%") print(f"[+] Signature : {gossip_msg.signature}") print(f"[+] EigenTrust : Global trust vectors re-converged across mesh.") else: print(f"[!] Gossip Error : {msg}") print("=" * 70) def cmd_bridge_transfer(args): from src.settlement.cross_chain_bridge import CrossChainBridgeRelay bridge = CrossChainBridgeRelay() ok, msg, voucher = bridge.lock_source_escrow( source_chain=args.source, target_chain=args.target, depositor=args.depositor, recipient=args.recipient, amount_usd=args.amount ) print("=" * 70) print("BTP v5.4 ATOMIC CROSS-CHAIN ESCROW BRIDGE: LOCK") print("=" * 70) if ok: print(f"[+] Voucher ID : {voucher.voucher_id}") print(f"[+] Route : {voucher.source_chain} -> {voucher.target_chain}") print(f"[+] Amount Locked: ${voucher.amount_usd:.2f} USD") print(f"[+] Lock Hash : {voucher.lock_hash}") print(f"[+] Preimage Key : {voucher.preimage}") print(f"[+] Relayer Sig : {voucher.signature}") print(f"[+] Status : {voucher.status}") else: print(f"[!] Bridge Error : {msg}") print("=" * 70) def cmd_bridge_claim(args): from src.settlement.cross_chain_bridge import CrossChainBridgeRelay bridge = CrossChainBridgeRelay() ok, msg, voucher = bridge.claim_target_escrow( voucher_id=args.voucher, secret_preimage=args.preimage ) print("=" * 70) print("BTP v5.4 ATOMIC CROSS-CHAIN ESCROW BRIDGE: CLAIM") print("=" * 70) if ok: print(f"[+] Voucher ID : {voucher.voucher_id}") print(f"[+] Target Rail : {voucher.target_chain}") print(f"[+] Disbursed To : {voucher.recipient}") print(f"[+] Amount : ${voucher.amount_usd:.2f} USD") print(f"[+] Status : {voucher.status}") else: print(f"[!] Claim Error : {msg}") print("=" * 70) def cmd_activate(args): """Activates Bartholomew Pro ($49/mo) or Enterprise ($199/mo) License.""" import webbrowser from src.usage_tracker import ( STRIPE_PRO_URL, STRIPE_ENTERPRISE_URL, STORE_URL, save_license, load_license ) print("=" * 70) print("[BTP GUARD] BARTHOLOMEW PROTOCOL (BTP v3.0) LICENSE ACTIVATION") print("=" * 70) current = load_license() if current.get("licensed"): print(f"[STATUS] Active License: Tier = {current.get('tier')} ({current.get('status')})") print(f"Features: {', '.join(current.get('features', []))}") print("-" * 70) if getattr(args, "key", None): res = save_license(args.key) print(f"\n[SUCCESS] License activated successfully!") print(f" -> Tier: {res['tier']}") print(f" -> Status: {res['status']}") print(f" -> Merkle Receipts Stamped with Verified {res['tier']} status.") return print("\nChoose an option to activate:") print(" [1] Pro Developer Tier ($49/mo) - Unlimited local evals & cloud policy editor") print(" [2] Enterprise SOC 2 Tier ($199/mo) - Continuous SOC 2/ISO 27001 evidence bundles") print(" [3] Enter Existing License Key - Activate key received via email/Stripe") print(" [4] Visit Storefront - https://bartholomew.info/store/") print("-" * 70) try: choice = input("Enter selection [1-4]: ").strip() except (EOFError, KeyboardInterrupt): print("\nActivation cancelled.") return if choice == "1": print(f"[*] Opening Pro checkout in your browser: {STRIPE_PRO_URL}") webbrowser.open(STRIPE_PRO_URL) print("\nAfter completing checkout, enter your license key below (or run 'python cli.py activate --key '):") try: key = input("Enter License Key: ").strip() if key: res = save_license(key) print(f"\n[SUCCESS] Activated {res['tier']} license!") except Exception: pass elif choice == "2": print(f"[*] Opening Enterprise checkout in your browser: {STRIPE_ENTERPRISE_URL}") webbrowser.open(STRIPE_ENTERPRISE_URL) print("\nAfter completing checkout, enter your license key below:") try: key = input("Enter License Key: ").strip() if key: res = save_license(key) print(f"\n[SUCCESS] Activated {res['tier']} license!") except Exception: pass elif choice == "3": try: key = input("Enter License Key: ").strip() if key: res = save_license(key) print(f"\n[SUCCESS] Activated {res['tier']} license!") else: print("[ERROR] No key provided.") except Exception as e: print(f"[ERROR] Activation failed: {e}") elif choice == "4": webbrowser.open(STORE_URL) else: print("Invalid selection.") def main(): parser = argparse.ArgumentParser(description="Bartholomew AI Agent Guardrail CLI") subparsers = parser.add_subparsers(dest="command", help="Available commands") # version subparsers.add_parser("version", help="Display BTP protocol version") # activate act_p = subparsers.add_parser("activate", help="Activate Bartholomew Pro ($49/mo) or Enterprise ($199/mo) License") act_p.add_argument("--key", "-k", type=str, default=None, help="License token received upon subscription checkout") # upgrade upg_p = subparsers.add_parser("upgrade", help="Upgrade to Bartholomew Pro ($49/mo) or Enterprise Fleet ($199/mo)") upg_p.add_argument("--tier", "-t", choices=["pro", "enterprise"], default="pro", help="Target commercial subscription tier (default: pro)") # pricing subparsers.add_parser("pricing", help="Display commercial tiers, fleet features, and direct Stripe checkout links") # init init_parser = subparsers.add_parser("init", help="10-second interactive project initialization & framework detection") init_parser.add_argument("--dir", "-d", default=".", help="Target project directory") init_parser.add_argument("--framework", "-f", choices=["crewai", "langgraph", "autogen", "openai", "anthropic", "generic"], help="Explicitly specify agent framework") init_parser.add_argument("--org", "-o", default="my-org", help="Organization / company name") init_parser.add_argument("--project", "-p", default="my-agent-swarm", help="Project / swarm name") init_parser.add_argument("--env", "-e", default="dev", choices=["dev", "staging", "prod"], help="Target environment") init_parser.add_argument("--pair", type=str, help="Legacy framework target alias") # onboard onboard_parser = subparsers.add_parser("onboard", help="Interactive 30-second developer fast-onboarding wizard for Cursor, LangGraph, CrewAI, OpenAI, and Escrows") onboard_parser.add_argument("--target", "-t", choices=["cursor", "windsurf", "vscode", "langchain", "crewai", "openai", "escrow", "license"], help="Directly configure target setup") # daemon daemon_parser = subparsers.add_parser("daemon", help="Manage background guard daemon") daemon_sub = daemon_parser.add_subparsers(dest="daemon_cmd") start_p = daemon_sub.add_parser("start", help="Start local daemon") start_p.add_argument("--port", type=int, default=8080, help="Daemon port (default: 8080)") start_p.add_argument("--host", type=str, default="127.0.0.1", help="Daemon host") start_p.add_argument("--background", "-b", action="store_true", help="Run in background") status_p = daemon_sub.add_parser("status", help="Query local daemon heartbeat & telemetry") status_p.add_argument("--port", type=int, default=8080, help="Daemon port") # mcp mcp_parser = subparsers.add_parser("mcp", help="Manage Model Context Protocol (MCP) server for Claude Desktop / Cursor / Astra") mcp_sub = mcp_parser.add_subparsers(dest="mcp_cmd") mcp_start_p = mcp_sub.add_parser("start", help="Start MCP stdio JSON-RPC server") mcp_start_p.add_argument("--workspace", type=str, default=None, help="Custom sandbox workspace root directory") mcp_run_p = mcp_sub.add_parser("run", help="Start MCP stdio JSON-RPC server (alias for start)") mcp_run_p.add_argument("--workspace", type=str, default=None, help="Custom sandbox workspace root directory") mcp_inst_p = mcp_sub.add_parser("install", help="1-Click auto-install into Claude Desktop / Cursor / Astra config") mcp_inst_p.add_argument("--target", type=str, default="claude", choices=["claude", "cursor", "astra", "all"], help="Target agent or IDE") mcp_inst_p.add_argument("--dry-run", action="store_true", help="Print configuration without writing to disk") mcp_inst_p.add_argument("--path", type=str, default=None, help="Custom configuration file path override") mcp_stat_p = mcp_sub.add_parser("status", help="Inspect registered MCP invariant tools and cryptographic capabilities") # policy policy_parser = subparsers.add_parser("policy", help="Manage declarative security policies") policy_sub = policy_parser.add_subparsers(dest="policy_cmd") val_p = policy_sub.add_parser("validate", help="Validate declarative YAML policy") val_p.add_argument("--file", "-f", type=str, default="policies/default_security_policy.yaml", help="Path to policy YAML") syn_p = policy_sub.add_parser("synthesize", help="Auto-synthesize least-privilege policy from traces") syn_p.add_argument("--output", "-o", type=str, default="policies/synthesized_policy.yaml", help="Output YAML file path") # demo demo_p = subparsers.add_parser("demo", help="Run high-impact interactive real-time invariant showcase") demo_p.add_argument("--speed", type=float, default=0.35, help="Simulation delay in seconds per step (default: 0.35)") # demo-v24 demo24_p = subparsers.add_parser("demo-v24", help="Run Bartholomew v2.4 Resilient MCP & Rollback Engine showcase") # proxy (MCP stdio proxy) proxy_p = subparsers.add_parser("proxy", help="Run Bartholomew as an inline MCP security proxy") proxy_p.add_argument("--server-cmd", nargs="+", required=True, help="Downstream MCP server command to launch") proxy_p.add_argument("--workspace", default=None, help="Root workspace directory to bound tool mutations") # agent (Interactive REPL) agent_p = subparsers.add_parser("agent", help="Launch interactive live agent REPL protected by Bartholomew") agent_p.add_argument("--interactive", "-i", action="store_true", default=True, help="Run in interactive REPL mode") # keygen (Ed25519) subparsers.add_parser("keygen", help="Generate a fresh sovereign Ed25519 keypair") # threshold-keygen (FROST RFC 9591) tk_p = subparsers.add_parser("threshold-keygen", help="Generate (t, n) FROST threshold shares and group public key (RFC 9591)") tk_p.add_argument("--threshold", "-t", type=int, default=3, help="Signing threshold: any t+1 agents can sign (default: 3)") tk_p.add_argument("--participants", "-n", type=int, default=5, help="Total swarm participants (default: 5)") tk_p.add_argument("--out", "-o", type=str, default=None, help="Directory to save group key and participant shares") # threshold-sign (FROST RFC 9591) ts_p = subparsers.add_parser("threshold-sign", help="Execute 2-round FROST threshold signing across agent shares") ts_p.add_argument("--shares", "-s", nargs="+", required=True, help="Paths to participant share JSON files") ts_p.add_argument("--payload", "-p", required=True, help="Payload string or path to JSON/binary file") ts_p.add_argument("--out", "-o", type=str, default=None, help="Output file to write signature JSON") # threshold-verify (FROST RFC 9591) tv_p = subparsers.add_parser("threshold-verify", help="Verify aggregate FROST threshold signature against group key") tv_p.add_argument("--sig", "-s", required=True, help="Path to signature JSON file") tv_p.add_argument("--payload", "-p", default=None, help="Payload string or file path to verify digest against") tv_p.add_argument("--pubkey", default=None, help="Group public key hex override (optional)") # threshold namespace subparser t_ns_p = subparsers.add_parser("threshold", help="FROST RFC 9591 & BIP 327 threshold signature engine") t_sub = t_ns_p.add_subparsers(dest="threshold_cmd") t_ns_k = t_sub.add_parser("keygen", help="Generate (t, n) FROST shares and group public key") t_ns_k.add_argument("--threshold", "-t", type=int, default=3, help="Signing threshold t (default: 3)") t_ns_k.add_argument("--participants", "-n", type=int, default=5, help="Total swarm participants n (default: 5)") t_ns_k.add_argument("--out", "-o", type=str, default=None, help="Directory to save group key and participant shares") t_ns_s = t_sub.add_parser("sign", help="Execute 2-round FROST threshold signing") t_ns_s.add_argument("--shares", "-s", nargs="+", required=True, help="Paths to participant share JSON files") t_ns_s.add_argument("--payload", "-p", required=True, help="Payload string or path to file") t_ns_s.add_argument("--out", "-o", type=str, default=None, help="Output signature file") t_ns_v = t_sub.add_parser("verify", help="Verify aggregate FROST threshold signature") t_ns_v.add_argument("--sig", "-s", required=True, help="Path to signature JSON file") t_ns_v.add_argument("--payload", "-p", default=None, help="Payload string or file path") t_ns_v.add_argument("--pubkey", default=None, help="Group public key hex override") # audit aud_p = subparsers.add_parser("audit", help="Audit local codebase for OWASP Agentic AI vulnerabilities") aud_p.add_argument("path", nargs="?", default=".", help="Target directory to audit (default: .)") aud_p.add_argument("--certify", action="store_true", help="Generate verifiable SOC 2 / OWASP compliance certificate with Merkle root & signature") aud_p.add_argument("--dossier", action="store_true", help="Export comprehensive CISO-ready SOC 2 / EU AI Act cryptographic dossier") aud_p.add_argument("--tenant", "-t", default="ten_default_enterprise", help="Tenant workspace ID for dossier") aud_p.add_argument("--format", "-F", choices=["md", "json"], default="md", help="Export format for compliance dossier") aud_p.add_argument("--org", type=str, default="Autonomous AI Deployment", help="Organization name for audit certificate") aud_p.add_argument("--out", "-o", type=str, default=None, help="Output path for certificate HTML or JSON package") # check chk_p = subparsers.add_parser("check", help="Statically verify policy for contradictions and invariant coverage") chk_p.add_argument("--file", "-f", default=".btp/policy.yaml", help="Path to policy YAML file") # sync sync_p = subparsers.add_parser("sync", help="Push verified policy to live agent workers via hot reload") sync_p.add_argument("--config", "-c", default=".btp/policy.yaml", help="Path to policy YAML file") sync_p.add_argument("--target", "-t", default="http://127.0.0.1:8000", help="Target daemon URL") sync_p.add_argument("--dry-run", action="store_true", help="Validate and fingerprint without dispatching") # verify-offline v_off_p = subparsers.add_parser("verify-offline", help="Independently verify an offline BTP receipt") v_off_p.add_argument("--receipt", "-r", required=True, help="Path to receipt JSON file") v_off_p.add_argument("--pubkey", "-p", help="Trusted authority public key hex (optional)") # hybrid-sign (BTP v2.9 FROST + Post-Quantum WOTS+) hs_p = subparsers.add_parser("hybrid-sign", help="Generate BTP v2.9 dual-layer FROST + Post-Quantum hybrid threshold signature") hs_p.add_argument("--shares", "-s", nargs="+", required=True, help="Paths to participant share JSON files") hs_p.add_argument("--payload", "-p", required=True, help="Payload string or path to JSON/binary file") hs_p.add_argument("--out", "-o", type=str, default=None, help="Output file to write hybrid envelope JSON") # hybrid-verify (BTP v2.9) hv_p = subparsers.add_parser("hybrid-verify", help="Verify BTP v2.9 dual-layer FROST + Post-Quantum hybrid envelope") hv_p.add_argument("--envelope", "-e", required=True, help="Path to hybrid envelope JSON file") hv_p.add_argument("--payload", "-p", default=None, help="Payload string or file path to verify digest against") # zk-prove (BTP v3.0) zkp_p = subparsers.add_parser("zk-prove", help="Generate BTP v3.0 Zero-Knowledge Invariant Compliance Proof") zkp_p.add_argument("--session-id", default=None, help="Agent session identifier") zkp_p.add_argument("--actions", "-a", nargs="*", default=None, help="List of tool actions/calls executed") zkp_p.add_argument("--actions-file", default=None, help="JSON file containing array of tool calls") zkp_p.add_argument("--policy", default="urn:btp:policy:standard-agent-invariants", help="Policy URI") zkp_p.add_argument("--out", "-o", type=str, default=None, help="Output receipt JSON file") # zk-verify (BTP v3.0) zkv_p = subparsers.add_parser("zk-verify", help="Verify BTP v3.0 Zero-Knowledge Invariant Compliance Receipt") zkv_p.add_argument("--receipt", "-r", required=True, help="Path to compliance receipt JSON file") # bond (BTP v3.1 Bonded Execution Warranty & Invariant Slashing) bond_p = subparsers.add_parser("bond", help="BTP v3.1 Bonded Execution Warranty & Invariant Slashing Engine") bond_sub = bond_p.add_subparsers(dest="bond_cmd") b_issue_p = bond_sub.add_parser("issue", help="Issue an execution warranty bond for an autonomous agent action") b_issue_p.add_argument("--agent", "-a", required=True, help="Agent identifier") b_issue_p.add_argument("--action", default="EXECUTE_TOOL", help="Action type or tool category") b_issue_p.add_argument("--amount", type=float, default=10000.0, help="Bond amount in USD (default: 10000.0)") b_issue_p.add_argument("--attestation", help="Attestation hash (optional)") b_issue_p.add_argument("--out", "-o", help="Output JSON file for bond artifact") b_slash_p = bond_sub.add_parser("slash", help="Arbitrate and slash an agent bond upon verified invariant breach") b_slash_p.add_argument("--bond-id", "-b", required=True, help="Bond ID string or path to bond JSON file") b_slash_p.add_argument("--proof", "-p", help="Path to breach receipt or failed ZK receipt JSON") b_slash_p.add_argument("--reason", "-r", help="Slashing reason description") # enclave (BTP v3.2 Confidential Computing Enclave Attestation) enc_p = subparsers.add_parser("enclave", help="BTP v3.2 Confidential Computing & Enclave Attestation Engine (AWS Nitro / AMD SEV-SNP)") enc_sub = enc_p.add_subparsers(dest="enclave_cmd") e_attest_p = enc_sub.add_parser("attest", help="Generate hardware-rooted confidential enclave attestation document") e_attest_p.add_argument("--module-id", default=None, help="Enclave module identifier") e_attest_p.add_argument("--nonce", default=None, help="Anti-replay freshness challenge nonce") e_attest_p.add_argument("--out", "-o", default=None, help="Output JSON file for attestation document") e_verify_p = enc_sub.add_parser("verify", help="Verify hardware attestation document against golden PCR baselines") e_verify_p.add_argument("--document", "-d", required=True, help="Path to enclave attestation document JSON") e_verify_p.add_argument("--nonce", default=None, help="Expected anti-replay challenge nonce (optional)") e_stat_p = enc_sub.add_parser("status", help="Display confidential enclave hardware telemetry and golden PCR baselines") # passport (BTP v3.1 Sovereign Digital Passports for Non-Human Workers) pass_p = subparsers.add_parser("passport", help="BTP v3.1 Sovereign Digital Passports for Non-Human Workers") pass_sub = pass_p.add_subparsers(dest="passport_cmd") p_issue_p = pass_sub.add_parser("issue", help="Issue an Ed25519-signed sovereign passport for an agent worker") p_issue_p.add_argument("--agent", "-a", required=True, help="Agent identifier") p_issue_p.add_argument("--model", "-m", required=True, help="Worker model or engine (e.g. gpt-4o, claude-3-5)") p_issue_p.add_argument("--capabilities", "-c", help="Comma-separated capability scopes (e.g. data:read,code:mutate)") p_issue_p.add_argument("--bond", "-b", type=float, default=0.0, help="Bonded warranty balance in USD") p_issue_p.add_argument("--out", "-o", help="Output JSON file path") p_verify_p = pass_sub.add_parser("verify", help="Cryptographically verify a sovereign passport file") p_verify_p.add_argument("--file", "-f", required=True, help="Path to passport JSON file") p_verify_p.add_argument("--capability", "-c", help="Optional capability scope to check authorization for") # peers (BTP v3.1 Autonomous Agent Peer Discovery Mesh) peer_p = subparsers.add_parser("peers", help="BTP v3.1 Autonomous Agent Peer Discovery Mesh") peer_sub = peer_p.add_subparsers(dest="peers_cmd") pr_disc_p = peer_sub.add_parser("discover", help="Discover peer agents matching capability and trust thresholds") pr_disc_p.add_argument("--capability", "-c", help="Required capability scope") pr_disc_p.add_argument("--min-reputation", type=float, help="Minimum trust reputation score (0.0 to 1.0)") pr_disc_p.add_argument("--min-bond", type=float, help="Minimum staked warranty bond in USD") pr_disc_p.add_argument("--model", help="Filter by model family (e.g. claude, gpt, gemini)") # escrow (BTP v4.0 Autonomous Micro-Escrow & Slashing Pool) escrow_p = subparsers.add_parser("escrow", help="BTP v4.0 Autonomous Micro-Escrow & Automated Slashing Pool") escrow_sub = escrow_p.add_subparsers(dest="escrow_cmd") esc_lock_p = escrow_sub.add_parser("lock", help="Lock collateral into autonomous micro-escrow before high-risk execution") esc_lock_p.add_argument("--agent", "-a", required=True, help="Agent identifier") esc_lock_p.add_argument("--action", required=True, help="Action type or operation being guarded") esc_lock_p.add_argument("--amount", type=float, default=100.0, help="Collateral amount in USD") esc_lock_p.add_argument("--passport", "-p", help="Path to sovereign agent passport JSON") esc_lock_p.add_argument("--rail", default="L402_LIGHTNING", help="Settlement rail (e.g. L402_LIGHTNING, SMART_CONTRACT)") esc_lock_p.add_argument("--out", "-o", help="Output deposit receipt JSON file path") esc_slash_p = escrow_sub.add_parser("slash", help="Liquidate and slash locked micro-escrow upon cryptographic regression proof") esc_slash_p.add_argument("--escrow-id", "-e", required=True, help="Escrow deposit ID to slash") esc_slash_p.add_argument("--proof", "-p", required=True, help="Path to cryptographic regression proof JSON") esc_slash_p.add_argument("--payee", required=True, help="Payee destination (Lightning invoice, wallet, or account)") esc_slash_p.add_argument("--passport", help="Path to sovereign agent passport JSON to trip circuit breaker") esc_slash_p.add_argument("--agent", default="Target-Agent", help="Target agent name if escrow not in active pool") esc_slash_p.add_argument("--amount", type=float, default=1000.0, help="Fallback indemnity amount if creating ad-hoc deposit") esc_stat_p = escrow_sub.add_parser("status", help="Display autonomous escrow liquidity and settlement telemetry") # rollup (BTP v3.5 Recursive Zero-Knowledge Rollup Batching & Enclave Anchoring) rollup_p = subparsers.add_parser("rollup", help="BTP v3.5 Recursive Zero-Knowledge Rollup Batching & Enclave Anchoring") rollup_sub = rollup_p.add_subparsers(dest="rollup_cmd") r_create_p = rollup_sub.add_parser("create", help="Batch multiple ZK-compliance proofs into a single recursive rollup") r_create_p.add_argument("--proofs", "-p", nargs="+", required=True, help="One or more ZK compliance receipt JSON files") r_create_p.add_argument("--out", "-o", help="Output sealed rollup batch JSON file path") r_verify_p = rollup_sub.add_parser("verify", help="Recursively verify a sealed ZK-Rollup batch") r_verify_p.add_argument("--rollup", "-r", required=True, help="Path to sealed rollup JSON file") r_anchor_p = rollup_sub.add_parser("anchor", help="Anchor a sealed ZK-Rollup to a confidential hardware enclave") r_anchor_p.add_argument("--rollup", "-r", required=True, help="Path to sealed rollup JSON file") r_anchor_p.add_argument("--out", "-o", help="Output hardware enclave anchor JSON file path") # arbitration (BTP v4.1 Decentralized Swarm Slashing Arbitration & ZK-Fault Proofs) arb_p = subparsers.add_parser("arbitration", help="BTP v4.1 Decentralized Swarm Slashing Arbitration & ZK-Fault Proofs") arb_sub = arb_p.add_subparsers(dest="arbitration_cmd") arb_prove_p = arb_sub.add_parser("prove-fault", help="Generate a Zero-Knowledge Fault Proof (zk-FP) for invariant breach") arb_prove_p.add_argument("--agent", "-a", required=True, help="Prover agent identifier") arb_prove_p.add_argument("--action", required=True, help="Target action name") arb_prove_p.add_argument("--violation", required=True, help="Violated invariant identifier") arb_prove_p.add_argument("--payload", required=True, help="Private payload text triggering violation") arb_prove_p.add_argument("--pre-hash", help="Optional pre-state hash") arb_prove_p.add_argument("--out", "-o", help="Output ZK-Fault Proof JSON file path") arb_chal_p = arb_sub.add_parser("challenge", help="Open a decentralized dispute challenging an escrow deposit") arb_chal_p.add_argument("--escrow-id", "-e", required=True, help="Escrow deposit ID to challenge") arb_chal_p.add_argument("--challenger", required=True, help="Challenger agent identifier") arb_chal_p.add_argument("--target-agent", required=True, help="Target agent identifier") arb_chal_p.add_argument("--target-action", required=True, help="Target action being challenged") arb_chal_p.add_argument("--amount", type=float, default=1000.0, help="Disputed amount in USD") arb_chal_p.add_argument("--fault-proof", "-f", required=True, help="Path to ZK-Fault Proof JSON file") arb_chal_p.add_argument("--out", "-o", help="Output dispute state JSON file path") arb_vote_p = arb_sub.add_parser("vote", help="Cast a signed juror vote in a swarm slashing dispute") arb_vote_p.add_argument("--dispute-id", "-d", required=True, help="Dispute ID to vote on") arb_vote_p.add_argument("--passport", "-p", required=True, help="Path to voter sovereign passport JSON file") arb_vote_p.add_argument("--vote", "-v", required=True, choices=["APPROVE_SLASH", "REJECT_SLASH"], help="Vote decision") arb_vote_p.add_argument("--dispute-file", help="Path to existing dispute JSON file to update") arb_vote_p.add_argument("--out", "-o", help="Output updated dispute JSON file path") arb_res_p = arb_sub.add_parser("resolve", help="Resolve swarm dispute and seal Arbitration Resolution Certificate") arb_res_p.add_argument("--dispute-id", "-d", required=True, help="Dispute ID to resolve") arb_res_p.add_argument("--dispute-file", help="Path to existing dispute JSON file") arb_res_p.add_argument("--out", "-o", help="Output Arbitration Resolution Certificate JSON path") # benchmark (BTP v4.4 High-Concurrency Chaos Fuzzing & Latency Benchmarks) bench_p = subparsers.add_parser("benchmark", help="BTP Performance & Chaos Latency Benchmarks") bench_sub = bench_p.add_subparsers(dest="benchmark_cmd") b_chaos_p = bench_sub.add_parser("swarm-chaos", help="Run high-concurrency cross-framework chaos fuzzing benchmark") b_chaos_p.add_argument("--iterations", "-i", type=int, default=50, help="Number of benchmark iterations") b_chaos_p.add_argument("--concurrency", "-c", type=int, default=4, help="Worker thread concurrency") b_chaos_p.add_argument("--collateral", type=float, default=250.0, help="Collateral per action in USD") b_chaos_p.add_argument("--out", "-o", help="Output benchmark report JSON file path") # settlement (BTP v4.3 Multi-Chain EVM & L402 Settlement Gateway) settle_p = subparsers.add_parser("settlement", help="BTP Multi-Chain Settlement & Contract Deployment") settle_sub = settle_p.add_subparsers(dest="settlement_cmd") s_evm_p = settle_sub.add_parser("deploy-evm", help="Deploy BartholomewEscrowPool.sol to EVM L2s") s_evm_p.add_argument("--network", "-n", default="base-sepolia", choices=["base-sepolia", "arbitrum-sepolia", "anvil-local"], help="Target EVM network") s_evm_p.add_argument("--rpc-url", help="Custom RPC endpoint URL") s_evm_p.add_argument("--private-key", help="Deployer private key (defaults to dry-run simulation)") s_evm_p.add_argument("--dry-run", action="store_true", default=True, help="Simulate deployment without spending live gas") s_evm_p.add_argument("--out", "-o", help="Output deployment receipt JSON file path") # workspace (BTP v5.0 Multi-Tenant Enterprise Workspaces & Scoped Projects) ws_p = subparsers.add_parser("workspace", help="BTP v5.0 Multi-Tenant Enterprise Workspaces & Scoped Projects") ws_sub = ws_p.add_subparsers(dest="workspace_cmd") ws_create_p = ws_sub.add_parser("create", help="Create a scoped organization and project workspace") ws_create_p.add_argument("--org", "-o", required=True, help="Organization / company identifier") ws_create_p.add_argument("--project", "-p", required=True, help="Project or swarm name") ws_create_p.add_argument("--env", "-e", default="dev", choices=["dev", "staging", "prod"], help="Deployment environment") ws_create_p.add_argument("--display-name", help="Friendly human-readable display name") ws_create_p.add_argument("--out", help="Output workspace configuration JSON file path") ws_list_p = ws_sub.add_parser("list", help="List registered workspace tenants") ws_key_p = ws_sub.add_parser("keygen", help="Generate a cryptographically scoped API key for this workspace") ws_key_p.add_argument("--org", "-o", required=True, help="Organization / company identifier") ws_key_p.add_argument("--project", "-p", required=True, help="Project or swarm name") ws_key_p.add_argument("--env", "-e", default="dev", choices=["dev", "staging", "prod"], help="Deployment environment") ws_key_p.add_argument("--role", default="developer", choices=["admin", "developer", "auditor"], help="API key role") # webhook (BTP v5.1 Real-Time Incident Webhooks & SecOps Alerts) wh_p = subparsers.add_parser("webhook", help="BTP v5.1 Real-Time Incident Webhooks & SecOps Alerts") wh_sub = wh_p.add_subparsers(dest="webhook_cmd") wh_add_p = wh_sub.add_parser("add", help="Register new webhook alert subscriber") wh_add_p.add_argument("--tenant", "-t", default="*", help="Tenant ID to bind alerts to (default: * for global)") wh_add_p.add_argument("--platform", "-p", choices=["slack", "discord", "pagerduty", "generic"], default="generic", help="Platform format") wh_add_p.add_argument("--url", "-u", required=True, help="Target destination webhook URL") wh_add_p.add_argument("--secret", "-s", default=None, help="Custom HMAC-SHA256 signing secret") wh_add_p.add_argument("--severity", choices=["LOW", "MEDIUM", "HIGH", "CRITICAL"], default="LOW", help="Minimum severity threshold") wh_list_p = wh_sub.add_parser("list", help="List registered webhook subscriptions") wh_list_p.add_argument("--tenant", "-t", default=None, help="Filter by tenant ID") wh_test_p = wh_sub.add_parser("test", help="Dispatch test incident alert to registered webhooks") wh_test_p.add_argument("--tenant", "-t", default="*", help="Target tenant ID") wh_test_p.add_argument("--severity", choices=["LOW", "MEDIUM", "HIGH", "CRITICAL"], default="HIGH", help="Severity level for test event") # immune (BTP v5.2 Auto-Immunity Engine & Self-Healing Invariant Synthesizer) immune_p = subparsers.add_parser("immune", help="BTP v5.2 Auto-Immunity Engine & Self-Healing Invariant Synthesizer") immune_sub = immune_p.add_subparsers(dest="immune_cmd") im_run_p = immune_sub.add_parser("run", help="Execute adversarial red-teaming fuzz cycle and auto-heal gaps") im_run_p.add_argument("--iterations", "-i", type=int, default=20, help="Number of adversarial mutations to generate (default: 20)") im_run_p.add_argument("--no-auto-heal", dest="auto_heal", action="store_false", default=True, help="Disable atomic policy hot-reload") # marketplace (BTP v5.3 Cross-Tenant Autonomous Agent Marketplace & SLA Escrows) mkt_p = subparsers.add_parser("marketplace", help="BTP v5.3 Cross-Tenant Autonomous Agent Marketplace & SLA Escrows") mkt_sub = mkt_p.add_subparsers(dest="marketplace_cmd") mkt_list_p = mkt_sub.add_parser("list", help="List registered cross-tenant specialist agents") mkt_list_p.add_argument("--capability", "-c", help="Filter by required capability") mkt_create_p = mkt_sub.add_parser("contract-create", help="Create cross-tenant SLA contract & lock two-sided escrow") mkt_create_p.add_argument("--client-tenant", required=True, help="Hiring client tenant ID") mkt_create_p.add_argument("--client-org", required=True, help="Hiring client organization") mkt_create_p.add_argument("--client-agent", required=True, help="Hiring client agent ID") mkt_create_p.add_argument("--provider-agent", required=True, help="Specialist provider agent ID") mkt_create_p.add_argument("--capability", required=True, help="Required capability scope") mkt_create_p.add_argument("--budget", type=float, required=True, help="Budget amount in USD") mkt_create_p.add_argument("--bond", type=float, default=25.0, help="Provider performance bond USD") mkt_create_p.add_argument("--rail", default="L402_LIGHTNING", choices=["L402_LIGHTNING", "EVM_BASE", "EVM_ARBITRUM"], help="Settlement rail") mkt_fulfill_p = mkt_sub.add_parser("contract-fulfill", help="Submit zk-TCP proof and settle cross-tenant SLA escrow") mkt_fulfill_p.add_argument("--contract-id", "-i", required=True, help="Contract ID to fulfill") # billing (BTP Multi-Tenant Metered Billing & Usage Engine) bill_p = subparsers.add_parser("billing", help="BTP Multi-Tenant Usage Metering & Invoicing") bill_sub = bill_p.add_subparsers(dest="billing_cmd") b_usage_p = bill_sub.add_parser("usage", help="Display metered usage consumption statement") b_usage_p.add_argument("--tenant", "-t", required=True, help="Tenant workspace ID") b_inv_p = bill_sub.add_parser("invoice", help="Generate itemized, cryptographically signed invoice") b_inv_p.add_argument("--tenant", "-t", required=True, help="Tenant workspace ID") b_inv_p.add_argument("--rail", default="STRIPE_METERED", choices=["STRIPE_METERED", "L402_LIGHTNING"], help="Settlement payment rail") # gossip (BTP v5.4 Decentralized P2P Peer Reputation Mesh) gossip_p = subparsers.add_parser("gossip", help="BTP v5.4 Decentralized P2P Peer Reputation Gossip") gossip_sub = gossip_p.add_subparsers(dest="gossip_cmd") g_list_p = gossip_sub.add_parser("peer-list", help="List active P2P mesh peers and EigenTrust scores") g_rate_p = gossip_sub.add_parser("rate", help="Broadcast signed peer performance rating across gossip mesh") g_rate_p.add_argument("--rater", "-r", required=True, help="Rater agent ID") g_rate_p.add_argument("--target", "-t", required=True, help="Target agent ID to rate") g_rate_p.add_argument("--score", "-s", type=float, required=True, help="Performance rating (0.0 to 1.0)") g_rate_p.add_argument("--contract-id", "-c", default="SLA-AUTO-101", help="Associated SLA task contract ID") # bridge (BTP v5.4 Atomic Cross-Chain Escrow Bridge Relay) bridge_p = subparsers.add_parser("bridge", help="BTP v5.4 Cross-Chain Escrow Bridge Relay") bridge_sub = bridge_p.add_subparsers(dest="bridge_cmd") b_lock_p = bridge_sub.add_parser("transfer", help="Initiate hash-locked cross-chain bridge transfer") b_lock_p.add_argument("--source", "-s", required=True, choices=["EVM_BASE", "EVM_ARBITRUM", "L402_LIGHTNING"], help="Source settlement rail") b_lock_p.add_argument("--target", "-t", required=True, choices=["EVM_BASE", "EVM_ARBITRUM", "L402_LIGHTNING"], help="Target settlement rail") b_lock_p.add_argument("--depositor", "-d", required=True, help="Depositor address/pubkey") b_lock_p.add_argument("--recipient", "-r", required=True, help="Recipient address/pubkey") b_lock_p.add_argument("--amount", "-a", type=float, required=True, help="Transfer amount in USD") b_claim_p = bridge_sub.add_parser("claim", help="Claim locked bridge escrow using cryptographic preimage") b_claim_p.add_argument("--voucher", "-v", required=True, help="Bridge voucher ID") b_claim_p.add_argument("--preimage", "-p", required=True, help="Secret preimage to unlock voucher") args = parser.parse_args() if args.command == "version": cmd_version(args) elif args.command == "upgrade": cmd_upgrade(args) elif args.command == "pricing": cmd_pricing(args) elif args.command == "activate": cmd_activate(args) elif args.command == "init": cmd_init(args) elif args.command == "gossip": if args.gossip_cmd == "peer-list": cmd_gossip_peer_list(args) elif args.gossip_cmd == "rate": cmd_gossip_rate(args) else: gossip_p.print_help() elif args.command == "bridge": if args.bridge_cmd == "transfer": cmd_bridge_transfer(args) elif args.bridge_cmd == "claim": cmd_bridge_claim(args) else: bridge_p.print_help() elif args.command == "billing": if args.billing_cmd == "usage": cmd_billing_usage(args) elif args.billing_cmd == "invoice": cmd_billing_invoice(args) else: bill_p.print_help() elif args.command == "marketplace": if args.marketplace_cmd == "list": cmd_marketplace_list(args) elif args.marketplace_cmd == "contract-create": cmd_marketplace_contract_create(args) elif args.marketplace_cmd == "contract-fulfill": cmd_marketplace_contract_fulfill(args) else: mkt_p.print_help() elif args.command == "workspace": if args.workspace_cmd == "create": cmd_workspace_create(args) elif args.workspace_cmd == "list": cmd_workspace_list(args) elif args.workspace_cmd == "keygen": cmd_workspace_keygen(args) else: ws_p.print_help() elif args.command == "webhook": if args.webhook_cmd == "add": cmd_webhook_add(args) elif args.webhook_cmd == "list": cmd_webhook_list(args) elif args.webhook_cmd == "test": cmd_webhook_test(args) else: wh_p.print_help() elif args.command == "immune": if args.immune_cmd == "run": cmd_immune_run(args) elif args.immune_cmd == "status": cmd_immune_status(args) elif args.immune_cmd == "rules": cmd_immune_rules(args) else: immune_p.print_help() elif args.command == "benchmark": if args.benchmark_cmd == "swarm-chaos": cmd_benchmark_chaos(args) else: bench_p.print_help() elif args.command == "settlement": if args.settlement_cmd == "deploy-evm": cmd_settlement_deploy_evm(args) else: settle_p.print_help() elif args.command == "arbitration": if args.arbitration_cmd == "prove-fault": cmd_arbitration_prove_fault(args) elif args.arbitration_cmd == "challenge": cmd_arbitration_challenge(args) elif args.arbitration_cmd == "vote": cmd_arbitration_vote(args) elif args.arbitration_cmd == "resolve": cmd_arbitration_resolve(args) else: arb_p.print_help() elif args.command == "escrow": if args.escrow_cmd == "lock": cmd_escrow_lock(args) elif args.escrow_cmd == "slash": cmd_escrow_slash(args) elif args.escrow_cmd == "status": cmd_escrow_status(args) else: escrow_p.print_help() elif args.command == "rollup": if args.rollup_cmd == "create": cmd_rollup_create(args) elif args.rollup_cmd == "verify": cmd_rollup_verify(args) elif args.rollup_cmd == "anchor": cmd_rollup_anchor(args) else: rollup_p.print_help() elif args.command == "enclave": if args.enclave_cmd == "attest": cmd_enclave_attest(args) elif args.enclave_cmd == "verify": cmd_enclave_verify(args) elif args.enclave_cmd == "status": cmd_enclave_status(args) else: enc_p.print_help() elif args.command == "bond": if args.bond_cmd == "issue": cmd_bond_issue(args) elif args.bond_cmd == "slash": cmd_bond_slash(args) else: bond_p.print_help() elif args.command == "passport": if args.passport_cmd == "issue": cmd_passport_issue(args) elif args.passport_cmd == "verify": cmd_passport_verify(args) else: pass_p.print_help() elif args.command == "peers": if args.peers_cmd == "discover": cmd_peers_discover(args) else: peer_p.print_help() elif args.command == "demo": from src.interactive_demo import run_interactive_demo run_interactive_demo(speed=args.speed) elif args.command == "demo-v24": from src.demo_v24 import run_demo_v24 run_demo_v24() elif args.command == "proxy": from src.mcp_gateway import MCPProxyGateway gateway = MCPProxyGateway(workspace_root=args.workspace) gateway.run_stdio_proxy(args.server_cmd) elif args.command == "agent": from src.interactive_agent_repl import run_agent_repl run_agent_repl() elif args.command == "keygen": cmd_keygen(args) elif args.command == "threshold-keygen": cmd_threshold_keygen(args) elif args.command == "threshold-sign": cmd_threshold_sign(args) elif args.command == "threshold-verify": cmd_threshold_verify(args) elif args.command == "threshold": if args.threshold_cmd == "keygen": cmd_threshold_keygen(args) elif args.threshold_cmd == "sign": cmd_threshold_sign(args) elif args.threshold_cmd == "verify": cmd_threshold_verify(args) else: t_ns_p.print_help() elif args.command == "audit": cmd_audit(args) elif args.command == "check": cmd_check(args) elif args.command == "sync": cmd_sync(args) elif args.command == "verify-offline": cmd_verify_offline(args) elif args.command == "hybrid-sign": cmd_hybrid_sign(args) elif args.command == "hybrid-verify": cmd_hybrid_verify(args) elif args.command == "zk-prove": cmd_zk_prove(args) elif args.command == "zk-verify": cmd_zk_verify(args) elif args.command == "init": cmd_init(args) elif args.command == "onboard": cmd_onboard(args) elif args.command == "daemon": if args.daemon_cmd == "start": cmd_daemon_start(args) elif args.daemon_cmd == "status": cmd_daemon_status(args) else: daemon_parser.print_help() elif args.command == "mcp": if args.mcp_cmd in ("start", "run") or not args.mcp_cmd: cmd_mcp_start(args) elif args.mcp_cmd == "install": cmd_mcp_install(args) elif args.mcp_cmd == "status": cmd_mcp_status(args) else: mcp_parser.print_help() elif args.command == "policy": if args.policy_cmd == "validate": cmd_policy_validate(args) elif args.policy_cmd == "synthesize": cmd_policy_synthesize(args) else: policy_parser.print_help() else: parser.print_help() if __name__ == "__main__": main()