proxmox/scripts/verify/audit-cusdc-cwusdc-etherscan-feeds.py

#!/usr/bin/env python3
"""Audit Chain 138 cUSDC and Ethereum cWUSDC explorer feeds.

This produces an evidence packet for Etherscan/listing submissions. It does not
ask Etherscan to merge Chain 138 traffic into the Ethereum token tracker; rather,
it documents that Ethereum Mainnet cWUSDC is the wrapped public-network transport
representation of canonical Chain 138 cUSDC and summarizes both API feeds.
"""

from __future__ import annotations

import argparse
import datetime as dt
import json
import os
import sys
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any


CHAIN138_CUSDC = "0xf22258f57794CC8E06237084b353Ab30fFfa640b"
MAINNET_CWUSDC = "0x2de5F116bFcE3d0f922d9C8351e0c5Fc24b9284a"
CHAIN138_EXPLORER_API = "https://explorer.d-bis.org/api/v2"
ETHERSCAN_V2_API = "https://api.etherscan.io/v2/api"
REPORT_BASE = Path("reports/status/cusdc-cwusdc-etherscan-feed-audit-latest")


def fetch_json(url: str, timeout: int = 30) -> Any:
    req = urllib.request.Request(url, headers={"User-Agent": "dbis-cusdc-cwusdc-feed-audit/1.0"})
    with urllib.request.urlopen(req, timeout=timeout) as response:
        payload = response.read().decode("utf-8")
    return json.loads(payload)


def human_units(raw: int, decimals: int) -> str:
    sign = "-" if raw < 0 else ""
    raw = abs(raw)
    whole = raw // (10**decimals)
    frac = str(raw % (10**decimals)).rjust(decimals, "0").rstrip("0")
    return f"{sign}{whole:,}" + (f".{frac}" if frac else "")


def addresses_from_transfer(item: dict[str, Any], style: str) -> set[str]:
    if style == "blockscout":
        values = [
            item.get("from", {}).get("hash"),
            item.get("to", {}).get("hash"),
        ]
    else:
        values = [item.get("from"), item.get("to")]
    return {str(v).lower() for v in values if v}


def summarize_blockscout_transfers(items: list[dict[str, Any]], decimals: int) -> dict[str, Any]:
    total_raw = 0
    addresses: set[str] = set()
    methods: dict[str, int] = {}
    latest = items[0] if items else None
    for item in items:
        value = item.get("total", {}).get("value", "0")
        try:
            total_raw += int(value)
        except (TypeError, ValueError):
            pass
        addresses.update(addresses_from_transfer(item, "blockscout"))
        method = item.get("method") or "unknown"
        methods[method] = methods.get(method, 0) + 1
    return {
        "sample_count": len(items),
        "sample_volume_raw": str(total_raw),
        "sample_volume_units": human_units(total_raw, decimals),
        "unique_addresses_in_sample": len(addresses),
        "method_counts": methods,
        "latest_transfer": {
            "hash": latest.get("transaction_hash"),
            "timestamp": latest.get("timestamp"),
            "from": latest.get("from", {}).get("hash"),
            "to": latest.get("to", {}).get("hash"),
            "value_raw": latest.get("total", {}).get("value"),
            "value_units": human_units(int(latest.get("total", {}).get("value", "0")), decimals),
            "method": latest.get("method"),
        }
        if latest
        else None,
        "addresses": sorted(addresses),
    }


def summarize_etherscan_transfers(items: list[dict[str, Any]], decimals: int) -> dict[str, Any]:
    total_raw = 0
    addresses: set[str] = set()
    methods: dict[str, int] = {}
    latest = items[0] if items else None
    for item in items:
        try:
            total_raw += int(item.get("value", "0"))
        except (TypeError, ValueError):
            pass
        addresses.update(addresses_from_transfer(item, "etherscan"))
        method = item.get("methodId") or item.get("functionName") or "unknown"
        methods[method] = methods.get(method, 0) + 1
    return {
        "sample_count": len(items),
        "sample_volume_raw": str(total_raw),
        "sample_volume_units": human_units(total_raw, decimals),
        "unique_addresses_in_sample": len(addresses),
        "method_counts": methods,
        "latest_transfer": {
            "hash": latest.get("hash"),
            "blockNumber": latest.get("blockNumber"),
            "timeStamp": latest.get("timeStamp"),
            "from": latest.get("from"),
            "to": latest.get("to"),
            "value_raw": latest.get("value"),
            "value_units": human_units(int(latest.get("value", "0")), decimals),
            "methodId": latest.get("methodId"),
            "functionName": latest.get("functionName"),
        }
        if latest
        else None,
        "addresses": sorted(addresses),
    }


