jmcore.podle

jmcore.podle

Proof of Discrete Log Equivalence (PoDLE) for JoinMarket.

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.

This module provides both generation (for takers) and verification (for makers) of PoDLE proofs.

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)

For detailed cryptographic documentation including the NUMS point generation algorithm and secp256k1 parameters, see docs/technical/privacy.md#podle.

Reference: https://gist.github.com/AdamISZ/9cbba5e9408d23813ca8 Reference: joinmarket-clientserver/src/jmclient/podle.py

Attributes

G_COMPRESSED = bytes.fromhex('0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798') module-attribute

G_UNCOMPRESSED = bytes.fromhex('0479be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798483ada7726a3c4655da4fbfc0e1108a8fd17b448a68554199c47d08ffb10d4b8') module-attribute

NUMS_TEST_VECTORS = {0: '0296f47ec8e6d6a9c3379c2ce983a6752bcfa88d46f2a6ffe0dd12c9ae76d01a1f', 1: '023f9976b86d3f1426638da600348d96dc1f1eb0bd5614cc50db9e9a067c0464a2', 5: '02bbc5c4393395a38446e2bd4d638b7bfd864afb5ffaf4bed4caf797df0e657434', 9: '021b739f21b981c2dcbaf9af4d89223a282939a92aee079e94a46c273759e5b42e', 100: '02aacc3145d04972d0527c4458629d328219feda92bef6ef6025878e3a252e105a', 255: '02a0a8694820c794852110e5939a2c03f8482f81ed57396042c6b34557f6eb430a'} module-attribute

Classes

PoDLECommitment

PoDLE commitment data generated by taker.

Source code in jmcore/src/jmcore/podle.py

@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()

Attributes

commitment: bytes instance-attribute

e: bytes instance-attribute

index: int instance-attribute

p: bytes instance-attribute

p2: bytes instance-attribute

sig: bytes instance-attribute

utxo: str instance-attribute

Methods:

to_commitment_str() -> 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)

Source code in jmcore/src/jmcore/podle.py

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()

to_revelation() -> dict[str, str]

Convert to revelation format for sending to maker.

Source code in jmcore/src/jmcore/podle.py

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,
    }

PoDLEError

Bases: Exception

PoDLE generation or verification error.

Source code in jmcore/src/jmcore/podle.py

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

    pass

Functions:

deserialize_revelation(revelation_str: str) -> dict[str, Any] | None

Deserialize PoDLE revelation from wire format.

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

Source code in jmcore/src/jmcore/podle.py

def deserialize_revelation(revelation_str: str) -> dict[str, Any] | None:
    """
    Deserialize PoDLE revelation from wire format.

    Format: utxo|P|P2|sig|e (pipe-separated strings)
    """
    try:
        if len(revelation_str) > _MAX_PODLE_REVELATION_LENGTH:
            logger.warning("PoDLE revelation exceeds maximum length")
            return None

        parts = revelation_str.split("|")
        if len(parts) != 5:
            logger.warning(f"Invalid revelation format: expected 5 parts, got {len(parts)}")
            return None

        revelation = {
            "utxo": parts[0],
            "P": parts[1],
            "P2": parts[2],
            "sig": parts[3],
            "e": parts[4],
        }
        if not _has_valid_revelation_field_lengths(revelation):
            return None

        return revelation

    except Exception as e:
        logger.error("Failed to deserialize PoDLE revelation")
        logger.bind(sensitive=True).error(f"Failed to deserialize PoDLE revelation: {e}")
        return None

generate_nums_point(index: int) -> CPubKey

Generate a Nothing-Up-My-Sleeve (NUMS) point deterministically.

The algorithm takes secp256k1's generator G (both compressed and uncompressed), appends the index byte and a counter byte, then SHA256 hashes the result. It tries to create a valid curve point by prepending 0x02 to the hash. The first valid point found is returned as the NUMS point for that index.

This ensures NUMS points are generated transparently with no hidden discrete log.

Reference: https://gist.github.com/AdamISZ/9cbba5e9408d23813ca8 Reference: joinmarket-clientserver/src/jmclient/podle.py getNUMS()

