Module taker.podle

Proof of Discrete Log Equivalence (PoDLE) generation for takers.

This module re-exports PoDLE generation functions from jmcore and provides taker-specific utilities for UTXO selection and commitment generation.

PoDLE is used to prevent sybil attacks in JoinMarket by requiring takers to prove ownership of a UTXO without revealing which UTXO until after the maker commits to participate.

Protocol flow: 1. Taker generates commitment C = H(P2) where P2 = kJ (k = private key, J = NUMS point) 2. Taker sends commitment C to maker 3. Maker accepts and sends pubkey 4. Taker reveals P, P2, sig, e as the "revelation" 5. Maker verifies: P = kG and P2 = k*J (same k)

Reference: https://gist.github.com/AdamISZ/9cbba5e9408d23813ca8

Functions

def generate_podle(private_key_bytes: bytes, utxo_str: str, index: int = 0) ‑> PoDLECommitment

Expand source code

def generate_podle(
    private_key_bytes: bytes,
    utxo_str: str,
    index: int = 0,
) -> PoDLECommitment:
    """
    Generate a PoDLE commitment for a UTXO.

    The PoDLE proves that the taker owns the UTXO without revealing
    the private key. It creates a zero-knowledge proof that:
    P = k*G and P2 = k*J have the same discrete log k.

    Args:
        private_key_bytes: 32-byte private key
        utxo_str: UTXO reference as "txid:vout"
        index: NUMS point index (0-9)

    Returns:
        PoDLECommitment with all proof data
    """
    if len(private_key_bytes) != 32:
        raise PoDLEError(f"Invalid private key length: {len(private_key_bytes)}")

    if index not in PRECOMPUTED_NUMS:
        raise PoDLEError(f"Invalid NUMS index: {index}")

    # Get private key as integer
    k = int.from_bytes(private_key_bytes, "big")
    if k == 0 or k >= SECP256K1_N:
        raise PoDLEError("Invalid private key value")

    # Calculate P = k*G (standard public key)
    p_point = scalar_mult_g(k)
    p_bytes = point_to_bytes(p_point)

    # Get NUMS point J
    j_point = get_nums_point(index)

    # Calculate P2 = k*J
    p2_point = point_mult(k, j_point)
    p2_bytes = point_to_bytes(p2_point)

    # Generate commitment C = H(P2)
    commitment = hashlib.sha256(p2_bytes).digest()

    # Generate Schnorr-like proof
    # Choose random nonce k_proof
    k_proof = int.from_bytes(secrets.token_bytes(32), "big") % SECP256K1_N
    if k_proof == 0:
        k_proof = 1

    # Kg = k_proof * G
    kg_point = scalar_mult_g(k_proof)
    kg_bytes = point_to_bytes(kg_point)

    # Kj = k_proof * J
    kj_point = point_mult(k_proof, j_point)
    kj_bytes = point_to_bytes(kj_point)

    # Challenge e = H(Kg || Kj || P || P2)
    e_bytes = hashlib.sha256(kg_bytes + kj_bytes + p_bytes + p2_bytes).digest()
    e = int.from_bytes(e_bytes, "big") % SECP256K1_N

    # Response s = k_proof + e * k (mod n) - JAM compatible
    s = (k_proof + e * k) % SECP256K1_N
    s_bytes = s.to_bytes(32, "big")

    logger.debug(
        f"Generated PoDLE for {utxo_str} using NUMS index {index}, "
        f"commitment={commitment.hex()[:16]}..."
    )

    return PoDLECommitment(
        commitment=commitment,
        p=p_bytes,
        p2=p2_bytes,
        sig=s_bytes,
        e=e_bytes,
        utxo=utxo_str,
        index=index,
    )

Generate a PoDLE commitment for a UTXO.

The PoDLE proves that the taker owns the UTXO without revealing the private key. It creates a zero-knowledge proof that: P = kG and P2 = kJ have the same discrete log k.

Args

private_key_bytes

32-byte private key

utxo_str

UTXO reference as "txid:vout"

index

NUMS point index (0-9)

Returns

PoDLECommitment with all proof data

def get_eligible_podle_utxos(utxos: list[UTXOInfo],
cj_amount: int,
min_confirmations: int = 5,
min_percent: int = 20) ‑> list[UTXOInfo]

Expand source code

def get_eligible_podle_utxos(
    utxos: list[UTXOInfo],
    cj_amount: int,
    min_confirmations: int = 5,
    min_percent: int = 20,
) -> list[UTXOInfo]:
    """
    Get all eligible UTXOs for PoDLE commitment, sorted by preference.

    Criteria:
    - Must have at least min_confirmations
    - Must be at least min_percent of cj_amount

    Returns:
        List of eligible UTXOs sorted by (confirmations, value) descending
    """
    min_value = int(cj_amount * min_percent / 100)

    eligible = [u for u in utxos if u.confirmations >= min_confirmations and u.value >= min_value]

    # Prefer older UTXOs with more value
    eligible.sort(key=lambda u: (u.confirmations, u.value), reverse=True)
    return eligible

