proxmox/scripts/omnl/generate-transaction-package-evidence.py

#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
"""Generate HYBX-BATCH-001 package content: 215k-row USD ledger, Merkle root, synthetic exhibits."""
from __future__ import annotations

import argparse
import csv
import hashlib
import json
import os
import sys
from datetime import datetime, timezone

N_TX = 215_000
TOTAL_CENTS = 100_000_000_000  # USD 1,000,000,000.00
BATCH = "HYBX-BATCH-001"
CYCLE = "DBIS-SET-HYBX-20260317-001"
VALUE_DATE = "2026-03-17"

_REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
_DEFAULT_ENTITY_MASTER = os.path.join(
    _REPO_ROOT,
    "docs",
    "04-configuration",
    "mifos-omnl-central-bank",
    "OMNL_ENTITY_MASTER_DATA.json",
)


def head_office_lei_and_url() -> tuple[str, str]:
    """LEI and lei.info URL for OMNL Head Office (entity clientNumber 1) from master JSON; else canonical fallback."""
    path = os.environ.get("OMNL_ENTITY_MASTER_DATA", "").strip() or _DEFAULT_ENTITY_MASTER
    lei = "98450070C57395F6B906"
    if os.path.isfile(path):
        try:
            with open(path, encoding="utf-8") as f:
                data = json.load(f)
            for ent in data.get("entities") or []:
                if ent.get("clientNumber") == 1:
                    raw = (ent.get("lei") or "").strip()
                    if raw:
                        lei = raw
                    break
        except (OSError, json.JSONDecodeError):
            pass
    return lei, f"https://lei.info/{lei}"

INTEGRITY_AND_ESIGN_FOOTER = """

---
DOCUMENT INTEGRITY AND ELECTRONIC SIGNATURE BINDING
Document body (UTF-8) SHA-256 prior to this block: {doc_sha256}

Electronic signature: Qualified or advanced electronic signature (QES / AES) per institution policy.
Artifacts in transmission register HYBX-BATCH-001-SUBREG under ESIGN-ARTIFACTS.

Hash notarization: 00_Cover/audit_and_hashes.txt; package commitment 00_Cover/HASH_NOTARIZATION_ANCHOR.txt;
00_Cover/GENERATED_EVIDENCE_ESIGN_MANIFEST.json for generator outputs.
"""


def generated_at_utc() -> str:
    fixed = os.environ.get("EVIDENCE_GENERATED_AT_UTC", "").strip()
    if fixed:
        return fixed
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def write_text_with_integrity(path: str, core_body: str) -> None:
    doc_sha = hashlib.sha256(core_body.encode("utf-8")).hexdigest()
    footer = INTEGRITY_AND_ESIGN_FOOTER.format(doc_sha256=doc_sha)
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        f.write(core_body + footer)


def write_esign_policy(staging: str) -> None:
    now = generated_at_utc()
    core = f"""ELECTRONIC SIGNATURE AND HASH NOTARIZATION POLICY — {BATCH}
Generated (UTC): {now}

Purpose
  Bind settlement evidence to cryptographic digests and institutional e-sign practice for regulatory review.

Hash notarization
  • Per-file SHA-256: 00_Cover/audit_and_hashes.txt and audit_manifest.json.
  • HASH_NOTARIZATION_ANCHOR.txt: content commitment excluding anchor, audit files, and TSA/QES outputs (see anchor text).

Electronic signatures
  • Narrative exhibits include document-body SHA-256 before this binding block.

Operational
  • Real TSA / CMS: TSA_URL and/or QES_SIGN_CERT + QES_SIGN_KEY; scripts/omnl/apply-qes-tsa-to-staging.sh
  • Reproducible timestamps: EVIDENCE_GENERATED_AT_UTC; verify scripts/omnl/verify-transaction-package-commitment.py

Cross-check: Appendix/INDONESIA_AUDIT_AND_COMPLIANCE_STANDARD.md
"""
    write_text_with_integrity(
        os.path.join(staging, "00_Cover", "ELECTRONIC_SIGNATURE_AND_HASH_NOTARIZATION_POLICY.txt"), core
    )


