Files
proxmox/scripts/verify/preflight-dbis-engine-x-virtual-batch-canary.py
defiQUG 55b3a1ed9e
All checks were successful
Deploy to Phoenix / validate (push) Successful in 1m16s
Deploy to Phoenix / deploy (push) Successful in 44s
Deploy to Phoenix / deploy-atomic-swap-dapp (push) Successful in 1m28s
phoenix-deploy Deployed to cloudflare-sync
Deploy to Phoenix / cloudflare (push) Successful in 39s
Add Engine X recipient proof package
2026-05-07 07:09:43 -07:00

457 lines
21 KiB
Python

#!/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())