#!/usr/bin/env python3 """ app.py — Strategy Factory: Profitability-Gated EVM Deployer ----------------------------------------------------------- - Runs on Hugging Face / any container - Uses RPC + keys only from env/secrets (no user input) - Auto-creates a local wallet (EOA), private key never leaves disk - Accepts Solidity project ZIPs - Compiles with Foundry (forge) - Discovers deployable artifacts - Runs a profitability check on the project - Only allows deployment if profitability gate passes - No hardcoded RPC URL or key (HF/Secrets-safe) """ import os import json import base64 import zipfile import subprocess import logging import uuid from pathlib import Path from typing import Dict, Any, List, Tuple from flask import Flask, render_template_string, request, abort from web3 import Web3 from eth_account import Account # ============================================================================== # 0. CONFIG / ENV # ============================================================================== BASE_DIR = Path(__file__).resolve().parent UPLOAD_DIR = BASE_DIR / "uploads" BUILD_ROOT = BASE_DIR / "builds" WALLET_FILE = BASE_DIR / "wallet.json" LOG_FILE = BASE_DIR / "app.log" UPLOAD_DIR.mkdir(exist_ok=True) BUILD_ROOT.mkdir(exist_ok=True) # RPC resolution: # 1) Prefer explicit RPC_URL secret # 2) Else build from YOUR_KEY + INFURA_PROJECT (also secrets) RPC_URL = os.getenv("RPC_URL") if not RPC_URL: key = os.getenv("YOUR_KEY") # HF secret: Infura key, for example project = os.getenv("INFURA_PROJECT") # HF secret: e.g. "base-mainnet" if not key or not project: raise SystemExit( "[Config] Set either RPC_URL, or YOUR_KEY + INFURA_PROJECT as secrets." ) # NOTE: no literal full URL or key in code; built only from env RPC_URL = f"https://{project}.infura.io/v3/{key}" DEFAULT_GAS_LIMIT = int(os.getenv("DEPLOY_GAS_LIMIT", "5000000")) DEFAULT_GAS_PRICE_GWEI = os.getenv("DEPLOY_GAS_PRICE_GWEI", "15") PREFERRED_CONTRACT_NAME = os.getenv("DEPLOY_CONTRACT_NAME", "").strip() # optional # Profitability gate config MIN_EXPECTED_PROFIT = float(os.getenv("MIN_EXPECTED_PROFIT", "0.0")) # in your units # ============================================================================== # 1. LOGGING # ============================================================================== logger = logging.getLogger("strategy_factory") logger.setLevel(logging.INFO) if not logger.handlers: fh = logging.FileHandler(LOG_FILE) fh.setLevel(logging.INFO) fmt = logging.Formatter( fmt="%(asctime)s | %(levelname)s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) fh.setFormatter(fmt) logger.addHandler(fh) def log_event(event: str, **fields: Any) -> None: parts = [f"event={event}"] for k, v in fields.items(): parts.append(f"{k}={v}") logger.info(" ".join(parts)) # ============================================================================== # 2. WALLET MANAGEMENT # ============================================================================== def load_or_create_wallet() -> Account: if WALLET_FILE.exists(): data = json.loads(WALLET_FILE.read_text()) key = base64.b64decode(data["private_key"]) acct = Account.from_key(key) print(f"[Wallet] Loaded: {acct.address}") log_event("wallet_loaded", address=acct.address) return acct acct = Account.create() key_data = { "address": acct.address, "private_key": base64.b64encode(acct.key).decode(), } WALLET_FILE.write_text(json.dumps(key_data)) print(f"[Wallet] Created: {acct.address}") log_event("wallet_created", address=acct.address) return acct wallet = load_or_create_wallet() # ============================================================================== # 3. RPC / WEB3 # ============================================================================== w3 = Web3(Web3.HTTPProvider(RPC_URL)) if not w3.is_connected(): log_event("rpc_connect_failed", rpc_url=RPC_URL) raise SystemExit(f"[RPC] Cannot connect to {RPC_URL}") try: CHAIN_ID = w3.eth.chain_id print(f"[RPC] Connected: chain_id={CHAIN_ID}") log_event("rpc_connected", rpc_url=RPC_URL, chain_id=CHAIN_ID) except Exception as e: CHAIN_ID = None log_event("rpc_chainid_error", error=str(e)) # ============================================================================== # 4. TOOLING CHECKS # ============================================================================== def check_forge() -> None: try: subprocess.run( ["forge", "--version"], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) log_event("forge_available") except Exception as e: log_event("forge_missing", error=str(e)) raise SystemExit("[Setup] Foundry 'forge' binary not found in PATH.") check_forge() # ============================================================================== # 5. BUILD PIPELINE # ============================================================================== def extract_project(zip_path: Path, build_id: str) -> Path: target_dir = BUILD_ROOT / build_id target_dir.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(zip_path, "r") as zf: zf.extractall(target_dir) log_event("project_extracted", build_id=build_id, zip=str(zip_path), dest=str(target_dir)) return target_dir def compile_project(project_path: Path) -> None: log_event("compile_start", path=str(project_path)) print("[Compile] forge build ...") try: subprocess.run( ["forge", "build", "--force"], cwd=project_path, check=True, ) except subprocess.CalledProcessError as e: log_event("compile_error", path=str(project_path), error=str(e)) raise RuntimeError(f"forge build failed: {e}") log_event("compile_success", path=str(project_path)) def discover_artifacts(project_path: Path) -> List[Dict[str, Any]]: """ Deployable artifacts: JSON with non-empty 'bytecode.object' and non-empty 'abi'. """ out_dir = project_path / "out" results: List[Dict[str, Any]] = [] if not out_dir.exists(): log_event("artifact_out_missing", out=str(out_dir)) return results for json_path in out_dir.rglob("*.json"): try: manifest = json.loads(json_path.read_text()) except Exception: continue bytecode_obj = (manifest.get("bytecode") or {}).get("object") or "" abi = manifest.get("abi") or [] contract_name = manifest.get("contractName") if bytecode_obj and abi and contract_name: relpath = json_path.relative_to(project_path) results.append( { "path": json_path, "relpath": relpath, "contractName": contract_name, "bytecodePrefix": bytecode_obj[:32], "abiLength": len(abi), } ) log_event("artifact_scan_done", path=str(project_path), count=len(results)) return results def select_artifact(artifacts: List[Dict[str, Any]]) -> Dict[str, Any]: """ Selection strategy: - If PREFERRED_CONTRACT_NAME is set and present, use it. - If only one artifact, use it. - Else pick lexicographically smallest relpath. """ if not artifacts: raise RuntimeError("No deployable artifacts found.") if PREFERRED_CONTRACT_NAME: for art in artifacts: if art["contractName"] == PREFERRED_CONTRACT_NAME: log_event("artifact_selected_preferred", contract=art["contractName"], relpath=str(art["relpath"])) return art if len(artifacts) == 1: art = artifacts[0] log_event("artifact_selected_single", contract=art["contractName"], relpath=str(art["relpath"])) return art artifacts_sorted = sorted(artifacts, key=lambda a: str(a["relpath"])) art = artifacts_sorted[0] log_event( "artifact_selected_default", contract=art["contractName"], relpath=str(art["relpath"]), candidates=len(artifacts), ) return art def write_build_metadata(build_id: str, artifact_relpath: str) -> None: meta = { "build_id": build_id, "artifact_relpath": artifact_relpath, } meta_path = BUILD_ROOT / build_id / "deploy_meta.json" meta_path.write_text(json.dumps(meta)) log_event("build_meta_written", build_id=build_id, meta=str(meta_path)) def read_build_metadata(build_id: str) -> Dict[str, Any]: meta_path = BUILD_ROOT / build_id / "deploy_meta.json" if not meta_path.exists(): log_event("build_meta_missing", build_id=build_id, path=str(meta_path)) raise RuntimeError("Build metadata not found.") return json.loads(meta_path.read_text()) # ============================================================================== # 6. PROFITABILITY VERIFIER (FACTORY GATE) # ============================================================================== def verify_profitability(project_path: Path) -> Tuple[bool, str]: """ Concrete, deterministic gate: - Looks for project_path / profitability.json - Expects: { "expectedProfit": , ... } - Accepts only if expectedProfit >= MIN_EXPECTED_PROFIT This gives you a real, pluggable contract: - Your MEV backtester / ZK verifier / simulator can emit this file. - The deployer refuses to deploy projects that don't meet the threshold. """ metrics_path = project_path / "profitability.json" if not metrics_path.exists(): msg = "profitability.json not found in project root" log_event("profitability_missing", path=str(metrics_path)) return False, msg try: data = json.loads(metrics_path.read_text()) except Exception as e: msg = f"invalid profitability.json: {e}" log_event("profitability_parse_error", path=str(metrics_path), error=str(e)) return False, msg expected_profit = float(data.get("expectedProfit", 0.0)) log_event("profitability_read", expected_profit=expected_profit, min_required=MIN_EXPECTED_PROFIT) if expected_profit < MIN_EXPECTED_PROFIT: msg = f"expectedProfit {expected_profit} < required {MIN_EXPECTED_PROFIT}" log_event("profitability_rejected", expected_profit=expected_profit) return False, msg msg = f"expectedProfit {expected_profit} >= required {MIN_EXPECTED_PROFIT}" log_event("profitability_accepted", expected_profit=expected_profit) return True, msg # ============================================================================== # 7. DEPLOYMENT # ============================================================================== def load_manifest(project_path: Path, artifact_relpath: str) -> Dict[str, Any]: manifest_path = project_path / artifact_relpath if not manifest_path.exists(): log_event("manifest_missing", path=str(manifest_path)) raise RuntimeError("Manifest file missing on disk.") return json.loads(manifest_path.read_text()) def build_deploy_tx(manifest: Dict[str, Any]) -> Dict[str, Any]: abi = manifest["abi"] bytecode = manifest["bytecode"]["object"] contract = w3.eth.contract(abi=abi, bytecode=bytecode) nonce = w3.eth.get_transaction_count(wallet.address) gas_price = w3.to_wei(DEFAULT_GAS_PRICE_GWEI, "gwei") tx = contract.constructor().build_transaction({ "from": wallet.address, "nonce": nonce, "gas": DEFAULT_GAS_LIMIT, "gasPrice": gas_price, "chainId": CHAIN_ID, }) log_event( "tx_built", wallet=wallet.address, nonce=nonce, gas=tx.get("gas"), gas_price=gas_price, ) return tx def deploy_manifest(project_path: Path, artifact_relpath: str) -> str: log_event( "deploy_start", project=str(project_path), artifact_relpath=artifact_relpath, wallet=wallet.address, ) manifest = load_manifest(project_path, artifact_relpath) tx = build_deploy_tx(manifest) signed = wallet.sign_transaction(tx) try: tx_hash = w3.eth.send_raw_transaction(signed.rawTransaction) log_event("tx_sent", tx_hash=tx_hash.hex()) receipt = w3.eth.wait_for_transaction_receipt(tx_hash) addr = receipt.contractAddress status = receipt.status log_event( "deploy_success", contract_address=addr, tx_hash=tx_hash.hex(), status=status, ) if status != 1: raise RuntimeError(f"Deployment transaction reverted, status={status}") return addr except Exception as e: log_event("deploy_error", error=str(e)) raise # ============================================================================== # 8. FLASK APP (FACTORY UI) # ============================================================================== app = Flask(__name__) INDEX_HTML = """