Args: index: Index of the NUMS point to generate (0-255)

Returns: The NUMS point as a CPubKey

Raises: PoDLEError: If index is out of range or no valid point found (should never happen)

Source code in jmcore/src/jmcore/podle.py

def generate_nums_point(index: int) -> CPubKey:
    """
    Generate a Nothing-Up-My-Sleeve (NUMS) point deterministically.

    The algorithm takes secp256k1's generator G (both compressed and uncompressed),
    appends the index byte and a counter byte, then SHA256 hashes the result.
    It tries to create a valid curve point by prepending 0x02 to the hash.
    The first valid point found is returned as the NUMS point for that index.

    This ensures NUMS points are generated transparently with no hidden discrete log.

    Reference: https://gist.github.com/AdamISZ/9cbba5e9408d23813ca8
    Reference: joinmarket-clientserver/src/jmclient/podle.py getNUMS()

    Args:
        index: Index of the NUMS point to generate (0-255)

    Returns:
        The NUMS point as a CPubKey

    Raises:
        PoDLEError: If index is out of range or no valid point found (should never happen)
    """
    if not 0 <= index <= 255:
        raise PoDLEError(f"NUMS point index {index} must be in range 0-255")

    # Try both compressed and uncompressed G as seeds
    for g_bytes in [G_COMPRESSED, G_UNCOMPRESSED]:
        seed = g_bytes + bytes([index])
        for counter in range(256):
            seed_with_counter = seed + bytes([counter])
            hashed_seed = hashlib.sha256(seed_with_counter).digest()
            # Try to create a valid point with 02 prefix (even y-coordinate)
            claimed_point = b"\x02" + hashed_seed
            try:
                nums_point = CPubKey(claimed_point)
                if nums_point.is_fullyvalid():
                    return nums_point
            except Exception:
                continue

    # This should never happen given the search space
    raise PoDLEError(f"Failed to generate NUMS point for index {index}")  # pragma: no cover

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 = 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-255)

Returns: PoDLECommitment with all proof data

Source code in jmcore/src/jmcore/podle.py

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-255)

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

    if not 0 <= index <= 255:
        raise PoDLEError(f"Invalid NUMS index: {index} (must be 0-255)")

    try:
        private_key = CKey(private_key_bytes)
    except ValueError as exc:
        raise PoDLEError("Invalid private key value") from exc

    # Calculate P = k*G (standard public key)
    p_point = private_key.pub
    p_bytes = point_to_bytes(p_point)

    # Get NUMS point J
    j_point = get_nums_point(index)

    # Calculate P2 = k*J
    p2_point = j_point.multiply(private_key_bytes)
    p2_bytes = point_to_bytes(p2_point)

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

    # Derive the nonce from the secret key and proof transcript. This removes
    # runtime RNG failures from a Schnorr-style operation where nonce reuse
    # across two proofs would reveal the UTXO private key.
    s_bytes: bytes | None = None
    for k_proof in _podle_nonce_candidates(
        private_key_bytes,
        utxo_str,
        index,
        p_bytes,
        p2_bytes,
    ):
        kg_bytes = point_to_bytes(scalar_mult_g(k_proof))
        kj_bytes = point_to_bytes(point_mult(k_proof, j_point))
        e_bytes = hashlib.sha256(kg_bytes + kj_bytes + p_bytes + p2_bytes).digest()
        s_bytes = _compute_podle_response(private_key, k_proof, e_bytes)
        if s_bytes is not None:
            break
    if s_bytes is None:
        raise PoDLEError("PoDLE nonce candidates exhausted")

    logger.bind(sensitive=True).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,
    )

get_nums_point(index: int) -> CPubKey

Get Nothing-Up-My-Sleeve (NUMS) generator point J for given index.

NUMS points are generated deterministically and cached for efficiency. Supports indices 0-255.

Args: index: Index of the NUMS point (0-255)

Returns: The NUMS point as a CPubKey

Source code in jmcore/src/jmcore/podle.py