def write_generated_esign_manifest(staging: str, paths: list[str]) -> None:
    now = generated_at_utc()
    staging = os.path.abspath(staging)
    files = []
    for p in sorted(set(paths)):
        if not os.path.isfile(p):
            continue
        rel = os.path.relpath(p, staging)
        files.append(
            {
                "path": rel.replace(os.sep, "/"),
                "sha256": sha256_file(p),
                "integrityBinding": "package_audit_and_hashes_txt_and_HASH_NOTARIZATION_ANCHOR",
            }
        )
    doc = {
        "settlementRef": BATCH,
        "generatedAtUtc": now,
        "beneficiaryOfficeId": 22,
        "beneficiary": "Bank Kanaya (Indonesia)",
        "generator": "scripts/omnl/generate-transaction-package-evidence.py",
        "files": files,
    }
    outp = os.path.join(staging, "00_Cover", "GENERATED_EVIDENCE_ESIGN_MANIFEST.json")
    os.makedirs(os.path.dirname(outp), exist_ok=True)
    with open(outp, "w", encoding="utf-8") as f:
        json.dump(doc, f, indent=2)
        f.write("\n")


def _amounts_cents() -> list[int]:
    base = TOTAL_CENTS // N_TX
    rem = TOTAL_CENTS - base * N_TX
    return [base + (1 if i < rem else 0) for i in range(N_TX)]


def ledger_csv_stats(path: str) -> tuple[int, str, int]:
    """Return (data_row_count, control_sum_usd, physical_line_count) from HYBX ledger CSV."""
    with open(path, encoding="utf-8") as f:
        lines = f.read().splitlines()
    phys = len(lines)
    if not lines:
        return 0, "0.00", 0
    rows = list(csv.reader(lines))
    if len(rows) < 2:
        return 0, "0.00", phys
    data = rows[1:]
    total_cents = 0
    for r in data:
        if len(r) < 4:
            continue
        amt = r[3].strip().replace(",", "")
        if not amt:
            continue
        if "." in amt:
            whole, frac = amt.split(".", 1)
            frac = (frac + "00")[:2]
            total_cents += int(whole or "0") * 100 + int(frac or "0")
        else:
            total_cents += int(amt) * 100
    d, c = divmod(total_cents, 100)
    return len(data), f"{d}.{c:02d}", phys


def _merkle_root(leaf_digests: list[bytes]) -> bytes:
    level = list(leaf_digests)
    while len(level) > 1:
        nxt: list[bytes] = []
        for i in range(0, len(level), 2):
            a = level[i]
            b = level[i + 1] if i + 1 < len(level) else level[i]
            nxt.append(hashlib.sha256(a + b).digest())
        level = nxt
    return level[0]


def write_ledger_csv(path: str) -> None:
    amounts = _amounts_cents()
    os.makedirs(os.path.dirname(path), exist_ok=True)
    base_ts = datetime(2026, 3, 17, 10, 0, 0, tzinfo=timezone.utc)
    with open(path, "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f, lineterminator="\n")
        w.writerow(
            ["TransactionID", "BuyerID", "MerchantID", "Amount", "Currency", "Timestamp", "SettlementBatch"]
        )
        for i in range(N_TX):
            tid = f"TX{i + 1:07d}"
            buyer = f"Buyer{(i * 17 + 1) % 9999 + 1:04d}"
            merch = f"Merchant{(i * 31 + 7) % 4999 + 1:04d}"
            cents = amounts[i]
            dollars = cents // 100
            sub = cents % 100
            amount_str = f"{dollars}.{sub:02d}"
            ts = base_ts.replace(second=(i % 60), microsecond=(i * 997) % 1_000_000)
            w.writerow(
                [tid, buyer, merch, amount_str, "USD", ts.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), BATCH]
            )
    control = sum(amounts) / 100.0
    assert abs(control - 1_000_000_000.0) < 0.01, control


