#!/usr/bin/env python3 """ Paritr Full Node ================ Self-contained blockchain node built on Bitcoin's design: * Memory-hard scrypt proof of work over an 80-byte block header (ASIC resistant) * Merkle root over every transaction in a block * Compact ``nBits`` difficulty format with hourly retargeting * Tail emission: halving down to a permanent minimum subsidy * P2Pool-style shares: lightweight proofs of work for small nodes * On-chain PPLNS - 95 % of the subsidy strictly by share weight, 5 % to the finder (inclusion incentive), 100 % of the fees to the finder * Coinbase maturity (rewards are spendable only after N confirmations) * Mempool with a fee market and a buffer for nonce gaps (out-of-order P2P) * Most-work-chain consensus with incremental reorganisation (headers-first IBD) * Side-branch store so delivered blocks never lose their rewards * Committed state root per block plus signed state snapshots for fast sync * P2P peer discovery via seed nodes, DNS seeds and a gossip protocol * Public JSON/REST interface for light wallets, compact binary framing for node-to-node share and block gossip * Hardware usage (cores + intensity) adjustable at runtime Account model: Paritr uses an account/nonce model instead of UTXO so light wallets work without input selection. Every other consensus rule follows Bitcoin's example. Platforms: Linux (x86_64 and ARM/aarch64), macOS and Windows. On systems without OpenSSL RIPEMD-160 (Windows, recent Linux distributions) a bundled Python implementation is used automatically. Start: python3 node.py # configuration from config.json python3 node.py --port 5051 # alternative RPC port python node.py --server waitress # Windows production server gunicorn -k gthread -w 1 --threads 8 -b 0.0.0.0:5050 wsgi:app """ from __future__ import annotations import argparse import hashlib import hmac import json import logging import multiprocessing import os import platform import queue import secrets import socket import sqlite3 import threading import time import zlib from multiprocessing import cpu_count from typing import Any, Dict, Iterable, List, Optional, Tuple from urllib.parse import urlparse import ecdsa import requests from flask import Flask, jsonify, request # --------------------------------------------------------------------------- # # 1. Chain parameters (consensus - identical on every node) # --------------------------------------------------------------------------- # NODE_VERSION = "5.0.0" PROTOCOL_VERSION = 6 CHAIN_ID = "paritr-mainnet" COIN = 100_000_000 # smallest unit: 0.00000001 PAR INITIAL_SUBSIDY = 10 * COIN # block reward for block 1 HALVING_INTERVAL = 1_600_000 # halving every 1,600,000 blocks (~3 years) MIN_SUBSIDY = int(0.5 * COIN) # tail emission: subsidy floor MIN_RELAY_FEE = 100_000 # 0.001 PAR minimum fee DUST_LIMIT = 1_000 # 0.00001 PAR smallest amount MAX_MONEY = 1 << 62 # sanity ceiling against integer overflow # There is deliberately no hard supply cap: after the last halving the tail # emission of MIN_SUBSIDY per block keeps funding network security forever. # Inflation therefore falls asymptotically towards zero without the security # budget ever depending on transaction fees alone. TAIL_EMISSION = True # P2Pool-style reward sharing: # The finder receives a fixed inclusion incentive plus all fees, while the bulk # of the subsidy is paid strictly by the share of valid shares in the measuring # window. Shares are lightweight proofs of work at 1/SHARE_TARGET_MULTIPLIER of # the block difficulty, so even small nodes prove their work continuously # instead of waiting for a rare block find. REWARD_WINDOW = 120 # measuring window in blocks (~2 hours) FINDER_SHARE_PERCENT = 5 # finder's inclusion incentive in percent SHARE_TARGET_MULTIPLIER = 100 # share target = block target * 100 MAX_SHARES_PER_BLOCK = 32 # shares a single block may confirm SHARE_TTL_BLOCKS = REWARD_WINDOW * 2 # lifetime of a share in blocks MAX_SHARE_POOL = 4096 # ceiling of the local share store MAX_COINBASE_OUTPUTS = 32 # ceiling of coinbase recipients per block MIN_SHARE_PAYOUT = DUST_LIMIT # smallest distributed reward slice TARGET_BLOCK_TIME = 60 # target block time in seconds RETARGET_INTERVAL = 60 # difficulty retarget every 60 blocks (~1 h) MAX_RETARGET_FACTOR = 4 # adjustment capped at factor 4 resp. 0.25 COINBASE_MATURITY = 10 # rewards spendable after 10 blocks MAX_BLOCK_TRANSACTIONS = 500 # capacity of a single block MAX_FUTURE_BLOCK_TIME = 2 * 60 * 60 # 2 hours timestamp tolerance MEDIAN_TIME_SPAN = 11 # median-time-past window MAX_SIDE_BLOCKS = 500 # buffer for blocks of competing branches MAX_ORPHAN_TRANSACTIONS = 512 # buffer for transactions with a nonce gap ORPHAN_TX_TTL = 900 # seconds a nonce gap may stay open MAX_NONCE_GAP = 64 # largest bridgeable nonce gap # Deterministic state commitment. Recomputing a root over every account on # every block would grow linearly with the number of accounts, so a checkpoint # is committed every STATE_ROOT_INTERVAL blocks instead. That is enough for new # nodes to verify a state snapshot instead of replaying the whole history, and # it keeps the per-block cost at zero for the other 719 blocks. STATE_ROOT_INTERVAL = 720 # state checkpoint every 720 blocks (~12 h) MAX_STATE_SNAPSHOTS = 2 # retained snapshot files on disk # --------------------------------------------------------------------------- # # Proof of work: memory-hard scrypt # --------------------------------------------------------------------------- # # SHA256d is trivially accelerated by the ASIC fleets already built for Bitcoin. # Pointing a fraction of that hardware at a young chain would raise difficulty # far beyond what CPUs can answer, and the promise of solo mining on everyday # devices would be gone. scrypt binds every hash to a 2 MiB scratchpad, so the # cost is dominated by memory latency instead of raw gate count - the same # footprint that kept ASICs away from CryptoNight for years. # # Parameters are consensus critical: changing any of them forks the chain. # 128 * N * r = 128 * 2048 * 8 = 2 MiB of scratchpad per hash. POW_SCRYPT_N = 2048 POW_SCRYPT_R = 8 POW_SCRYPT_P = 1 POW_SCRYPT_MAXMEM = 512 * 1024 * 1024 # OpenSSL guard, well above the 2 MiB need # Block identity stays SHA256d (cheap lookups, Merkle trees, transaction ids); # only the proof of work uses scrypt. Litecoin and Dogecoin split the two the # same way. POW_LIMIT = (1 << 240) - 1 # easiest permitted target = difficulty 1.0 INITIAL_TARGET = (1 << 239) - 1 # chain start = difficulty 2 GENESIS_TARGET = (1 << 252) - 1 # genesis is mined locally by every node ADDRESS_VERSION = 55 # Base58 prefix -> addresses start with "P" ADDRESS_PREFIX_HINT = "P" COINBASE_SENDER = "coinbase" GENESIS_TIMESTAMP = 1767225600 # 2026-01-01 00:00:00 UTC GENESIS_MESSAGE = "Paritr Genesis - an open network for everyone" # Hardcoded seed nodes for automatic bootstrapping SEED_NODES = [ "https://node0.oe-net.de", "http://node0.oe-net.de:5050", ] # DNS seeds: resolved A records are used as additional peer candidates DNS_SEEDS = ["node0.oe-net.de"] SEED_FALLBACK_PORT = 5050 BASE_DIR = os.path.dirname(os.path.abspath(__file__)) CONFIG_FILE = os.path.join(BASE_DIR, "config.json") PLATFORM_LABEL = f"{platform.system()} {platform.machine()}".strip() DEFAULT_CONFIG: Dict[str, Any] = { "network": CHAIN_ID, "node_id": "", "rpc_host": "0.0.0.0", "rpc_port": 5050, "public_url": "", "admin_secret": "", "miner_address": "", "mining_enabled": True, "mining_processes": 0, "mining_intensity": 100, "seed_nodes": SEED_NODES, "peers": [], "data_dir": "data", "rpc_cors_origins": "*", "log_level": "INFO", } log = logging.getLogger("paritr") # --------------------------------------------------------------------------- # # 2. Krypto- und Serialisierungs-Hilfsfunktionen # --------------------------------------------------------------------------- # B58_ALPHABET = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" def sha256d(data: bytes) -> bytes: return hashlib.sha256(hashlib.sha256(data).digest()).digest() def pow_hash(header: bytes) -> bytes: """Memory-hard proof-of-work digest of an 80-byte header. The header doubles as the salt so a single input fully determines the result and no extra field has to travel over the wire. Roughly 2 MiB of scratchpad per hash keeps commodity ASICs out while a CPU still manages a few hundred hashes per second. """ return hashlib.scrypt(header, salt=header, n=POW_SCRYPT_N, r=POW_SCRYPT_R, p=POW_SCRYPT_P, dklen=32, maxmem=POW_SCRYPT_MAXMEM) def verify_pow_available() -> None: """Fails fast when the runtime cannot compute the proof of work. ``hashlib.scrypt`` only exists when Python was built against OpenSSL 1.1+. Without this check the node would start, serve /status, and then die inside the mining thread with a message nobody sees. """ if not hasattr(hashlib, "scrypt"): raise SystemExit( "This Python build has no hashlib.scrypt (needs OpenSSL 1.1+).\n" "Paritr cannot compute its proof of work without it. Install a\n" "standard python3 package for your distribution and retry." ) try: digest = pow_hash(bytes(80)) except Exception as exc: raise SystemExit(f"hashlib.scrypt is present but unusable: {exc}") if len(digest) != 32: raise SystemExit("hashlib.scrypt returned an unexpected digest length.") def pow_value(header: bytes) -> int: """Proof-of-work digest as the integer that is compared against a target.""" return int.from_bytes(pow_hash(header), "little") class _PowCache: """Bounded cache of headers whose proof of work already checked out. A share is verified once when it arrives over gossip and again when a block confirms it. With a memory-hard digest that second pass is expensive, so the result is remembered. The key is the full header, therefore a cache hit can never accept a header that was not actually verified. """ def __init__(self, capacity: int = 8192): self.capacity = capacity self._seen: Dict[bytes, int] = {} self._lock = threading.Lock() def value(self, header: bytes) -> int: with self._lock: cached = self._seen.get(header) if cached is not None: return cached result = pow_value(header) with self._lock: if len(self._seen) >= self.capacity: # Drop the oldest insertions; dict keeps insertion order. for key in list(self._seen)[:self.capacity // 4]: self._seen.pop(key, None) self._seen[header] = result return result _pow_cache = _PowCache() def cached_pow_value(header: bytes) -> int: return _pow_cache.value(header) # RIPEMD-160 only exists in OpenSSL 3 behind the legacy provider and is # therefore missing on Windows and many current Linux distributions. A Python # reference is bundled so the node derives identical addresses everywhere. _RMD_RL = ( 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 7, 4, 13, 1, 10, 6, 15, 3, 12, 0, 9, 5, 2, 14, 11, 8, 3, 10, 14, 4, 9, 15, 8, 1, 2, 7, 0, 6, 13, 11, 5, 12, 1, 9, 11, 10, 0, 8, 12, 4, 13, 3, 7, 15, 14, 5, 6, 2, 4, 0, 5, 9, 7, 12, 2, 10, 14, 1, 3, 8, 11, 6, 15, 13, ) _RMD_RR = ( 5, 14, 7, 0, 9, 2, 11, 4, 13, 6, 15, 8, 1, 10, 3, 12, 6, 11, 3, 7, 0, 13, 5, 10, 14, 15, 8, 12, 4, 9, 1, 2, 15, 5, 1, 3, 7, 14, 6, 9, 11, 8, 12, 2, 10, 0, 4, 13, 8, 6, 4, 1, 3, 11, 15, 0, 5, 12, 2, 13, 9, 7, 10, 14, 12, 15, 10, 4, 1, 5, 8, 7, 6, 2, 13, 14, 0, 3, 9, 11, ) _RMD_SL = ( 11, 14, 15, 12, 5, 8, 7, 9, 11, 13, 14, 15, 6, 7, 9, 8, 7, 6, 8, 13, 11, 9, 7, 15, 7, 12, 15, 9, 11, 7, 13, 12, 11, 13, 6, 7, 14, 9, 13, 15, 14, 8, 13, 6, 5, 12, 7, 5, 11, 12, 14, 15, 14, 15, 9, 8, 9, 14, 5, 6, 8, 6, 5, 12, 9, 15, 5, 11, 6, 8, 13, 12, 5, 12, 13, 14, 11, 8, 5, 6, ) _RMD_SR = ( 8, 9, 9, 11, 13, 15, 15, 5, 7, 7, 8, 11, 14, 14, 12, 6, 9, 13, 15, 7, 12, 8, 9, 11, 7, 7, 12, 7, 6, 15, 13, 11, 9, 7, 15, 11, 8, 6, 6, 14, 12, 13, 5, 14, 13, 13, 7, 5, 15, 5, 8, 11, 14, 14, 6, 14, 6, 9, 12, 9, 12, 5, 15, 8, 8, 5, 12, 9, 12, 5, 14, 6, 8, 13, 6, 5, 15, 13, 11, 11, ) _RMD_KL = (0x00000000, 0x5A827999, 0x6ED9EBA1, 0x8F1BBCDC, 0xA953FD4E) _RMD_KR = (0x50A28BE6, 0x5C4DD124, 0x6D703EF3, 0x7A6D76E9, 0x00000000) _MASK32 = 0xFFFFFFFF def _rmd_f(round_index: int, x: int, y: int, z: int) -> int: if round_index < 16: return x ^ y ^ z if round_index < 32: return (x & y) | (~x & _MASK32 & z) if round_index < 48: return (x | (~y & _MASK32)) ^ z if round_index < 64: return (x & z) | (y & (~z & _MASK32)) return x ^ (y | (~z & _MASK32)) def _rmd_rol(value: int, bits: int) -> int: value &= _MASK32 return ((value << bits) | (value >> (32 - bits))) & _MASK32 def ripemd160_python(data: bytes) -> bytes: """Reference implementation per ISO/IEC 10118-3 (RIPEMD-160).""" h0, h1, h2, h3, h4 = 0x67452301, 0xEFCDAB89, 0x98BADCFE, 0x10325476, 0xC3D2E1F0 message = bytearray(data) bit_length = (len(data) * 8) & 0xFFFFFFFFFFFFFFFF message.append(0x80) while len(message) % 64 != 56: message.append(0x00) message += bit_length.to_bytes(8, "little") for offset in range(0, len(message), 64): words = [int.from_bytes(message[offset + i * 4:offset + i * 4 + 4], "little") for i in range(16)] al, bl, cl, dl, el = h0, h1, h2, h3, h4 ar, br, cr, dr, er = h0, h1, h2, h3, h4 for j in range(80): block = j // 16 tmp = _rmd_rol( (al + _rmd_f(j, bl, cl, dl) + words[_RMD_RL[j]] + _RMD_KL[block]) & _MASK32, _RMD_SL[j], ) al, bl, cl, dl, el = el, (tmp + el) & _MASK32, bl, _rmd_rol(cl, 10), dl tmp = _rmd_rol( (ar + _rmd_f(79 - j, br, cr, dr) + words[_RMD_RR[j]] + _RMD_KR[block]) & _MASK32, _RMD_SR[j], ) ar, br, cr, dr, er = er, (tmp + er) & _MASK32, br, _rmd_rol(cr, 10), dr h0, h1, h2, h3, h4 = ( (h1 + cl + dr) & _MASK32, (h2 + dl + er) & _MASK32, (h3 + el + ar) & _MASK32, (h4 + al + br) & _MASK32, (h0 + bl + cr) & _MASK32, ) return b"".join(word.to_bytes(4, "little") for word in (h0, h1, h2, h3, h4)) def _select_ripemd160(): try: if hashlib.new("ripemd160", b"").digest().hex() == "9c1185a5c5e9fc54612808977ee8f548b2258d31": return lambda payload: hashlib.new("ripemd160", payload).digest() except Exception: pass return ripemd160_python ripemd160 = _select_ripemd160() def hash160(data: bytes) -> bytes: return ripemd160(hashlib.sha256(data).digest()) def b58encode(raw: bytes) -> str: num = int.from_bytes(raw, "big") out = "" while num > 0: num, rem = divmod(num, 58) out = B58_ALPHABET[rem] + out for byte in raw: if byte != 0: break out = B58_ALPHABET[0] + out return out or B58_ALPHABET[0] def b58decode(text: str) -> bytes: num = 0 for char in text: idx = B58_ALPHABET.find(char) if idx < 0: raise ValueError("Invalid Base58 character") num = num * 58 + idx body = num.to_bytes((num.bit_length() + 7) // 8, "big") if num else b"" pad = 0 for char in text: if char != B58_ALPHABET[0]: break pad += 1 return b"\x00" * pad + body def normalize_pubkey(pubkey_hex: str) -> bytes: """Accepts 128 (raw) or 130 (with 04 prefix) hex characters.""" raw = bytes.fromhex(pubkey_hex.strip()) if len(raw) == 64: return b"\x04" + raw if len(raw) == 65 and raw[0] == 0x04: return raw raise ValueError("Invalid public key") def pubkey_to_address(pubkey_hex: str) -> str: payload = bytes([ADDRESS_VERSION]) + hash160(normalize_pubkey(pubkey_hex)) return b58encode(payload + sha256d(payload)[:4]) def is_valid_address(address: Any) -> bool: if not isinstance(address, str) or not 26 <= len(address) <= 40: return False try: raw = b58decode(address) except ValueError: return False if len(raw) != 25 or raw[0] != ADDRESS_VERSION: return False return sha256d(raw[:21])[:4] == raw[21:] def target_to_bits(target: int) -> int: """Compact Bitcoin target representation (nBits).""" if target <= 0: return 0 raw = target.to_bytes((target.bit_length() + 7) // 8, "big") if raw[0] & 0x80: raw = b"\x00" + raw size = len(raw) word = int.from_bytes(raw[:3].ljust(3, b"\x00"), "big") return (size << 24) | word def bits_to_target(bits: int) -> int: size = bits >> 24 word = bits & 0x007FFFFF if size <= 3: return word >> (8 * (3 - size)) return word << (8 * (size - 3)) def bits_to_hex(bits: int) -> str: return f"{bits:08x}" def work_from_bits(bits: int) -> int: """Expected number of hashes for one block (Bitcoin work metric).""" target = bits_to_target(bits) if target <= 0: return 0 return (1 << 256) // (target + 1) def difficulty_from_bits(bits: int) -> float: target = bits_to_target(bits) if target <= 0: return 0.0 return round(POW_LIMIT / target, 6) def canonical_json(obj: Any) -> bytes: return json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8") def block_subsidy(height: int) -> int: """Block subsidy with tail emission (block 1 = INITIAL_SUBSIDY). The subsidy halves every ``HALVING_INTERVAL`` blocks but never drops below ``MIN_SUBSIDY``. Network security therefore stays funded permanently instead of depending on fees alone at some point. Pure integer arithmetic - every node arrives at the identical result. """ if height < 1: return 0 halvings = (height - 1) // HALVING_INTERVAL if halvings >= 64: return MIN_SUBSIDY return max(INITIAL_SUBSIDY >> halvings, MIN_SUBSIDY) def share_target_from_bits(bits: int) -> int: """Target of a share: easier than the block target, but never easier than ``POW_LIMIT`` (a share would otherwise be free to produce).""" return min(POW_LIMIT, bits_to_target(bits) * SHARE_TARGET_MULTIPLIER) # --------------------------------------------------------------------------- # # 3. Transaktionen # --------------------------------------------------------------------------- # def signing_payload(sender: str, recipient: str, amount: int, fee: int, nonce: int) -> bytes: return canonical_json({ "amount": int(amount), "chain": CHAIN_ID, "fee": int(fee), "nonce": int(nonce), "recipient": recipient, "sender": sender, }) def signature_hash(sender: str, recipient: str, amount: int, fee: int, nonce: int) -> bytes: return sha256d(signing_payload(sender, recipient, amount, fee, nonce)) def transaction_id(tx: Dict[str, Any]) -> str: body = {k: v for k, v in tx.items() if k != "txid"} return sha256d(canonical_json(body))[::-1].hex() def is_coinbase(tx: Dict[str, Any]) -> bool: return bool(tx.get("coinbase")) or tx.get("sender") == COINBASE_SENDER def verify_transaction_signature(tx: Dict[str, Any]) -> bool: try: pubkey = normalize_pubkey(str(tx.get("pubkey", ""))) signature = bytes.fromhex(str(tx.get("signature", ""))) if len(signature) != 64: return False if pubkey_to_address(pubkey.hex()) != tx.get("sender"): return False digest = signature_hash( tx["sender"], tx["recipient"], int(tx["amount"]), int(tx["fee"]), int(tx["nonce"]) ) vk = ecdsa.VerifyingKey.from_string(pubkey[1:], curve=ecdsa.SECP256k1) return vk.verify_digest(signature, digest, sigdecode=ecdsa.util.sigdecode_string) except Exception: return False def build_coinbase(height: int, address: str, total_reward: int, extranonce: int, note: str = "", shares_root: Optional[str] = None, committed_state_root: Optional[str] = None) -> Dict[str, Any]: tx: Dict[str, Any] = { "coinbase": True, "sender": COINBASE_SENDER, "recipient": address, "amount": int(total_reward), "fee": 0, "nonce": 0, "height": height, "extranonce": int(extranonce), "note": note or f"Paritr block #{height}", "pubkey": "", "signature": "", } # The share commitment sits in the first coinbase and therefore, via the # Merkle root, inside the block hash: confirmed shares cannot be swapped. if shares_root is not None: tx["shares_root"] = shares_root # Same mechanism for the periodic account-state checkpoint. if committed_state_root is not None: tx["state_root"] = committed_state_root tx["txid"] = transaction_id(tx) return tx def block_finder(block: Dict[str, Any]) -> str: """Address of the node that found the block. The first coinbase recipient is authoritative: it sits inside the block hash via the Merkle root and therefore cannot be altered afterwards (the ``miner`` field is for display only). """ for tx in block.get("transactions", []): if is_coinbase(tx): return str(tx.get("recipient") or "") return "" def share_weights(parent_chain: List[Dict[str, Any]], new_shares: Iterable[Dict[str, Any]]) -> Dict[str, int]: """Valid shares per address inside the measuring window. Counted are the shares confirmed by the last ``REWARD_WINDOW`` blocks plus the shares the freshly found block confirms. Because a share costs exactly 1/``SHARE_TARGET_MULTIPLIER`` of the block work, the share count is a direct measure of contributed hash power. """ weights: Dict[str, int] = {} for block in parent_chain[-REWARD_WINDOW:]: for share in block.get("shares", []): address = str(share.get("miner_address") or "") if is_valid_address(address): weights[address] = weights.get(address, 0) + 1 for share in new_shares: address = str(share.get("miner_address") or "") if is_valid_address(address): weights[address] = weights.get(address, 0) + 1 return weights def coinbase_payouts(height: int, total_fees: int, finder: str, parent_chain: List[Dict[str, Any]], new_shares: Optional[Iterable[Dict[str, Any]]] = None ) -> List[Tuple[str, int]]: """Deterministic payout plan for the coinbase of a block. * ``FINDER_SHARE_PERCENT`` % of the subsidy go to the finder as an inclusion incentive - withholding foreign shares therefore never pays. * The remainder is distributed strictly by share weight in the window. * All transaction fees go to the finder in full. * Rounding remainders and amounts below ``DUST_LIMIT`` fall back to the finder so ``sum(outputs) == subsidy + fees`` works out exactly. Every node computes the same result, which is why the plan is verifiable without any extra data. """ subsidy = block_subsidy(height) fees = max(0, int(total_fees)) if not is_valid_address(finder): total = subsidy + fees return [(finder, total)] if total > 0 else [] finder_base = subsidy * FINDER_SHARE_PERCENT // 100 pool = subsidy - finder_base weights = share_weights(parent_chain, new_shares or ()) total_weight = sum(weights.values()) # Only the strongest miners get an output of their own; that bounds the # block size and avoids dust payouts. Whatever is left over by that cap or # by rounding falls back to the finder. ranked = sorted(weights.items(), key=lambda kv: (-kv[1], kv[0]))[:MAX_COINBASE_OUTPUTS] shares: Dict[str, int] = {} distributed = 0 if total_weight > 0 and pool > 0: for address, weight in ranked: if address == finder: continue # settled with the base amount below amount = pool * weight // total_weight if amount >= MIN_SHARE_PAYOUT: shares[address] = amount distributed += amount rest = pool - distributed # Rundungsreste + Kleinstanteile finder_amount = finder_base + fees + rest payouts = [(finder, finder_amount)] payouts.extend((address, shares[address]) for address in sorted(shares)) return [(address, amount) for address, amount in payouts if amount > 0] def build_coinbase_transactions(height: int, payouts: List[Tuple[str, int]], extranonce: int, shares_root: Optional[str] = None, committed_state_root: Optional[str] = None ) -> List[Dict[str, Any]]: """Builds the coinbase outputs; the finder always comes first.""" txs: List[Dict[str, Any]] = [] for index, (address, amount) in enumerate(payouts): note = (f"Paritr block #{height}" if index == 0 else f"Paritr block #{height} - work share") txs.append(build_coinbase(height, address, amount, extranonce, note, shares_root if index == 0 else None, committed_state_root if index == 0 else None)) return txs # --------------------------------------------------------------------------- # # 4. Blockheader & Merkle-Baum # --------------------------------------------------------------------------- # def merkle_root(txids: List[str]) -> str: if not txids: return "00" * 32 layer = [bytes.fromhex(t)[::-1] for t in txids] while len(layer) > 1: if len(layer) % 2: layer.append(layer[-1]) layer = [sha256d(layer[i] + layer[i + 1]) for i in range(0, len(layer), 2)] return layer[0][::-1].hex() def merkle_branch(txids: List[str], index: int) -> List[str]: """Merkle path of an entry - proves it belongs to the root.""" if not txids or not 0 <= index < len(txids): return [] layer = [bytes.fromhex(t)[::-1] for t in txids] branch: List[str] = [] while len(layer) > 1: if len(layer) % 2: layer.append(layer[-1]) sibling = index ^ 1 branch.append(layer[sibling][::-1].hex()) layer = [sha256d(layer[i] + layer[i + 1]) for i in range(0, len(layer), 2)] index //= 2 return branch def merkle_root_from_branch(txid: str, branch: List[str], index: int) -> str: """Rebuilds the Merkle root from an entry and its path.""" node = bytes.fromhex(txid)[::-1] for step in branch: sibling = bytes.fromhex(step)[::-1] node = sha256d(node + sibling) if index % 2 == 0 else sha256d(sibling + node) index //= 2 return node[::-1].hex() def state_leaves(balances: Dict[str, int], nonces: Dict[str, int]) -> List[str]: """Canonical leaf list of the account state, ordered by address. Accounts that hold nothing and never sent anything are omitted, so the result depends only on the reachable state and not on how a node happened to arrive at it. """ leaves: List[str] = [] for address in sorted(set(balances) | set(nonces)): balance = int(balances.get(address, 0)) nonce = int(nonces.get(address, 0)) if balance == 0 and nonce == 0: continue payload = canonical_json({"address": address, "balance": balance, "nonce": nonce}) leaves.append(sha256d(payload)[::-1].hex()) return leaves def state_root(balances: Dict[str, int], nonces: Dict[str, int]) -> str: """Merkle commitment over the full account state. Committed in the coinbase every ``STATE_ROOT_INTERVAL`` blocks. It lets a fresh node check a downloaded state snapshot against the chain instead of replaying every historical balance change. """ return merkle_root(state_leaves(balances, nonces)) def is_state_checkpoint(height: int) -> bool: return height > 0 and height % STATE_ROOT_INTERVAL == 0 def serialize_header(block: Dict[str, Any]) -> bytes: """80-byte block header, matching Bitcoin (little-endian fields).""" return ( int(block["version"]).to_bytes(4, "little") + bytes.fromhex(block["previous_hash"])[::-1] + bytes.fromhex(block["merkle_root"])[::-1] + int(block["timestamp"]).to_bytes(4, "little") + int(block["bits"], 16).to_bytes(4, "little") + int(block["nonce"]).to_bytes(4, "little") ) def compute_block_hash(block: Dict[str, Any]) -> str: return sha256d(serialize_header(block))[::-1].hex() def header_only(block: Dict[str, Any]) -> Dict[str, Any]: return { "height": block["height"], "version": block["version"], "previous_hash": block["previous_hash"], "merkle_root": block["merkle_root"], "timestamp": block["timestamp"], "bits": block["bits"], "nonce": block["nonce"], "hash": block["hash"], "tx_count": len(block.get("transactions", [])), "share_count": len(block.get("shares", [])), } # --------------------------------------------------------------------------- # # 4b. Shares (P2Pool-artiger Leistungsnachweis) # --------------------------------------------------------------------------- # class ConsensusError(Exception): """A block, transaction or share violates the consensus rules.""" MAX_SHARE_BRANCH = 16 # deckt bis zu 65.536 Transaktionen ab def serialize_share_header(share: Dict[str, Any]) -> bytes: """80-byte header of a share - identical layout to a block header. A share therefore falls out of the same hash loop as the block search: whoever looks for a block keeps finding shares along the way. """ return ( int(share["version"]).to_bytes(4, "little") + bytes.fromhex(share["parent_hash"])[::-1] + bytes.fromhex(share["merkle_root"])[::-1] + int(share["timestamp"]).to_bytes(4, "little") + int(share["bits"], 16).to_bytes(4, "little") + int(share["nonce"]).to_bytes(4, "little") ) def share_id(share: Dict[str, Any]) -> str: """Identifier of a share - the header hash is already unique.""" return str(share.get("header_hash") or "") def build_share(block: Dict[str, Any], nonce: int, miner_address: str, txids: List[str]) -> Dict[str, Any]: """Builds a share from a candidate block and the nonce that was found.""" share = { "miner_address": miner_address, "version": int(block["version"]), "parent_hash": str(block["previous_hash"]), "merkle_root": str(block["merkle_root"]), "timestamp": int(block["timestamp"]), "bits": str(block["bits"]), "nonce": int(nonce), "coinbase": block["transactions"][0], "coinbase_branch": merkle_branch(txids, 0), } share["header_hash"] = sha256d(serialize_share_header(share))[::-1].hex() return share def _is_hex(value: Any, length: int) -> bool: if not isinstance(value, str) or len(value) != length: return False try: bytes.fromhex(value) except ValueError: return False return True def validate_share_pow(share: Any) -> None: """Checks structure, proof of work and address binding of a share. The payout address sits in the candidate's coinbase and therefore, via the Merkle path, inside the hashed header. A relay cannot swap it for its own without destroying the proof of work. The call is deliberately cheap and serves as the first denial-of-service barrier for incoming shares. """ if not isinstance(share, dict): raise ConsensusError("Share has an invalid format.") address = share.get("miner_address") if not is_valid_address(address): raise ConsensusError("Share names no valid miner address.") if not _is_hex(share.get("parent_hash"), 64) or not _is_hex(share.get("merkle_root"), 64): raise ConsensusError("Share header is incomplete.") if not _is_hex(share.get("bits"), 8) or not _is_hex(share.get("header_hash"), 64): raise ConsensusError("Share header is incomplete.") try: version = int(share["version"]) timestamp = int(share["timestamp"]) nonce = int(share["nonce"]) except (KeyError, TypeError, ValueError): raise ConsensusError("Share header is incomplete.") if not 0 <= nonce <= 0xFFFFFFFF or not 0 <= version <= 0xFFFFFFFF: raise ConsensusError("Share header out of range.") if not 0 <= timestamp <= 0xFFFFFFFF: raise ConsensusError("Share timestamp out of range.") header = serialize_share_header(share) if sha256d(header)[::-1].hex() != share["header_hash"]: raise ConsensusError("Share hash does not match the header.") bits = int(share["bits"], 16) if cached_pow_value(header) > share_target_from_bits(bits): raise ConsensusError("Share does not meet the share target.") coinbase = share.get("coinbase") branch = share.get("coinbase_branch") if not isinstance(coinbase, dict) or not is_coinbase(coinbase): raise ConsensusError("Share carries no coinbase proof.") if not isinstance(branch, list) or len(branch) > MAX_SHARE_BRANCH: raise ConsensusError("Merkle path of the share is invalid.") if any(not _is_hex(step, 64) for step in branch): raise ConsensusError("Merkle path of the share is invalid.") if str(coinbase.get("recipient") or "") != address: raise ConsensusError("Share coinbase does not belong to the miner address.") if merkle_root_from_branch(transaction_id(coinbase), branch, 0) != share["merkle_root"]: raise ConsensusError("Merkle path does not prove the share coinbase.") def compact_share(share: Dict[str, Any]) -> Dict[str, Any]: """Consensus-only view of a share, used for storage inside a block.""" return { "miner_address": str(share["miner_address"]), "version": int(share["version"]), "parent_hash": str(share["parent_hash"]), "merkle_root": str(share["merkle_root"]), "timestamp": int(share["timestamp"]), "bits": str(share["bits"]).lower(), "nonce": int(share["nonce"]), "header_hash": str(share["header_hash"]), "coinbase": share["coinbase"], "coinbase_branch": [str(step) for step in share["coinbase_branch"]], } # --------------------------------------------------------------------------- # # 4c. Compact binary wire format (node-to-node gossip) # --------------------------------------------------------------------------- # # # REST/JSON stays the interface for light clients and debugging, but shares are # the most frequent message on the network and JSON is a poor carrier for them: # 64 hex characters per hash plus quoting roughly triples every field. The # binary framing below packs the fixed-width header fields as raw bytes and # carries the coinbase as deflate-compressed canonical JSON, so the transaction # id it hashes to round-trips byte for byte. SHARE_WIRE_MAGIC = b"PSH1" BLOCK_WIRE_MAGIC = b"PBK1" MAX_WIRE_MESSAGE = 4 * 1024 * 1024 # hard ceiling for an inbound frame class WireFormatError(Exception): """A binary frame is malformed and cannot be decoded.""" def encode_share(share: Dict[str, Any]) -> bytes: """Packs a share into the compact gossip frame.""" address = str(share["miner_address"]).encode("ascii") if len(address) > 255: raise WireFormatError("Miner address too long.") branch = [str(step) for step in share["coinbase_branch"]] if len(branch) > 255: raise WireFormatError("Merkle path too long.") coinbase = zlib.compress(canonical_json(share["coinbase"]), 6) out = bytearray(SHARE_WIRE_MAGIC) out.append(len(address)) out += address out += int(share["version"]).to_bytes(4, "little") out += bytes.fromhex(str(share["parent_hash"])) out += bytes.fromhex(str(share["merkle_root"])) out += int(share["timestamp"]).to_bytes(4, "little") out += int(str(share["bits"]), 16).to_bytes(4, "little") out += int(share["nonce"]).to_bytes(4, "little") out.append(len(branch)) for step in branch: out += bytes.fromhex(step) out += len(coinbase).to_bytes(4, "little") out += coinbase return bytes(out) def decode_share(raw: bytes) -> Dict[str, Any]: """Rebuilds a share from its compact frame. Every length is validated against the remaining buffer before it is used, so a truncated or hostile frame raises instead of allocating wildly. """ if len(raw) > MAX_WIRE_MESSAGE: raise WireFormatError("Frame exceeds the size limit.") view = memoryview(raw) if len(view) < 5 or bytes(view[:4]) != SHARE_WIRE_MAGIC: raise WireFormatError("Not a share frame.") pos = 4 def take(count: int) -> bytes: nonlocal pos if count < 0 or pos + count > len(view): raise WireFormatError("Frame truncated.") chunk = bytes(view[pos:pos + count]) pos += count return chunk address_len = take(1)[0] try: address = take(address_len).decode("ascii") except UnicodeDecodeError: raise WireFormatError("Miner address is not ASCII.") version = int.from_bytes(take(4), "little") parent_hash = take(32).hex() merkle = take(32).hex() timestamp = int.from_bytes(take(4), "little") bits = f"{int.from_bytes(take(4), 'little'):08x}" nonce = int.from_bytes(take(4), "little") branch = [take(32).hex() for _ in range(take(1)[0])] coinbase_len = int.from_bytes(take(4), "little") if coinbase_len > MAX_WIRE_MESSAGE: raise WireFormatError("Coinbase section too large.") try: coinbase = json.loads(zlib.decompress(take(coinbase_len), bufsize=MAX_WIRE_MESSAGE)) except (zlib.error, ValueError): raise WireFormatError("Coinbase section is unreadable.") if not isinstance(coinbase, dict): raise WireFormatError("Coinbase section is not an object.") share = { "miner_address": address, "version": version, "parent_hash": parent_hash, "merkle_root": merkle, "timestamp": timestamp, "bits": bits, "nonce": nonce, "coinbase": coinbase, "coinbase_branch": branch, } # Derived rather than transmitted: one field less on the wire, and the # value can never contradict the header it is supposed to describe. share["header_hash"] = sha256d(serialize_share_header(share))[::-1].hex() return share def encode_block(block: Dict[str, Any]) -> bytes: """Packs a block into a compressed gossip frame. A block is dominated by its transaction list, which has no fixed shape, so the win here comes from compression rather than field packing. Typical saving over raw JSON is 70-85 %. """ return BLOCK_WIRE_MAGIC + zlib.compress(canonical_json(block), 6) def decode_block(raw: bytes) -> Dict[str, Any]: if len(raw) > MAX_WIRE_MESSAGE: raise WireFormatError("Frame exceeds the size limit.") if len(raw) < 5 or raw[:4] != BLOCK_WIRE_MAGIC: raise WireFormatError("Not a block frame.") try: block = json.loads(zlib.decompress(raw[4:], bufsize=MAX_WIRE_MESSAGE)) except (zlib.error, ValueError): raise WireFormatError("Block frame is unreadable.") if not isinstance(block, dict): raise WireFormatError("Block frame is not an object.") return block # --------------------------------------------------------------------------- # # 5. Persistenz (SQLite Blockstore) # --------------------------------------------------------------------------- # class BlockStore: """Block store with support for reorganisations.""" def __init__(self, path: str): self.path = path self._local = threading.local() conn = self._connect() conn.execute( "CREATE TABLE IF NOT EXISTS blocks (" " height INTEGER PRIMARY KEY," " hash TEXT NOT NULL UNIQUE," " data TEXT NOT NULL)" ) conn.commit() def _connect(self) -> sqlite3.Connection: conn = getattr(self._local, "conn", None) if conn is None: conn = sqlite3.connect(self.path, check_same_thread=False, timeout=30) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA synchronous=NORMAL") self._local.conn = conn return conn def load_all(self) -> List[Dict[str, Any]]: rows = self._connect().execute("SELECT data FROM blocks ORDER BY height ASC").fetchall() return [json.loads(row[0]) for row in rows] def append(self, block: Dict[str, Any]) -> None: conn = self._connect() conn.execute( "INSERT OR REPLACE INTO blocks (height, hash, data) VALUES (?, ?, ?)", (block["height"], block["hash"], json.dumps(block, separators=(",", ":"))), ) conn.commit() def replace_from(self, height: int, blocks: Iterable[Dict[str, Any]]) -> None: conn = self._connect() conn.execute("DELETE FROM blocks WHERE height >= ?", (height,)) conn.executemany( "INSERT OR REPLACE INTO blocks (height, hash, data) VALUES (?, ?, ?)", [(b["height"], b["hash"], json.dumps(b, separators=(",", ":"))) for b in blocks], ) conn.commit() # --------------------------------------------------------------------------- # # 6. Blockchain-Kern # --------------------------------------------------------------------------- # class Blockchain: def __init__(self, config: Dict[str, Any], data_dir: str): self.config = config self.data_dir = data_dir self.lock = threading.RLock() self.tip_changed = threading.Event() self.store = BlockStore(os.path.join(data_dir, "chain.sqlite")) self.chain: List[Dict[str, Any]] = self.store.load_all() self.mempool: Dict[str, Dict[str, Any]] = {} # Transactions with a nonce gap (out-of-order delivery over P2P) self.orphan_txs: Dict[str, Dict[str, Any]] = {} # Received, not yet confirmed shares (in memory, short lived) self.shares: Dict[str, Dict[str, Any]] = {} # Identifier -> height of every confirmed share (O(1) replay protection) self.confirmed_shares: Dict[str, int] = {} self.balances: Dict[str, int] = {} self.nonces: Dict[str, int] = {} self.history: Dict[str, List[Tuple[int, str]]] = {} self.coinbase_heights: Dict[str, List[Tuple[int, int]]] = {} # Blocks of competing branches, until a complete branch is available self.side_blocks: Dict[str, Dict[str, Any]] = {} self.total_work = 0 if not self.chain: self._install_genesis() else: self._assert_matching_chain() self._rebuild_state() # ---------------------------------------------------------------- Genesis @staticmethod def build_genesis() -> Dict[str, Any]: """The genesis block, computed identically on every node.""" coinbase = build_coinbase(0, "", 0, 0, GENESIS_MESSAGE) block: Dict[str, Any] = { "height": 0, "version": 1, "previous_hash": "00" * 32, "merkle_root": merkle_root([coinbase["txid"]]), "timestamp": GENESIS_TIMESTAMP, "bits": bits_to_hex(target_to_bits(GENESIS_TARGET)), "nonce": 0, "transactions": [coinbase], "miner": "", "mining_duration": 0.0, } target = bits_to_target(int(block["bits"], 16)) nonce = 0 while True: block["nonce"] = nonce if pow_value(serialize_header(block)) <= target: break nonce += 1 block["hash"] = compute_block_hash(block) block["chain_work"] = work_from_bits(int(block["bits"], 16)) return block def _assert_matching_chain(self) -> None: """Refuses to start on a data directory from a different chain. Changing the proof of work, the address version or the genesis message produces a different genesis block. Silently continuing would leave the node mining a chain nobody else follows - which looks exactly like "everything is zero" from the outside. Better to stop and say so. """ expected = self.build_genesis()["hash"] actual = str(self.chain[0].get("hash") or "") if actual == expected: return raise SystemExit( f"The data directory belongs to a different chain.\n" f" expected genesis: {expected}\n" f" found genesis: {actual}\n" f" data directory: {self.data_dir}\n\n" f"This happens after a protocol change or a chain restart.\n" f"Stop the node, delete the data directory and start again." ) def _install_genesis(self) -> None: block = self.build_genesis() self.chain = [block] self.store.replace_from(0, self.chain) log.info("Genesis block created: %s", block["hash"]) # ------------------------------------------------------------- Chain-Info @property def tip(self) -> Dict[str, Any]: return self.chain[-1] @property def height(self) -> int: return self.chain[-1]["height"] def block_at(self, height: int) -> Optional[Dict[str, Any]]: if 0 <= height < len(self.chain): return self.chain[height] return None def block_by_hash(self, block_hash: str) -> Optional[Dict[str, Any]]: for block in reversed(self.chain): if block["hash"] == block_hash: return block return None def headers(self, start: int, count: int) -> List[Dict[str, Any]]: return [header_only(b) for b in self.chain[start:start + count]] def blocks(self, start: int, count: int) -> List[Dict[str, Any]]: return self.chain[start:start + count] # ------------------------------------------------------------- Difficulty @staticmethod def next_bits(chain: List[Dict[str, Any]]) -> str: """Retargeting after Bitcoin's model, damped to a factor of 4.""" tip = chain[-1] next_height = tip["height"] + 1 current_bits = int(tip["bits"], 16) if next_height < RETARGET_INTERVAL: return bits_to_hex(target_to_bits(INITIAL_TARGET)) if next_height % RETARGET_INTERVAL != 0: return bits_to_hex(current_bits) # The genesis block carries a hardcoded timestamp and would skew the # first measuring window - so measure from block 1 onwards. first = chain[max(1, next_height - RETARGET_INTERVAL)] actual_span = max(1, int(tip["timestamp"]) - int(first["timestamp"])) target_span = (tip["height"] - first["height"]) * TARGET_BLOCK_TIME actual_span = max(target_span // MAX_RETARGET_FACTOR, min(target_span * MAX_RETARGET_FACTOR, actual_span)) new_target = bits_to_target(current_bits) * actual_span // target_span new_target = max(1, min(POW_LIMIT, new_target)) return bits_to_hex(target_to_bits(new_target)) @staticmethod def median_time_past(chain: List[Dict[str, Any]]) -> int: window = sorted(int(b["timestamp"]) for b in chain[-MEDIAN_TIME_SPAN:]) return window[len(window) // 2] # ------------------------------------------------------------ State-Index def _rebuild_state(self) -> None: self.balances = {} self.nonces = {} self.history = {} self.coinbase_heights = {} self.confirmed_shares = {} self.total_work = 0 for block in self.chain: self._apply_block_to_state(block) self._prune_mempool() def _apply_block_to_state(self, block: Dict[str, Any]) -> None: self.total_work += work_from_bits(int(block["bits"], 16)) height = block["height"] for share in block.get("shares", []): self.confirmed_shares[share_id(share)] = height for tx in block.get("transactions", []): txid = tx.get("txid") or transaction_id(tx) recipient = tx.get("recipient") or "" amount = int(tx.get("amount", 0)) fee = int(tx.get("fee", 0)) if recipient: self.balances[recipient] = self.balances.get(recipient, 0) + amount self.history.setdefault(recipient, []).append((height, txid)) if is_coinbase(tx): if recipient and amount > 0: self.coinbase_heights.setdefault(recipient, []).append((height, amount)) continue sender = tx["sender"] self.balances[sender] = self.balances.get(sender, 0) - (amount + fee) self.nonces[sender] = int(tx["nonce"]) + 1 if sender != recipient: self.history.setdefault(sender, []).append((height, txid)) def _undo_block_from_state(self, block: Dict[str, Any]) -> None: """Exact inverse of ``_apply_block_to_state``. This lets a reorganisation be unwound incrementally from the fork point instead of rebuilding the entire state from block 0. Without it the cost of every reorg would grow with the chain length - a welcome target for denial of service. """ height = block["height"] for share in block.get("shares", []): self.confirmed_shares.pop(share_id(share), None) for tx in reversed(block.get("transactions", [])): txid = tx.get("txid") or transaction_id(tx) recipient = tx.get("recipient") or "" amount = int(tx.get("amount", 0)) fee = int(tx.get("fee", 0)) if is_coinbase(tx): if recipient and amount > 0: entries = self.coinbase_heights.get(recipient) if entries and entries[-1] == (height, amount): entries.pop() if entries is not None and not entries: self.coinbase_heights.pop(recipient, None) else: sender = tx["sender"] self.balances[sender] = self.balances.get(sender, 0) + (amount + fee) # Nonces are strictly ascending per sender, so undoing them in # reverse order restores exactly the previous state. self.nonces[sender] = int(tx["nonce"]) if self.nonces[sender] <= 0: self.nonces.pop(sender, None) if sender != recipient: self._pop_history(sender, height, txid) if recipient: self.balances[recipient] = self.balances.get(recipient, 0) - amount self._pop_history(recipient, height, txid) if self.balances.get(recipient) == 0: self.balances.pop(recipient, None) self.total_work -= work_from_bits(int(block["bits"], 16)) def _pop_history(self, address: str, height: int, txid: str) -> None: entries = self.history.get(address) if entries and entries[-1] == (height, txid): entries.pop() if entries is not None and not entries: self.history.pop(address, None) def confirmed_balance(self, address: str) -> int: return self.balances.get(address, 0) # ------------------------------------------------------- State snapshots def _snapshot_files(self) -> List[str]: try: names = [name for name in os.listdir(self.data_dir) if name.startswith("state-") and name.endswith(".json")] except OSError: return [] return sorted(names, key=lambda name: int(name[6:-5]) if name[6:-5].isdigit() else 0) def _write_state_snapshot(self, committed_at: int, root: str) -> None: """Persists the account state that a checkpoint block commits to. Called just before the checkpoint block is applied, so ``self.balances`` still holds the parent state - exactly what ``state_root`` covered. """ payload = { "chain_id": CHAIN_ID, "height": committed_at - 1, "committed_at": committed_at, "state_root": root, "accounts": [ {"address": address, "balance": int(self.balances.get(address, 0)), "nonce": int(self.nonces.get(address, 0))} for address in sorted(set(self.balances) | set(self.nonces)) if self.balances.get(address, 0) or self.nonces.get(address, 0) ], } path = os.path.join(self.data_dir, f"state-{committed_at}.json") try: tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as handle: json.dump(payload, handle, separators=(",", ":")) os.replace(tmp, path) except OSError as exc: log.warning("Could not write state snapshot: %s", exc) return for name in self._snapshot_files()[:-MAX_STATE_SNAPSHOTS]: try: os.remove(os.path.join(self.data_dir, name)) except OSError: pass def latest_state_snapshot(self) -> Optional[Dict[str, Any]]: """Newest persisted state checkpoint, or None if none exists yet.""" files = self._snapshot_files() if not files: return None try: with open(os.path.join(self.data_dir, files[-1]), "r", encoding="utf-8") as handle: return json.load(handle) except (OSError, ValueError): return None def immature_balance(self, address: str) -> int: tip = self.height return sum(amount for h, amount in self.coinbase_heights.get(address, []) if tip - h < COINBASE_MATURITY) def pending_outgoing(self, address: str) -> int: return sum(int(tx["amount"]) + int(tx["fee"]) for tx in self.mempool.values() if tx["sender"] == address) def pending_incoming(self, address: str) -> int: return sum(int(tx["amount"]) for tx in self.mempool.values() if tx["recipient"] == address) def spendable_balance(self, address: str) -> int: return (self.confirmed_balance(address) - self.immature_balance(address) - self.pending_outgoing(address)) def next_nonce(self, address: str) -> int: nonce = self.nonces.get(address, 0) for tx in self.mempool.values(): if tx["sender"] == address: nonce = max(nonce, int(tx["nonce"]) + 1) return nonce def circulating_supply(self) -> int: return sum(int(tx.get("amount", 0)) for block in self.chain for tx in block.get("transactions", []) if is_coinbase(tx)) # --------------------------------------------------------------- Historie def address_history(self, address: str, limit: int = 50, offset: int = 0) -> List[Dict[str, Any]]: entries = list(reversed(self.history.get(address, []))) tip = self.height result: List[Dict[str, Any]] = [] if offset == 0: for tx in self.mempool.values(): if tx["sender"] == address or tx["recipient"] == address: result.append(self._describe_tx(tx, address, None, tip)) for height, txid in entries[offset:offset + limit]: block = self.block_at(height) if not block: continue for tx in block["transactions"]: if (tx.get("txid") or transaction_id(tx)) == txid: result.append(self._describe_tx(tx, address, block, tip)) break return result def _describe_tx(self, tx: Dict[str, Any], address: str, block: Optional[Dict[str, Any]], tip: int) -> Dict[str, Any]: coinbase = is_coinbase(tx) incoming = tx.get("recipient") == address height = block["height"] if block else None confirmations = (tip - height + 1) if height is not None else 0 if coinbase: direction = "reward" elif incoming: direction = "in" else: direction = "out" return { "txid": tx.get("txid") or transaction_id(tx), "direction": direction, "coinbase": coinbase, "counterparty": tx.get("sender") if incoming else tx.get("recipient"), "amount": int(tx.get("amount", 0)), "fee": int(tx.get("fee", 0)), "nonce": int(tx.get("nonce", 0)), "height": height, "confirmations": confirmations, "matured": (not coinbase) or confirmations > COINBASE_MATURITY, "timestamp": int(block["timestamp"]) if block else int(tx.get("received", time.time())), "status": "confirmed" if block else "pending", "note": tx.get("note", ""), } def find_transaction(self, txid: str) -> Optional[Dict[str, Any]]: if txid in self.mempool: return {"transaction": self.mempool[txid], "height": None, "confirmations": 0} for block in reversed(self.chain): for tx in block["transactions"]: if (tx.get("txid") or transaction_id(tx)) == txid: return { "transaction": tx, "height": block["height"], "confirmations": self.height - block["height"] + 1, } return None # ---------------------------------------------------------------- Mempool def accept_transaction(self, payload: Dict[str, Any]) -> Dict[str, Any]: """Accepts a signed transaction. Returns ``{"transaction": ..., "status": ...}``. The status is ``accepted`` (in the mempool), ``known`` (already seen) or ``buffered`` (nonce gap, see ``_buffer_orphan_tx``). """ try: sender = str(payload["sender"]).strip() recipient = str(payload["recipient"]).strip() amount = int(payload["amount"]) fee = int(payload["fee"]) nonce = int(payload["nonce"]) pubkey = str(payload["pubkey"]).strip() signature = str(payload["signature"]).strip() except (KeyError, TypeError, ValueError): raise ConsensusError("Incomplete transaction data.") if not is_valid_address(sender) or not is_valid_address(recipient): raise ConsensusError("Invalid wallet address.") if sender == recipient: raise ConsensusError("Sender and recipient are identical.") if amount < DUST_LIMIT: raise ConsensusError(f"Minimum amount: {DUST_LIMIT / COIN:.8f} PAR.") if fee < MIN_RELAY_FEE: raise ConsensusError(f"Minimum fee: {MIN_RELAY_FEE / COIN:.8f} PAR.") if amount > MAX_MONEY or fee > MAX_MONEY or nonce < 0: raise ConsensusError("Amount or nonce out of range.") tx = { "sender": sender, "recipient": recipient, "amount": amount, "fee": fee, "nonce": nonce, "pubkey": pubkey, "signature": signature, "received": int(payload.get("received") or time.time()), } if not verify_transaction_signature(tx): raise ConsensusError("Signature is invalid or does not match the sender address.") tx["txid"] = transaction_id(tx) with self.lock: if tx["txid"] in self.mempool: return {"transaction": self.mempool[tx["txid"]], "status": "known"} if tx["txid"] in self.orphan_txs: return {"transaction": self.orphan_txs[tx["txid"]], "status": "buffered"} expected = self.next_nonce(sender) if nonce < expected: raise ConsensusError( f"Nonce already used. Expected: {expected}, received: {nonce}.") if nonce > expected: self._buffer_orphan_tx(tx, expected) return {"transaction": tx, "status": "buffered"} if self.spendable_balance(sender) < amount + fee: raise ConsensusError("Available balance does not cover amount and fee.") self.mempool[tx["txid"]] = tx promoted = self._promote_orphan_txs(sender) log.info("Transaction in mempool: %s (%.8f PAR)%s", tx["txid"][:16], amount / COIN, f" - {promoted} promoted" if promoted else "") return {"transaction": tx, "status": "accepted"} def _buffer_orphan_tx(self, tx: Dict[str, Any], expected: int) -> None: """Parks a transaction whose predecessor nonce is still missing. Over P2P, transactions from the same sender do not necessarily arrive in order. Dropping a transaction with nonce n+1 immediately would tear valid chains apart and force wallets to resend. The buffer holds it for ``ORPHAN_TX_TTL`` seconds instead and promotes it as soon as the gap closes. """ if int(tx["nonce"]) - expected > MAX_NONCE_GAP: raise ConsensusError( f"Nonce gap too large. Expected: {expected}, received: {tx['nonce']}.") if len(self.orphan_txs) >= MAX_ORPHAN_TRANSACTIONS: oldest = min(self.orphan_txs.items(), key=lambda kv: int(kv[1]["received"]))[0] self.orphan_txs.pop(oldest, None) self.orphan_txs[tx["txid"]] = tx log.debug("Transaction %s buffered (nonce %s, expected %s)", tx["txid"][:16], tx["nonce"], expected) def _promote_orphan_txs(self, sender: Optional[str] = None) -> int: """Promotes buffered transactions as soon as they fit. Without ``sender`` every sender is checked (after a block), otherwise only the one just affected - that keeps the accept path cheap. """ if not self.orphan_txs: return 0 by_sender: Dict[str, List[Tuple[int, str, Dict[str, Any]]]] = {} for txid, tx in self.orphan_txs.items(): if sender is not None and tx["sender"] != sender: continue by_sender.setdefault(tx["sender"], []).append((int(tx["nonce"]), txid, tx)) promoted = 0 for address, entries in by_sender.items(): entries.sort(key=lambda entry: entry[0]) expected = self.next_nonce(address) for nonce, txid, tx in entries: if nonce != expected: break # the gap still stands if self.spendable_balance(address) < int(tx["amount"]) + int(tx["fee"]): break self.orphan_txs.pop(txid, None) self.mempool[txid] = tx expected = nonce + 1 promoted += 1 return promoted def _prune_orphan_txs(self) -> None: now = int(time.time()) self.orphan_txs = { txid: tx for txid, tx in self.orphan_txs.items() if now - int(tx.get("received", now)) <= ORPHAN_TX_TTL and int(tx["nonce"]) >= self.nonces.get(tx["sender"], 0) and txid not in self.mempool } def _prune_mempool(self) -> None: """Removes confirmed, stale or no longer valid transactions.""" confirmed = { tx.get("txid") or transaction_id(tx) for block in self.chain[-max(COINBASE_MATURITY * 10, 20):] for tx in block.get("transactions", []) } now = int(time.time()) keep: Dict[str, Dict[str, Any]] = {} for txid, tx in sorted(self.mempool.items(), key=lambda kv: int(kv[1]["nonce"])): if txid in confirmed: continue if now - int(tx.get("received", now)) > 86400: continue if int(tx["nonce"]) < self.nonces.get(tx["sender"], 0): continue keep[txid] = tx self.mempool = keep self.orphan_txs = {txid: tx for txid, tx in self.orphan_txs.items() if txid not in confirmed} self._prune_orphan_txs() self._promote_orphan_txs() def select_block_transactions(self, reserved: int = 1) -> List[Dict[str, Any]]: """Picks transactions by highest fee; an empty mempool means an empty block.""" candidates = sorted(self.mempool.values(), key=lambda t: (-int(t["fee"]), int(t["nonce"]))) balances = dict(self.balances) nonces = dict(self.nonces) selected: List[Dict[str, Any]] = [] for tx in candidates: if len(selected) >= MAX_BLOCK_TRANSACTIONS - max(1, reserved): break sender = tx["sender"] total = int(tx["amount"]) + int(tx["fee"]) if nonces.get(sender, 0) != int(tx["nonce"]): continue if balances.get(sender, 0) - self.immature_balance(sender) < total: continue balances[sender] = balances.get(sender, 0) - total balances[tx["recipient"]] = balances.get(tx["recipient"], 0) + int(tx["amount"]) nonces[sender] = int(tx["nonce"]) + 1 selected.append(tx) return selected # ----------------------------------------------------------------- Shares @staticmethod def confirmed_share_ids(parent_chain: List[Dict[str, Any]]) -> set: """All share identifiers already confirmed inside the lifetime window. Basis of the replay protection: a share counts exactly once, no matter how many miners include it in their candidates. """ return { str(share.get("header_hash") or "") for block in parent_chain[-SHARE_TTL_BLOCKS:] for share in block.get("shares", []) } @staticmethod def _share_parent_index(parent_chain: List[Dict[str, Any]]) -> Dict[str, int]: """Hash -> index of the blocks a share may build on. Built once per pass so checking many shares does not search the chain over and over. """ offset = max(0, len(parent_chain) - SHARE_TTL_BLOCKS) return {block["hash"]: offset + index for index, block in enumerate(parent_chain[offset:])} def validate_share(self, share: Dict[str, Any], parent_chain: List[Dict[str, Any]], reference_time: Optional[int] = None, index_by_hash: Optional[Dict[str, int]] = None) -> str: """Checks a share against the proof of work and the chain context. The context is deliberately deterministic: parent block, difficulty and time window follow from ``parent_chain`` alone. Every node therefore reaches the same verdict when validating a block. """ validate_share_pow(share) if index_by_hash is None: index_by_hash = self._share_parent_index(parent_chain) index = index_by_hash.get(str(share["parent_hash"])) if index is None: raise ConsensusError("Share does not reference any block of this chain.") parent = parent_chain[index] if str(share["bits"]).lower() != self.next_bits(parent_chain[:index + 1]): raise ConsensusError("Share uses the wrong difficulty.") coinbase = share["coinbase"] if int(coinbase.get("height", -1)) != int(parent["height"]) + 1: raise ConsensusError("Share coinbase states the wrong height.") timestamp = int(share["timestamp"]) limit = int(reference_time if reference_time is not None else time.time()) if timestamp > limit + MAX_FUTURE_BLOCK_TIME: raise ConsensusError("Share is too far in the future.") if timestamp + MAX_FUTURE_BLOCK_TIME < int(parent["timestamp"]): raise ConsensusError("Share is too old.") return share_id(share) def accept_share(self, payload: Any) -> str: """Takes a foreign share into the local store. The order of checks is intentional: first the proof of work, which is cheap to verify but expensive to forge, then duplicates and chain context. A flooding attempt therefore costs the attacker far more than the node - and every lookup is constant cost rather than dependent on the chain length. """ validate_share_pow(payload) share = compact_share(payload) identifier = share_id(share) with self.lock: if identifier in self.shares or identifier in self.confirmed_shares: return "known" self.validate_share(share, self.chain) if len(self.shares) >= MAX_SHARE_POOL: oldest = min(self.shares.items(), key=lambda kv: int(kv[1]["timestamp"]))[0] self.shares.pop(oldest, None) self.shares[identifier] = share return "accepted" def select_block_shares(self, parent_chain: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Picks up to ``MAX_SHARES_PER_BLOCK`` not yet confirmed shares.""" confirmed = self.confirmed_share_ids(parent_chain) index_by_hash = self._share_parent_index(parent_chain) selected: List[Dict[str, Any]] = [] for identifier, share in sorted(self.shares.items(), key=lambda kv: (int(kv[1]["timestamp"]), kv[0])): if identifier in confirmed: continue try: self.validate_share(share, parent_chain, None, index_by_hash) except ConsensusError: continue selected.append(share) if len(selected) >= MAX_SHARES_PER_BLOCK: break return selected def _prune_shares(self) -> None: """Removes confirmed and stale shares from the store.""" index_by_hash = self._share_parent_index(self.chain) keep: Dict[str, Dict[str, Any]] = {} for identifier, share in self.shares.items(): if identifier in self.confirmed_shares: continue try: self.validate_share(share, self.chain, None, index_by_hash) except ConsensusError: continue keep[identifier] = share self.shares = keep # Confirmed shares outside the lifetime window can never be included # again - their index entry is dispensable. cutoff = self.height - SHARE_TTL_BLOCKS if cutoff > 0: self.confirmed_shares = {key: height for key, height in self.confirmed_shares.items() if height > cutoff} def share_statistics(self, parent_chain: Optional[List[Dict[str, Any]]] = None ) -> Dict[str, Any]: """Aggregated metrics of the measuring window - basis of the mining view.""" chain = parent_chain if parent_chain is not None else self.chain window = chain[-REWARD_WINDOW:] weights = share_weights(window, ()) total = sum(weights.values()) span = max(1, int(window[-1]["timestamp"]) - int(window[0]["timestamp"])) \ if len(window) > 1 else TARGET_BLOCK_TIME return { "window": REWARD_WINDOW, "window_blocks": len(window), "window_seconds": span, "total_shares": total, "active_miners": len(weights), "share_rate": total / span if span else 0.0, "weights": weights, } # ------------------------------------------------------------ Validierung def validate_block(self, block: Dict[str, Any], parent_chain: List[Dict[str, Any]], balances: Dict[str, int], nonces: Dict[str, int], coinbases: Dict[str, List[Tuple[int, int]]]) -> None: """Checks a block against every consensus rule (state = end of parent_chain).""" required = ("height", "version", "previous_hash", "merkle_root", "timestamp", "bits", "nonce", "transactions") for key in required: if key not in block: raise ConsensusError(f"Incomplete block: '{key}' is missing.") parent = parent_chain[-1] height = int(block["height"]) if height != parent["height"] + 1: raise ConsensusError("Wrong block height.") if block["previous_hash"] != parent["hash"]: raise ConsensusError("Previous block hash does not match.") if str(block["bits"]).lower() != self.next_bits(parent_chain): raise ConsensusError("Invalid difficulty (nBits).") # The committed state root describes the account state as of the parent # block, so it has to be taken before this block mutates anything. checkpoint = is_state_checkpoint(height) expected_state_root = state_root(balances, nonces) if checkpoint else None timestamp = int(block["timestamp"]) if timestamp > time.time() + MAX_FUTURE_BLOCK_TIME: raise ConsensusError("Block is too far in the future.") if timestamp <= self.median_time_past(parent_chain): raise ConsensusError("Timestamp violates the median-time-past rule.") txs = block["transactions"] if not txs or not is_coinbase(txs[0]): raise ConsensusError("The first transaction must be the coinbase.") if len(txs) > MAX_BLOCK_TRANSACTIONS: raise ConsensusError("Maximum block size exceeded.") # The coinbase may carry several outputs (work shares). They have to sit # together at the very start of the block. coinbase_count = 0 for tx in txs: if not is_coinbase(tx): break coinbase_count += 1 if any(is_coinbase(tx) for tx in txs[coinbase_count:]): raise ConsensusError("Coinbase outputs must come first in the block.") if coinbase_count > MAX_COINBASE_OUTPUTS: raise ConsensusError("Too many coinbase outputs.") coinbase_txs = txs[:coinbase_count] txids: List[str] = [] for tx in txs: txid = transaction_id(tx) if tx.get("txid") and tx["txid"] != txid: raise ConsensusError("Transaction id was tampered with.") if txid in txids: raise ConsensusError("Duplicate transaction in the block.") txids.append(txid) if merkle_root(txids) != block["merkle_root"]: raise ConsensusError("Merkle root does not match.") if compute_block_hash(block) != block.get("hash"): raise ConsensusError("Block hash does not match the header.") target = bits_to_target(int(block["bits"], 16)) if cached_pow_value(serialize_header(block)) > target: raise ConsensusError("Proof of work does not meet the target.") total_fees = 0 for tx in txs[coinbase_count:]: if not verify_transaction_signature(tx): raise ConsensusError("Invalid transaction signature in the block.") sender = tx["sender"] amount, fee, nonce = int(tx["amount"]), int(tx["fee"]), int(tx["nonce"]) if amount < DUST_LIMIT or fee < MIN_RELAY_FEE: raise ConsensusError("Transaction violates the amount or fee rule.") if amount > MAX_MONEY or fee > MAX_MONEY or nonce < 0: raise ConsensusError("Amount or nonce out of range.") if not is_valid_address(tx["recipient"]) or sender == tx["recipient"]: raise ConsensusError("Invalid recipient address in the block.") if nonces.get(sender, 0) != nonce: raise ConsensusError("Nonce ordering violated inside the block.") immature = sum(a for h, a in coinbases.get(sender, []) if height - 1 - h < COINBASE_MATURITY) if balances.get(sender, 0) - immature < amount + fee: raise ConsensusError("Insufficient funds inside the block.") balances[sender] = balances.get(sender, 0) - (amount + fee) balances[tx["recipient"]] = balances.get(tx["recipient"], 0) + amount nonces[sender] = nonce + 1 total_fees += fee if total_fees > MAX_MONEY: raise ConsensusError("Total fees out of range.") seen_recipients = set() for coinbase in coinbase_txs: if int(coinbase.get("height", -1)) != height: raise ConsensusError("Coinbase states the wrong height.") if int(coinbase["fee"]) != 0 or coinbase.get("signature"): raise ConsensusError("Invalid coinbase structure.") amount = int(coinbase["amount"]) if amount < DUST_LIMIT or amount > MAX_MONEY: raise ConsensusError("Coinbase output violates the amount rules.") recipient = coinbase.get("recipient") or "" if height > 0 and not is_valid_address(recipient): raise ConsensusError("Coinbase has no valid recipient address.") if recipient in seen_recipients: raise ConsensusError("Duplicate coinbase recipient.") seen_recipients.add(recipient) committed_state = str(coinbase_txs[0].get("state_root") or "") if checkpoint: if committed_state != expected_state_root: raise ConsensusError("State root does not match the account state.") elif committed_state: raise ConsensusError("State root committed outside a checkpoint height.") shares = self._validate_block_shares(block, parent_chain, coinbase_txs[0]) subsidy = block_subsidy(height) paid = sum(int(tx["amount"]) for tx in coinbase_txs) if paid > subsidy + total_fees: raise ConsensusError("Coinbase exceeds the permitted payout.") finder = str(coinbase_txs[0].get("recipient") or "") expected = coinbase_payouts(height, total_fees, finder, parent_chain, shares) actual = [(str(tx.get("recipient") or ""), int(tx["amount"])) for tx in coinbase_txs] if actual != expected: raise ConsensusError("Coinbase does not follow the share distribution.") if paid != subsidy + total_fees: raise ConsensusError("Coinbase does not spend subsidy and fees exactly.") for coinbase in coinbase_txs: recipient = coinbase.get("recipient") or "" if recipient: balances[recipient] = balances.get(recipient, 0) + int(coinbase["amount"]) coinbases.setdefault(recipient, []).append((height, int(coinbase["amount"]))) def _validate_block_shares(self, block: Dict[str, Any], parent_chain: List[Dict[str, Any]], first_coinbase: Dict[str, Any]) -> List[Dict[str, Any]]: """Checks the shares a block confirms, including replay protection. The commitment hash sits in the first coinbase and therefore, via the Merkle root, inside the block hash - confirmed shares cannot be swapped out afterwards. """ shares = block.get("shares", []) if not isinstance(shares, list): raise ConsensusError("Share list of the block is invalid.") if len(shares) > MAX_SHARES_PER_BLOCK: raise ConsensusError("Too many shares in the block.") confirmed = self.confirmed_share_ids(parent_chain) identifiers: List[str] = [] for share in shares: identifier = self.validate_share(share, parent_chain, int(block["timestamp"])) if identifier in confirmed: raise ConsensusError("An already confirmed share was included again.") if identifier in identifiers: raise ConsensusError("Duplicate share in the block.") identifiers.append(identifier) if str(first_coinbase.get("shares_root") or "") != merkle_root(identifiers): raise ConsensusError("Share commitment does not match the shares.") return shares def validate_chain(self, chain: List[Dict[str, Any]]) -> int: """Validates a complete chain and returns its cumulative work. Used for diagnostics only - reorganisations run incrementally from the fork point via ``replace_chain``. Because it starts from an empty state, it cannot verify state root checkpoints. """ if not chain or chain[0]["hash"] != self.chain[0]["hash"]: raise ConsensusError("Different genesis block.") balances: Dict[str, int] = {} nonces: Dict[str, int] = {} coinbases: Dict[str, List[Tuple[int, int]]] = {} work = work_from_bits(int(chain[0]["bits"], 16)) for index in range(1, len(chain)): self.validate_block(chain[index], chain[:index], balances, nonces, coinbases) work += work_from_bits(int(chain[index]["bits"], 16)) return work # --------------------------------------------------------- Blockannahme def add_block(self, block: Dict[str, Any]) -> bool: """Appends a block to the tip (only if it connects directly).""" with self.lock: if block.get("previous_hash") != self.tip["hash"]: return False if block.get("hash") == self.tip["hash"]: return False balances = dict(self.balances) nonces = dict(self.nonces) coinbases = {k: list(v) for k, v in self.coinbase_heights.items()} self.validate_block(block, self.chain, balances, nonces, coinbases) if is_state_checkpoint(int(block["height"])): self._write_state_snapshot( int(block["height"]), str(block["transactions"][0].get("state_root") or "")) block["chain_work"] = self.total_work + work_from_bits(int(block["bits"], 16)) self.chain.append(block) self.store.append(block) self._apply_block_to_state(block) self._prune_mempool() self._prune_shares() self.tip_changed.set() log.info("Block #%s accepted (%s tx, %s shares, difficulty %s)", block["height"], len(block["transactions"]), len(block.get("shares", [])), difficulty_from_bits(int(block["bits"], 16))) return True @staticmethod def _fork_height(current: List[Dict[str, Any]], candidate: List[Dict[str, Any]]) -> int: """Number of blocks both chains share at the start.""" fork = 0 for index in range(min(len(current), len(candidate))): if current[index]["hash"] != candidate[index]["hash"]: break fork = index + 1 return fork def replace_chain(self, new_chain: List[Dict[str, Any]]) -> bool: """Adopts a competing chain when it carries more work. The switch happens incrementally from the fork point: the blocks of the old branch are unwound, then only the new branch is checked and applied. A full rebuild from block 0 would become an easily triggered denial of service as the chain grows. """ if not new_chain or new_chain[0]["hash"] != self.chain[0]["hash"]: raise ConsensusError("Different genesis block.") work = sum(work_from_bits(int(b["bits"], 16)) for b in new_chain) with self.lock: if work <= self.total_work: return False fork_height = self._fork_height(self.chain, new_chain) if fork_height < 1: raise ConsensusError("Different genesis block.") branch = list(new_chain[fork_height:]) if not branch: return False old_tail = self.chain[fork_height:] for block in reversed(old_tail): self._undo_block_from_state(block) # Check against a copy first: if a block of the new branch fails, # the existing state stays untouched. balances = dict(self.balances) nonces = dict(self.nonces) coinbases = {k: list(v) for k, v in self.coinbase_heights.items()} working = self.chain[:fork_height] try: for block in branch: self.validate_block(block, working, balances, nonces, coinbases) working = working + [block] except ConsensusError: for block in old_tail: self._apply_block_to_state(block) raise for block in branch: block["chain_work"] = self.total_work + work_from_bits(int(block["bits"], 16)) self._apply_block_to_state(block) self.chain = self.chain[:fork_height] + branch self.store.replace_from(fork_height, branch) for tx in (t for b in old_tail for t in b["transactions"] if not is_coinbase(t)): self.mempool.setdefault(tx.get("txid") or transaction_id(tx), tx) # Shares of the discarded branch may be confirmed again. for block in old_tail: for share in block.get("shares", []): self.shares.setdefault(share_id(share), share) self._prune_mempool() self._prune_shares() self.tip_changed.set() log.info("Chain reorganised -> height %s (fork at %s, %s blocks replaced)", self.height, fork_height, len(old_tail)) return True # ------------------------------------------------------------- Seitenaeste def _store_side_block(self, block: Dict[str, Any]) -> None: """Remembers a block that does not (yet) fit onto the tip.""" self.side_blocks[block["hash"]] = block if len(self.side_blocks) <= MAX_SIDE_BLOCKS: return stale = sorted(self.side_blocks.values(), key=lambda b: int(b.get("height", 0))) for old in stale[:len(self.side_blocks) - MAX_SIDE_BLOCKS]: self.side_blocks.pop(old["hash"], None) def _assemble_branch(self, block: Dict[str, Any]) -> Optional[List[Dict[str, Any]]]: """Assembles a branch from stored side blocks back to the main chain.""" index_by_hash = {b["hash"]: index for index, b in enumerate(self.chain)} branch = [block] for _ in range(MAX_SIDE_BLOCKS): parent_hash = branch[0].get("previous_hash") fork_index = index_by_hash.get(parent_hash) if fork_index is not None: return self.chain[:fork_index + 1] + branch parent = self.side_blocks.get(parent_hash) if parent is None: return None # ancestors are still missing branch.insert(0, parent) return None def submit_block(self, block: Dict[str, Any]) -> str: """Accepts a delivered block. If it does not attach to the tip directly it lands in the side-branch store. As soon as a branch collected there carries more work than the local chain, a reorganisation happens. Without this step a node without outbound peers would sit on its own fork forever - the blocks, and with them the rewards, of the other side would be lost. """ if not isinstance(block, dict) or not block.get("hash"): raise ConsensusError("No valid block supplied.") # Check header and proof of work up front so the side-branch store # cannot be flooded with cheaply produced nonsense. try: header = serialize_header(block) target = bits_to_target(int(block["bits"], 16)) except (KeyError, TypeError, ValueError) as exc: raise ConsensusError(f"Unusable block header: {exc}") if sha256d(header)[::-1].hex() != block["hash"]: raise ConsensusError("Block hash does not match the header.") if cached_pow_value(header) > target: raise ConsensusError("Proof of work does not meet the target.") if self.block_by_hash(block["hash"]): return "known" if self.add_block(block): return "accepted" with self.lock: self._store_side_block(block) candidate = self._assemble_branch(block) if candidate is None: return "orphan" work = sum(work_from_bits(int(b["bits"], 16)) for b in candidate) if work <= self.total_work: return "stored" return "reorg" if self.replace_chain(candidate) else "stored" # --------------------------------------------------------------------------- # # 7. Proof-of-Work Miner # --------------------------------------------------------------------------- # # Nonces between two checks of the abort signal. A memory-hard hash costs # milliseconds rather than microseconds, so the stride is small: the check # itself is free compared to a single hash, and the miner must react to a new # chain tip within a fraction of a second. ABORT_CHECK_STRIDE = 16 def scan_nonce_range(header_prefix: bytes, start: int, count: int, target: int, share_target: int, epoch: int, current_epoch=None) -> List[Tuple[bool, int]]: """Scans a nonce range and reports both block and share hits. Runs identically in the main process (single core) and in the worker processes. Each hash is computed exactly once and checked against both targets - shares therefore cost no extra computation. """ found: List[Tuple[bool, int]] = [] end = min(start + count, 0x100000000) for nonce in range(start, end): value = pow_value(header_prefix + nonce.to_bytes(4, "little")) if value <= share_target: is_block = value <= target found.append((is_block, nonce)) if is_block: break if current_epoch is not None and (nonce - start) % ABORT_CHECK_STRIDE == 0: if current_epoch.value != epoch: break return found def _mining_worker(jobs, results, epoch_flag) -> None: # pragma: no cover - Subprozess """Long-running compute process. The process stays alive for the whole runtime and waits for jobs. A pool that is created and torn down on every iteration costs more time on small systems than it spends computing. """ while True: job = jobs.get() if job is None: return epoch, header_prefix, start, count, target, share_target = job try: found = scan_nonce_range(header_prefix, start, count, target, share_target, epoch, epoch_flag) except Exception: found = [] results.put((epoch, start, count, found)) class MinerPool: """Persistent process pool with a job and a result queue. The processes are started once and only terminated on shutdown. A shared epoch flag (shared memory) makes every worker abort immediately as soon as a hit is in or the chain tip changes. """ def __init__(self, size: int): ctx = multiprocessing.get_context() self.size = size self.jobs = ctx.Queue() self.results = ctx.Queue() self.epoch_flag = ctx.Value("q", 0, lock=False) self.workers = [ ctx.Process(target=_mining_worker, args=(self.jobs, self.results, self.epoch_flag), daemon=True, name=f"paritr-hash-{index}") for index in range(size) ] for worker in self.workers: worker.start() def dispatch(self, epoch: int, header_prefix: bytes, base: int, chunk: int, target: int, share_target: int) -> int: self.epoch_flag.value = epoch for index in range(self.size): self.jobs.put((epoch, header_prefix, base + index * chunk, chunk, target, share_target)) return self.size def abort(self) -> None: self.epoch_flag.value = -1 def close(self) -> None: self.abort() for _ in self.workers: try: self.jobs.put_nowait(None) except Exception: pass for worker in self.workers: worker.join(timeout=2) if worker.is_alive(): worker.terminate() self.workers = [] class Miner(threading.Thread): """Continuous mining worker - also produces empty blocks (coinbase only). Besides whole blocks, the same hash loop keeps yielding a multiple of that in shares. They prove the work done even when the node never finds a block itself, and secure it a slice of the subsidy of every block in the window. Hardware usage is adjustable at runtime: * ``mining_processes`` number of parallel compute processes (0 = automatic) * ``mining_intensity`` duty cycle in percent (5-100) """ # A scrypt hash costs milliseconds, so a batch is sized in hundreds of # nonces rather than the hundreds of thousands a SHA256d miner would use. # Roughly two seconds of work per worker and round. CHUNK = 512 CANDIDATE_TTL = 60 MIN_INTENSITY = 5 def __init__(self, chain: Blockchain, config: Dict[str, Any], on_block, on_share=None): super().__init__(name="paritr-miner", daemon=True) self.chain = chain self.config = config self.on_block = on_block self.on_share = on_share or (lambda _share: None) self.hashrate = 0.0 self.blocks_found = 0 self.shares_found = 0 self.last_block_at: Optional[int] = None self._stop = threading.Event() self.cpu_total = max(1, cpu_count() or 1) self._pool: Optional[MinerPool] = None self._epoch = 0 # ------------------------------------------------------- Hardware-Budget @property def processes(self) -> int: """Effectively used compute processes (never more than available cores).""" try: configured = int(self.config.get("mining_processes") or 0) except (TypeError, ValueError): configured = 0 if configured <= 0: configured = max(1, self.cpu_total - 1) return max(1, min(self.cpu_total, configured)) @property def intensity(self) -> int: """Duty cycle in percent - the miner pauses for the remaining time.""" try: value = int(self.config.get("mining_intensity", 100)) except (TypeError, ValueError): value = 100 return max(self.MIN_INTENSITY, min(100, value)) def _throttle(self, busy_seconds: float) -> None: intensity = self.intensity if intensity >= 100: return self._stop.wait(min(2.0, busy_seconds * (100 - intensity) / intensity)) def _close_pool(self) -> None: if self._pool is not None: self._pool.close() self._pool = None def _ensure_pool(self, workers: int) -> Optional[MinerPool]: """Keeps exactly one pool alive; restarted only when the size changes.""" if workers <= 1: self._close_pool() return None if self._pool is None or self._pool.size != workers: self._close_pool() self._pool = MinerPool(workers) return self._pool def stop(self) -> None: self._stop.set() self._close_pool() @property def enabled(self) -> bool: return bool(self.config.get("mining_enabled")) and is_valid_address( str(self.config.get("miner_address", "")) ) def run(self) -> None: while not self._stop.is_set(): if not self.enabled: self.hashrate = 0.0 time.sleep(3) continue try: self._mine_once() except ConsensusError as exc: log.warning("Own block discarded: %s", exc) time.sleep(1) except Exception as exc: # pragma: no cover - the miner must never die log.error("Mining error: %s", exc) time.sleep(3) def _build_candidate(self) -> Tuple[Dict[str, Any], List[str], int, int]: """Builds a candidate block including the share commitment. The order is mandatory: first pick the shares to confirm, derive the distribution key and the commitment from them, then build the coinbase. Only that way do both end up inside the later block hash. """ miner_address = str(self.config.get("miner_address", "")) with self.chain.lock: parent = self.chain.tip height = parent["height"] + 1 bits = self.chain.next_bits(self.chain.chain) selected = self.chain.select_block_transactions(MAX_COINBASE_OUTPUTS) median = self.chain.median_time_past(self.chain.chain) shares = self.chain.select_block_shares(self.chain.chain) fees = sum(int(t["fee"]) for t in selected) payouts = coinbase_payouts(height, fees, miner_address, self.chain.chain, shares) # Commits the account state as of the parent block, so it has to be # read while the tip is still held. committed_state = (state_root(self.chain.balances, self.chain.nonces) if is_state_checkpoint(height) else None) shares_root = merkle_root([share_id(share) for share in shares]) txs = build_coinbase_transactions(height, payouts, secrets.randbits(32), shares_root, committed_state) + selected txids = [t.get("txid") or transaction_id(t) for t in txs] block = { "height": height, "version": 1, "previous_hash": parent["hash"], "merkle_root": merkle_root(txids), "timestamp": max(int(time.time()), median + 1), "bits": bits, "nonce": 0, "transactions": txs, "shares": shares, "miner": miner_address, "mining_duration": 0.0, } target = bits_to_target(int(bits, 16)) return block, txids, target, share_target_from_bits(int(bits, 16)) def _mine_once(self) -> None: self.chain.tip_changed.clear() block, txids, target, share_target = self._build_candidate() miner_address = str(block["miner"]) header_prefix = serialize_header(block)[:76] started = time.perf_counter() hashes = 0 nonce_base = 0 while not self._stop.is_set(): if self.chain.tip_changed.is_set() or not self.enabled: return if nonce_base >= 0xFFFFFFFF: return # Extranonce erneuern -> neuer Kandidat workers = self.processes chunk_started = time.perf_counter() hits = self._scan(header_prefix, nonce_base, target, share_target, workers) hashes += self.CHUNK * workers self._throttle(time.perf_counter() - chunk_started) elapsed = max(1e-6, time.perf_counter() - started) self.hashrate = hashes / elapsed solution: Optional[int] = None for is_block, nonce in hits: if is_block: solution = nonce else: self._publish_share(block, nonce, miner_address, txids) if solution is not None: block["nonce"] = solution block["hash"] = compute_block_hash(block) block["mining_duration"] = round(elapsed, 3) if self.chain.add_block(block): self.blocks_found += 1 self.last_block_at = int(time.time()) self.on_block(block) return nonce_base += self.CHUNK * workers if elapsed > self.CANDIDATE_TTL: return # refresh the candidate (new transactions / timestamp) def _publish_share(self, block: Dict[str, Any], nonce: int, miner_address: str, txids: List[str]) -> None: """Stores a found share locally and distributes it across the network.""" share = build_share(block, nonce, miner_address, txids) try: if self.chain.accept_share(share) != "accepted": return except ConsensusError as exc: log.debug("Own share discarded: %s", exc) return self.shares_found += 1 self.on_share(share) def _scan(self, header_prefix: bytes, nonce_base: int, target: int, share_target: int, workers: int) -> List[Tuple[bool, int]]: """Spreads a nonce range across the compute processes.""" pool = self._ensure_pool(workers) if pool is None: return scan_nonce_range(header_prefix, nonce_base, self.CHUNK, target, share_target, 0) self._epoch += 1 pending = pool.dispatch(self._epoch, header_prefix, nonce_base, self.CHUNK, target, share_target) hits: List[Tuple[bool, int]] = [] while pending: try: epoch, _start, _count, found = pool.results.get(timeout=120) except Exception: break pending -= 1 if epoch != self._epoch: continue # result of a discarded round hits.extend(found) if any(is_block for is_block, _ in found): pool.abort() # pull the remaining workers out of the race hits.sort(key=lambda hit: hit[1]) return hits # --------------------------------------------------------------------------- # # 8. P2P-Netzwerk (Seeds, Gossip, Headers-First IBD, Block-Relay) # --------------------------------------------------------------------------- # class OutboundQueue: """Decoupled work queue for every outbound peer call. Network I/O must never happen inside a caller that holds a lock on the blockchain or the peer manager: a slow counterpart would otherwise stall the entire node. Neither may a thread be spawned per peer - that would be a free lever for denial of service. Instead a few long-running workers pick up the jobs; under overload new jobs are dropped rather than queued up. """ def __init__(self, workers: int = 8, capacity: int = 2048): self.queue: "queue.Queue" = queue.Queue(maxsize=capacity) self.dropped = 0 for index in range(workers): threading.Thread(target=self._worker, daemon=True, name=f"paritr-outbound-{index}").start() def submit(self, func, *args) -> bool: try: self.queue.put_nowait((func, args)) return True except queue.Full: self.dropped += 1 log.debug("Outbound queue full - job dropped") return False def _worker(self) -> None: while True: func, args = self.queue.get() try: func(*args) except Exception as exc: # pragma: no cover - a worker must never die log.debug("Outbound job failed: %s", exc) finally: self.queue.task_done() @property def pending(self) -> int: return self.queue.qsize() class PeerManager: MAX_PEERS = 64 HEADER_BATCH = 500 BLOCK_BATCH = 50 INBOUND_TTL = 900 # inbound contacts count as active for this long MAX_INBOUND = 512 def __init__(self, chain: Blockchain, config: Dict[str, Any], data_dir: str): self.chain = chain self.config = config self.peer_file = os.path.join(data_dir, "peers.json") self.lock = threading.RLock() # Every outbound HTTP call runs through this queue - never directly # from a path that holds a lock. self.outbound = OutboundQueue() self.peers: Dict[str, Dict[str, Any]] = {} # Nodes that contacted us - the key is their IP self.inbound: Dict[str, float] = {} self._seed_cache: Tuple[float, List[str]] = (0.0, []) self._own_host_cache: Tuple[float, set] = (0.0, set()) self._identity_cache: Dict[str, Tuple[float, str]] = {} # Candidates derived from source IPs that are currently being probed self._probing: set = set() # Prevents parallel chain syncs (each one downloads the whole chain) self.sync_wanted = threading.Event() # Network-wide node count including this node - kept by the discovery worker. self.connected_nodes = 1 # Number of deduplicated outbound peers - also kept by the worker. self.connected_peers = 0 self._load() # -------------------------------------------------------------- Speicher def _load(self) -> None: stored: List[str] = [] if os.path.exists(self.peer_file): try: with open(self.peer_file, "r", encoding="utf-8") as handle: stored = json.load(handle).get("peers", []) except Exception: stored = [] for url in stored + list(self.config.get("peers", [])) + self.seed_candidates(): self.add(url, persist=False) self._save() def _save(self) -> None: try: with open(self.peer_file, "w", encoding="utf-8") as handle: json.dump({"peers": sorted(self.peers)}, handle, indent=2) except Exception as exc: log.warning("Could not write peers.json: %s", exc) # ----------------------------------------------------------------- Seeds def seed_candidates(self) -> List[str]: cached_at, cached = self._seed_cache if time.time() - cached_at < 300 and cached: return cached candidates = list(self.config.get("seed_nodes") or SEED_NODES) configured_hosts = {urlparse(url).hostname for url in (self.normalize(u) or "" for u in candidates) if url} for host in DNS_SEEDS: # If the host is already configured as a seed URL, its A records add # nothing: behind a reverse proxy or CDN they point at edge # addresses where the RPC port is not reachable at all. That is # exactly how dead entries accumulate in the peer set. if host in configured_hosts: continue try: for info in socket.getaddrinfo(host, None, socket.AF_INET, socket.SOCK_STREAM): candidates.append(f"http://{info[4][0]}:{SEED_FALLBACK_PORT}") except OSError: continue normalized = [] for url in candidates: norm = self.normalize(url) if norm and norm not in normalized: normalized.append(norm) self._seed_cache = (time.time(), normalized) return normalized # ----------------------------------------------------------------- Peers @staticmethod def normalize(url: Any) -> Optional[str]: if not isinstance(url, str) or not url.strip(): return None url = url.strip().rstrip("/") if "://" not in url: url = "http://" + url parsed = urlparse(url) if parsed.scheme not in ("http", "https") or not parsed.hostname: return None port = f":{parsed.port}" if parsed.port else "" return f"{parsed.scheme}://{parsed.hostname}{port}" def own_url(self) -> str: return self.normalize(self.config.get("public_url", "")) or "" def own_hosts(self) -> set: """Own hostname plus resolved IPs - prevents connecting to ourselves.""" cached_at, cached = self._own_host_cache if time.time() - cached_at < 300 and cached: return cached hosts = set() own = self.own_url() host = urlparse(own).hostname if own else None if host: hosts.add(host) try: for info in socket.getaddrinfo(host, None, socket.AF_INET, socket.SOCK_STREAM): hosts.add(info[4][0]) except OSError: pass self._own_host_cache = (time.time(), hosts) return hosts def add(self, url: Any, persist: bool = True) -> bool: normalized = self.normalize(url) if not normalized or normalized == self.own_url(): return False if urlparse(normalized).hostname in self.own_hosts(): return False with self.lock: if normalized in self.peers or len(self.peers) >= self.MAX_PEERS: return False self.peers[normalized] = {"url": normalized, "failures": 0, "height": None, "last_seen": 0} if persist: self._save() log.info("Peer added: %s", normalized) return True def remove(self, url: str) -> None: with self.lock: self.peers.pop(self.normalize(url) or url, None) self._save() def learn_peer(self, node_url: Any, client_ip: Optional[str] = None, rpc_port: Any = None) -> None: """Takes the counterpart of an inbound call on as a peer. If it reports no ``public_url``, a candidate is formed from the source IP and the reported RPC port and probed in the background. Without that return path a node that never dials anyone would not know its counterparts and would permanently report zero peers - blocks could then only flow in one direction. """ if self.normalize(node_url): self.add(node_url) return host = (client_ip or "").strip() if not host or host.startswith("127.") or host in ("::1", "localhost"): return if host in self.own_hosts(): return try: port = int(rpc_port or SEED_FALLBACK_PORT) except (TypeError, ValueError): return if not 1024 <= port <= 65535: return candidate = self.normalize(f"http://{host}:{port}") if not candidate: return with self.lock: if candidate in self.peers or candidate in self._probing: return self._probing.add(candidate) self.outbound.submit(self._probe_candidate, candidate) def _probe_candidate(self, url: str) -> None: """Checks a candidate derived from the source IP.""" try: info = self._get(url, "/p2p/info", timeout=5) if not info or info.get("chain_id") != CHAIN_ID: return if info.get("node_id") == self.own_id(): return if self.add(url): self.remember_node_id(url, info.get("node_id")) self.handshake(url) except Exception as exc: log.debug("Peer candidate %s unreachable: %s", url, exc) finally: with self.lock: self._probing.discard(url) def list_urls(self) -> List[str]: with self.lock: return sorted(self.peers) def active_urls(self) -> List[str]: """Peer list without duplicates - one URL per actual counterpart. The same server is often known under several addresses (seed URL, port variant, resolved IP). Gossip, relay and sync only need the best of them; otherwise every block is sent to the same node repeatedly and dead variants keep spreading through gossip. """ best: Dict[str, Tuple[Tuple[int, int, int], str]] = {} for peer in self.snapshot(): url = peer["url"] identity = peer.get("node_id") or self._identity(url) parsed = urlparse(url) rank = ( 1 if peer.get("last_seen") and not peer.get("failures") else 0, 1 if parsed.scheme == "https" else 0, 0 if (parsed.hostname or "").replace(".", "").isdigit() else 1, ) current = best.get(identity) if current is None or rank > current[0]: best[identity] = (rank, url) return sorted(url for _, url in best.values()) def snapshot(self) -> List[Dict[str, Any]]: with self.lock: return [dict(peer) for peer in self.peers.values()] def _identity(self, url: str) -> str: """Identity of a peer: resolved IP, otherwise the hostname. The same server appears in the peer list several times (seed URL, DNS seed, port variant). Through the identity it counts only once. """ host = urlparse(url).hostname or url cached = self._identity_cache.get(host) if cached and time.time() - cached[0] < 300: return cached[1] identity = host try: addresses = sorted({ info[4][0] for info in socket.getaddrinfo(host, None, socket.AF_INET, socket.SOCK_STREAM) }) if addresses: identity = addresses[0] except OSError: pass self._identity_cache[host] = (time.time(), identity) return identity def refresh_connected_count(self, max_age: int = 600) -> int: """Counts active nodes in the network, this node included. Takes both directions into account: peers we reach ourselves and nodes that contact us. The latter matters because nodes behind a router without port forwarding do participate but cannot be dialled from outside. Counted by node identity, so the same node reached via different addresses (seed URL, DNS seed, port variant) counts only once. Runs in the discovery worker so /status answers without DNS lookups. """ now = time.time() cutoff = now - max_age identities = set() self.connected_peers = len(self.active_urls()) for peer in self.snapshot(): if peer.get("failures") or not peer.get("last_seen"): continue if peer["last_seen"] < cutoff: continue identities.add(peer.get("node_id") or self._identity(peer["url"])) with self.lock: self.inbound = { key: seen for key, seen in self.inbound.items() if seen >= now - self.INBOUND_TTL } identities.update(self.inbound) identities.discard(self.own_id()) self.connected_nodes = len(identities) + 1 return self.connected_nodes def own_id(self) -> str: return str(self.config.get("node_id", "")) def rpc_port(self) -> int: """Own RPC port - counterparts without a ``public_url`` are reachable there.""" try: return int(self.config.get("rpc_port") or SEED_FALLBACK_PORT) except (TypeError, ValueError): return SEED_FALLBACK_PORT def note_inbound(self, node_id: Optional[str], ip: Optional[str]) -> None: """Notes a node that contacted us. The node identity that is sent along is more reliable than the IP: it also works when several nodes sit behind the same address or run on the same server. """ key = (node_id or "").strip() if key: if key == self.own_id(): return else: # Older nodes send no identity - then the IP has to do. key = (ip or "").strip() if not key or key.startswith("127.") or key in ("::1", "localhost"): return if key in self.own_hosts(): return with self.lock: self.inbound[key] = time.time() if len(self.inbound) > self.MAX_INBOUND: cutoff = time.time() - self.INBOUND_TTL self.inbound = {k: v for k, v in self.inbound.items() if v >= cutoff} def remember_node_id(self, url: str, node_id: Optional[str]) -> None: """Remembers the identity of a peer we reach ourselves.""" if not node_id: return with self.lock: peer = self.peers.get(url) if peer: peer["node_id"] = str(node_id) def _mark(self, url: str, ok: bool, height: Optional[int] = None) -> None: with self.lock: peer = self.peers.get(url) if not peer: return if ok: peer["failures"] = 0 peer["last_seen"] = int(time.time()) if height is not None: peer["height"] = height else: peer["failures"] += 1 if peer["failures"] >= 8 and url not in self.seed_candidates(): self.peers.pop(url, None) # ------------------------------------------------------------------ HTTP def _headers(self) -> Dict[str, str]: return { "User-Agent": f"Paritr/{NODE_VERSION}", "X-Node-Id": self.own_id(), } def _get(self, url: str, path: str, timeout: float = 8.0) -> Optional[Any]: try: res = requests.get(f"{url}{path}", timeout=timeout, headers=self._headers()) if res.status_code != 200: return None self._mark(url, True) return res.json() except Exception: self._mark(url, False) return None def _post(self, url: str, path: str, payload: Dict[str, Any], timeout: float = 8.0) -> Optional[Any]: try: res = requests.post(f"{url}{path}", json=payload, timeout=timeout, headers=self._headers()) if res.status_code >= 400: return None self._mark(url, True) return res.json() except Exception: self._mark(url, False) return None def _post_binary(self, url: str, path: str, payload: bytes, timeout: float = 8.0) -> Optional[Any]: try: res = requests.post(f"{url}{path}", data=payload, timeout=timeout, headers={**self._headers(), "Content-Type": "application/octet-stream", "X-Node-Url": self.own_url(), "X-Rpc-Port": str(self.rpc_port())}) if res.status_code >= 400: # 404/405 means the peer predates binary framing. return None self._mark(url, True) return res.json() except Exception: self._mark(url, False) return None def supports_binary(self, url: str) -> bool: """True while a peer has not refused the binary endpoints.""" with self.lock: peer = self.peers.get(url) return bool(peer is None or peer.get("binary", True)) def _note_binary_unsupported(self, url: str) -> None: with self.lock: peer = self.peers.get(url) if peer is not None: peer["binary"] = False # ------------------------------------------------------------- Protokoll def handshake(self, url: str) -> Optional[Dict[str, Any]]: info = self._post(url, "/p2p/handshake", { "node_url": self.own_url(), "node_id": self.own_id(), "rpc_port": self.rpc_port(), "chain_id": CHAIN_ID, "protocol": PROTOCOL_VERSION, "height": self.chain.height, }) if not info or info.get("chain_id") != CHAIN_ID: return None for peer_url in info.get("peers", [])[:self.MAX_PEERS]: self.add(peer_url) self._mark(url, True, info.get("height")) self.remember_node_id(url, info.get("node_id")) return info def discover(self) -> None: """Gossip: ask known peers for their peers (getaddr).""" for url in self.active_urls(): if self.handshake(url) is None: data = self._get(url, "/p2p/peers") if data: for peer_url in data.get("peers", []): self.add(peer_url) self.refresh_connected_count() def best_peer(self) -> Optional[Tuple[str, Dict[str, Any]]]: best: Optional[Tuple[str, Dict[str, Any]]] = None for url in self.active_urls(): info = self._get(url, "/p2p/info", timeout=6) if not info or info.get("chain_id") != CHAIN_ID: continue self._mark(url, True, info.get("height")) self.remember_node_id(url, info.get("node_id")) work = int(info.get("chain_work", 0)) if work > self.chain.total_work and (best is None or work > int(best[1]["chain_work"])): best = (url, info) return best def sync(self) -> bool: """Headers-first initial block download resp. ongoing chain sync.""" target = self.best_peer() if not target: return False url, info = target log.info("Synchronising with %s (height %s)", url, info.get("height")) headers: List[Dict[str, Any]] = [] start = 0 while True: batch = self._get(url, f"/p2p/headers?from={start}&count={self.HEADER_BATCH}", timeout=25) if not batch or not batch.get("headers"): break headers.extend(batch["headers"]) start += len(batch["headers"]) if len(batch["headers"]) < self.HEADER_BATCH: break if not headers or headers[0]["hash"] != self.chain.chain[0]["hash"]: log.warning("Peer %s belongs to a different chain - removing it", url) self.remove(url) return False remote_work = sum(work_from_bits(int(h["bits"], 16)) for h in headers) if remote_work <= self.chain.total_work: return False fork = 0 for index in range(min(len(headers), len(self.chain.chain))): if headers[index]["hash"] != self.chain.chain[index]["hash"]: break fork = index + 1 downloaded: List[Dict[str, Any]] = [] cursor = fork while cursor < len(headers): batch = self._get(url, f"/p2p/blocks?from={cursor}&count={self.BLOCK_BATCH}", timeout=60) if not batch or not batch.get("blocks"): break downloaded.extend(batch["blocks"]) cursor += len(batch["blocks"]) if not downloaded: return False candidate = self.chain.chain[:fork] + downloaded try: return self.chain.replace_chain(candidate) except ConsensusError as exc: log.warning("Chain from %s rejected: %s", url, exc) self._mark(url, False) return False def broadcast_block(self, block: Dict[str, Any]) -> None: payload = {"block": block, "node_url": self.own_url(), "node_id": self.own_id(), "rpc_port": self.rpc_port()} try: frame = encode_block(block) except WireFormatError: frame = None for url in self.active_urls(): self.outbound.submit(self._relay_block, url, payload, frame) def _relay_block(self, url: str, payload: Dict[str, Any], frame: Optional[bytes] = None) -> None: """Delivers a block and logs a rejection. A rejection is the most common reason for missing rewards (a skewed system clock, for instance) and must therefore never be swallowed silently. """ answer = None if frame is not None and self.supports_binary(url): answer = self._post_binary(url, "/p2p/block/bin", frame) if answer is None: self._note_binary_unsupported(url) if answer is None: answer = self._post(url, "/p2p/block", payload) if answer is None: log.debug("Block #%s not deliverable to %s", payload["block"].get("height"), url) elif answer.get("status") not in ("accepted", "reorg", "known", "stored"): log.warning("Peer %s rejected block #%s: %s", url, payload["block"].get("height"), answer.get("error") or answer.get("status")) def broadcast_transaction(self, tx: Dict[str, Any]) -> None: payload = {"transaction": tx, "node_url": self.own_url(), "node_id": self.own_id(), "rpc_port": self.rpc_port()} for url in self.active_urls(): self.outbound.submit(self._post, url, "/p2p/tx", payload) def broadcast_share(self, share: Dict[str, Any]) -> None: """Distributes a proof of work to every known counterpart. Shares are small and frequent, so they go out over the compact binary framing where the peer supports it. Like every other peer call they run through the bounded work queue and can never stall the node, however sluggish a counterpart is. """ payload = {"share": share, "node_url": self.own_url(), "node_id": self.own_id(), "rpc_port": self.rpc_port()} try: frame = encode_share(share) except (WireFormatError, ValueError): frame = None for url in self.active_urls(): self.outbound.submit(self._relay_share, url, payload, frame) def _relay_share(self, url: str, payload: Dict[str, Any], frame: Optional[bytes] = None) -> None: if frame is not None and self.supports_binary(url): if self._post_binary(url, "/p2p/share/bin", frame) is not None: return self._note_binary_unsupported(url) self._post(url, "/p2p/share", payload) def request_sync(self) -> None: """Wakes the sync worker. Deliberately just a signal instead of a dedicated thread: otherwise every incoming orphan block could start another full chain sync - a cheap lever to bring the node down. """ self.sync_wanted.set() # --------------------------------------------------------------------------- # # 9. Konfiguration & Node-Lebenszyklus # --------------------------------------------------------------------------- # def load_config() -> Dict[str, Any]: config = dict(DEFAULT_CONFIG) if os.path.exists(CONFIG_FILE): try: with open(CONFIG_FILE, "r", encoding="utf-8-sig") as handle: config.update(json.load(handle) or {}) except Exception as exc: log.warning("config.json unreadable (%s) - falling back to defaults", exc) # Migration of older configurations if config.get("port") and not config.get("rpc_port"): config["rpc_port"] = config["port"] config.pop("port", None) config.pop("initial_difficulty", None) config.pop("auto_mine", None) config.pop("auto_mine_interval", None) config.pop("genesis_initialized", None) config.pop("chain_file", None) if config.pop("peer_nodes", None) and not config.get("peers"): config["peers"] = [] if not config.get("admin_secret"): config["admin_secret"] = secrets.token_hex(24) # Stable identity: identifies this node independently of IP and port. if not config.get("node_id"): config["node_id"] = secrets.token_hex(16) if not config.get("seed_nodes"): config["seed_nodes"] = list(SEED_NODES) if not is_valid_address(config.get("miner_address", "")): config["miner_address"] = "" try: config["mining_processes"] = max(0, int(config.get("mining_processes") or 0)) except (TypeError, ValueError): config["mining_processes"] = 0 try: config["mining_intensity"] = max(5, min(100, int(config.get("mining_intensity", 100)))) except (TypeError, ValueError): config["mining_intensity"] = 100 config["network"] = CHAIN_ID return config def save_config(config: Dict[str, Any]) -> None: tmp = CONFIG_FILE + ".tmp" with open(tmp, "w", encoding="utf-8") as handle: json.dump(config, handle, indent=2) os.replace(tmp, CONFIG_FILE) # --------------------------------------------------------------------------- # # Public address discovery (used by the installers) # --------------------------------------------------------------------------- # # Fallback echo services, only consulted when no seed node answers. PUBLIC_IP_SERVICES = [ "https://api.ipify.org", "https://ifconfig.me/ip", "https://icanhazip.com", ] def _is_public_ipv4(value: str) -> bool: """True for a syntactically valid, globally routable IPv4 address.""" parts = value.split(".") if len(parts) != 4: return False try: octets = [int(part) for part in parts] except ValueError: return False if any(not 0 <= octet <= 255 for octet in octets): return False if any(part != str(octet) for part, octet in zip(parts, octets)): return False # reject leading zeros and other odd spellings a, b = octets[0], octets[1] if a in (0, 10, 127) or a >= 224: return False if a == 172 and 16 <= b <= 31: return False if a == 192 and b == 168: return False if a == 169 and b == 254: return False if a == 100 and 64 <= b <= 127: return False return True def detect_public_ip(seeds: Optional[List[str]] = None, timeout: float = 5.0) -> str: """Best-effort discovery of this machine's public IPv4 address. Seed nodes are asked first: they already see our source address, so no third-party service is involved and the answer comes from our own network. Only if none of them answers do the public echo services get a turn. The result is always re-validated locally, so a hostile answer can at worst stop auto-detection - never inject a private or malformed address. """ for url in (seeds if seeds is not None else SEED_NODES): try: res = requests.get(f"{url.rstrip('/')}/p2p/whoami", timeout=timeout) if res.status_code == 200: candidate = str((res.json() or {}).get("ip", "")).strip() if _is_public_ipv4(candidate): return candidate except Exception: continue for service in PUBLIC_IP_SERVICES: try: res = requests.get(service, timeout=timeout) if res.status_code == 200: candidate = res.text.strip() if _is_public_ipv4(candidate): return candidate except Exception: continue return "" def suggest_public_url(port: int, seeds: Optional[List[str]] = None) -> str: """Pre-fill value for the ``public_url`` prompt during installation.""" ip = detect_public_ip(seeds) return f"http://{ip}:{int(port)}" if ip else "" class Node: def __init__(self, config: Dict[str, Any]): self.config = config self.data_dir = os.path.join(BASE_DIR, str(config.get("data_dir", "data"))) os.makedirs(self.data_dir, exist_ok=True) self.chain = Blockchain(config, self.data_dir) self.peers = PeerManager(self.chain, config, self.data_dir) self.miner = Miner(self.chain, config, self._on_block_mined, self._on_share_found) self.started_at = int(time.time()) self._threads_started = False def _on_block_mined(self, block: Dict[str, Any]) -> None: self.peers.broadcast_block(block) def _on_share_found(self, share: Dict[str, Any]) -> None: self.peers.broadcast_share(share) # --------------------------------------------------------------- Threads def start_background(self) -> None: if self._threads_started: return self._threads_started = True threading.Thread(target=self._bootstrap_worker, daemon=True, name="bootstrap").start() threading.Thread(target=self._discovery_worker, daemon=True, name="discovery").start() threading.Thread(target=self._sync_worker, daemon=True, name="sync").start() self.miner.start() def _bootstrap_worker(self) -> None: time.sleep(1) seeds = self.peers.seed_candidates() log.info("Bootstrapping via seed nodes: %s", ", ".join(seeds) or "-") for url in seeds: self.peers.add(url) self.peers.handshake(url) self.peers.refresh_connected_count() try: self.peers.sync() except Exception as exc: log.warning("Initial block download failed: %s", exc) def _discovery_worker(self) -> None: while True: time.sleep(60) try: self.peers.discover() except Exception as exc: log.debug("Peer discovery: %s", exc) def _sync_worker(self) -> None: """Syncs the chain periodically - and immediately on request.""" while True: self.peers.sync_wanted.wait(timeout=20) self.peers.sync_wanted.clear() try: self.peers.sync() except Exception as exc: log.debug("Sync: %s", exc) # ------------------------------------------------------------------ Info def status(self) -> Dict[str, Any]: tip = self.chain.tip height = tip["height"] bits = int(tip["bits"], 16) next_height = height + 1 durations = [b.get("mining_duration", 0) for b in self.chain.chain[-30:] if b.get("mining_duration")] recent = self.chain.chain[-min(len(self.chain.chain), RETARGET_INTERVAL + 1):] if len(recent) > 1: avg_block_time = (int(recent[-1]["timestamp"]) - int(recent[0]["timestamp"])) / (len(recent) - 1) else: avg_block_time = float(TARGET_BLOCK_TIME) stats = self.chain.share_statistics() circulating = self.chain.circulating_supply() return { "status": "online", "chain_id": CHAIN_ID, "node_version": NODE_VERSION, "protocol_version": PROTOCOL_VERSION, "uptime_seconds": int(time.time()) - self.started_at, "height": height, "best_block_hash": tip["hash"], "chain_work": self.chain.total_work, "difficulty": difficulty_from_bits(bits), "share_difficulty": round(difficulty_from_bits(bits) / SHARE_TARGET_MULTIPLIER, 8), "network_hashrate": self.network_hashrate(), "bits": tip["bits"], "target_block_time": TARGET_BLOCK_TIME, "avg_block_time": round(avg_block_time, 2), "avg_mining_duration": round(sum(durations) / len(durations), 2) if durations else 0.0, "next_block_reward": block_subsidy(next_height), "initial_subsidy": INITIAL_SUBSIDY, "min_subsidy": MIN_SUBSIDY, "halving_interval": HALVING_INTERVAL, "halvings": max(0, height) // HALVING_INTERVAL, "blocks_until_halving": HALVING_INTERVAL - ((next_height - 1) % HALVING_INTERVAL), "circulating_supply": circulating, "tail_emission": TAIL_EMISSION, "max_supply": None, "annual_inflation": self.annual_inflation(circulating, next_height), "coin": COIN, "min_fee": MIN_RELAY_FEE, "coinbase_maturity": COINBASE_MATURITY, "mempool_size": len(self.chain.mempool), "orphan_tx_count": len(self.chain.orphan_txs), "peer_count": self.peers.connected_peers, "known_peers": len(self.peers.peers), "inbound_count": len(self.peers.inbound), "side_blocks": len(self.chain.side_blocks), "network_nodes": self.peers.connected_nodes, "reward_window": REWARD_WINDOW, "finder_share_percent": FINDER_SHARE_PERCENT, "share_multiplier": SHARE_TARGET_MULTIPLIER, "shares_in_window": stats["total_shares"], "active_miners": stats["active_miners"], "share_pool_size": len(self.chain.shares), "mining_enabled": bool(self.config.get("mining_enabled")), "mining_active": self.miner.enabled, "miner_address": str(self.config.get("miner_address", "")), "hashrate": round(self.miner.hashrate, 2), "blocks_found": self.miner.blocks_found, "shares_found": self.miner.shares_found, "last_block_time": int(tip["timestamp"]), "cpu_total": self.miner.cpu_total, "mining_processes": self.miner.processes, "mining_processes_config": int(self.config.get("mining_processes") or 0), "mining_intensity": self.miner.intensity, "platform": PLATFORM_LABEL, } def network_hashrate(self) -> float: """Estimated total network hash power: work per second in the window.""" window = self.chain.chain[-min(len(self.chain.chain), RETARGET_INTERVAL + 1):] if len(window) < 2: return 0.0 span = int(window[-1]["timestamp"]) - int(window[0]["timestamp"]) if span <= 0: return 0.0 work = sum(work_from_bits(int(b["bits"], 16)) for b in window[1:]) return round(work / span, 2) @staticmethod def annual_inflation(circulating: int, next_height: int) -> float: """Annual new issuance as a percentage of the circulating supply. Without a hard cap this is the meaningful figure: it falls with every halving and afterwards keeps falling towards zero through the growing supply, without emission ever drying up completely. """ if circulating <= 0: return 0.0 blocks_per_year = (365 * 24 * 60 * 60) // TARGET_BLOCK_TIME return round(blocks_per_year * block_subsidy(next_height) / circulating * 100, 4) # --------------------------------------------------------------------------- # # 10. RPC / REST Schnittstelle # --------------------------------------------------------------------------- # def create_app(node: "Node") -> Flask: app = Flask(__name__) allowed_origin = str(node.config.get("rpc_cors_origins", "*")) or "*" @app.after_request def apply_cors(response): origin = request.headers.get("Origin", "") if allowed_origin == "*": response.headers["Access-Control-Allow-Origin"] = "*" elif origin and origin in [o.strip() for o in allowed_origin.split(",")]: response.headers["Access-Control-Allow-Origin"] = origin response.headers["Vary"] = "Origin" response.headers["Access-Control-Allow-Headers"] = "Content-Type, X-Admin-Secret, Authorization" response.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS" response.headers["Access-Control-Max-Age"] = "600" response.headers["X-Content-Type-Options"] = "nosniff" response.headers["Referrer-Policy"] = "no-referrer" return response @app.route("/", methods=["OPTIONS"]) @app.route("/", methods=["OPTIONS"]) def preflight(_ignored: str = ""): return "", 204 # ------------------------------------------------------------ Helpers def body() -> Dict[str, Any]: return request.get_json(force=True, silent=True) or {} def client_ip() -> str: """Caller IP - behind nginx/Caddy it sits in X-Forwarded-For.""" forwarded = request.headers.get("X-Forwarded-For", "") if forwarded: return forwarded.split(",")[0].strip() return request.remote_addr or "" def note_peer_contact() -> None: """Records that a node contacted us. The identity is in the header (all calls) or in the body (handshake and relay). The IP only serves as a fallback for older nodes. """ node_id = request.headers.get("X-Node-Id", "") if not node_id and request.method == "POST": node_id = str(body().get("node_id", "")) node.peers.note_inbound(node_id, client_ip()) def admin_ok() -> bool: secret = str(node.config.get("admin_secret", "")) if not secret: return False token = request.headers.get("X-Admin-Secret", "") if not token: auth = request.headers.get("Authorization", "") if auth.startswith("Bearer "): token = auth[7:] if not token: token = str(body().get("admin_secret", "")) return bool(token) and hmac.compare_digest(token.strip(), secret.strip()) def require_admin(): if not admin_ok(): return jsonify({"error": "Not authorised. A valid admin secret key is required."}), 403 return None # -------------------------------------------------------- Oeffentliche API @app.route("/", methods=["GET"]) @app.route("/status", methods=["GET"]) def rpc_status(): return jsonify(node.status()) @app.route("/health", methods=["GET"]) def rpc_health(): return jsonify({"status": "online", "chain_id": CHAIN_ID, "height": node.chain.height}) @app.route("/chain/params", methods=["GET"]) def rpc_params(): return jsonify({ "chain_id": CHAIN_ID, "coin": COIN, "initial_subsidy": INITIAL_SUBSIDY, "min_subsidy": MIN_SUBSIDY, "halving_interval": HALVING_INTERVAL, "tail_emission": TAIL_EMISSION, "max_supply": None, "target_block_time": TARGET_BLOCK_TIME, "retarget_interval": RETARGET_INTERVAL, "coinbase_maturity": COINBASE_MATURITY, "min_fee": MIN_RELAY_FEE, "dust_limit": DUST_LIMIT, "max_block_transactions": MAX_BLOCK_TRANSACTIONS, "reward_window": REWARD_WINDOW, "finder_share_percent": FINDER_SHARE_PERCENT, "share_multiplier": SHARE_TARGET_MULTIPLIER, "max_shares_per_block": MAX_SHARES_PER_BLOCK, "share_ttl_blocks": SHARE_TTL_BLOCKS, "max_coinbase_outputs": MAX_COINBASE_OUTPUTS, "address_prefix": ADDRESS_PREFIX_HINT, "pow_algorithm": "scrypt", "pow_scrypt_n": POW_SCRYPT_N, "pow_scrypt_r": POW_SCRYPT_R, "pow_scrypt_p": POW_SCRYPT_P, "pow_scratchpad_bytes": 128 * POW_SCRYPT_N * POW_SCRYPT_R, "state_root_interval": STATE_ROOT_INTERVAL, "seed_nodes": node.config.get("seed_nodes", SEED_NODES), }) @app.route("/address/
", methods=["GET"]) def rpc_address(address: str): address = address.strip() if not is_valid_address(address): return jsonify({"error": "Invalid wallet address."}), 400 return jsonify({ "address": address, "confirmed": node.chain.confirmed_balance(address), "immature": node.chain.immature_balance(address), "pending_in": node.chain.pending_incoming(address), "pending_out": node.chain.pending_outgoing(address), "spendable": node.chain.spendable_balance(address), "next_nonce": node.chain.next_nonce(address), "height": node.chain.height, "coin": COIN, }) @app.route("/address/
/transactions", methods=["GET"]) def rpc_address_txs(address: str): address = address.strip() if not is_valid_address(address): return jsonify({"error": "Invalid wallet address."}), 400 try: limit = max(1, min(200, int(request.args.get("limit", 50)))) offset = max(0, int(request.args.get("offset", 0))) except ValueError: return jsonify({"error": "Invalid parameters."}), 400 return jsonify({ "address": address, "height": node.chain.height, "transactions": node.chain.address_history(address, limit, offset), }) @app.route("/transaction", methods=["POST"]) def rpc_send_transaction(): try: result = node.chain.accept_transaction(body()) except ConsensusError as exc: return jsonify({"error": str(exc)}), 400 tx = result["transaction"] if result["status"] == "accepted": node.peers.broadcast_transaction(tx) return jsonify({"status": result["status"], "txid": tx["txid"]}), 201 @app.route("/transaction/", methods=["GET"]) def rpc_transaction(txid: str): found = node.chain.find_transaction(txid.strip()) if not found: return jsonify({"error": "Transaction not found."}), 404 return jsonify(found) @app.route("/fee/estimate", methods=["GET"]) def rpc_fee(): pending = len(node.chain.mempool) multiplier = 1 + min(4, pending // MAX_BLOCK_TRANSACTIONS) return jsonify({ "min_fee": MIN_RELAY_FEE, "recommended_fee": MIN_RELAY_FEE * multiplier, "mempool_size": pending, }) @app.route("/block/", methods=["GET"]) def rpc_block(identifier: str): block = (node.chain.block_at(int(identifier)) if identifier.isdigit() else node.chain.block_by_hash(identifier.strip())) if not block: return jsonify({"error": "Block not found."}), 404 return jsonify({ "block": block, "confirmations": node.chain.height - block["height"] + 1, "difficulty": difficulty_from_bits(int(block["bits"], 16)), "subsidy": block_subsidy(block["height"]), }) @app.route("/mining/info", methods=["GET"]) def rpc_mining_info(): status = node.status() return jsonify({ "height": status["height"], "difficulty": status["difficulty"], "share_difficulty": status["share_difficulty"], "share_multiplier": SHARE_TARGET_MULTIPLIER, "shares_in_window": status["shares_in_window"], "node_shares_submitted": node.miner.shares_found, "share_pool_size": status["share_pool_size"], "active_miners": status["active_miners"], "network_hashrate": status["network_hashrate"], "bits": status["bits"], "next_block_reward": status["next_block_reward"], "blocks_until_halving": status["blocks_until_halving"], "target_block_time": TARGET_BLOCK_TIME, "avg_block_time": status["avg_block_time"], "mempool_size": status["mempool_size"], "mining_enabled": status["mining_enabled"], "mining_active": status["mining_active"], "miner_address": status["miner_address"], "hashrate": status["hashrate"], "blocks_found": status["blocks_found"], "coinbase_maturity": COINBASE_MATURITY, "processes": node.miner.processes, "cpu_total": node.miner.cpu_total, "mining_processes": status["mining_processes"], "mining_processes_config": status["mining_processes_config"], "mining_intensity": status["mining_intensity"], "min_intensity": Miner.MIN_INTENSITY, "platform": PLATFORM_LABEL, "node_version": NODE_VERSION, "uptime_seconds": status["uptime_seconds"], "peer_count": status["peer_count"], "network_nodes": status["network_nodes"], "reward_window": REWARD_WINDOW, "finder_share_percent": FINDER_SHARE_PERCENT, "coin": COIN, }) @app.route("/mining/distribution", methods=["GET"]) def rpc_mining_distribution(): """Aggregated metrics of the measuring window. Deliberately without a list of the participating addresses: the distribution is recomputable from the chain anyway, and a ready-made ranking of all miners would only be an unnecessary privacy risk. With ``?address=`` the endpoint additionally returns that address's share. """ with node.chain.lock: parent_chain = list(node.chain.chain) stats = node.chain.share_statistics(parent_chain) height = parent_chain[-1]["height"] + 1 subsidy = block_subsidy(height) pool = subsidy - subsidy * FINDER_SHARE_PERCENT // 100 total = stats["total_shares"] address = str(request.args.get("address", "")).strip() own_shares = stats["weights"].get(address, 0) if is_valid_address(address) else 0 return jsonify({ "height": height, "window": REWARD_WINDOW, "window_blocks": stats["window_blocks"], "window_seconds": stats["window_seconds"], "finder_share_percent": FINDER_SHARE_PERCENT, "share_multiplier": SHARE_TARGET_MULTIPLIER, "subsidy": subsidy, "shared_pool": pool, "total_shares": total, "active_miners": stats["active_miners"], "share_rate": round(stats["share_rate"], 4), "network_hashrate": node.network_hashrate(), "coin": COIN, "address": address if own_shares else "", "address_shares": own_shares, "address_share_percent": round(own_shares / total * 100, 4) if total else 0.0, "address_estimated_payout": pool * own_shares // total if total else 0, }) @app.route("/mining/rewards/
", methods=["GET"]) def rpc_mining_rewards(address: str): address = address.strip() if not is_valid_address(address): return jsonify({"error": "Invalid wallet address."}), 400 tip = node.chain.height entries = node.chain.coinbase_heights.get(address, []) return jsonify({ "address": address, "total_mined": sum(amount for _, amount in entries), "blocks_mined": len(entries), "immature": node.chain.immature_balance(address), "coin": COIN, "rewards": [{ "height": height, "amount": amount, "confirmations": tip - height + 1, "matured": tip - height >= COINBASE_MATURITY, } for height, amount in reversed(entries[-50:])], }) @app.route("/supply", methods=["GET"]) def rpc_supply(): """Offenes Emissionsmodell: keine Obergrenze, aber sinkende Inflation.""" circulating = node.chain.circulating_supply() next_height = node.chain.height + 1 blocks_per_year = (365 * 24 * 60 * 60) // TARGET_BLOCK_TIME subsidy = block_subsidy(next_height) return jsonify({ "circulating": circulating, "tail_emission": TAIL_EMISSION, "max_supply": None, "block_subsidy": subsidy, "min_subsidy": MIN_SUBSIDY, "emission_per_year": blocks_per_year * subsidy, "annual_inflation": Node.annual_inflation(circulating, next_height), "blocks_until_halving": HALVING_INTERVAL - ((next_height - 1) % HALVING_INTERVAL), "halvings": max(0, node.chain.height) // HALVING_INTERVAL, "coin": COIN, }) @app.route("/state/snapshot", methods=["GET"]) def rpc_state_snapshot(): """Latest committed account state. The ``state_root`` in the answer is the value the coinbase of block ``committed_at`` commits to, so a fresh node can verify the whole snapshot against the chain instead of replaying every balance change. """ snapshot = node.chain.latest_state_snapshot() if not snapshot: return jsonify({"error": "No state snapshot available yet."}), 404 return jsonify(snapshot) # ---------------------------------------------------------------- P2P API @app.route("/p2p/info", methods=["GET"]) def p2p_info(): # Deliberately without note_peer_contact: the endpoint is exposed through # the web proxy, so a browser call would otherwise count as a node. tip = node.chain.tip return jsonify({ "chain_id": CHAIN_ID, "protocol": PROTOCOL_VERSION, "node_version": NODE_VERSION, "node_id": node.peers.own_id(), "height": tip["height"], "best_hash": tip["hash"], "chain_work": node.chain.total_work, "peer_count": len(node.peers.peers), "node_url": node.peers.own_url(), }) @app.route("/p2p/whoami", methods=["GET"]) def p2p_whoami(): """Reports the caller's source address. A node behind NAT cannot know its own public address. Asking a peer is the cheapest reliable way to find it, and it keeps installation free of any third-party IP lookup service. """ return jsonify({"ip": client_ip(), "chain_id": CHAIN_ID}) @app.route("/p2p/handshake", methods=["POST"]) def p2p_handshake(): data = body() if data.get("chain_id") not in (None, CHAIN_ID): return jsonify({"error": "Foreign chain id."}), 400 note_peer_contact() node.peers.learn_peer(data.get("node_url", ""), client_ip(), data.get("rpc_port")) tip = node.chain.tip return jsonify({ "chain_id": CHAIN_ID, "protocol": PROTOCOL_VERSION, "node_id": node.peers.own_id(), "height": tip["height"], "best_hash": tip["hash"], "chain_work": node.chain.total_work, "peers": node.peers.active_urls()[:PeerManager.MAX_PEERS], "node_url": node.peers.own_url(), }) @app.route("/p2p/peers", methods=["GET"]) def p2p_peers(): note_peer_contact() return jsonify({"peers": node.peers.active_urls()}) @app.route("/p2p/headers", methods=["GET"]) def p2p_headers(): note_peer_contact() try: start = max(0, int(request.args.get("from", 0))) count = max(1, min(PeerManager.HEADER_BATCH, int(request.args.get("count", 200)))) except ValueError: return jsonify({"error": "Invalid parameters."}), 400 return jsonify({"headers": node.chain.headers(start, count), "height": node.chain.height}) @app.route("/p2p/blocks", methods=["GET"]) def p2p_blocks(): note_peer_contact() try: start = max(0, int(request.args.get("from", 0))) count = max(1, min(PeerManager.BLOCK_BATCH, int(request.args.get("count", 20)))) except ValueError: return jsonify({"error": "Invalid parameters."}), 400 return jsonify({"blocks": node.chain.blocks(start, count), "height": node.chain.height}) @app.route("/p2p/block", methods=["POST"]) def p2p_receive_block(): data = body() note_peer_contact() node.peers.learn_peer(data.get("node_url", ""), client_ip(), data.get("rpc_port")) block = data.get("block") if not isinstance(block, dict): return jsonify({"error": "No block supplied."}), 400 try: result = node.chain.submit_block(block) except ConsensusError as exc: log.warning("Delivered block #%s rejected: %s", block.get("height"), exc) return jsonify({"status": "rejected", "error": str(exc)}), 400 if result in ("accepted", "reorg"): node.peers.broadcast_block(block) elif result == "orphan": node.peers.request_sync() # Vorfahren fehlen -> Abgleich anstossen return jsonify({"status": result, "height": node.chain.height}) @app.route("/p2p/tx", methods=["POST"]) def p2p_receive_tx(): data = body() note_peer_contact() node.peers.learn_peer(data.get("node_url", ""), client_ip(), data.get("rpc_port")) tx = data.get("transaction") if not isinstance(tx, dict): return jsonify({"error": "No transaction supplied."}), 400 try: result = node.chain.accept_transaction(tx) except ConsensusError as exc: return jsonify({"status": "rejected", "error": str(exc)}), 400 if result["status"] == "accepted": node.peers.broadcast_transaction(result["transaction"]) return jsonify({"status": result["status"], "txid": result["transaction"]["txid"]}) @app.route("/p2p/share", methods=["POST"]) def p2p_receive_share(): """Accepts a proof of work and relays it onwards. Validation happens entirely in ``accept_share``; only genuinely new and valid shares cause any further network traffic. """ data = body() note_peer_contact() node.peers.learn_peer(data.get("node_url", ""), client_ip(), data.get("rpc_port")) share = data.get("share") if not isinstance(share, dict): return jsonify({"error": "No share supplied."}), 400 try: result = node.chain.accept_share(share) except ConsensusError as exc: return jsonify({"status": "rejected", "error": str(exc)}), 400 if result == "accepted": node.peers.broadcast_share(share) return jsonify({"status": result, "share_id": share.get("header_hash", "")}) @app.route("/p2p/share/bin", methods=["POST"]) def p2p_receive_share_binary(): """Compact binary variant of ``/p2p/share`` for node-to-node gossip.""" note_peer_contact() node.peers.learn_peer(request.headers.get("X-Node-Url", ""), client_ip(), request.headers.get("X-Rpc-Port")) raw = request.get_data(cache=False) if len(raw) > MAX_WIRE_MESSAGE: return jsonify({"error": "Frame too large."}), 413 try: share = decode_share(raw) except WireFormatError as exc: return jsonify({"error": str(exc)}), 400 try: result = node.chain.accept_share(share) except ConsensusError as exc: return jsonify({"status": "rejected", "error": str(exc)}), 400 if result == "accepted": node.peers.broadcast_share(share) return jsonify({"status": result, "share_id": share.get("header_hash", "")}) @app.route("/p2p/block/bin", methods=["POST"]) def p2p_receive_block_binary(): """Compact binary variant of ``/p2p/block``.""" note_peer_contact() node.peers.learn_peer(request.headers.get("X-Node-Url", ""), client_ip(), request.headers.get("X-Rpc-Port")) raw = request.get_data(cache=False) if len(raw) > MAX_WIRE_MESSAGE: return jsonify({"error": "Frame too large."}), 413 try: block = decode_block(raw) except WireFormatError as exc: return jsonify({"error": str(exc)}), 400 try: result = node.chain.submit_block(block) except ConsensusError as exc: log.warning("Delivered block #%s rejected: %s", block.get("height"), exc) return jsonify({"status": "rejected", "error": str(exc)}), 400 if result in ("accepted", "reorg"): node.peers.broadcast_block(block) elif result == "orphan": node.peers.request_sync() return jsonify({"status": result, "height": node.chain.height}) # -------------------------------------------------------------- Admin API @app.route("/admin/auth", methods=["GET", "POST"]) def admin_auth(): if not admin_ok(): return jsonify({"authenticated": False, "error": "Invalid secret key."}), 403 return jsonify({ "authenticated": True, "miner_address": str(node.config.get("miner_address", "")), "mining_enabled": bool(node.config.get("mining_enabled")), "public_url": str(node.config.get("public_url", "")), "node_version": NODE_VERSION, }) @app.route("/admin/mining", methods=["POST"]) def admin_mining(): denied = require_admin() if denied: return denied data = body() if "miner_address" in data: address = str(data["miner_address"]).strip() if address and not is_valid_address(address): return jsonify({"error": "Invalid wallet address."}), 400 node.config["miner_address"] = address if "mining_enabled" in data: node.config["mining_enabled"] = bool(data["mining_enabled"]) if "mining_processes" in data: try: requested = max(0, int(data["mining_processes"])) except (TypeError, ValueError): return jsonify({"error": "Invalid process count."}), 400 node.config["mining_processes"] = min(requested, node.miner.cpu_total) if "mining_intensity" in data: try: requested = int(data["mining_intensity"]) except (TypeError, ValueError): return jsonify({"error": "Invalid intensity."}), 400 node.config["mining_intensity"] = max(Miner.MIN_INTENSITY, min(100, requested)) save_config(node.config) node.chain.tip_changed.set() return jsonify({ "status": "success", "miner_address": node.config.get("miner_address", ""), "mining_enabled": bool(node.config.get("mining_enabled")), "mining_processes": node.miner.processes, "mining_processes_config": int(node.config.get("mining_processes") or 0), "mining_intensity": node.miner.intensity, "cpu_total": node.miner.cpu_total, }) @app.route("/admin/peers", methods=["GET"]) def admin_peers(): denied = require_admin() if denied: return denied return jsonify({ "peers": node.peers.snapshot(), "active": node.peers.active_urls(), "seeds": node.peers.seed_candidates(), "node_id": node.peers.own_id(), "inbound_count": len(node.peers.inbound), "connected_nodes": node.peers.connected_nodes, }) @app.route("/admin/peers/add", methods=["POST"]) def admin_peers_add(): denied = require_admin() if denied: return denied url = str(body().get("url", "")).strip() if not node.peers.add(url): return jsonify({"error": "Peer invalid or already known."}), 400 node.peers.handshake(PeerManager.normalize(url) or url) return jsonify({"status": "success", "peers": node.peers.list_urls()}) @app.route("/admin/peers/remove", methods=["POST"]) def admin_peers_remove(): denied = require_admin() if denied: return denied node.peers.remove(str(body().get("url", "")).strip()) return jsonify({"status": "success", "peers": node.peers.list_urls()}) @app.route("/admin/sync", methods=["POST"]) def admin_sync(): denied = require_admin() if denied: return denied return jsonify({"status": "success", "replaced": node.peers.sync(), "height": node.chain.height}) @app.route("/admin/mempool", methods=["GET"]) def admin_mempool(): denied = require_admin() if denied: return denied return jsonify({"transactions": list(node.chain.mempool.values()), "count": len(node.chain.mempool)}) @app.route("/admin/blocks", methods=["GET"]) def admin_blocks(): denied = require_admin() if denied: return denied try: limit = max(1, min(100, int(request.args.get("limit", 25)))) except ValueError: return jsonify({"error": "Invalid parameters."}), 400 return jsonify({ "height": node.chain.height, "blocks": [{ **header_only(block), "miner": block.get("miner", ""), "mining_duration": block.get("mining_duration", 0), "reward": sum(int(t["amount"]) for t in block["transactions"] if is_coinbase(t)), "difficulty": difficulty_from_bits(int(block["bits"], 16)), } for block in node.chain.chain[-limit:][::-1]], }) @app.errorhandler(404) def not_found(_error): return jsonify({"error": "Endpoint not found."}), 404 @app.errorhandler(500) def server_error(_error): # pragma: no cover return jsonify({"error": "Internal node error."}), 500 return app # --------------------------------------------------------------------------- # # 11. Einstiegspunkt # --------------------------------------------------------------------------- # def configure_logging(config: Dict[str, Any]) -> None: logging.basicConfig( level=getattr(logging, str(config.get("log_level", "INFO")).upper(), logging.INFO), format="%(asctime)s [%(levelname)s] %(message)s", ) def build_node(config: Optional[Dict[str, Any]] = None) -> Tuple[Node, Flask]: config = config or load_config() configure_logging(config) verify_pow_available() save_config(config) instance = Node(config) flask_app = create_app(instance) instance.start_background() log.info("Paritr node %s ready - height %s, peers %s", NODE_VERSION, instance.chain.height, len(instance.peers.peers)) return instance, flask_app def serve(flask_app: Flask, host: str, port: int, preferred: str = "auto") -> None: """Starts the HTTP server. ``waitress`` is the production-grade choice on Windows (gunicorn does not run there). If it is missing, the built-in Flask server is used instead. """ if preferred in ("auto", "waitress"): try: from waitress import serve as waitress_serve except ImportError: if preferred == "waitress": log.error("waitress is not installed: pip install waitress") raise SystemExit(1) else: log.info("HTTP server: waitress") waitress_serve(flask_app, host=host, port=port, threads=8) return log.warning("HTTP server: Flask development server (for continuous " "operation please install waitress or gunicorn)") flask_app.run(host=host, port=port, threaded=True, use_reloader=False) def main() -> None: multiprocessing.freeze_support() parser = argparse.ArgumentParser(description="Paritr Full Node") parser.add_argument("--port", type=int, help="RPC port (default from config.json)") parser.add_argument("--host", type=str, help="Bind address (default 0.0.0.0)") parser.add_argument("--miner-address", type=str, help="Coinbase address for block rewards") parser.add_argument("--public-url", type=str, help="Publicly reachable URL of this node") parser.add_argument("--detect-public-url", action="store_true", help="Print the detected public URL and exit (used by the installers)") parser.add_argument("--no-mining", action="store_true", help="Disable mining") parser.add_argument("--cores", type=int, help="Mining worker processes (0 = automatic)") parser.add_argument("--intensity", type=int, help="Mining load in percent (5-100)") parser.add_argument("--server", choices=["auto", "waitress", "flask"], default="auto", help="HTTP server (default: waitress, otherwise Flask)") args = parser.parse_args() config = load_config() if args.port: config["rpc_port"] = args.port if args.detect_public_url: # Runs before anything else is started: the installers call this to # pre-fill the public address prompt. print(suggest_public_url(int(config["rpc_port"]), config.get("seed_nodes"))) return if args.host: config["rpc_host"] = args.host if args.miner_address: config["miner_address"] = args.miner_address.strip() if args.public_url: config["public_url"] = args.public_url.strip() if args.no_mining: config["mining_enabled"] = False if args.cores is not None: config["mining_processes"] = max(0, args.cores) if args.intensity is not None: config["mining_intensity"] = max(Miner.MIN_INTENSITY, min(100, args.intensity)) instance, flask_app = build_node(config) log.info("Platform: %s | CPU cores: %s | Mining processes: %s | Intensity: %s %%", PLATFORM_LABEL, instance.miner.cpu_total, instance.miner.processes, instance.miner.intensity) log.info("RPC interface: http://%s:%s", config["rpc_host"], config["rpc_port"]) serve(flask_app, str(config["rpc_host"]), int(config["rpc_port"]), args.server) if __name__ == "__main__": main()