def blockscout_token_metadata(address: str) -> dict[str, Any]:
    return fetch_json(f"{CHAIN138_EXPLORER_API}/tokens/{address}")


def blockscout_transfers(address: str, pages: int) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    params: dict[str, Any] | None = None
    for _ in range(pages):
        url = f"{CHAIN138_EXPLORER_API}/tokens/{address}/transfers"
        if params:
            url += "?" + urllib.parse.urlencode(params)
        payload = fetch_json(url)
        items.extend(payload.get("items", []))
        params = payload.get("next_page_params")
        if not params:
            break
    return items


def etherscan_call(params: dict[str, str], api_key: str) -> Any:
    query = {"chainid": "1", **params, "apikey": api_key}
    payload = fetch_json(f"{ETHERSCAN_V2_API}?{urllib.parse.urlencode(query)}")
    if payload.get("status") == "0" and "No transactions found" not in str(payload.get("message")):
        raise RuntimeError(f"Etherscan API error: {payload.get('message')} {payload.get('result')}")
    return payload.get("result", [])


def build_report(args: argparse.Namespace) -> dict[str, Any]:
    api_key = args.etherscan_api_key or os.environ.get("ETHERSCAN_API_KEY", "")
    if not api_key:
        raise SystemExit("ETHERSCAN_API_KEY is required for Ethereum cWUSDC Etherscan API checks")

    c138_meta = blockscout_token_metadata(args.chain138_cusdc)
    c138_decimals = int(c138_meta.get("decimals") or 6)
    c138_transfers = blockscout_transfers(args.chain138_cusdc, args.chain138_pages)

    cw_supply_raw = etherscan_call(
        {
            "module": "stats",
            "action": "tokensupply",
            "contractaddress": args.mainnet_cwusdc,
        },
        api_key,
    )
    cw_transfers = etherscan_call(
        {
            "module": "account",
            "action": "tokentx",
            "contractaddress": args.mainnet_cwusdc,
            "page": "1",
            "offset": str(args.etherscan_offset),
            "sort": "desc",
        },
        api_key,
    )
    if not isinstance(cw_transfers, list):
        cw_transfers = []

    c138_summary = summarize_blockscout_transfers(c138_transfers, c138_decimals)
    cw_summary = summarize_etherscan_transfers(cw_transfers, 6)
    common_addresses = sorted(set(c138_summary["addresses"]) & set(cw_summary["addresses"]))

    c138_summary_public = {k: v for k, v in c138_summary.items() if k != "addresses"}
    cw_summary_public = {k: v for k, v in cw_summary.items() if k != "addresses"}

    return {
        "generatedAt": dt.datetime.now(dt.UTC).isoformat().replace("+00:00", "Z"),
        "purpose": "Evidence packet for Etherscan/listing feeds: Chain 138 cUSDC is the canonical source asset; Ethereum cWUSDC is the wrapped transport representation.",
        "canonicalRelationship": {
            "sourceChainId": 138,
            "sourceToken": {
                "symbol": "cUSDC",
                "name": "USD Coin (Compliant)",
                "address": args.chain138_cusdc,
                "explorer": f"https://explorer.d-bis.org/token/{args.chain138_cusdc}",
                "api": f"{CHAIN138_EXPLORER_API}/tokens/{args.chain138_cusdc}",
            },
            "wrappedChainId": 1,
            "wrappedToken": {
                "symbol": "cWUSDC",
                "name": "Wrapped cUSDC",
                "address": args.mainnet_cwusdc,
                "explorer": f"https://etherscan.io/token/{args.mainnet_cwusdc}",
                "api": ETHERSCAN_V2_API,
            },
            "mappingSource": "config/token-mapping-multichain.json: 138 cUSDC -> Ethereum Mainnet cWUSDC",
            "trackerLanguage": "cWUSDC is the Ethereum Mainnet compliant wrapped transport representation of canonical Chain 138 cUSDC. It is not Circle-issued USDC.",
        },
        "chain138Cusdc": {
            "metadata": {
                "name": c138_meta.get("name"),
                "symbol": c138_meta.get("symbol"),
                "decimals": c138_meta.get("decimals"),
                "holders": c138_meta.get("holders"),
                "totalSupplyRaw": c138_meta.get("total_supply"),
                "totalSupplyUnits": human_units(int(c138_meta.get("total_supply") or 0), c138_decimals),
            },
            "transferFeed": c138_summary_public,
        },
        "mainnetCwusdc": {
            "metadata": {
                "name": "Wrapped cUSDC",
                "symbol": "cWUSDC",
                "decimals": "6",
                "totalSupplyRaw": str(cw_supply_raw),
                "totalSupplyUnits": human_units(int(cw_supply_raw or 0), 6),
            },
            "transferFeed": cw_summary_public,
        },
        "crossFeedSignals": {
            "commonAddressesInRecentSamples": common_addresses,
            "commonAddressCount": len(common_addresses),
            "interpretation": "Common addresses are supporting evidence only. Canonical linkage is established by the token mapping, metadata registry, and bridge/listing documentation; Etherscan itself will only index Ethereum Mainnet cWUSDC traffic for the token page.",
        },
        "etherscanSubmissionNote": "Ask Etherscan to list the Ethereum token as Wrapped cUSDC (cWUSDC), with Chain 138 cUSDC identified as the canonical source asset in the description/supporting links. Do not ask Etherscan to add Chain 138 transfer counts to the Ethereum token tracker totals.",
    }