def get_nums_point(index: int) -> CPubKey:
    """
    Get Nothing-Up-My-Sleeve (NUMS) generator point J for given index.

    NUMS points are generated deterministically and cached for efficiency.
    Supports indices 0-255.

    Args:
        index: Index of the NUMS point (0-255)

    Returns:
        The NUMS point as a CPubKey
    """
    if index in _nums_cache:
        return _nums_cache[index]

    nums_point = generate_nums_point(index)
    _nums_cache[index] = nums_point
    return nums_point

parse_podle_revelation(revelation: dict[str, Any]) -> dict[str, Any] | None

Parse and validate PoDLE revelation structure.

Expected format from taker: { 'P': , 'P2': , 'sig': , 'e': , 'utxo': }

Returns parsed structure with bytes, or None if invalid. Extended format includes scriptpubkey and blockheight for neutrino_compat feature.

Source code in jmcore/src/jmcore/podle.py

def parse_podle_revelation(revelation: dict[str, Any]) -> dict[str, Any] | None:
    """
    Parse and validate PoDLE revelation structure.

    Expected format from taker:
    {
        'P': <hex string>,
        'P2': <hex string>,
        'sig': <hex string>,
        'e': <hex string>,
        'utxo': <txid:vout or txid:vout:scriptpubkey:blockheight string>
    }

    Returns parsed structure with bytes, or None if invalid.
    Extended format includes scriptpubkey and blockheight for neutrino_compat feature.
    """
    try:
        required_fields = ["P", "P2", "sig", "e", "utxo"]
        for field in required_fields:
            if field not in revelation:
                logger.warning(f"Missing required field in PoDLE revelation: {field}")
                return None

        if not _has_valid_revelation_field_lengths(revelation):
            return None

        p_bytes = bytes.fromhex(revelation["P"])
        p2_bytes = bytes.fromhex(revelation["P2"])
        sig_bytes = bytes.fromhex(revelation["sig"])
        e_bytes = bytes.fromhex(revelation["e"])

        try:
            utxo = UTXOMetadata.from_str(revelation["utxo"])
            txid = utxo.txid
            vout = utxo.vout
            scriptpubkey = utxo.scriptpubkey
            blockheight = utxo.blockheight
            if scriptpubkey:
                logger.bind(sensitive=True).debug(
                    f"Parsed extended UTXO format: {txid}:{vout} with metadata"
                )
        except (ValueError, ValidationError):
            logger.warning("Invalid UTXO format in PoDLE revelation")
            logger.bind(sensitive=True).warning(
                f"Invalid UTXO format in PoDLE revelation: {revelation['utxo']}"
            )
            return None

        result: dict[str, Any] = {
            "P": p_bytes,
            "P2": p2_bytes,
            "sig": sig_bytes,
            "e": e_bytes,
            "txid": txid,
            "vout": vout,
        }

        # Add extended metadata if present
        if scriptpubkey is not None:
            result["scriptpubkey"] = scriptpubkey
        if blockheight is not None:
            result["blockheight"] = blockheight

        return result

    except Exception as e:
        logger.error("Failed to parse PoDLE revelation")
        logger.bind(sensitive=True).error(f"Failed to parse PoDLE revelation: {e}")
        return None

point_add(p1: CPubKey, p2: CPubKey) -> CPubKey

Add two EC points using libsecp256k1.

Source code in jmcore/src/jmcore/podle.py

def point_add(p1: CPubKey, p2: CPubKey) -> CPubKey:
    """Add two EC points using libsecp256k1."""
    return CPubKey.combine(p1, p2, compressed=p1.is_compressed())

point_mult(scalar: int, point: CPubKey) -> CPubKey

Multiply an EC point by a scalar using libsecp256k1.

Source code in jmcore/src/jmcore/podle.py

def point_mult(scalar: int, point: CPubKey) -> CPubKey:
    """Multiply an EC point by a scalar using libsecp256k1."""
    scalar = scalar % SECP256K1_N
    if scalar == 0:
        raise PoDLEError("Scalar cannot be zero")
    scalar_bytes = scalar.to_bytes(32, "big")
    return point.multiply(scalar_bytes)

point_to_bytes(point: CPubKey) -> bytes

Convert EC point to compressed bytes.

Source code in jmcore/src/jmcore/podle.py