🧱 Strategy Factory: Profitability-Gated Deployer

Wallet: {{ wallet }}

RPC: {{ rpc }}

Chain ID: {{ chain_id }}

Min expectedProfit: {{ min_profit }}

""" REVIEW_HTML = """

📦 Strategy Preview

Build ID: {{ build_id }}

Artifact: {{ artifact_relpath }}

Profitability Gate: {{ profit_msg }}

{% if can_deploy %} 
{{ preview | tojson(indent=2) }}
{% else %}

Deployment blocked by profitability gate.

{{ preview | tojson(indent=2) }}
Back {% endif %} """ ERROR_HTML = """

❌ Error

{{ message }}
Back """ @app.route("/") def index(): return render_template_string( INDEX_HTML, wallet=wallet.address, rpc=RPC_URL, chain_id=CHAIN_ID, min_profit=MIN_EXPECTED_PROFIT, ) @app.route("/health") def health(): return { "status": "ok", "chainId": CHAIN_ID, "wallet": wallet.address, "rpcConfigured": bool(RPC_URL), } @app.route("/upload", methods=["POST"]) def upload(): if "zipfile" not in request.files: abort(400, "missing file") f = request.files["zipfile"] if not f.filename.lower().endswith(".zip"): abort(400, "only .zip supported") build_id = uuid.uuid4().hex zip_path = UPLOAD_DIR / f"{build_id}.zip" f.save(zip_path) log_event("upload_received", filename=f.filename, saved_as=str(zip_path), build_id=build_id) try: project_path = extract_project(zip_path, build_id) compile_project(project_path) artifacts = discover_artifacts(project_path) artifact = select_artifact(artifacts) # profitability gate ok, profit_msg = verify_profitability(project_path) write_build_metadata(build_id, str(artifact["relpath"])) preview = { "contract": artifact["contractName"], "artifactRelpath": str(artifact["relpath"]), "bytecodePrefix": artifact["bytecodePrefix"], "abiLength": artifact["abiLength"], } log_event( "preview_ready", build_id=build_id, contract=preview["contract"], artifact_relpath=str(artifact["artifactRelpath"]) if "artifactRelpath" in preview else preview["artifactRelpath"] if "artifactRelpath" in preview else "", abi_len=preview["abiLength"], profitability_ok=ok, ) return render_template_string( REVIEW_HTML, preview=preview, build_id=build_id, artifact_relpath=str(artifact["relpath"]), profit_msg=profit_msg, can_deploy=ok, ) except Exception as e: log_event("upload_processing_error", build_id=build_id, error=str(e)) return render_template_string(ERROR_HTML, message=str(e)), 500 @app.route("/deploy", methods=["POST"]) def deploy(): build_id = request.form.get("build_id", "").strip() if not build_id: abort(400, "missing build id") try: meta = read_build_metadata(build_id) artifact_relpath = meta["artifact_relpath"] project_path = BUILD_ROOT / build_id if not project_path.exists(): raise RuntimeError("Build directory not found on server.") # Optional: re-run profitability gate here if you want to be strict ok, _ = verify_profitability(project_path) if not ok: raise RuntimeError("Profitability gate failed at deploy time.") address = deploy_manifest(project_path, artifact_relpath) except Exception as e: msg = f"Deployment failed: {e}" log_event("deploy_exception", build_id=build_id, error=str(e)) return render_template_string(ERROR_HTML, message=msg), 500 return ( f"

✅ Deployed at: {address}

" f"

Wallet: {wallet.address}

" f"

Back

" ) if __name__ == "__main__": print("\n=== Strategy Factory Ready ===") print(f"Wallet: {wallet.address}") print(f"RPC: {RPC_URL}") print(f"Chain ID: {CHAIN_ID}") print(f"Min expectedProfit: {MIN_EXPECTED_PROFIT}") print("Upload strategy ZIP via http://localhost:7860") app.run(host="0.0.0.0", port=7860)