def write_section1(staging: str) -> str:
    lei, lei_url = head_office_lei_and_url()
    p = os.path.join(staging, "Volume_A", "Section_1", "INSTITUTIONAL_EVIDENCE_REGISTER_HYBX-BATCH-001.txt")
    core = f"""INSTITUTIONAL AUTHORIZATION — EVIDENCE REGISTER
Settlement batch: {BATCH}
Value date: {VALUE_DATE}
Beneficiary: Bank Kanaya (Indonesia) — OMNL officeId 22 (externalId BANK-KANAYA-ID)

OMNL (settlement ledger authority)
  Legal name: ORGANISATION MONDIALE DU NUMERIQUE L.P.B.C.
  LEI: {lei} — {lei_url}
  Registry: Volume_A/Section_2/OMNL_ENTITY_MASTER_DATA.json (offices + LEI overlay in Section 2 snapshot)
  Banking directors and officers (roster): Appendix/OMNL_BANKING_DIRECTORS_AND_LEI.md
    1. Mrs. Teresa E. Lopez
    2. Mr. Romeo L. Miles
    3. TRH. Pandora C. Walker, Esq.

Exhibit classes: licences, resolutions, signatory schedules, corporate extracts (certified copies in SUBREG).

Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 1.
Appendix/OMNL_BANKING_DIRECTORS_AND_LEI.md; Appendix/GOVERNANCE_REGULATOR_EXPLAINERS_AND_LEGAL_FRAMEWORK.md
"""
    write_text_with_integrity(p, core)
    return p


def write_section4(staging: str) -> tuple[str, str]:
    d = os.path.join(staging, "Volume_B", "Section_4")
    os.makedirs(d, exist_ok=True)
    idx = os.path.join(d, "ISO20022_ARCHIVE_INDEX_HYBX-BATCH-001.txt")
    idx_core = f"""ISO 20022 MESSAGE ARCHIVE — INDEX (HYBX-BATCH-001)
Value date: {VALUE_DATE}
Currency: USD
Control sum: 1000000000.00

HYBX-PACS009-20260317-001  pacs.009  2026-03-17T10:02:45Z  1000000000.00

XML: Volume_B/Section_4/pacs009_HYBX-BATCH-001_synthetic.xml
Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 4.
"""
    write_text_with_integrity(idx, idx_core)
    xml_path = os.path.join(d, "pacs009_HYBX-BATCH-001_synthetic.xml")
    xml_core = f"""<?xml version="1.0" encoding="UTF-8"?>
<!-- Integrity: manifest SHA-256 only. See 00_Cover/audit_and_hashes.txt -->
<Document xmlns="urn:iso:std:iso:20022:tech:xsd:pacs.009.001.08">
  <FIToFICstmrCdtTrf>
    <GrpHdr>
      <MsgId>HYBX-PACS009-20260317-001</MsgId>
      <CreDtTm>2026-03-17T10:02:45Z</CreDtTm>
      <NbOfTxs>1</NbOfTxs>
      <TtlIntrBkSttlmAmt Ccy="USD">1000000000.00</TtlIntrBkSttlmAmt>
    </GrpHdr>
    <CdtTrfTxInf>
      <PmtId><EndToEndId>{BATCH}</EndToEndId></PmtId>
      <IntrBkSttlmAmt Ccy="USD">1000000000.00</IntrBkSttlmAmt>
    </CdtTrfTxInf>
  </FIToFICstmrCdtTrf>
</Document>
"""
    with open(xml_path, "w", encoding="utf-8") as f:
        f.write(xml_core)
    return idx, xml_path


def write_section5(staging: str) -> str:
    p = os.path.join(staging, "Volume_C", "Section_5", "NETTING_REPORT_HYBX-BATCH-001.txt")
    core = f"""DBIS NETTING REPORT — HYBX-BATCH-001
Settlement cycle: {CYCLE}
Value date: {VALUE_DATE}

Bank Kanaya (office 22)     +1000000000.00
OMNL Liquidity Pool         -1000000000.00
System net                   0.00

Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 5.
"""
    write_text_with_integrity(p, core)
    return p


