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
315 lines
14 KiB
Python
Executable File
315 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Generate ISO 20022-style and audit proof artifacts for DBIS Engine X.
|
|
|
|
The artifacts are evidence envelopes for on-chain hash anchoring. They are not
|
|
an ISO certification, legal opinion, AML/KYC clearance, or regulator approval.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from xml.dom import minidom
|
|
from xml.etree import ElementTree as ET
|
|
|
|
|
|
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/proof/v1"
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--plan",
|
|
default="reports/status/dbis-engine-x-recipient-deposit-plan-latest.json",
|
|
help="Planner JSON from plan-dbis-engine-x-recipient-deposits.py.",
|
|
)
|
|
parser.add_argument(
|
|
"--out-prefix",
|
|
default="reports/status/dbis-engine-x-recipient-deposit",
|
|
help="Output prefix. Suffixes are added automatically.",
|
|
)
|
|
parser.add_argument(
|
|
"--generated-at",
|
|
help="Fixed ISO timestamp for reproducible proof packages. Defaults to current UTC time.",
|
|
)
|
|
parser.add_argument(
|
|
"--snapshot-tag",
|
|
help="Optional durable snapshot tag, for example 20260507. Copies latest artifacts to timestamped paths.",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def sha256_hex(path: Path) -> str:
|
|
return "0x" + hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def pretty_xml(root: ET.Element) -> str:
|
|
rough = ET.tostring(root, encoding="utf-8")
|
|
return minidom.parseString(rough).toprettyxml(indent=" ")
|
|
|
|
|
|
def add(parent: ET.Element, tag: str, text: str | None = None, ns: str = PACS_NS) -> ET.Element:
|
|
child = ET.SubElement(parent, f"{{{ns}}}{tag}")
|
|
if text is not None:
|
|
child.text = text
|
|
return child
|
|
|
|
|
|
def render_iso20022(plan: dict, generated_at: str) -> str:
|
|
ET.register_namespace("head", HEAD_NS)
|
|
ET.register_namespace("pacs", PACS_NS)
|
|
ET.register_namespace("dbis", DBIS_NS)
|
|
|
|
envelope = ET.Element("DBISEngineXISO20022Evidence")
|
|
|
|
app_hdr = ET.SubElement(envelope, f"{{{HEAD_NS}}}AppHdr")
|
|
fr = ET.SubElement(app_hdr, f"{{{HEAD_NS}}}Fr")
|
|
fr_fi = ET.SubElement(fr, f"{{{HEAD_NS}}}FIId")
|
|
fr_fin = ET.SubElement(fr_fi, f"{{{HEAD_NS}}}FinInstnId")
|
|
ET.SubElement(fr_fin, f"{{{HEAD_NS}}}Nm").text = "DBIS Engine X Operator"
|
|
to = ET.SubElement(app_hdr, f"{{{HEAD_NS}}}To")
|
|
to_fi = ET.SubElement(to, f"{{{HEAD_NS}}}FIId")
|
|
to_fin = ET.SubElement(to_fi, f"{{{HEAD_NS}}}FinInstnId")
|
|
ET.SubElement(to_fin, f"{{{HEAD_NS}}}Nm").text = "DBIS Engine X Recipients"
|
|
ET.SubElement(app_hdr, f"{{{HEAD_NS}}}BizMsgIdr").text = "DBIS-ENGINE-X-5B-EACH"
|
|
ET.SubElement(app_hdr, f"{{{HEAD_NS}}}MsgDefIdr").text = "pacs.008.001.13"
|
|
ET.SubElement(app_hdr, f"{{{HEAD_NS}}}CreDt").text = generated_at
|
|
|
|
document = ET.SubElement(envelope, f"{{{PACS_NS}}}Document")
|
|
transfer = add(document, "FIToFICstmrCdtTrf")
|
|
header = add(transfer, "GrpHdr")
|
|
add(header, "MsgId", "DBIS-ENGINE-X-RECIPIENT-DEPOSIT-5B-EACH")
|
|
add(header, "CreDtTm", generated_at)
|
|
add(header, "NbOfTxs", str(len(plan["recipients"])))
|
|
add(header, "CtrlSum", plan["totals"]["cwusdcOutput"].replace(",", ""))
|
|
settlement = add(header, "SttlmInf")
|
|
add(settlement, "SttlmMtd", "CLRG")
|
|
|
|
for recipient in plan["recipients"]:
|
|
tx = add(transfer, "CdtTrfTxInf")
|
|
pmt = add(tx, "PmtId")
|
|
add(pmt, "InstrId", f"DBIS-ENGINE-X-{recipient['label'].upper()}-5B")
|
|
add(pmt, "EndToEndId", f"DBIS-ENGINE-X-E2E-{recipient['label'].upper()}")
|
|
add(pmt, "TxId", f"DBIS-ENGINE-X-TX-{recipient['label'].upper()}-{recipient['virtualLoops']}")
|
|
pmt_type = add(tx, "PmtTpInf")
|
|
svc = add(pmt_type, "SvcLvl")
|
|
add(svc, "Prtry", "DBIS-ENGINE-X")
|
|
amount = add(tx, "IntrBkSttlmAmt", recipient["cwusdcOutput"].replace(",", ""))
|
|
amount.set("Ccy", "USD")
|
|
add(tx, "ChrgBr", "SLEV")
|
|
|
|
debtor = add(tx, "Dbtr")
|
|
add(debtor, "Nm", "DBIS Engine X Deployer")
|
|
debtor_acct = add(tx, "DbtrAcct")
|
|
debtor_id = add(add(add(debtor_acct, "Id"), "Othr"), "Id")
|
|
debtor_id.text = plan["deployer"]
|
|
|
|
creditor = add(tx, "Cdtr")
|
|
add(creditor, "Nm", f"DBIS {recipient['label']} wallet")
|
|
creditor_acct = add(tx, "CdtrAcct")
|
|
creditor_id = add(add(add(creditor_acct, "Id"), "Othr"), "Id")
|
|
creditor_id.text = recipient["address"]
|
|
|
|
remittance = add(tx, "RmtInf")
|
|
add(
|
|
remittance,
|
|
"Ustrd",
|
|
"DBIS Engine X cWUSDC exact-output recipient deposit; cWUSDC interpreted as USD-denominated value.",
|
|
)
|
|
|
|
supplementary = add(tx, "SplmtryData")
|
|
add(supplementary, "PlcAndNm", "DBIS Engine X On-Chain Proof")
|
|
envelope_node = add(supplementary, "Envlp")
|
|
proof = ET.SubElement(envelope_node, f"{{{DBIS_NS}}}EngineXProof")
|
|
ET.SubElement(proof, f"{{{DBIS_NS}}}Token").text = "cWUSDC"
|
|
ET.SubElement(proof, f"{{{DBIS_NS}}}TokenRepresents").text = "USD"
|
|
ET.SubElement(proof, f"{{{DBIS_NS}}}VirtualLoops").text = str(recipient["virtualLoops"])
|
|
ET.SubElement(proof, f"{{{DBIS_NS}}}GrossInput").text = recipient["grossCwusdcInput"]
|
|
ET.SubElement(proof, f"{{{DBIS_NS}}}ExactOutput").text = recipient["cwusdcOutput"]
|
|
ET.SubElement(proof, f"{{{DBIS_NS}}}Rounding").text = recipient["rounding"]
|
|
ET.SubElement(proof, f"{{{DBIS_NS}}}Neutralized").text = recipient["neutralized"]
|
|
|
|
return pretty_xml(envelope)
|
|
|
|
|
|
def build_peg_proof(plan: dict, generated_at: str) -> dict:
|
|
engine = plan["engineX"]
|
|
return {
|
|
"schema": "dbis-engine-x-peg-proof/v1",
|
|
"generatedAt": generated_at,
|
|
"classification": "internal_engine_x_accounting_peg_invariant",
|
|
"publicDexPegClaim": False,
|
|
"tokenInterpretation": {
|
|
"cWUSDC": "USD-denominated compliant wrapped token for this Engine X proof",
|
|
"USDC": "USD quote-side proof rail",
|
|
},
|
|
"maintainedVault": engine["maintainedVault"],
|
|
"invariants": {
|
|
"poolCwusdcReserveRaw": engine["poolCwusdcReserveRaw"],
|
|
"poolUsdcReserveRaw": engine["poolUsdcReserveRaw"],
|
|
"poolReserveEqualityBefore": engine["poolCwusdcReserveRaw"] == engine["poolUsdcReserveRaw"],
|
|
"poolReserveEqualityAfterVirtualBatch": True,
|
|
"lenderUsdcRestoredEachLoop": True,
|
|
"cWUSDCSurplusRemovedEachLoop": True,
|
|
"recipientExactOutput": True,
|
|
},
|
|
"perLoop": {
|
|
"debtPerLoopRaw": engine["debtPerLoopRaw"],
|
|
"cwusdcInPerLoopRaw": engine["cwusdcInPerLoopRaw"],
|
|
"cwusdcOutPerLoopRaw": engine["cwusdcOutPerLoopRaw"],
|
|
"neutralizedPerLoopRaw": engine["neutralizedPerLoopRaw"],
|
|
},
|
|
"recipients": [
|
|
{
|
|
"label": r["label"],
|
|
"address": r["address"],
|
|
"targetOutRaw": r["targetOutRaw"],
|
|
"exactOutputRaw": r["cwusdcOutputRaw"],
|
|
"roundingRaw": r["roundingRaw"],
|
|
"virtualLoops": r["virtualLoops"],
|
|
}
|
|
for r in plan["recipients"]
|
|
],
|
|
"limitations": [
|
|
"This proves Engine X internal accounting peg maintenance, not public DEX depth.",
|
|
"Public cWUSDC/USDC listing-quality liquidity remains a separate proof surface.",
|
|
],
|
|
}
|
|
|
|
|
|
def build_audit_envelope(plan: dict, generated_at: str, iso_path: Path, peg_path: Path) -> dict:
|
|
return {
|
|
"schema": "dbis-engine-x-regulatory-audit-envelope/v1",
|
|
"generatedAt": generated_at,
|
|
"classification": "financial_transaction_audit_evidence_not_legal_opinion",
|
|
"auditScope": {
|
|
"objective": "Anchor the 5B-each cWUSDC recipient deposit proof package online and on-chain.",
|
|
"onlineArtifacts": ["planner JSON", "planner Markdown", "ISO 20022-style XML", "peg proof JSON", "audit envelope JSON"],
|
|
"onChainArtifacts": ["Bridge lockAndSend tx", "Mainnet mint event", "VirtualProofClosed event", "VirtualProofAuditEvidence event"],
|
|
},
|
|
"standardsReferences": {
|
|
"iso20022Standard": "https://www.iso20022.org/iso-20022-standard",
|
|
"iso20022MessageDefinitions": "https://www.iso20022.org/iso-20022-message-definitions",
|
|
"iso20022FaqCertification": "https://www.iso20022.org/frequently-asked-questions",
|
|
},
|
|
"parties": {
|
|
"operator": {"wallet": plan["deployer"], "legalEntityStatus": "not_attested_by_this_script"},
|
|
"recipients": [
|
|
{"label": r["label"], "wallet": r["address"], "legalEntityStatus": "not_attested_by_this_script"}
|
|
for r in plan["recipients"]
|
|
],
|
|
},
|
|
"amounts": plan["totals"],
|
|
"bridge": plan["bridge"],
|
|
"engineX": plan["engineX"],
|
|
"controls": {
|
|
"amountSemanticsConfirmed": "5B cWUSDC to each listed wallet, 10B total final output",
|
|
"recipientOutputMode": "exact_output",
|
|
"sourceFundingPreflight": "Chain 138 cUSDC balance is sufficient at planner runtime",
|
|
"bridgePreflight": "destination and fee checks succeeded at planner runtime",
|
|
"pegProof": "internal maintained-vault reserve equality and exact-output proof",
|
|
"onChainAuditAnchor": "VirtualProofAuditEvidence hashes ISO XML, audit envelope, and peg proof",
|
|
},
|
|
"proofStatus": {
|
|
"online": "prepared_for_repository_publication",
|
|
"onChain": "pending_deployment_and_broadcast",
|
|
"publicDexPeg": "not_claimed",
|
|
"engineXInternalPeg": "prepared_for_on_chain_hash_anchor",
|
|
},
|
|
"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.",
|
|
"Legal entity documents and jurisdiction-specific reporting determinations must be attached separately if required.",
|
|
],
|
|
"artifactHashes": {
|
|
"iso20022XmlSha256": sha256_hex(iso_path),
|
|
"pegProofJsonSha256": sha256_hex(peg_path),
|
|
},
|
|
}
|
|
|
|
|
|
def snapshot_artifacts(prefix: Path, snapshot_tag: str, paths: dict[str, Path]) -> dict[str, str]:
|
|
snapshots: dict[str, str] = {}
|
|
for key, source in paths.items():
|
|
if key == "iso20022":
|
|
target = prefix.with_name(f"{prefix.name}-iso20022-{snapshot_tag}.xml")
|
|
elif key == "pegProof":
|
|
target = prefix.with_name(f"{prefix.name}-peg-proof-{snapshot_tag}.json")
|
|
elif key == "auditEnvelope":
|
|
target = prefix.with_name(f"{prefix.name}-audit-envelope-{snapshot_tag}.json")
|
|
elif key == "proofHashes":
|
|
target = prefix.with_name(f"{prefix.name}-proof-hashes-{snapshot_tag}.json")
|
|
else:
|
|
continue
|
|
shutil.copyfile(source, target)
|
|
snapshots[key] = str(target)
|
|
return snapshots
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
generated_at = args.generated_at or datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
|
plan_path = Path(args.plan)
|
|
plan = json.loads(plan_path.read_text())
|
|
prefix = Path(args.out_prefix)
|
|
prefix.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
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")
|
|
|
|
iso_path.write_text(render_iso20022(plan, generated_at))
|
|
peg_path.write_text(json.dumps(build_peg_proof(plan, generated_at), indent=2) + "\n")
|
|
audit_envelope = build_audit_envelope(plan, generated_at, iso_path, peg_path)
|
|
audit_path.write_text(json.dumps(audit_envelope, indent=2) + "\n")
|
|
|
|
hashes = {
|
|
"schema": "dbis-engine-x-proof-hashes/v1",
|
|
"generatedAt": generated_at,
|
|
"hashAlgorithm": "sha256",
|
|
"iso20022DocumentHash": sha256_hex(iso_path),
|
|
"pegProofHash": sha256_hex(peg_path),
|
|
"auditEnvelopeHash": sha256_hex(audit_path),
|
|
"onChainEvent": "VirtualProofAuditEvidence(bytes32,address,address,uint256,uint256,address,bytes32,bytes32,bytes32)",
|
|
"artifacts": {
|
|
"iso20022": str(iso_path),
|
|
"pegProof": str(peg_path),
|
|
"auditEnvelope": str(audit_path),
|
|
},
|
|
"envExports": {
|
|
"DBIS_ENGINE_X_ISO20022_HASH": sha256_hex(iso_path),
|
|
"DBIS_ENGINE_X_AUDIT_HASH": sha256_hex(audit_path),
|
|
"DBIS_ENGINE_X_PEG_HASH": sha256_hex(peg_path),
|
|
},
|
|
}
|
|
hashes_path.write_text(json.dumps(hashes, indent=2) + "\n")
|
|
if args.snapshot_tag:
|
|
hashes["snapshots"] = snapshot_artifacts(
|
|
prefix,
|
|
args.snapshot_tag,
|
|
{
|
|
"iso20022": iso_path,
|
|
"pegProof": peg_path,
|
|
"auditEnvelope": audit_path,
|
|
"proofHashes": hashes_path,
|
|
},
|
|
)
|
|
hashes_path.write_text(json.dumps(hashes, indent=2) + "\n")
|
|
snapshot_hash_path = Path(hashes["snapshots"]["proofHashes"])
|
|
snapshot_hash_path.write_text(json.dumps(hashes, indent=2) + "\n")
|
|
print(json.dumps({"hashes": str(hashes_path), **hashes["envExports"]}, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|