def point_to_bytes(point: CPubKey) -> bytes:
    """Convert EC point to compressed bytes."""
    return bytes(point)

scalar_mult_g(scalar: int) -> CPubKey

Multiply generator G by scalar (creates public key from private key).

Source code in jmcore/src/jmcore/podle.py

def scalar_mult_g(scalar: int) -> CPubKey:
    """Multiply generator G by scalar (creates public key from private key)."""
    scalar = scalar % SECP256K1_N
    if scalar == 0:
        raise PoDLEError("Scalar cannot be zero")
    scalar_bytes = scalar.to_bytes(32, "big")
    return CKey(scalar_bytes).pub

serialize_revelation(commitment: PoDLECommitment) -> str

Serialize PoDLE revelation to wire format.

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

Source code in jmcore/src/jmcore/podle.py

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

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

verify_podle(p: bytes, p2: bytes, sig: bytes, e: bytes, commitment: bytes, index_range: range = range(10)) -> tuple[bool, str]

Verify PoDLE proof.

Verifies that P and P2 have the same discrete log (private key) without revealing the private key itself.

Args: p: Public key bytes (33 bytes compressed) p2: Commitment public key bytes (33 bytes compressed) sig: Signature s value (32 bytes) e: Challenge e value (32 bytes) commitment: sha256(P2) commitment (32 bytes) index_range: Allowed NUMS indices to try

Returns: (is_valid, error_message)

Source code in jmcore/src/jmcore/podle.py

def verify_podle(
    p: bytes,
    p2: bytes,
    sig: bytes,
    e: bytes,
    commitment: bytes,
    index_range: range = range(10),
) -> tuple[bool, str]:
    """
    Verify PoDLE proof.

    Verifies that P and P2 have the same discrete log (private key)
    without revealing the private key itself.

    Args:
        p: Public key bytes (33 bytes compressed)
        p2: Commitment public key bytes (33 bytes compressed)
        sig: Signature s value (32 bytes)
        e: Challenge e value (32 bytes)
        commitment: sha256(P2) commitment (32 bytes)
        index_range: Allowed NUMS indices to try

    Returns:
        (is_valid, error_message)
    """
    try:
        if len(p) != 33:
            return False, f"Invalid P length: {len(p)}, expected 33"
        if len(p2) != 33:
            return False, f"Invalid P2 length: {len(p2)}, expected 33"
        if len(sig) != 32:
            return False, f"Invalid sig length: {len(sig)}, expected 32"
        if len(e) != 32:
            return False, f"Invalid e length: {len(e)}, expected 32"
        if len(commitment) != 32:
            return False, f"Invalid commitment length: {len(commitment)}, expected 32"

        expected_commitment = hashlib.sha256(p2).digest()
        if commitment != expected_commitment:
            return False, "Commitment does not match H(P2)"

        p_point = CPubKey(p)
        p2_point = CPubKey(p2)
        if not p_point.is_fullyvalid() or not p2_point.is_fullyvalid():
            return False, "Invalid public key"

        s_int = int.from_bytes(sig, "big")
        e_int = int.from_bytes(e, "big")

        if s_int == 0 or s_int >= SECP256K1_N:
            return False, "Signature value out of range"

        # sg = s * G
        sg = scalar_mult_g(s_int) if s_int > 0 else None

        # Compute -e mod N for subtraction (JAM compatible: s = k + e*x, verify Kg = s*G - e*P)
        minus_e_int = (-e_int) % SECP256K1_N

        for index in index_range:
            try:
                j = get_nums_point(index)

                # Kg = s*G - e*P = s*G + (-e)*P (JAM compatible verification)
                minus_e_p = point_mult(minus_e_int, p_point)
                kg = point_add(sg, minus_e_p) if sg is not None else minus_e_p

                # Kj = s*J - e*P2 = s*J + (-e)*P2
                minus_e_p2 = point_mult(minus_e_int, p2_point)
                if s_int > 0:
                    sj = point_mult(s_int, j)
                    kj = point_add(sj, minus_e_p2)
                else:
                    kj = minus_e_p2

                kg_bytes = point_to_bytes(kg)
                kj_bytes = point_to_bytes(kj)

                e_check = hashlib.sha256(kg_bytes + kj_bytes + p + p2).digest()

                if e_check == e:
                    logger.debug(f"PoDLE verification successful at index {index}")
                    return True, ""

            except Exception as ex:
                logger.bind(sensitive=True).debug(
                    f"PoDLE verification failed at index {index}: {ex}"
                )
                continue

        return False, f"PoDLE verification failed for all indices in {index_range}"

    except Exception as ex:
        logger.error("PoDLE verification error")
        logger.bind(sensitive=True).error(f"PoDLE verification error: {ex}")
        return False, f"Verification error: {ex}"