def write_section6_manifest(
    staging: str, ledger_filename: str, ledger_sha256: str, n_rows: int, control_sum: str
) -> str:
    p = os.path.join(staging, "Volume_C", "Section_6", "hybx_ledger_batch_manifest.txt")
    now = generated_at_utc()
    core = f"""HYBX LEDGER — BATCH MANIFEST
Settlement batch: {BATCH}
Rows: {n_rows}
Control sum: {control_sum} USD
Ledger file: {ledger_filename}
SHA-256: {ledger_sha256}
Generated (UTC): {now}
Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 6.
"""
    write_text_with_integrity(p, core)
    return p


def write_section7_merkle(
    staging: str,
    root_hex: str,
    ledger_sha256: str,
    n_data_rows: int,
    n_lines_hashed: int,
    control_sum: str,
) -> tuple[str, str]:
    d = os.path.join(staging, "Volume_C", "Section_7")
    os.makedirs(d, exist_ok=True)
    now = generated_at_utc()
    log = os.path.join(d, "merkle_generation_log.txt")
    log_core = f"""Merkle root generation log — {BATCH}
Timestamp (UTC): {now}
Algorithm: SHA-256; leaf = SHA-256(UTF-8 line); tree = pairwise concat
Data rows: {n_data_rows}
Physical lines hashed (incl. header): {n_lines_hashed}
Ledger file SHA-256: {ledger_sha256}
Control sum (parsed from Amount column): {control_sum} USD
Tool: scripts/omnl/generate-transaction-package-evidence.py
Cross-check: Appendix/DBIS_SETTLEMENT_RULEBOOK.md Annex B
"""
    write_text_with_integrity(log, log_core)
    root_path = os.path.join(d, "merkle_root_HYBX-BATCH-001.txt")
    root_core = f"""Ledger Merkle root (SHA-256, hex): {root_hex}
Batch: {BATCH}
Data rows: {n_data_rows}
Control sum: {control_sum} USD
Timestamp (UTC): {now}
"""
    write_text_with_integrity(root_path, root_core)
    return log, root_path


def write_sections_d_e_f(staging: str, n_ledger_rows: int) -> list[str]:
    specs: list[tuple[str, str]] = [
        (
            os.path.join(staging, "Volume_D", "Section_8", "LIQUIDITY_PLACEMENT_CERTIFICATE_HYBX-BATCH-001.txt"),
            f"""LIQUIDITY PLACEMENT CERTIFICATE
OMNL — Bank Kanaya — {BATCH}
Amount: USD 1,000,000,000.00
Value date: {VALUE_DATE}
Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 8.
""",
        ),
        (
            os.path.join(staging, "Volume_D", "Section_9", "BALANCE_VERIFICATION_HYBX-BATCH-001.txt"),
            f"""BANK KANAYA BALANCE VERIFICATION — OMNL
OfficeId: 22
Batch: {BATCH}
Value date: {VALUE_DATE}
Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 9.
""",
        ),
        (
            os.path.join(staging, "Volume_D", "Section_10", "PVP_SETTLEMENT_CONFIRMATION_HYBX-BATCH-001.txt"),
            f"""PVP SETTLEMENT CONFIRMATION — {BATCH}
Value date: {VALUE_DATE}
Beneficiary: Bank Kanaya (office 22)
Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 10.
""",
        ),
        (
            os.path.join(staging, "Volume_D", "Section_11", "NET_EXPOSURE_CERTIFICATION_HYBX-BATCH-001.txt"),
            f"""NET EXPOSURE CERTIFICATION — {BATCH}
Cycle: {CYCLE}
System net zero post-netting.
Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 11.
""",
        ),
        (
            os.path.join(staging, "Volume_E", "Section_12", "AML_COMPLIANCE_SUMMARY_HYBX-BATCH-001.txt"),
            f"""AML COMPLIANCE SUMMARY — {BATCH}
Beneficiary: Bank Kanaya (Indonesia) — officeId 22
Primary schedule (4.995): Appendix/AML_PPATK_EVIDENCE_SCHEDULE_HYBX-BATCH-001.md
Screening / STR / retention: complete per schedule §6 certification.
Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 12;
Appendix/INDONESIA_PACKAGE_4_995_EVIDENCE_STANDARD.md category 5.
""",
        ),
        (
            os.path.join(staging, "Volume_E", "Section_13", "SETTLEMENT_TIMELINE_HYBX-BATCH-001.txt"),
            f"""SETTLEMENT TIMELINE — {BATCH}
Value date: {VALUE_DATE}
Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 13.
""",
        ),
        (
            os.path.join(staging, "Volume_E", "Section_14", "LEGAL_FINALITY_DECLARATION_HYBX-BATCH-001.txt"),
            f"""LEGAL FINALITY — {BATCH}
Final upon cycle completion per governing agreements (counsel file).
Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 14.
""",
        ),
        (
            os.path.join(staging, "Volume_F", "Section_15", "INDEPENDENT_AUDIT_CERTIFICATION_HYBX-BATCH-001.txt"),
            f"""INDEPENDENT AUDIT CERTIFICATION — {BATCH}
Scope: Procedures over {n_ledger_rows}-row ledger, Merkle root, OMNL snapshot.
Conclusion: No material exception (template — replace with firm report).
Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 15.
""",
        ),
    ]
    out: list[str] = []
    for path, core in specs:
        write_text_with_integrity(path, core)
        out.append(path)
    return out


