I installed Bitcoin Core on my laptop, synchronized it, and successfully ran it. Additionally, I developed a Python program to perform solo mining by connecting to Bitcoin Core via RPC.
In this program, I implemented an algorithm that generates various nonce values and applies them to the mining process. While I am aware that the probability of successfully mining a block is extremely low, I would like to confirm whether my solo mining Python program is technically correct and operates without errors on the actual Bitcoin network.
It would be greatly helpful if you could review my program and let me know if there are any issues or potential improvements. I sincerely appreciate your time and assistance with this request.
Thank you once again, and I look forward to your response.
import hashlib
import requests
import json
import struct
import time
import random
# RPC settings
RPC_USER = "user"
RPC_PASSWORD = "passwd"
RPC_URL = "
http://127.0.0.1:8332/"
# User-defined address
MINING_ADDRESS = "your_bitcoin_address"
def is_valid_bitcoin_address(address):
"""Validate the Bitcoin address"""
try:
alphabet = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
base58_decoded = 0
for char in address:
base58_decoded = base58_decoded * 58 + alphabet.index(char)
address_bytes = base58_decoded.to_bytes(25, byteorder="big")
checksum = address_bytes[-4:]
hash_checksum = hashlib.sha256(hashlib.sha256(address_bytes[:-4]).digest()).digest()[:4]
return checksum == hash_checksum
except Exception as e:
print(f"Address validation failed: {e}")
return False
# Validate MINING_ADDRESS
if not is_valid_bitcoin_address(MINING_ADDRESS):
raise ValueError(f"Invalid Bitcoin address: {MINING_ADDRESS}")
def rpc_request(method, params=None):
"""Send an RPC request"""
payload = json.dumps({"jsonrpc": "2.0", "id": "mining", "method": method, "params": params or []})
response = requests.post(RPC_URL, auth=(RPC_USER, RPC_PASSWORD), data=payload)
try:
response.raise_for_status()
result = response.json()
if "result" not in result:
raise ValueError("The 'result' field is missing in the RPC response.")
return result["result"]
except Exception as e:
print(f"RPC request failed: {e}")
print(f"Response content: {response.text}")
raise
def address_to_script_pubkey(address):
"""Convert Bitcoin address to ScriptPubKey"""
try:
alphabet = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
base58_decoded = 0
for char in address:
base58_decoded = base58_decoded * 58 + alphabet.index(char)
address_bytes = base58_decoded.to_bytes(25, byteorder="big")
print(f"Base58 decoding successful: {address_bytes.hex()}")
pubkey_hash = address_bytes[1:-4] # Exclude network byte and checksum
script_pubkey = b"\x76\xa9" + bytes([len(pubkey_hash)]) + pubkey_hash + b"\x88\xac"
print(f"ScriptPubKey generated successfully: {script_pubkey.hex()}")
return script_pubkey
except Exception as e:
print(f"Address conversion failed: {e}")
raise
def modify_coinbase_tx(block_template, extra_nonce):
"""Generate a coinbase transaction"""
coinbase_value = block_template["coinbasevalue"]
script_pubkey = address_to_script_pubkey(MINING_ADDRESS)
extra_nonce_script = struct.pack("<Q", extra_nonce)
script_pubkey = extra_nonce_script + script_pubkey
coinbase_tx = (
b"\x01"
b"\x00"
b"\x01"
+ b"\x00" * 32
+ b"\xff" * 4
+ b"\x01"
+ struct.pack("<Q", coinbase_value)
+ struct.pack("B", len(script_pubkey))
+ script_pubkey
)
return coinbase_tx
def calculate_merkle_root(transactions):
"""Calculate the Merkle root"""
transaction_hashes = [hashlib.sha256(bytes.fromhex(tx["data"])).digest() for tx in transactions]
while len(transaction_hashes) > 1:
if len(transaction_hashes) % 2 == 1: # Duplicate the last hash if odd
transaction_hashes.append(transaction_hashes[-1])
transaction_hashes = [
hashlib.sha256(transaction_hashes
+ transaction_hashes[i + 1]).digest()
for i in range(0, len(transaction_hashes), 2)
]
return transaction_hashes[0][::-1] # Return little-endian
def calculate_block_hash(header):
"""Generate a block hash by hashing the block header"""
hash1 = hashlib.sha256(header).digest()
hash2 = hashlib.sha256(hash1).digest()
return hash2[::-1] # Return little-endian
def mine():
"""Mining loop"""
print("Starting Bitcoin mining...")
extra_nonce = 0 # Initialize Extra Nonce
while True:
try:
block_template = rpc_request("getblocktemplate", [{"rules": ["segwit"]}])
previous_block = block_template["previousblockhash"]
target = int(block_template["target"], 16)
print(f"Mining target: {block_template['target']}")
while True: # Extra Nonce loop
coinbase_tx = modify_coinbase_tx(block_template, extra_nonce)
transactions = [{"data": coinbase_tx.hex()}] + block_template["transactions"]
merkle_root = calculate_merkle_root(transactions)
version = struct.pack("<I", block_template["version"])
prev_block_hash = bytes.fromhex(previous_block)[::-1]
ntime = struct.pack("<I", int(block_template["curtime"]))
nbits = bytes.fromhex(block_template["bits"])[::-1]
nonce = 0
while nonce < 2**32: # Nonce range
header = (
version
+ prev_block_hash
+ merkle_root
+ ntime
+ nbits
+ struct.pack("<I", nonce)
)
block_hash = calculate_block_hash(header)
if int(block_hash.hex(), 16) < target:
print(f"Mining successful! Block hash: {block_hash.hex()}")
print(f"Extra Nonce: {extra_nonce}, Nonce: {nonce}")
result = rpc_request("submitblock", [header.hex()])
if result is None:
print("Block successfully submitted!")
else:
print(f"Block submission failed: {result}")
return # Exit mining loop
nonce += 1
# Increase Extra Nonce when nonce is exhausted
print(f"Nonce exhausted - Increasing Extra Nonce: {extra_nonce}")
extra_nonce += 1
except Exception as e:
print(f"Error occurred: {e}")
time.sleep(5)
if __name__ == "__main__":
mine()