Get all eligible UTXOs for PoDLE commitment, sorted by preference.

Criteria: - Must have at least min_confirmations - Must be at least min_percent of cj_amount

Returns

List of eligible UTXOs sorted by (confirmations, value) descending

def select_podle_utxo(utxos: list[UTXOInfo],
cj_amount: int,
min_confirmations: int = 5,
min_percent: int = 20) ‑> UTXOInfo | None

Expand source code

def select_podle_utxo(
    utxos: list[UTXOInfo],
    cj_amount: int,
    min_confirmations: int = 5,
    min_percent: int = 20,
) -> UTXOInfo | None:
    """
    Select the best UTXO for PoDLE commitment.

    Args:
        utxos: Available UTXOs
        cj_amount: CoinJoin amount
        min_confirmations: Minimum confirmations required
        min_percent: Minimum value as percentage of cj_amount

    Returns:
        Best UTXO for PoDLE or None if no suitable UTXO
    """
    eligible = get_eligible_podle_utxos(utxos, cj_amount, min_confirmations, min_percent)

    if not eligible:
        min_value = int(cj_amount * min_percent / 100)
        logger.warning(
            f"No suitable UTXOs for PoDLE: need {min_confirmations}+ confirmations "
            f"and value >= {min_value} sats ({min_percent}% of {cj_amount})"
        )
        return None

    selected = eligible[0]
    logger.info(
        f"Selected UTXO for PoDLE: {selected.txid}:{selected.vout} "
        f"(value={selected.value}, confs={selected.confirmations})"
    )

    return selected

Select the best UTXO for PoDLE commitment.

Args

utxos

Available UTXOs

cj_amount

CoinJoin amount

min_confirmations

Minimum confirmations required

min_percent

Minimum value as percentage of cj_amount

Returns

Best UTXO for PoDLE or None if no suitable UTXO

def serialize_revelation(commitment: PoDLECommitment) ‑> str

Expand source code

def serialize_revelation(commitment: PoDLECommitment) -> str:
    """
    Serialize PoDLE revelation to wire format.

    Format: P|P2|sig|e|utxo (pipe-separated hex strings)
    """
    return "|".join(
        [
            commitment.p.hex(),
            commitment.p2.hex(),
            commitment.sig.hex(),
            commitment.e.hex(),
            commitment.utxo,
        ]
    )

Serialize PoDLE revelation to wire format.

Format: P|P2|sig|e|utxo (pipe-separated hex strings)

Classes

class ExtendedPoDLECommitment (*args: Any, **kwargs: Any)

Expand source code

@dataclass
class ExtendedPoDLECommitment:
    """
    PoDLE commitment with extended UTXO metadata for neutrino_compat feature.

    This extends the base PoDLECommitment with scriptpubkey and blockheight
    for Neutrino-compatible UTXO verification.
    """

    commitment: PoDLECommitment
    scriptpubkey: str | None = None  # Hex-encoded scriptPubKey
    blockheight: int | None = None  # Block height where UTXO was confirmed

    # Expose underlying commitment properties for compatibility
    @property
    def p(self) -> bytes:
        """Public key P = k*G"""
        return self.commitment.p

    @property
    def p2(self) -> bytes:
        """Commitment point P2 = k*J"""
        return self.commitment.p2

    @property
    def sig(self) -> bytes:
        """Schnorr signature s"""
        return self.commitment.sig

    @property
    def e(self) -> bytes:
        """Challenge e"""
        return self.commitment.e

    @property
    def utxo(self) -> str:
        """UTXO reference txid:vout"""
        return self.commitment.utxo

    @property
    def index(self) -> int:
        """NUMS point index used"""
        return self.commitment.index

    def to_revelation(self, extended: bool = False) -> dict[str, str]:
        """
        Convert to revelation format for sending to maker.

        Args:
            extended: If True, include scriptpubkey:blockheight in utxo string
        """
        rev = self.commitment.to_revelation()
        if extended and self.scriptpubkey and self.blockheight is not None:
            # Replace utxo with extended format: txid:vout:scriptpubkey:blockheight
            txid, vout = self.commitment.utxo.split(":")
            rev["utxo"] = f"{txid}:{vout}:{self.scriptpubkey}:{self.blockheight}"
        return rev

    def to_commitment_str(self) -> str:
        """Get commitment as hex string."""
        return self.commitment.to_commitment_str()

    def has_neutrino_metadata(self) -> bool:
        """Check if we have metadata for Neutrino-compatible verification."""
        return self.scriptpubkey is not None and self.blockheight is not None

PoDLE commitment with extended UTXO metadata for neutrino_compat feature.

This extends the base PoDLECommitment with scriptpubkey and blockheight for Neutrino-compatible UTXO verification.

Instance variables

var blockheight : int | None

The type of the None singleton.

var commitment : PoDLECommitment

The type of the None singleton.

prop e : bytes

Expand source code

