#!/usr/bin/env python3 """Preflight the DBIS Engine X virtual-batch vault canary. Read-only by design. It prepares canary proof hashes and emits review-only commands; it does not deploy, swap, approve, seed, or run a canary. """ from __future__ import annotations import argparse import hashlib import json import os import re import subprocess from datetime import datetime, timezone from decimal import Decimal, getcontext from pathlib import Path from xml.dom import minidom from xml.etree import ElementTree as ET getcontext().prec = 80 CWUSDC = "0x2de5F116bFcE3d0f922d9C8351e0c5Fc24b9284a" USDC = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48" XAUT = "0x68749665FF8D2d112Fa859AA293F07A622782F38" UNISWAP_V3_QUOTER_V2 = "0x61fFE014bA17989E743c5F6cB21bF9697530B21e" UNISWAP_SWAP_ROUTER_02 = "0x68b3465833fb72A70ecDF485E0e4C7bD8665Fc45" DEPLOYER_FALLBACK = "0x4A666F96fC8764181194447A7dFdb7d471b301C8" XAUT_USD_PRICE6 = 3_226_640_000 LTV_BPS = 8_000 MAX_ROUND_TRIP_LOSS_BPS = 100 CANARY_POOL_RAW = 1_000_000 CANARY_LENDER_RAW = 10_000 CANARY_DEBT_RAW = 1_000 XAUT_SWAP_RAW = 250 XAUT_USDC_FEE = 500 HEAD_NS = "urn:iso:std:iso:20022:tech:xsd:head.001.001.03" PACS_NS = "urn:iso:std:iso:20022:tech:xsd:pacs.008.001.13" DBIS_NS = "https://d-bis.org/iso20022/engine-x/canary/v1" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--json-out", default="reports/status/dbis-engine-x-virtual-batch-canary-preflight-latest.json") parser.add_argument("--md-out", default="reports/status/dbis-engine-x-virtual-batch-canary-preflight-latest.md") parser.add_argument("--artifact-prefix", default="reports/status/dbis-engine-x-virtual-batch-canary") parser.add_argument("--skip-deploy-gas-estimate", action="store_true") return parser.parse_args() def run(cmd: list[str], *, allow_fail: bool = False, cwd: str | None = None) -> str: proc = subprocess.run(cmd, text=True, capture_output=True, check=False, cwd=cwd) if proc.returncode != 0: if allow_fail: return proc.stderr.strip() or proc.stdout.strip() raise RuntimeError(f"{' '.join(cmd)} failed: {proc.stderr.strip()}") return proc.stdout.strip() def first_uint(output: str) -> int | None: match = re.search(r"\b(\d+)\b", output) return int(match.group(1)) if match else None def human(raw: int, decimals: int = 6) -> str: return f"{Decimal(raw) / (Decimal(10) ** decimals):,.{decimals}f}" def eth(raw: int) -> str: return f"{Decimal(raw) / Decimal(10**18):,.18f}" def get_amount_in(amount_out: int, reserve_in: int, reserve_out: int) -> int: numerator = reserve_in * amount_out * 1000 denominator = (reserve_out - amount_out) * 997 return numerator // denominator + 1 def get_amount_out(amount_in: int, reserve_in: int, reserve_out: int) -> int: amount_in_with_fee = amount_in * 997 numerator = amount_in_with_fee * reserve_out denominator = reserve_in * 1000 + amount_in_with_fee return numerator // denominator def min_xaut_collateral(debt_usdc: int) -> int: numerator = debt_usdc * 1_000_000 * 10_000 denominator = XAUT_USD_PRICE6 * LTV_BPS return (numerator + denominator - 1) // denominator def sha256_hex(path: Path) -> str: return "0x" + hashlib.sha256(path.read_bytes()).hexdigest() def write_canary_artifacts(prefix: Path, payload: dict) -> dict[str, str]: prefix.parent.mkdir(parents=True, exist_ok=True) generated_at = payload["generatedAt"] iso_path = prefix.with_name(prefix.name + "-iso20022-latest.xml") peg_path = prefix.with_name(prefix.name + "-peg-proof-latest.json") audit_path = prefix.with_name(prefix.name + "-audit-envelope-latest.json") hashes_path = prefix.with_name(prefix.name + "-proof-hashes-latest.json") ET.register_namespace("head", HEAD_NS) ET.register_namespace("pacs", PACS_NS) ET.register_namespace("dbis", DBIS_NS) root = ET.Element("DBISEngineXCanaryISO20022Evidence") hdr = ET.SubElement(root, f"{{{HEAD_NS}}}AppHdr") ET.SubElement(hdr, f"{{{HEAD_NS}}}BizMsgIdr").text = "DBIS-ENGINE-X-CANARY" ET.SubElement(hdr, f"{{{HEAD_NS}}}MsgDefIdr").text = "pacs.008.001.13" ET.SubElement(hdr, f"{{{HEAD_NS}}}CreDt").text = generated_at doc = ET.SubElement(root, f"{{{PACS_NS}}}Document") tx = ET.SubElement(ET.SubElement(doc, f"{{{PACS_NS}}}FIToFICstmrCdtTrf"), f"{{{PACS_NS}}}CdtTrfTxInf") pmt = ET.SubElement(tx, f"{{{PACS_NS}}}PmtId") ET.SubElement(pmt, f"{{{PACS_NS}}}InstrId").text = "DBIS-ENGINE-X-CANARY" amount = ET.SubElement(tx, f"{{{PACS_NS}}}IntrBkSttlmAmt") amount.set("Ccy", "USD") amount.text = payload["canary"]["exactOutput"] supp = ET.SubElement(tx, f"{{{PACS_NS}}}SplmtryData") env = ET.SubElement(supp, f"{{{PACS_NS}}}Envlp") proof = ET.SubElement(env, f"{{{DBIS_NS}}}EngineXCanaryProof") for key in ("debtRaw", "cwusdcInRaw", "cwusdcOutRaw", "exactOutputRaw", "collateralXautRaw"): ET.SubElement(proof, f"{{{DBIS_NS}}}{key}").text = str(payload["canary"][key]) iso_path.write_text(minidom.parseString(ET.tostring(root, encoding="utf-8")).toprettyxml(indent=" ")) peg = { "schema": "dbis-engine-x-canary-peg-proof/v1", "generatedAt": generated_at, "classification": "internal_engine_x_accounting_peg_canary", "publicDexPegClaim": False, "vaultPoolBeforeRaw": {"cWUSDC": str(CANARY_POOL_RAW), "USDC": str(CANARY_POOL_RAW)}, "vaultPoolAfterRaw": {"cWUSDC": str(CANARY_POOL_RAW), "USDC": str(CANARY_POOL_RAW)}, "canary": payload["canary"], "limitations": [ "Canary proves the virtual-batch exact-output path and internal reserve invariant only.", "It does not prove public DEX price discovery or public cWUSDC/USDC market depth.", ], } peg_path.write_text(json.dumps(peg, indent=2) + "\n") audit = { "schema": "dbis-engine-x-canary-audit-envelope/v1", "generatedAt": generated_at, "classification": "canary_audit_evidence_not_legal_opinion", "deployer": payload["deployer"], "plannedVaultConstructor": payload["constructor"], "canary": payload["canary"], "controls": { "exactOutputHashAnchor": True, "poolReserveEqualityBefore": True, "poolReserveEqualityAfter": True, "noPublicDexPegClaim": True, }, "regulatoryLimitations": [ "This package is not AML/KYC clearance.", "This package is not sanctions-screening evidence.", "This package is not an external auditor opinion.", "This package is not regulator approval.", ], } audit_path.write_text(json.dumps(audit, indent=2) + "\n") hashes = { "schema": "dbis-engine-x-canary-proof-hashes/v1", "generatedAt": generated_at, "hashAlgorithm": "sha256", "iso20022DocumentHash": sha256_hex(iso_path), "pegProofHash": sha256_hex(peg_path), "auditEnvelopeHash": sha256_hex(audit_path), "artifacts": {"iso20022": str(iso_path), "pegProof": str(peg_path), "auditEnvelope": str(audit_path)}, "envExports": { "DBIS_ENGINE_X_CANARY_ISO20022_HASH": sha256_hex(iso_path), "DBIS_ENGINE_X_CANARY_AUDIT_HASH": sha256_hex(audit_path), "DBIS_ENGINE_X_CANARY_PEG_HASH": sha256_hex(peg_path), }, } hashes_path.write_text(json.dumps(hashes, indent=2) + "\n") return {"iso20022": str(iso_path), "pegProof": str(peg_path), "auditEnvelope": str(audit_path), "proofHashes": str(hashes_path), **hashes["envExports"]} def main() -> int: args = parse_args() rpc = os.environ.get("ETHEREUM_MAINNET_RPC") or os.environ.get("MAINNET_RPC_URL") if not rpc: raise SystemExit("ETHEREUM_MAINNET_RPC or MAINNET_RPC_URL is required") deployer = os.environ.get("DEPLOYER_ADDRESS") or DEPLOYER_FALLBACK private_key_present = bool(os.environ.get("PRIVATE_KEY")) if private_key_present: deployer = run(["cast", "wallet", "address", os.environ["PRIVATE_KEY"]]) cwusdc = os.environ.get("CWUSDC_MAINNET") or CWUSDC usdc = os.environ.get("USDC_MAINNET") or USDC xaut = os.environ.get("XAUT_MAINNET") or XAUT surplus_receiver = os.environ.get("DBIS_ENGINE_X_SURPLUS_RECEIVER") or deployer rounding_receiver = os.environ.get("DBIS_ENGINE_X_ROUNDING_RECEIVER") or deployer balances = { "ethWei": first_uint(run(["cast", "balance", deployer, "--rpc-url", rpc], allow_fail=True)) or 0, "cWUSDCRaw": first_uint(run(["cast", "call", cwusdc, "balanceOf(address)(uint256)", deployer, "--rpc-url", rpc], allow_fail=True)) or 0, "usdCRaw": first_uint(run(["cast", "call", usdc, "balanceOf(address)(uint256)", deployer, "--rpc-url", rpc], allow_fail=True)) or 0, "xautRaw": first_uint(run(["cast", "call", xaut, "balanceOf(address)(uint256)", deployer, "--rpc-url", rpc], allow_fail=True)) or 0, } gas_price = first_uint(run(["cast", "gas-price", "--rpc-url", rpc], allow_fail=True)) or 0 cw_in = get_amount_in(CANARY_DEBT_RAW, CANARY_POOL_RAW, CANARY_POOL_RAW) cw_reserve_after_in = CANARY_POOL_RAW + cw_in usdc_reserve_after_out = CANARY_POOL_RAW - CANARY_DEBT_RAW cw_out = get_amount_out(CANARY_DEBT_RAW, usdc_reserve_after_out, cw_reserve_after_in) collateral = min_xaut_collateral(CANARY_DEBT_RAW) proof_id = "0x" + hashlib.sha256(f"dbis-engine-x:canary:{deployer}:{CANARY_DEBT_RAW}".encode()).hexdigest() quote_raw = first_uint( run( [ "cast", "call", UNISWAP_V3_QUOTER_V2, "quoteExactInputSingle((address,address,uint256,uint24,uint160))(uint256,uint160,uint32,uint256)", f"({xaut},{usdc},{XAUT_SWAP_RAW},{XAUT_USDC_FEE},0)", "--rpc-url", rpc, ], allow_fail=True, ) ) or 0 min_out = quote_raw * 97 // 100 deploy_gas = 0 if not args.skip_deploy_gas_estimate: project_root = Path(__file__).resolve().parents[2] smom = project_root / "smom-dbis-138" run( ["bash", "scripts/forge/scope.sh", "build", "flash", "--contracts", "contracts/flash/DBISEngineXVirtualBatchVault.sol"], cwd=str(smom), allow_fail=False, ) artifact = smom / "out/scopes/flash/DBISEngineXVirtualBatchVault.sol/DBISEngineXVirtualBatchVault.json" bytecode = json.loads(artifact.read_text())["bytecode"]["object"] deploy_gas = first_uint( run( [ "cast", "estimate", "--rpc-url", rpc, "--from", deployer, "--create", bytecode, "constructor(address,address,address,address,address,uint256,uint256,uint256)", cwusdc, usdc, xaut, deployer, surplus_receiver, str(XAUT_USD_PRICE6), str(LTV_BPS), str(MAX_ROUND_TRIP_LOSS_BPS), ], allow_fail=True, ) ) or 0 usdc_required = CANARY_POOL_RAW + CANARY_LENDER_RAW needs_usdc_swap = balances["usdCRaw"] < usdc_required estimated_post_deploy_gas = 60_000 + 60_000 + 130_000 + 80_000 + 280_000 estimated_swap_gas = 250_000 if needs_usdc_swap else 0 gas_units = deploy_gas + estimated_post_deploy_gas + estimated_swap_gas estimated_cost = gas_units * gas_price min_eth_reserve = int(Decimal(os.environ.get("DBIS_ENGINE_X_MIN_ETH_RESERVE", "0.002")) * Decimal(10**18)) blockers: list[str] = [] warnings: list[str] = [] if not private_key_present: blockers.append("PRIVATE_KEY is not loaded; broadcast is impossible") if balances["cWUSDCRaw"] < CANARY_POOL_RAW + cw_in: blockers.append("deployer cWUSDC is below canary seed plus canary input") if balances["xautRaw"] < collateral + (XAUT_SWAP_RAW if needs_usdc_swap else 0): blockers.append("deployer XAUt is below canary collateral plus proposed XAUt->USDC swap") if needs_usdc_swap and quote_raw < usdc_required: blockers.append("quoted XAUt->USDC output is below canary USDC seed requirement") if balances["ethWei"] < estimated_cost + min_eth_reserve: blockers.append("deployer ETH is below estimated canary path cost plus reserve") if needs_usdc_swap: warnings.append("wallet USDC is zero/insufficient; canary requires a small XAUt->USDC swap before seed/fund") if not os.environ.get("ETHERSCAN_API_KEY"): warnings.append("ETHERSCAN_API_KEY is missing; source verification will need a manual follow-up") payload = { "schema": "dbis-engine-x-virtual-batch-canary-preflight/v1", "generatedAt": datetime.now(timezone.utc).replace(microsecond=0).isoformat(), "classification": "read_only_no_broadcast", "deployer": deployer, "tokens": {"cWUSDC": cwusdc, "USDC": usdc, "XAUt": xaut}, "constructor": { "owner": deployer, "surplusReceiver": surplus_receiver, "xautUsdPrice6": str(XAUT_USD_PRICE6), "ltvBps": str(LTV_BPS), "maxRoundTripLossBps": str(MAX_ROUND_TRIP_LOSS_BPS), }, "receivers": {"surplusReceiver": surplus_receiver, "roundingReceiver": rounding_receiver}, "balances": { **{k: str(v) for k, v in balances.items()}, "eth": eth(balances["ethWei"]), "cWUSDC": human(balances["cWUSDCRaw"]), "USDC": human(balances["usdCRaw"]), "XAUt": human(balances["xautRaw"]), "gasPriceWei": str(gas_price), }, "canary": { "poolSeedRaw": str(CANARY_POOL_RAW), "poolSeed": human(CANARY_POOL_RAW), "lenderSeedRaw": str(CANARY_LENDER_RAW), "lenderSeed": human(CANARY_LENDER_RAW), "debtRaw": str(CANARY_DEBT_RAW), "debt": human(CANARY_DEBT_RAW), "cwusdcInRaw": str(cw_in), "cwusdcIn": human(cw_in), "cwusdcOutRaw": str(cw_out), "cwusdcOut": human(cw_out), "exactOutputRaw": str(cw_out), "exactOutput": human(cw_out), "collateralXautRaw": str(collateral), "collateralXaut": human(collateral), "proofId": proof_id, }, "xautToUsdcPrep": { "needed": needs_usdc_swap, "router": UNISWAP_SWAP_ROUTER_02, "quoter": UNISWAP_V3_QUOTER_V2, "fee": XAUT_USDC_FEE, "xautInRaw": str(XAUT_SWAP_RAW), "xautIn": human(XAUT_SWAP_RAW), "quotedUsdcOutRaw": str(quote_raw), "quotedUsdcOut": human(quote_raw), "minUsdcOutRaw": str(min_out), "minUsdcOut": human(min_out), }, "gas": { "deployGasEstimate": str(deploy_gas), "estimatedPostDeployGas": str(estimated_post_deploy_gas), "estimatedSwapGas": str(estimated_swap_gas), "estimatedTotalGasUnits": str(gas_units), "estimatedCostWei": str(estimated_cost), "estimatedCostEth": eth(estimated_cost), "minimumEthReserveWei": str(min_eth_reserve), "minimumEthReserve": eth(min_eth_reserve), }, "blockers": blockers, "warnings": warnings, } payload["proofArtifacts"] = write_canary_artifacts(Path(args.artifact_prefix), payload) payload["operatorCommands"] = { "note": "Review-only. These commands intentionally omit --broadcast orchestration.", "deploy": ( "forge create --broadcast --verify --rpc-url \"$ETHEREUM_MAINNET_RPC\" --private-key \"$PRIVATE_KEY\" " "contracts/flash/DBISEngineXVirtualBatchVault.sol:DBISEngineXVirtualBatchVault " f"--constructor-args {cwusdc} {usdc} {xaut} {deployer} {surplus_receiver} " f"{XAUT_USD_PRICE6} {LTV_BPS} {MAX_ROUND_TRIP_LOSS_BPS}" ), "xautApproveForSwap": f'cast send {xaut} "approve(address,uint256)" {UNISWAP_SWAP_ROUTER_02} {XAUT_SWAP_RAW} --rpc-url "$ETHEREUM_MAINNET_RPC" --private-key "$PRIVATE_KEY"', "xautSwapForUsdc": ( f'cast send {UNISWAP_SWAP_ROUTER_02} ' '"exactInputSingle((address,address,uint24,address,uint256,uint256,uint256,uint160))(uint256)" ' f'"({xaut},{usdc},{XAUT_USDC_FEE},{deployer},$(($(date +%s)+600)),{XAUT_SWAP_RAW},{min_out},0)" ' '--rpc-url "$ETHEREUM_MAINNET_RPC" --private-key "$PRIVATE_KEY"' ), "approveSeedTokens": [ f'cast send {cwusdc} "approve(address,uint256)" "$DBIS_ENGINE_X_VIRTUAL_BATCH_VAULT" {CANARY_POOL_RAW + cw_in} --rpc-url "$ETHEREUM_MAINNET_RPC" --private-key "$PRIVATE_KEY"', f'cast send {usdc} "approve(address,uint256)" "$DBIS_ENGINE_X_VIRTUAL_BATCH_VAULT" {usdc_required} --rpc-url "$ETHEREUM_MAINNET_RPC" --private-key "$PRIVATE_KEY"', f'cast send {xaut} "approve(address,uint256)" "$DBIS_ENGINE_X_VIRTUAL_BATCH_VAULT" {collateral} --rpc-url "$ETHEREUM_MAINNET_RPC" --private-key "$PRIVATE_KEY"', ], "seedAndFund": [ f'cast send "$DBIS_ENGINE_X_VIRTUAL_BATCH_VAULT" "seedPool(uint256,uint256)" {CANARY_POOL_RAW} {CANARY_POOL_RAW} --rpc-url "$ETHEREUM_MAINNET_RPC" --private-key "$PRIVATE_KEY"', f'cast send "$DBIS_ENGINE_X_VIRTUAL_BATCH_VAULT" "fundLender(uint256)" {CANARY_LENDER_RAW} --rpc-url "$ETHEREUM_MAINNET_RPC" --private-key "$PRIVATE_KEY"', ], "runExactOutputCanary": ( 'cast send "$DBIS_ENGINE_X_VIRTUAL_BATCH_VAULT" ' '"runVirtualProofExactOutTo(bytes32,uint256,uint256,address,uint256,address,bytes32,bytes32,bytes32)" ' f'{proof_id} {CANARY_DEBT_RAW} 1 {deployer} {cw_out} {rounding_receiver} ' '"$DBIS_ENGINE_X_CANARY_ISO20022_HASH" "$DBIS_ENGINE_X_CANARY_AUDIT_HASH" "$DBIS_ENGINE_X_CANARY_PEG_HASH" ' '--rpc-url "$ETHEREUM_MAINNET_RPC" --private-key "$PRIVATE_KEY"' ), } json_path = Path(args.json_out) md_path = Path(args.md_out) json_path.parent.mkdir(parents=True, exist_ok=True) json_path.write_text(json.dumps(payload, indent=2) + "\n") md_path.write_text(render_markdown(payload) + "\n") print(json.dumps({"json": str(json_path), "markdown": str(md_path), "blockers": blockers, "warnings": warnings}, indent=2)) return 0 def render_markdown(payload: dict) -> str: lines = [ "# DBIS Engine X Virtual Batch Canary Preflight", "", f"Generated: `{payload['generatedAt']}`", "", "Classification: read-only preflight. No transaction was broadcast.", "", "## Summary", "", f"- Deployer: `{payload['deployer']}`", f"- ETH: `{payload['balances']['eth']}`", f"- cWUSDC: `{payload['balances']['cWUSDC']}`", f"- USDC: `{payload['balances']['USDC']}`", f"- XAUt: `{payload['balances']['XAUt']}`", f"- Estimated full canary path cost: `{payload['gas']['estimatedCostEth']} ETH`", f"- Minimum ETH reserve: `{payload['gas']['minimumEthReserve']} ETH`", "", "## Canary", "", f"- Seed: `{payload['canary']['poolSeed']} cWUSDC / {payload['canary']['poolSeed']} USDC`", f"- Lender: `{payload['canary']['lenderSeed']} USDC`", f"- Debt: `{payload['canary']['debt']} USDC`", f"- cWUSDC in/out: `{payload['canary']['cwusdcIn']} / {payload['canary']['cwusdcOut']}`", f"- XAUt collateral: `{payload['canary']['collateralXaut']} XAUt`", f"- Proof ID: `{payload['canary']['proofId']}`", "", "## Canary Proof Hashes", "", f"- ISO 20022-style XML: `{payload['proofArtifacts']['DBIS_ENGINE_X_CANARY_ISO20022_HASH']}`", f"- Audit envelope: `{payload['proofArtifacts']['DBIS_ENGINE_X_CANARY_AUDIT_HASH']}`", f"- Peg proof: `{payload['proofArtifacts']['DBIS_ENGINE_X_CANARY_PEG_HASH']}`", "", "## Blockers", "", ] if payload["blockers"]: lines.extend([f"- {item}" for item in payload["blockers"]]) else: lines.append("- No read-only preflight blockers found.") lines.extend(["", "## Warnings", ""]) if payload["warnings"]: lines.extend([f"- {item}" for item in payload["warnings"]]) else: lines.append("- None.") lines.extend( [ "", "## Notes", "", "- Wallet USDC must be present before seed/fund, or the tiny quoted XAUt->USDC prep swap must be broadcast first.", "- The canary proves the new virtual-batch exact-output and hash-anchor path, not public DEX peg repair.", "- Keep the 5B proof hashes separate from these canary hashes.", ] ) return "\n".join(lines) if __name__ == "__main__": raise SystemExit(main())