verify_podle_binding(p: bytes, scriptpubkey: bytes | str) -> tuple[bool, str]

Verify that a PoDLE public key P actually controls a given UTXO.

The PoDLE proof on its own only shows that P and P2 share a discrete log; it does NOT tie P to the UTXO the taker claims to be spending. Without this binding a taker could replay a stranger's UTXO as its anti-snooping commitment. This helper closes that gap by reconstructing the scriptPubKey that P would produce for each supported output type and comparing it against the UTXO's on-chain scriptPubKey.

Supported script types: P2PKH, P2SH-P2WPKH, and P2WPKH (all using the 33-byte compressed P).

Args: p: PoDLE public key P (33-byte compressed SEC1 encoding). scriptpubkey: UTXO scriptPubKey as raw bytes or a hex string.

Returns: (is_bound, error_message)

Source code in jmcore/src/jmcore/podle.py

def verify_podle_binding(p: bytes, scriptpubkey: bytes | str) -> tuple[bool, str]:
    """
    Verify that a PoDLE public key ``P`` actually controls a given UTXO.

    The PoDLE proof on its own only shows that ``P`` and ``P2`` share a
    discrete log; it does NOT tie ``P`` to the UTXO the taker claims to be
    spending. Without this binding a taker could replay a stranger's UTXO as
    its anti-snooping commitment. This helper closes that gap by reconstructing
    the scriptPubKey that ``P`` would produce for each supported output type and
    comparing it against the UTXO's on-chain scriptPubKey.

    Supported script types: P2PKH, P2SH-P2WPKH, and P2WPKH (all using the
    33-byte compressed ``P``).

    Args:
        p: PoDLE public key ``P`` (33-byte compressed SEC1 encoding).
        scriptpubkey: UTXO scriptPubKey as raw bytes or a hex string.

    Returns:
        (is_bound, error_message)
    """
    if isinstance(scriptpubkey, str):
        try:
            spk = bytes.fromhex(scriptpubkey)
        except ValueError:
            return False, "scriptpubkey is not valid hex"
    else:
        spk = scriptpubkey

    if len(p) != 33:
        return False, f"Invalid P length: {len(p)}, expected 33 (compressed)"

    h160 = hash160(p)

    # P2WPKH: OP_0 <20-byte hash160(P)>
    if len(spk) == 22 and spk[0] == 0x00 and spk[1] == 0x14:
        if spk[2:] == h160:
            return True, ""
        return False, "P does not match P2WPKH scriptpubkey"

    # P2PKH: OP_DUP OP_HASH160 <20-byte hash160(P)> OP_EQUALVERIFY OP_CHECKSIG
    if (
        len(spk) == 25
        and spk[0] == 0x76
        and spk[1] == 0xA9
        and spk[2] == 0x14
        and spk[23] == 0x88
        and spk[24] == 0xAC
    ):
        if spk[3:23] == h160:
            return True, ""
        return False, "P does not match P2PKH scriptpubkey"

    # P2SH-P2WPKH: OP_HASH160 <hash160(0x0014||hash160(P))> OP_EQUAL
    if len(spk) == 23 and spk[0] == 0xA9 and spk[1] == 0x14 and spk[22] == 0x87:
        redeem_script = b"\x00\x14" + h160
        if spk[2:22] == hash160(redeem_script):
            return True, ""
        return False, "P does not match P2SH-P2WPKH scriptpubkey"

    family = _classify_unsupported_scriptpubkey(spk)
    return False, f"Unsupported {family} scriptpubkey for PoDLE binding ({len(spk)} bytes)"