def write_markdown(report: dict[str, Any], path: Path) -> None:
    rel = report["canonicalRelationship"]
    c138 = report["chain138Cusdc"]
    cw = report["mainnetCwusdc"]
    signals = report["crossFeedSignals"]
    lines = [
        "# cUSDC / cWUSDC Etherscan Feed Audit",
        "",
        f"Generated: `{report['generatedAt']}`",
        "",
        "## Relationship",
        "",
        f"- Source asset: Chain 138 `cUSDC` at `{rel['sourceToken']['address']}`",
        f"- Wrapped transport asset: Ethereum Mainnet `cWUSDC` at `{rel['wrappedToken']['address']}`",
        f"- Mapping source: `{rel['mappingSource']}`",
        f"- Tracker language: {rel['trackerLanguage']}",
        "",
        "## API Feed Summary",
        "",
        "| Feed | Supply | Recent sample transfers | Recent sample volume | Unique addresses in sample |",
        "|---|---:|---:|---:|---:|",
        f"| Chain 138 cUSDC Blockscout | {c138['metadata']['totalSupplyUnits']} | {c138['transferFeed']['sample_count']} | {c138['transferFeed']['sample_volume_units']} | {c138['transferFeed']['unique_addresses_in_sample']} |",
        f"| Ethereum cWUSDC Etherscan | {cw['metadata']['totalSupplyUnits']} | {cw['transferFeed']['sample_count']} | {cw['transferFeed']['sample_volume_units']} | {cw['transferFeed']['unique_addresses_in_sample']} |",
        "",
        "## Latest Transfers",
        "",
        f"- Chain 138 cUSDC latest: `{(c138['transferFeed']['latest_transfer'] or {}).get('hash')}`",
        f"- Ethereum cWUSDC latest: `{(cw['transferFeed']['latest_transfer'] or {}).get('hash')}`",
        "",
        "## Cross-Feed Signal",
        "",
        f"- Common addresses in recent API samples: `{signals['commonAddressCount']}`",
        f"- Interpretation: {signals['interpretation']}",
        "",
        "## Etherscan Submission Note",
        "",
        report["etherscanSubmissionNote"],
        "",
    ]
    path.write_text("\n".join(lines), encoding="utf-8")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--chain138-cusdc", default=CHAIN138_CUSDC)
    parser.add_argument("--mainnet-cwusdc", default=MAINNET_CWUSDC)
    parser.add_argument("--etherscan-api-key", default="")
    parser.add_argument("--chain138-pages", type=int, default=3)
    parser.add_argument("--etherscan-offset", type=int, default=150)
    parser.add_argument("--json-out", default=f"{REPORT_BASE}.json")
    parser.add_argument("--md-out", default=f"{REPORT_BASE}.md")
    args = parser.parse_args()

    report = build_report(args)
    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(report, indent=2) + "\n", encoding="utf-8")
    write_markdown(report, md_path)
    print(f"Wrote {json_path}")
    print(f"Wrote {md_path}")
    return 0


if __name__ == "__main__":
    sys.exit(main())