plm-lottery/app/rounds/draw.py

import hashlib
import secrets

# Byte offsets of a standard 80-byte block header: version(4) + prev_block(32) +
# merkle_root(32) + timestamp(4) + bits(4) + nonce(4).
_HEADER_LENGTH_BYTES = 80
_PREV_BLOCK_OFFSET = 4
_PREV_BLOCK_LENGTH = 32
_BITS_OFFSET = 72
_BITS_LENGTH = 4


# B-74 (ELC-11): the proof-of-work floor of PLM mainnet — `consensus.powLimit` in
# palladiumcore/src/chainparams.cpp, the same as Bitcoin's. A header claiming a target
# *easier* than this was not mined to the chain's rules, whatever its own hash says, so
# the claimed target alone (which is whoever fabricates the header's to choose) is not a
# proof of work. Read through the module at call time so a test can swap in a regtest
# limit to mine headers cheaply; production code never changes it.
MAINNET_POW_LIMIT = int("00000000" + "ff" * 28, 16)
POW_LIMIT = MAINNET_POW_LIMIT


class HeaderValidationError(Exception):
    """Raised by ElectrumListener._apply_header (B-28) when a header either doesn't
    satisfy the difficulty target it claims for itself, or doesn't chain from the
    previously accepted tip. Letting this propagate out of the header-consuming
    task ends the current Electrum session the same way a dropped connection would
    (see ElectrumListener._run_once), so the listener rotates to the next
    configured server instead of trusting a header a server just forged."""


def header_hex_to_block_hash(header_hex: str) -> str:
    """Block hash from a raw Electrum header: sha256d, byte-reversed, hex.
    Verified against a real mainnet block (blockchain.transaction.get's own
    reported blockhash) during development."""
    header_bytes = bytes.fromhex(header_hex)
    digest = hashlib.sha256(hashlib.sha256(header_bytes).digest()).digest()
    return digest[::-1].hex()


def header_prev_hash(header_hex: str) -> str:
    """The header's `prev_block` field, byte-reversed to the same conventional
    (display) order as header_hex_to_block_hash's return value, so the two can be
    compared directly to check that one header actually chains from another."""
    header_bytes = bytes.fromhex(header_hex)
    prev = header_bytes[_PREV_BLOCK_OFFSET : _PREV_BLOCK_OFFSET + _PREV_BLOCK_LENGTH]
    return prev[::-1].hex()


def _target_from_bits(bits: int) -> int:
    """Decompress Bitcoin-style compact `nBits` difficulty encoding into the full
    256-bit target a valid header's hash must be less than or equal to."""
    exponent = bits >> 24
    mantissa = bits & 0xFFFFFF
    if exponent <= 3:
        return mantissa >> (8 * (3 - exponent))
    return mantissa << (8 * (exponent - 3))


def header_meets_its_own_target(header_hex: str) -> bool:
    """Whether this header's hash satisfies the difficulty target *it claims for
    itself* (the `bits` field), and that target is no easier than the chain's powLimit
    (B-74, ELC-11 — without that, a fabricated header just claims `bits = 0x207fffff`). Rejects a header that was never actually mined —
    e.g. one fabricated wholesale by a hostile or MITM'd Electrum server (B-28),
    since satisfying a self-chosen target still requires real proof-of-work.

    This does NOT — and, short of downloading and validating the full header
    chain's difficulty-retarget history, cannot — catch a header honestly mined at
    a real but implausibly low self-chosen difficulty: a server could still declare
    an easy target and grind it out with modest hardware. That residual risk is why
    the draw additionally requires the winning block's header to be corroborated by
    the configured servers at a fixed height before using it as the seed (see
    ElectrumListener.agreed_block_hash and rounds/scheduler.py:_wait_for_draw_block)
    rather than relying on this check alone.
    """
    header_bytes = bytes.fromhex(header_hex)
    if len(header_bytes) != _HEADER_LENGTH_BYTES:
        return False
    bits = int.from_bytes(header_bytes[_BITS_OFFSET : _BITS_OFFSET + _BITS_LENGTH], "little")
    target = _target_from_bits(bits)
    if target <= 0 or target > POW_LIMIT:  # B-74: not easier than the chain's powLimit
        return False
    digest = hashlib.sha256(hashlib.sha256(header_bytes).digest()).digest()
    # The hash as the integer comparable against `target`: this is the same digest
    # header_hex_to_block_hash reverses into the conventional display hex, so
    # reading it byte-reversed as a big-endian int is equivalent to reading the
    # original digest bytes as little-endian — both give the same integer.
    hash_int = int.from_bytes(digest, "little")
    return hash_int <= target


def sha256_hex(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def new_server_seed() -> str:
    """B-73 (RND-34, DAT-64): a round's secret — 32 random bytes from the OS CSPRNG, as
    64 lowercase hex characters."""
    return secrets.token_bytes(32).hex()


def server_seed_commitment(server_seed_hex: str) -> str:
    """`server_seed_hash`: sha256 of the secret's 32 bytes, published from the opening of
    the round so the operator cannot change the secret afterwards (RND-34)."""
    return sha256_hex(bytes.fromhex(server_seed_hex))


def participants_hash(rows: list[tuple[str, int, str]]) -> str:
    """The fingerprint of the ordered draw list (RND-14): sha256 of one `txid:vout:address`
    line per participant, in `draw_position` order, each line ending in `\n`, UTF-8. `txid`
    is the bet transaction's, lowercase hex as explorers show it; `vout` is the output paying
    the round's pool address, in decimal; `address` is the participant's own."""
    text = "".join(f"{txid}:{vout}:{address}\n" for txid, vout, address in rows)
    return sha256_hex(text.encode("utf-8"))


def draw_seed(server_seed_hex: str, block_hash_hex: str, participants_hash_hex: str) -> int:
    """RND-17: sha256 of the secret (32 bytes), the hash of the block at `draw_height` (32
    bytes, in the order explorers show it) and `participants_hash` (32 bytes), read as an
    unsigned big-endian integer. Nobody who lacks the secret can predict it, and the
    operator, who knows the secret, cannot know the block when committing to the list."""
    digest = hashlib.sha256(
        bytes.fromhex(server_seed_hex) + bytes.fromhex(block_hash_hex) + bytes.fromhex(participants_hash_hex)
    ).digest()
    return int.from_bytes(digest, "big")


def draw_index(seed: int, participant_count: int) -> int:
    """The winner's `draw_position`: seed modulo the number of participants. Every
    participant holds exactly one position, so odds are equal whatever the amount."""
    if participant_count <= 0:
        raise ValueError("no participants to draw from")
    return seed % participant_count


def draw_winner(participants: list[str], server_seed_hex: str, block_hash_hex: str, participants_hash_hex: str) -> str:
    """The v1 draw algorithm (docs/ciclo-di-vita-del-round.md RND-17), recomputable by anyone
    from the public proof (GET /rounds/{id}/proof): `participants` are the entries in
    `draw_position` order, the winner is the one at `draw_index(draw_seed(...), n)`. Deliberately
    simple and replaceable."""
    if not participants:
        raise ValueError("no participants to draw from")
    seed = draw_seed(server_seed_hex, block_hash_hex, participants_hash_hex)
    return participants[draw_index(seed, len(participants))]