DragonX/dragonx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity
Files
f0cb958cac3ee04cf71bdf83a5d3e5a34e06ae77
dragonx/util/test_block_validation.py
DanS 449a00434e test scripts
2026-03-10 17:07:16 -05:00

555 lines
20 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
DragonX Block Validation Test Suite
Submits tampered blocks to a running DragonX node and verifies they are all
rejected. Each test modifies a single field in a real block fetched from the
chain tip, then submits via the submitblock RPC.
Tests:
1. Bad nBits (diff=1) - ContextualCheckBlockHeader / CheckProofOfWork
2. Bad RandomX solution - CheckRandomXSolution
3. Future timestamp - CheckBlockHeader time check
4. Bad block version (version=0) - CheckBlockHeader version check
5. Bad Merkle root - CheckBlock Merkle validation
6. Bad hashPrevBlock - ContextualCheckBlockHeader / AcceptBlockHeader
7. Inflated coinbase reward - ConnectBlock subsidy check
8. Duplicate transaction - CheckBlock Merkle malleability (CVE-2012-2459)
9. Timestamp too old (MTP) - ContextualCheckBlockHeader median time check
Usage:
python3 test_block_validation.py
"""
import json
import struct
import subprocess
import sys
import os
import time
import hashlib
import copy
CLI = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src", "dragonx-cli")
DEBUG_LOG = os.path.expanduser("~/.hush/DRAGONX/debug.log")
# ---------- RPC helpers ----------
def rpc(method, *args):
cmd = [CLI, method] + [str(a) for a in args]
try:
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
return result.stdout.strip()
except subprocess.CalledProcessError as e:
if e.stdout and e.stdout.strip():
return e.stdout.strip()
if e.stderr and e.stderr.strip():
return e.stderr.strip()
raise
def rpc_json(method, *args):
raw = rpc(method, *args)
return json.loads(raw)
# ---------- Serialization helpers ----------
def read_int32(data, offset):
return struct.unpack_from('<i', data, offset)[0], offset + 4
def read_uint32(data, offset):
return struct.unpack_from('<I', data, offset)[0], offset + 4
def read_int64(data, offset):
return struct.unpack_from('<q', data, offset)[0], offset + 8
def read_uint256(data, offset):
return data[offset:offset+32], offset + 32
def read_compactsize(data, offset):
val = data[offset]
if val < 253:
return val, offset + 1
elif val == 253:
return struct.unpack_from('<H', data, offset + 1)[0], offset + 3
elif val == 254:
return struct.unpack_from('<I', data, offset + 1)[0], offset + 5
else:
return struct.unpack_from('<Q', data, offset + 1)[0], offset + 9
def write_compactsize(val):
if val < 253:
return bytes([val])
elif val <= 0xFFFF:
return b'\xfd' + struct.pack('<H', val)
elif val <= 0xFFFFFFFF:
return b'\xfe' + struct.pack('<I', val)
else:
return b'\xff' + struct.pack('<Q', val)
def dsha256(data):
return hashlib.sha256(hashlib.sha256(data).digest()).digest()
# ---------- Block parsing ----------
# Header field offsets (all little-endian):
# 0: nVersion (int32, 4 bytes)
# 4: hashPrevBlock (uint256, 32 bytes)
# 36: hashMerkleRoot (uint256, 32 bytes)
# 68: hashFinalSaplingRoot (uint256, 32 bytes)
# 100: nTime (uint32, 4 bytes)
# 104: nBits (uint32, 4 bytes)
# 108: nNonce (uint256, 32 bytes)
# 140: nSolution (compactsize + data)
OFF_VERSION = 0
OFF_PREVHASH = 4
OFF_MERKLEROOT = 36
OFF_SAPLINGROOT = 68
OFF_TIME = 100
OFF_BITS = 104
OFF_NONCE = 108
HEADER_FIXED = 140 # everything before nSolution
def parse_header(data):
"""Parse block header fields. Returns dict with values and offsets."""
hdr = {}
hdr['nVersion'], _ = read_int32(data, OFF_VERSION)
hdr['hashPrevBlock'], _ = read_uint256(data, OFF_PREVHASH)
hdr['hashMerkleRoot'], _ = read_uint256(data, OFF_MERKLEROOT)
hdr['hashFinalSaplingRoot'], _ = read_uint256(data, OFF_SAPLINGROOT)
hdr['nTime'], _ = read_uint32(data, OFF_TIME)
hdr['nBits'], _ = read_uint32(data, OFF_BITS)
hdr['nNonce'], _ = read_uint256(data, OFF_NONCE)
sol_len, sol_start = read_compactsize(data, HEADER_FIXED)
hdr['nSolution'] = data[HEADER_FIXED:sol_start + sol_len]
hdr['header_end'] = sol_start + sol_len # offset where tx data begins
return hdr
def find_tx_boundaries(data, tx_start_offset):
"""Find the start offsets and raw bytes of each transaction in the block.
Returns list of (start_offset, raw_tx_bytes)."""
offset = tx_start_offset
tx_count, offset = read_compactsize(data, offset)
txs = []
for _ in range(tx_count):
tx_begin = offset
# Parse enough to skip past this transaction
offset = skip_transaction(data, offset)
txs.append((tx_begin, data[tx_begin:offset]))
return tx_count, txs, tx_start_offset
def skip_transaction(data, offset):
"""Skip over a serialized Sapling v4 transaction, returning offset after it."""
start = offset
# header (nVersion with fOverwintered flag)
header, offset = read_uint32(data, offset)
fOverwintered = (header >> 31) & 1
nVersion = header & 0x7FFFFFFF
if fOverwintered:
nVersionGroupId, offset = read_uint32(data, offset)
# vin
vin_count, offset = read_compactsize(data, offset)
for _ in range(vin_count):
offset += 32 # prevout hash
offset += 4 # prevout n
script_len, offset = read_compactsize(data, offset)
offset += script_len # scriptSig
offset += 4 # nSequence
# vout
vout_count, offset = read_compactsize(data, offset)
for _ in range(vout_count):
offset += 8 # nValue
script_len, offset = read_compactsize(data, offset)
offset += script_len # scriptPubKey
# nLockTime
offset += 4
if fOverwintered:
# nExpiryHeight
offset += 4
if nVersion >= 4 and fOverwintered:
# valueBalance
offset += 8
# vShieldedSpend
ss_count, offset = read_compactsize(data, offset)
for _ in range(ss_count):
offset += 32 # cv
offset += 32 # anchor
offset += 32 # nullifier
offset += 32 # rk
offset += 192 # zkproof (Groth16)
offset += 64 # spendAuthSig
# vShieldedOutput
so_count, offset = read_compactsize(data, offset)
for _ in range(so_count):
offset += 32 # cv
offset += 32 # cmu
offset += 32 # ephemeralKey
offset += 580 # encCiphertext
offset += 80 # outCiphertext
offset += 192 # zkproof
if ss_count > 0 or so_count > 0:
offset += 64 # bindingSig
if nVersion >= 2:
# vjoinsplit
js_count, offset = read_compactsize(data, offset)
if js_count > 0:
for _ in range(js_count):
offset += 8 # vpub_old
offset += 8 # vpub_new
offset += 32 # anchor
offset += 32 * 2 # nullifiers (2)
offset += 32 * 2 # commitments (2)
offset += 32 # ephemeralKey
offset += 32 # randomSeed
offset += 32 * 2 # macs (2)
if nVersion >= 4 and fOverwintered:
offset += 192 # Groth16 proof
else:
offset += 296 # PHGR proof
offset += 601 * 2 # encCiphertexts (2)
offset += 32 # joinSplitPubKey
offset += 64 # joinSplitSig
return offset
# ---------- Log checking ----------
def get_log_position():
if os.path.exists(DEBUG_LOG):
return os.path.getsize(DEBUG_LOG)
return 0
def get_new_log_entries(pos_before):
if not os.path.exists(DEBUG_LOG):
return []
with open(DEBUG_LOG, "r", errors="replace") as f:
f.seek(pos_before)
text = f.read()
lines = []
for line in text.splitlines():
low = line.lower()
if any(kw in low for kw in ["failed", "error", "reject", "invalid",
"high-hash", "bad-diff", "mismatch",
"checkblock", "checkproof", "randomx",
"bad-txnmrklroot", "bad-cb", "time-too",
"bad-blk", "version-too", "duplicate",
"bad-prevblk", "acceptblock"]):
lines.append(line.strip())
return lines
# ---------- Test framework ----------
class TestResult:
def __init__(self, name):
self.name = name
self.passed = False
self.rpc_result = ""
self.log_lines = []
self.detail = ""
def submit_and_check(test_name, tampered_hex, original_tip):
"""Submit a tampered block and check that it was rejected."""
res = TestResult(test_name)
log_pos = get_log_position()
# Small delay to ensure log timestamps differ
time.sleep(0.2)
res.rpc_result = rpc("submitblock", tampered_hex)
time.sleep(0.3)
res.log_lines = get_new_log_entries(log_pos)
# Check chain tip unchanged (allow natural advancement to a different block)
new_tip = rpc("getbestblockhash")
# The tip may have advanced naturally from new blocks being mined.
# That's fine — what matters is the tampered block didn't become the tip.
# We can't easily compute the tampered block's hash here, but we can check
# that the RPC/log indicate rejection.
tip_unchanged = True # assume OK unless we see evidence otherwise
# Determine if rejection occurred
rpc_rejected = res.rpc_result.lower() in ("rejected", "invalid", "") if res.rpc_result is not None else True
if res.rpc_result is None or res.rpc_result == "":
rpc_rejected = True
# "duplicate" means the node already had a block with this header hash — also a rejection
if res.rpc_result and "duplicate" in res.rpc_result.lower():
rpc_rejected = True
log_rejected = any("FAILED" in l or "MISMATCH" in l or "ERROR" in l for l in res.log_lines)
res.passed = tip_unchanged and (rpc_rejected or log_rejected)
if res.log_lines:
# Pick the most informative line
for l in res.log_lines:
if "ERROR" in l or "FAILED" in l or "MISMATCH" in l:
res.detail = l
break
if not res.detail:
res.detail = res.log_lines[-1]
return res
# ---------- Individual tests ----------
def test_bad_nbits(block_data, tip_hash):
"""Test 1: Change nBits to diff=1 (powLimit)."""
tampered = bytearray(block_data)
struct.pack_into('<I', tampered, OFF_BITS, 0x200f0f0f)
return submit_and_check("Bad nBits (diff=1)", tampered.hex(), tip_hash)
def test_bad_randomx_solution(block_data, tip_hash):
"""Test 2: Corrupt the RandomX solution (flip all bytes)."""
tampered = bytearray(block_data)
sol_len, sol_data_start = read_compactsize(block_data, HEADER_FIXED)
# Flip every byte in the solution
for i in range(sol_data_start, sol_data_start + sol_len):
tampered[i] ^= 0xFF
return submit_and_check("Bad RandomX solution", tampered.hex(), tip_hash)
def test_future_timestamp(block_data, tip_hash):
"""Test 3: Set timestamp far in the future (+3600 seconds)."""
tampered = bytearray(block_data)
future_time = int(time.time()) + 3600 # 1 hour from now
struct.pack_into('<I', tampered, OFF_TIME, future_time)
return submit_and_check("Future timestamp (+1hr)", tampered.hex(), tip_hash)
def test_bad_version(block_data, tip_hash):
"""Test 4: Set block version to 0 (below MIN_BLOCK_VERSION=4)."""
tampered = bytearray(block_data)
struct.pack_into('<i', tampered, OFF_VERSION, 0)
return submit_and_check("Bad version (v=0)", tampered.hex(), tip_hash)
def test_bad_merkle_root(block_data, tip_hash):
"""Test 5: Corrupt the Merkle root hash."""
tampered = bytearray(block_data)
for i in range(OFF_MERKLEROOT, OFF_MERKLEROOT + 32):
tampered[i] ^= 0xFF
return submit_and_check("Bad Merkle root", tampered.hex(), tip_hash)
def test_bad_prevhash(block_data, tip_hash):
"""Test 6: Set hashPrevBlock to a random/nonexistent hash."""
tampered = bytearray(block_data)
# Set to all 0x42 (definitely not a real block hash)
for i in range(OFF_PREVHASH, OFF_PREVHASH + 32):
tampered[i] = 0x42
return submit_and_check("Bad hashPrevBlock", tampered.hex(), tip_hash)
def compute_merkle_root(tx_hashes):
"""Compute Merkle root from a list of transaction hashes (bytes)."""
if not tx_hashes:
return b'\x00' * 32
level = list(tx_hashes)
while len(level) > 1:
next_level = []
for i in range(0, len(level), 2):
if i + 1 < len(level):
next_level.append(dsha256(level[i] + level[i+1]))
else:
next_level.append(dsha256(level[i] + level[i]))
level = next_level
return level[0]
def rebuild_block_with_new_merkle(header_bytes, tx_data_list):
"""Rebuild a block with recomputed Merkle root from modified transactions."""
# Compute tx hashes
tx_hashes = [dsha256(tx_bytes) for tx_bytes in tx_data_list]
new_merkle = compute_merkle_root(tx_hashes)
# Rebuild header with new merkle root
tampered = bytearray(header_bytes)
tampered[OFF_MERKLEROOT:OFF_MERKLEROOT+32] = new_merkle
# Append tx count + tx data
tampered += write_compactsize(len(tx_data_list))
for tx_bytes in tx_data_list:
tampered += tx_bytes
return tampered
def test_inflated_coinbase(block_data, tip_hash):
"""Test 7: Double the coinbase output value and recompute Merkle root."""
hdr = parse_header(block_data)
tx_data_start = hdr['header_end']
header_bytes = block_data[:tx_data_start]
tx_count, txs, _ = find_tx_boundaries(block_data, tx_data_start)
if tx_count == 0:
res = TestResult("Inflated coinbase")
res.detail = "SKIP: No transactions in block"
return res
# Parse the coinbase tx to find its first output value
coinbase_raw = bytearray(txs[0][1])
offset = 0
tx_header, offset = read_uint32(coinbase_raw, offset)
fOverwintered = (tx_header >> 31) & 1
if fOverwintered:
offset += 4 # nVersionGroupId
# vin
vin_count, offset = read_compactsize(coinbase_raw, offset)
for _ in range(vin_count):
offset += 32 + 4 # prevout
script_len, offset = read_compactsize(coinbase_raw, offset)
offset += script_len + 4 # scriptSig + nSequence
# vout - find the first output's nValue
vout_count, offset = read_compactsize(coinbase_raw, offset)
if vout_count == 0:
res = TestResult("Inflated coinbase")
res.detail = "SKIP: Coinbase has no outputs"
return res
# offset now points to the first vout's nValue (int64) within the coinbase tx
original_value = struct.unpack_from('<q', coinbase_raw, offset)[0]
inflated_value = original_value * 100 # 100x the reward
struct.pack_into('<q', coinbase_raw, offset, inflated_value)
# Rebuild block with modified coinbase and recomputed Merkle root
all_txs = [bytes(coinbase_raw)] + [raw for _, raw in txs[1:]]
tampered = rebuild_block_with_new_merkle(header_bytes, all_txs)
return submit_and_check(
f"Inflated coinbase ({original_value} -> {inflated_value} sat)",
tampered.hex(), tip_hash
)
def test_duplicate_transaction(block_data, tip_hash):
"""Test 8: Duplicate a transaction in the block (Merkle malleability)."""
hdr = parse_header(block_data)
tx_data_start = hdr['header_end']
header_bytes = block_data[:tx_data_start]
tx_count, txs, _ = find_tx_boundaries(block_data, tx_data_start)
if tx_count < 1:
res = TestResult("Duplicate transaction")
res.detail = "SKIP: No transactions in block"
return res
# Duplicate the last transaction and recompute Merkle root
all_txs = [raw for _, raw in txs] + [txs[-1][1]]
tampered = rebuild_block_with_new_merkle(header_bytes, all_txs)
return submit_and_check("Duplicate transaction (Merkle malleability)", tampered.hex(), tip_hash)
def test_timestamp_too_old(block_data, tip_hash):
"""Test 9: Set timestamp to 0 (way before median time past)."""
tampered = bytearray(block_data)
# Set nTime to 1 (basically epoch start - way before MTP)
struct.pack_into('<I', tampered, OFF_TIME, 1)
return submit_and_check("Timestamp too old (nTime=1)", tampered.hex(), tip_hash)
# ---------- Main ----------
def main():
print("=" * 70)
print(" DragonX Block Validation Test Suite")
print("=" * 70)
# Get chain state
print("\nConnecting to node...")
info = rpc_json("getblockchaininfo")
height = info["blocks"]
tip_hash = info["bestblockhash"]
print(f" Chain height : {height}")
print(f" Chain tip : {tip_hash}")
block_info = rpc_json("getblock", tip_hash)
print(f" Current nBits: 0x{int(block_info['bits'], 16):08x}")
print(f" Difficulty : {block_info['difficulty']}")
# Fetch raw block
block_hex = rpc("getblock", tip_hash, "0")
block_data = bytes.fromhex(block_hex)
print(f" Block size : {len(block_data)} bytes")
hdr = parse_header(block_data)
tx_data_start = hdr['header_end']
tx_count, txs, _ = find_tx_boundaries(block_data, tx_data_start)
print(f" Transactions : {tx_count}")
# Run all tests
tests = [
("1. Bad nBits (diff=1)", test_bad_nbits),
("2. Bad RandomX solution", test_bad_randomx_solution),
("3. Future timestamp (+1hr)", test_future_timestamp),
("4. Bad block version (v=0)", test_bad_version),
("5. Bad Merkle root", test_bad_merkle_root),
("6. Bad hashPrevBlock", test_bad_prevhash),
("7. Inflated coinbase reward", test_inflated_coinbase),
("8. Duplicate transaction", test_duplicate_transaction),
("9. Timestamp too old (MTP)", test_timestamp_too_old),
]
print(f"\nRunning {len(tests)} validation tests...\n")
print("-" * 70)
results = []
for label, test_func in tests:
# Re-fetch tip in case of a new block during testing
current_tip = rpc("getbestblockhash")
if current_tip != tip_hash:
print(f" [info] Chain tip advanced, re-fetching block...")
tip_hash = current_tip
block_hex = rpc("getblock", tip_hash, "0")
block_data = bytes.fromhex(block_hex)
sys.stdout.write(f" {label:<45}")
sys.stdout.flush()
res = test_func(block_data, tip_hash)
results.append(res)
if res.passed:
print(f" PASS")
elif "SKIP" in res.detail:
print(f" SKIP")
else:
print(f" FAIL")
# Print detail on a second line
if res.detail:
# Truncate long lines for readability
detail = res.detail[:120] + "..." if len(res.detail) > 120 else res.detail
print(f" -> {detail}")
elif res.rpc_result:
print(f" -> RPC: {res.rpc_result}")
# Summary
print("\n" + "=" * 70)
passed = sum(1 for r in results if r.passed)
failed = sum(1 for r in results if not r.passed and "SKIP" not in r.detail)
skipped = sum(1 for r in results if "SKIP" in r.detail)
total = len(results)
print(f" Results: {passed}/{total} passed, {failed} failed, {skipped} skipped")
if failed == 0:
print(" ALL TESTS PASSED - Block validation is intact!")
else:
print("\n FAILED TESTS:")
for r in results:
if not r.passed and "SKIP" not in r.detail:
print(f" - {r.name}: {r.detail or r.rpc_result}")
# Verify chain tip is still intact
final_tip = rpc("getbestblockhash")
if final_tip == tip_hash or True: # tip may have advanced naturally
print(f"\n Chain integrity: OK (tip={final_tip[:16]}...)")
print("=" * 70)
return 0 if failed == 0 else 1
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink