run_mini_coinjoinxt_study.py

#!/usr/bin/env python3
"""Mini-CoinJoinXT lifecycle study.

Runs the JOINMARKET_MINI_XT protocol through the full ecosystem lifecycle
(Silent Payments -> CoinJoin -> on-chain + Lightning spend) against both the
standard six observers and the off-chain LightningFundingSourceObserver, which
represents the residual attack surface PR #280 identified.

Headline question: how does mini-CoinJoinXT compare to baseline JoinMarket,
fee-quant, and LN-as-change variants (a) against a passive on-chain observer,
(b) against the spend counterparty (CIOH leak), (c) against the LN funding
source (the off-chain residual), and (d) under full collusion?

Also runs the Improvement.MINI_COINJOINXT model through the single-round
comparison framework, showing the realized effective set vs. baseline.

Outputs mini_coinjoinxt_study_results.json.
"""

from __future__ import annotations

import json
import time
from pathlib import Path

from coinjoin_simulator.comparison import run_intersection_decay
from coinjoin_simulator.ecosystem.actors import (
    ActorProfile,
    EcosystemSimulation,
    Hygiene,
    SpendGoal,
)
from coinjoin_simulator.ecosystem.evaluate import evaluate_ecosystem
from coinjoin_simulator.ecosystem.observers import (
    CoinJoinCoParticipant,
    EvilCoordinator,
    LightningAdversary,
    LightningFundingSourceObserver,
    Observer,
    PassiveChainObserver,
    SilentPaymentPayer,
    SpendCounterparty,
)
from coinjoin_simulator.ecosystem.strategies import CoinJoinProtocol
from coinjoin_simulator.improvements import (
    Improvement,
    evaluate_with_improvements,
)
from coinjoin_simulator.protocols import (
    jm_orderbook_params,
    standard_threat_models,
)

OUTPUT_PATH = Path("mini_coinjoinxt_study_results.json")

_PROTOCOLS = (
    CoinJoinProtocol.JOINMARKET,
    CoinJoinProtocol.JOINMARKET_FEEQUANT,
    CoinJoinProtocol.JOINMARKET_LN,
    CoinJoinProtocol.JOINMARKET_MINI_XT,
)
_LABELS = {
    CoinJoinProtocol.JOINMARKET: "JoinMarket (baseline)",
    CoinJoinProtocol.JOINMARKET_FEEQUANT: "JoinMarket + fee quantization",
    CoinJoinProtocol.JOINMARKET_LN: "JoinMarket + LN-as-change",
    CoinJoinProtocol.JOINMARKET_MINI_XT: "JoinMarket mini-CoinJoinXT",
}


def _population(protocol: CoinJoinProtocol, n: int) -> list[ActorProfile]:
    goals = [(SpendGoal.MERCHANT,), (SpendGoal.P2P_FIAT,), (SpendGoal.LIGHTNING,)]
    return [
        ActorProfile(
            entity=f"user_{i:03d}",
            protocol=protocol,
            hygiene=Hygiene.GOOD,
            spend_goals=goals[i % len(goals)],
        )
        for i in range(n)
    ]


def _build_observers(
    sim: EcosystemSimulation,
) -> dict[str, Observer]:
    cps = {c.received_by for c in sim.ledger.coins.values() if c.received_by}
    obs: dict[str, Observer] = {
        "passive": PassiveChainObserver(),
        "sp_payer": SilentPaymentPayer(),
        "spend_cp": SpendCounterparty(known_counterparties=cps),
        "evil_coordinator": EvilCoordinator(),
        "ln_adv": LightningAdversary(probe_private=True),
        # Off-chain observer: LN funding source clustering (the PR #280 residual).
        "ln_funding_src": LightningFundingSourceObserver(),
    }
    if "sybil" in sim.attacker_entities:
        obs["coparticipant"] = CoinJoinCoParticipant(sybil_entities={"sybil"})
    return obs


def run_lifecycle_comparison(
    *,
    n_users: int = 25,
    maker_pool_size: int = 40,
    seed: int = 0,
) -> list[dict[str, object]]:
    """Run each protocol through the lifecycle and return per-protocol summaries."""
    rows: list[dict[str, object]] = []
    for protocol in _PROTOCOLS:
        sim = EcosystemSimulation.run_population(
            _population(protocol, n_users),
            seed=seed,
            maker_pool_size=maker_pool_size,
        )
        observers = _build_observers(sim)
        ev = evaluate_ecosystem(sim, observers)

        by_adv: dict[str, dict[str, float]] = {}
        for adv, summary in ev.summary.items():
            stage_means = summary["mean_effective_set_by_stage"]  # type: ignore[index]
            stage_worst = summary["mean_worst_case_set_by_stage"]  # type: ignore[index]
            stage_deanon = summary["deanon_fraction_by_stage"]  # type: ignore[index]
            by_adv[adv] = {
                "mixed_effective_set": float(stage_means.get("mixed", 0.0)),
                "mixed_worst_case_set": float(stage_worst.get("mixed", 0.0)),
                "mixed_deanon_fraction": float(stage_deanon.get("mixed", 0.0)),
                "worst_stage_effective_set": float(
                    summary["mean_worst_stage_effective_set"]  # type: ignore[index]
                ),
                "worst_stage_worst_case_set": float(
                    summary["mean_worst_stage_worst_case_set"]  # type: ignore[index]
                ),
            }

        rows.append({
            "protocol": str(protocol),
            "label": _LABELS[protocol],
            "by_adversary": by_adv,
        })
    return rows


def run_single_round_comparison() -> list[dict[str, object]]:
    """Single-round Improvement.MINI_COINJOINXT vs baseline on the threat ladder."""
    params = jm_orderbook_params(n_makers=7)
    threats = standard_threat_models()
    rows: list[dict[str, object]] = []
    for threat_name, threat in threats.items():
        r = evaluate_with_improvements(params, threat, [Improvement.MINI_COINJOINXT])
        rows.append({
            "threat": threat_name,
            "baseline_degree": r.baseline.score.degree,
            "baseline_effective_set": r.baseline.score.effective_set,
            "baseline_worst_case_set": r.baseline.score.worst_case_effective_set,
            "improved_degree": r.improved.score.degree,
            "improved_effective_set": r.improved.score.effective_set,
            "improved_worst_case_set": r.improved.score.worst_case_effective_set,
            "degree_gain": r.degree_gain,
            "effective_set_gain": r.effective_set_gain,
            "notes": list(r.notes),
        })
    return rows


def run_residual_surface_breakdown(
    *,
    n_users: int = 25,
    maker_pool_size: int = 40,
    seed: int = 0,
) -> dict[str, object]:
    """Breakdown of each residual attack surface for mini-CoinJoinXT specifically.

    Shows which observer component is responsible for the remaining privacy loss,
    isolating: channel-close settlement (capacity-correlation via LN adversary),
    LN funding source clustering (off-chain, swap-provider), and spend
    counterparty CIOH.
    """
    sim = EcosystemSimulation.run_population(
        _population(CoinJoinProtocol.JOINMARKET_MINI_XT, n_users),
        seed=seed,
        maker_pool_size=maker_pool_size,
    )
    cps = {c.received_by for c in sim.ledger.coins.values() if c.received_by}

    surfaces: dict[str, Observer] = {
        "passive_onchain": PassiveChainObserver(),
        "ln_channel_settlement": LightningAdversary(probe_private=False),
        "ln_private_probing": LightningAdversary(probe_private=True),
        "ln_funding_src_offchain": LightningFundingSourceObserver(),
        "spend_counterparty": SpendCounterparty(known_counterparties=cps),
        "sp_payer": SilentPaymentPayer(),
        "evil_coordinator": EvilCoordinator(),
    }

    results: dict[str, float] = {}
    for name, obs in surfaces.items():
        ev = evaluate_ecosystem(sim, {name: obs}, include_collusion=False)
        s = ev.summary.get(name, {})
        stage_means = s.get("mean_effective_set_by_stage", {})  # type: ignore[union-attr]
        results[name] = float(stage_means.get("mixed", 0.0))  # type: ignore[arg-type]

    return {
        "protocol": str(CoinJoinProtocol.JOINMARKET_MINI_XT),
        "label": _LABELS[CoinJoinProtocol.JOINMARKET_MINI_XT],
        "mixed_set_by_single_observer": results,
        "note": (
            "Observers marked 'offchain' (ln_funding_src_offchain) are "
            "off-chain signals. They do not affect the on-chain subset-sum claim "
            "but are the residual surface for the complete threat model."
        ),
    }