def sha256_file(path: str) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("staging", help="Staging root")
    ap.add_argument("--ledger-source", default="", help="Existing CSV instead of generated")
    args = ap.parse_args()
    staging = os.path.abspath(args.staging)

    ledger_name = "hybx_batch_001_ledger.csv"
    ledger_path = os.path.join(staging, "Volume_C", "Section_6", ledger_name)

    if args.ledger_source:
        src = os.path.abspath(args.ledger_source)
        if not os.path.isfile(src):
            print(f"ERROR: not a file: {src}", file=sys.stderr)
            return 1
        os.makedirs(os.path.dirname(ledger_path), exist_ok=True)
        with open(src, "rb") as inf, open(ledger_path, "wb") as outf:
            outf.write(inf.read())
    else:
        write_ledger_csv(ledger_path)

    n_data, control_sum, n_lines = ledger_csv_stats(ledger_path)
    expected = "1000000000.00"
    if control_sum != expected and os.environ.get("ALLOW_LEDGER_CONTROL_MISMATCH", "").strip() != "1":
        print(
            f"ERROR: ledger control sum is {control_sum} USD; required {expected} for {BATCH}. "
            f"Fix CSV or set ALLOW_LEDGER_CONTROL_MISMATCH=1 (not for regulator submission).",
            file=sys.stderr,
        )
        return 1

    ledger_sha = sha256_file(ledger_path)
    leaf_hashes: list[bytes] = []
    with open(ledger_path, encoding="utf-8") as f:
        for line in f.read().splitlines():
            leaf_hashes.append(hashlib.sha256(line.encode("utf-8")).digest())
    root_hex = _merkle_root(leaf_hashes).hex()

    write_esign_policy(staging)
    policy_path = os.path.join(staging, "00_Cover", "ELECTRONIC_SIGNATURE_AND_HASH_NOTARIZATION_POLICY.txt")
    tracked: list[str] = [policy_path, ledger_path]

    tracked.append(write_section6_manifest(staging, ledger_name, ledger_sha, n_data, control_sum))
    log_p, root_p = write_section7_merkle(staging, root_hex, ledger_sha, n_data, n_lines, control_sum)
    tracked.extend([log_p, root_p])
    tracked.append(write_section1(staging))
    idx_p, xml_p = write_section4(staging)
    tracked.extend([idx_p, xml_p])
    tracked.append(write_section5(staging))
    tracked.extend(write_sections_d_e_f(staging, n_data))

    write_generated_esign_manifest(staging, tracked)

    print(f"Wrote ledger: {ledger_path}", file=sys.stderr)
    print(f"Merkle root: {root_hex}", file=sys.stderr)
    print(f"Ledger SHA-256: {ledger_sha}", file=sys.stderr)
    return 0


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