@property
def e(self) -> bytes:
    """Challenge e"""
    return self.commitment.e

Challenge e

prop index : int

Expand source code

@property
def index(self) -> int:
    """NUMS point index used"""
    return self.commitment.index

NUMS point index used

prop p : bytes

Expand source code

@property
def p(self) -> bytes:
    """Public key P = k*G"""
    return self.commitment.p

Public key P = k*G

prop p2 : bytes

Expand source code

@property
def p2(self) -> bytes:
    """Commitment point P2 = k*J"""
    return self.commitment.p2

Commitment point P2 = k*J

var scriptpubkey : str | None

The type of the None singleton.

prop sig : bytes

Expand source code

@property
def sig(self) -> bytes:
    """Schnorr signature s"""
    return self.commitment.sig

Schnorr signature s

prop utxo : str

Expand source code

@property
def utxo(self) -> str:
    """UTXO reference txid:vout"""
    return self.commitment.utxo

UTXO reference txid:vout

Methods

def has_neutrino_metadata(self) ‑> bool

Expand source code

def has_neutrino_metadata(self) -> bool:
    """Check if we have metadata for Neutrino-compatible verification."""
    return self.scriptpubkey is not None and self.blockheight is not None

Check if we have metadata for Neutrino-compatible verification.

def to_commitment_str(self) ‑> str

Expand source code

def to_commitment_str(self) -> str:
    """Get commitment as hex string."""
    return self.commitment.to_commitment_str()

Get commitment as hex string.

def to_revelation(self, extended: bool = False) ‑> dict[str, str]

Expand source code

def to_revelation(self, extended: bool = False) -> dict[str, str]:
    """
    Convert to revelation format for sending to maker.

    Args:
        extended: If True, include scriptpubkey:blockheight in utxo string
    """
    rev = self.commitment.to_revelation()
    if extended and self.scriptpubkey and self.blockheight is not None:
        # Replace utxo with extended format: txid:vout:scriptpubkey:blockheight
        txid, vout = self.commitment.utxo.split(":")
        rev["utxo"] = f"{txid}:{vout}:{self.scriptpubkey}:{self.blockheight}"
    return rev

Convert to revelation format for sending to maker.

Args

extended

If True, include scriptpubkey:blockheight in utxo string

class PoDLECommitment (*args: Any, **kwargs: Any)

Expand source code

@dataclass
class PoDLECommitment:
    """PoDLE commitment data generated by taker."""

    commitment: bytes  # H(P2) - 32 bytes
    p: bytes  # Public key P = k*G - 33 bytes compressed
    p2: bytes  # Commitment point P2 = k*J - 33 bytes compressed
    sig: bytes  # Schnorr signature s - 32 bytes
    e: bytes  # Challenge e - 32 bytes
    utxo: str  # UTXO reference "txid:vout"
    index: int  # NUMS point index used

    def to_revelation(self) -> dict[str, str]:
        """Convert to revelation format for sending to maker."""
        return {
            "P": self.p.hex(),
            "P2": self.p2.hex(),
            "sig": self.sig.hex(),
            "e": self.e.hex(),
            "utxo": self.utxo,
        }

    def to_commitment_str(self) -> str:
        """
        Get commitment as string with type prefix.

        JoinMarket requires a commitment type prefix to allow future
        commitment schemes. "P" indicates a standard PoDLE commitment.
        Format: "P" + hex(commitment)
        """
        return "P" + self.commitment.hex()

PoDLE commitment data generated by taker.

Instance variables

var commitment : bytes

The type of the None singleton.

var e : bytes

The type of the None singleton.

var index : int

The type of the None singleton.

var p : bytes

The type of the None singleton.

var p2 : bytes

The type of the None singleton.

var sig : bytes

The type of the None singleton.

var utxo : str

The type of the None singleton.

Methods

def to_commitment_str(self) ‑> str

Expand source code

def to_commitment_str(self) -> str:
    """
    Get commitment as string with type prefix.

    JoinMarket requires a commitment type prefix to allow future
    commitment schemes. "P" indicates a standard PoDLE commitment.
    Format: "P" + hex(commitment)
    """
    return "P" + self.commitment.hex()

Get commitment as string with type prefix.

JoinMarket requires a commitment type prefix to allow future commitment schemes. "P" indicates a standard PoDLE commitment. Format: "P" + hex(commitment)

def to_revelation(self) ‑> dict[str, str]

Expand source code

def to_revelation(self) -> dict[str, str]:
    """Convert to revelation format for sending to maker."""
    return {
        "P": self.p.hex(),
        "P2": self.p2.hex(),
        "sig": self.sig.hex(),
        "e": self.e.hex(),
        "utxo": self.utxo,
    }

Convert to revelation format for sending to maker.

class PoDLEError (*args, **kwargs)

Expand source code

class PoDLEError(Exception):
    """PoDLE generation or verification error."""

    pass

PoDLE generation or verification error.

Ancestors

• builtins.Exception
• builtins.BaseException