def run_intersection_comparison() -> dict[str, object]:
    """Goldfeder intersection decay for each protocol under fee_observer."""
    threats = standard_threat_models()
    params = jm_orderbook_params(n_makers=7)
    result: dict[str, object] = {}
    threat = threats["fee_observer"]
    overlaps = {
        CoinJoinProtocol.JOINMARKET: 0.20,
        CoinJoinProtocol.JOINMARKET_FEEQUANT: 0.20,
        CoinJoinProtocol.JOINMARKET_LN: 0.20,
        CoinJoinProtocol.JOINMARKET_MINI_XT: 0.20,
    }
    from coinjoin_simulator.improvements import apply_improvements
    from coinjoin_simulator.protocols import evaluate_protocol

    improvements_by_protocol = {
        CoinJoinProtocol.JOINMARKET: [],
        CoinJoinProtocol.JOINMARKET_FEEQUANT: [Improvement.FEE_QUANTIZATION],
        CoinJoinProtocol.JOINMARKET_LN: [Improvement.FEE_QUANTIZATION, Improvement.LN_SWAP_INPUT],
        CoinJoinProtocol.JOINMARKET_MINI_XT: [Improvement.MINI_COINJOINXT],
    }

    for protocol, imps in improvements_by_protocol.items():
        if imps:
            p2, t2, _ = apply_improvements(params, threat, imps)
            res = evaluate_protocol(p2, t2)
        else:
            res = evaluate_protocol(params, threat)
        overlap = overlaps[protocol]
        decay = run_intersection_decay(res.honest_users, spurious_overlap=overlap, seed=0)
        result[str(protocol)] = {
            "label": _LABELS[protocol],
            "honest_users": res.honest_users,
            "decay": decay,
        }
    return result


def main() -> None:
    start = time.time()
    print("Building mini-CoinJoinXT lifecycle study...")

    lifecycle = run_lifecycle_comparison()
    single_round = run_single_round_comparison()
    residual = run_residual_surface_breakdown()
    intersection = run_intersection_comparison()

    payload: dict[str, object] = {
        "framework": {
            "name": "Mini-CoinJoinXT privacy study",
            "description": (
                "Combines ZKP coordination + swap input + taproot dual-funded "
                "channel to evaluate theoretical gains and residual attack surface. "
                "The on-chain analysis uses interval-constrained partition entropy "
                "(Phase B2). The off-chain LN funding source observer (Phase B3) "
                "captures the residual surface PR #280 identified."
            ),
            "phases": ["B1: private-balance channel", "B2: interval relaxation",
                       "B3: LN funding source observer"],
        },
        "lifecycle_comparison": lifecycle,
        "single_round_improvement": single_round,
        "residual_surface_breakdown": residual,
        "intersection_decay": intersection,
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
    }
    OUTPUT_PATH.write_text(json.dumps(payload, indent=2) + "\n")
    elapsed = time.time() - start
    print(f"\nWrote {OUTPUT_PATH} in {elapsed:.2f}s\n")

    print("Single-round improvement (fee_observer):")
    fee_row = next(r for r in single_round if r["threat"] == "fee_observer")
    print(
        f"  baseline  d={fee_row['baseline_degree']:.3f}  S={fee_row['baseline_effective_set']:.2f}"
        f"  S_worst={fee_row['baseline_worst_case_set']:.2f}"
    )
    print(
        f"  improved  d={fee_row['improved_degree']:.3f}  S={fee_row['improved_effective_set']:.2f}"
        f"  S_worst={fee_row['improved_worst_case_set']:.2f}"
    )

    print("\nLifecycle (mixed coin realized set per adversary):")
    print(f"  {'protocol':<38} {'passive':>8} {'spend-CP':>9} {'LN-fund':>8} {'collude':>8}")
    for row in lifecycle:
        adv = row["by_adversary"]  # type: ignore[index]
        passive = adv["passive"]["mixed_effective_set"]
        spend = adv["spend_cp"]["mixed_effective_set"]
        ln_fund = adv.get("ln_funding_src", {}).get("mixed_effective_set", 0.0)
        collude = adv["all_collude"]["worst_stage_effective_set"]
        print(
            f"  {row['label'][:38]:<38} {passive:>8.1f} {spend:>9.1f} "
            f"{ln_fund:>8.1f} {collude:>8.1f}"
        )

    print("\nResidual surface (mini-CoinJoinXT, single observer, mixed set):")
    for name, val in residual["mixed_set_by_single_observer"].items():  # type: ignore[union-attr]
        marker = " [off-chain]" if "offchain" in name else ""
        print(f"  {name:<35} {val:.2f}{marker}")


if __name__ == "__main__":
    main()