wall.py

import hashlib, binascii, sys, os, hmac
from bitarray import bitarray
from bitarray.util import ba2int
from hashlib import sha256
from itertools import permutations
import csv
import time
from dotenv import dotenv_values
import concurrent.futures
from bip_utils import Bip39SeedGenerator, Bip44, Bip44Coins, Bip44Changes
from ecdsa import SECP256k1
from ecdsa.ecdsa import Public_key
from eth_utils import keccak, to_checksum_address
import base58
from pybip39 import Mnemonic, Seed
# from mnemonic import Mnemonic
from solders.keypair import Keypair
from solders.pubkey import Pubkey
from solana.rpc.async_api import AsyncClient
import asyncio
import threading
import queue
import logging

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(threadName)s - %(message)s')
logger = logging.getLogger(__name__)

seed_queue = queue.Queue(maxsize=100)  # Adjust maxsize as needed
found_wallet = threading.Event()
bals_checked = 0
seeds_generated = 0

config = dotenv_values(".env.txt")
# from bip_utils.bip.bip39 import Bip39MnemonicValidator

SIMULATE = True

hashesComputed = 0
testedduplicate = 0
testedcands = [""]
testedcandidates = 0

f = open("english.txt", "r")
vars = open(".env.txt", "r")
w1 = open("seedlist.txt", "w")
twords = f.readlines()

bals_checked = 0

words = []
words_fantasy = []
words_sign = []
words_arrow = []
words_trash = []
simplewords = []
i = 0
wrdindex = {}
for word in twords:
    word = word.replace("\n", "")
    wrdindex[word] = i
    words.append(word)
    i += 1

# mneN = config["SEEDPHRASE"]
CHAIN = config["CHAIN"]
DESCRAMBLE = eval(config["DESCRAMBLE"])

passphrase = ""


def generateSeed(mnemonicstring):
    salt = "mnemonic" + passphrase
    seed = hashlib.pbkdf2_hmac(
        "sha512", mnemonicstring.encode("utf-8"), salt.encode("utf-8"), 2048
    )
    return seed.hex()

def generate_seeds():
    while not found_wallet.is_set():
        # Your existing seed generation logic goes here
        # Instead of writing to a file, put the seed in the queue
        seed = " ".join(generate_seed_phrase())  # Implement this function based on your existing logic
        seed_queue.put(seed)



def isOk(ws):
    N = 0
    for w in ws:
        N = (N<<11) + wrdindex[w]

    nhex = format(N, '033x') # include leading zero if needed

    h = hashlib.sha256(binascii.unhexlify(nhex[:-1])).hexdigest()

    return h[0] == nhex[-1]




def isOk24(seed_phrase):
    # word_indices = {word: index for index, word in enumerate(words)}
    entropy_length = 264  # 24 words * 11 bits/word
    checksum_length = entropy_length // 32
    checksum_start = entropy_length - checksum_length

    entropy = ''.join(format(wrdindex[word], '011b') for word in seed_phrase)
    checksum = entropy[checksum_start:]
    entropy = entropy[:checksum_start]

    hash_bytes = hashlib.sha256(int(entropy, 2).to_bytes((len(entropy) + 7) // 8, 'big')).digest()
    hash_bits = ''.join(format(byte, '08b') for byte in hash_bytes)
    calculated_checksum = hash_bits[:checksum_length]

    return calculated_checksum == checksum



def generateRoot(seed):
    # the HMAC-SHA512 `key` and `data` must be bytes:
    seed_bytes = binascii.unhexlify(seed)
    I = hmac.new(b"Bitcoin seed", seed_bytes, hashlib.sha512).digest()
    L, R = I[:32], I[32:]
    master_private_key = int.from_bytes(L, "big")
    master_chain_code = R
    # print(str(master_private_key)+" CC:  "+str(master_chain_code))
    # return [master_private_key, master_chain_code]

    VERSION_BYTES = {
        "mainnet_public": binascii.unhexlify("0488b21e"),
        "mainnet_private": binascii.unhexlify("0488ade4"),
        "testnet_public": binascii.unhexlify("043587cf"),
        "testnet_private": binascii.unhexlify("04358394"),
    }
    version_bytes = VERSION_BYTES["mainnet_private"]
    depth_byte = b"\x00"
    parent_fingerprint = b"\x00" * 4
    child_number_bytes = b"\x00" * 4
    key_bytes = b"\x00" + L
    all_parts = (
        version_bytes,  # 4 bytes
        depth_byte,  # 1 byte
        parent_fingerprint,  # 4 bytes
        child_number_bytes,  # 4 bytes
        master_chain_code,  # 32 bytes
        key_bytes,  # 33 bytes
    )
    all_bytes = b"".join(all_parts)
    root_key = base58.b58encode_check(all_bytes).decode("utf8")
    # print(root_key)
    # xprv9s21ZrQH143K...T2emdEXVYsCzC2U
    # print(root_key)
    return [master_private_key, master_chain_code, root_key]


def deriveWallet(pkey, chain_code, root_key):
    # Break each depth into integers (m/44'/60'/0'/0/0)
    #    e.g. (44, 60, 0, 0, 0)# If hardened, add 2**31 to the number:
    #    e.g. (44 + 2**31, 60 + 2**31, 0 + 2**31, 0, 0)
    if CHAIN == "SOLANA":
        # use m/44'/501'/0'/0'
        path_numbers = (44 + 2**31, 501 + 2**31, 0 + 2**31, 0, 0)
    else:
        path_numbers = (2147483692, 2147483708, 2147483648, 0, 0)
    depth = 0
    parent_fingerprint = None
    child_number = None
    private_key = pkey
    chain_code = chain_code

    from ecdsa import SECP256k1
    from ecdsa.ecdsa import Public_key

    SECP256k1_GEN = SECP256k1.generator

    def serialize_curve_point(p):
        x, y = p.x(), p.y()
        if y & 1:
            return b"\x03" + x.to_bytes(32, "big")
        else:
            return b"\x02" + x.to_bytes(32, "big")

    def curve_point_from_int(k):
        return Public_key(SECP256k1_GEN, SECP256k1_GEN * k).point

    def fingerprint_from_private_key(k):
        K = curve_point_from_int(k)
        K_compressed = serialize_curve_point(K)

        identifier = hashlib.new(
            "ripemd160",
            hashlib.sha256(K_compressed).digest(),
        ).digest()

        return identifier[:4]

    SECP256k1_ORD = SECP256k1.order

    def derive_ext_private_key(private_key, chain_code, child_number):
        if child_number >= 2**31:
            # Generate a hardened key
            data = b"\x00" + private_key.to_bytes(32, "big")
        else:
            # Generate a non-hardened key
            p = curve_point_from_int(private_key)
            data = serialize_curve_point(p)

        data += child_number.to_bytes(4, "big")

        hmac_bytes = hmac.new(chain_code, data, hashlib.sha512).digest()
        L, R = hmac_bytes[:32], hmac_bytes[32:]

        L_as_int = int.from_bytes(L, "big")
        child_private_key = (L_as_int + private_key) % SECP256k1_ORD
        child_chain_code = R

        return (child_private_key, child_chain_code)

    def derivePublicKey(private_key):
        # Derive the public key Point:
        p = curve_point_from_int(private_key)
        # print(f'Point object: {p}\n')

        # Serialize the Point, p
        public_key_bytes = serialize_curve_point(p)

        # print(f'public key (hex): 0x{public_key_bytes.hex()}')
        return public_key_bytes.hex()

    def deriveWalletAddresss(private_key):
        from eth_utils import keccak

        p = curve_point_from_int(private_key)
        # Hash the concatenated x and y public key point values:
        digest = keccak(p.x().to_bytes(32, "big") + p.y().to_bytes(32, "big"))

        # Take the last 20 bytes and add '0x' to the front:
        address = "0x" + digest[-20:].hex()

        # print(f'address: {address}')
        return address

    for i in path_numbers:
        depth += 1
        child_number = i
        parent_fingerprint = fingerprint_from_private_key(private_key)
        private_key, chain_code = derive_ext_private_key(
            private_key, chain_code, child_number
        )

    # print(f'private key: {hex(private_key)}')
    pubkey = derivePublicKey(private_key)
    address = deriveWalletAddresss(private_key)
    return (private_key, pubkey, address)


def seed_to_keys_and_address(seed_phrase, coin_type='ETH'):
    # Generate seed from mnemonic
    seed_bytes = Bip39SeedGenerator(seed_phrase).Generate()

    if coin_type == 'SOL':
        # Solana derivation path: m/44'/501'/0'/0'

        # Use Solana's Keypair to derive public key and address
        mnemonic = Mnemonic.from_phrase(seed_phrase)
        seed = Seed(mnemonic, passphrase)

        account_index = 0
        derivation_path = f"m/44'/501'/0'/{account_index}'"
        keypair = Keypair.from_seed_and_derivation_path(seed_bytes, derivation_path)
        

        return {
            'keypair': keypair,
            
        }

    else:  # ETH
        # Ethereum derivation path: m/44'/60'/0'/0/0
        bip44_mst_ctx = Bip44.FromSeed(seed_bytes, Bip44Coins.ETHEREUM)
        bip44_acc_ctx = bip44_mst_ctx.Purpose().Coin().Account(0)
        bip44_chg_ctx = bip44_acc_ctx.Change(Bip44Changes.CHAIN_EXT)
        bip44_addr_ctx = bip44_chg_ctx.AddressIndex(0)
        private_key_bytes = bip44_addr_ctx.PrivateKey().Raw().ToBytes()
        private_key = int.from_bytes(private_key_bytes, 'big')

        # Generate public key
        public_key_point = SECP256k1.generator * private_key
        
        # Serialize public key
        x = public_key_point.x()
        y = public_key_point.y()
        if y & 1:  # odd
            public_key = b'\x03' + x.to_bytes(32, 'big')
        else:  # even
            public_key = b'\x02' + x.to_bytes(32, 'big')

        # Generate Ethereum address
        address = to_checksum_address(keccak(public_key_point.x().to_bytes(32, 'big') + 
                                             public_key_point.y().to_bytes(32, 'big'))[-20:])

        return {
            'private_key': private_key_bytes.hex(),
            'public_key': public_key.hex(),
            'address': address
        }

async def check_solana_balance(client, keypair):
    try:
        public_key = keypair.pubkey()
        balance = await client.get_balance(public_key)
        return balance
    except Exception as e:
        print(f"An error occurred while checking balance: {str(e)}")
        return None


targetWallet = config["TARGETWALLET"]
# targetWallet = targetWallet.lower()   

# given list of tuples in lengths of 2, 3, 4, 5, 6..
# assume order of tuples is correct, don't permute.
# take each tuple, permute internally, create candidate as list
ggCount = 0
dCount = 0


# def fillAddons(rcand):
#     global ggCount
#     # non-tuple
#     raw = rcand.copy()
#     missinglen = len(raw)
#     diff = SEEDLENGTH - missinglen
#     fillAddonsTuples(raw, diff)

def fillAddons(rcand):
    global seeds_generated
    # non-tuple
    raw = rcand.copy()
    missinglen = len(raw)
    diff = SEEDLENGTH - missinglen
    fillAddonsTuples(raw, diff)


def fillAddonsTuples(raw, diff):
    # permute all known tuples and condense into iterations of lists
    noTup = True
    # Test each element for tuples
    for x, ele in enumerate(raw):
        # If list contains tuple elements, condense tuple-lists into independant lists and recurse into this fn.
        if isinstance(ele, tuple):
            noTup = False
            # permute elements in tuple, recreate list each time and pass on.
            for atom in ele:
                cand = raw.copy()
                cand[x] = atom
                fillAddonsTuples(cand, diff)
            # Stop iterating over other main-cand elements since we've already recursed over the others.
            # Basically, stop at one found tuple element.
            break
    if noTup:
        # Take a cleaned list (ie. wo/tuples) and do the standard fill missing words and permute thing.
        # print("testing list cand: " + str(raw))
        fillAddonsPermute(raw, diff)


def fillAddonsPermute(raw, diff):
    global seeds_generated
    # The following block expects "raw" to be a list
    permB = permutations(words, diff)
    for i in permB:
        # generate seed
        tcand = raw + list(i)
        tc = " ".join(tcand)
        if SEEDLENGTH == 12:
            if isOk(tcand):
                if targetWallet:
                    kp = seed_to_keys_and_address(tc, 'SOL')
                    seeds_generated += 1
                    if seeds_generated % 100 == 0:
                        logger.info(f"{seeds_generated} valid seeds generated")
                    if kp['keypair'].pubkey() == targetWallet:
                        seed_queue.put(tc)
                        
                        
                else:
                    seed_queue.put(tc)
                    seeds_generated += 1
                    if seeds_generated % 1000 == 0:
                        logger.info(f"{seeds_generated} valid seeds generated")
        else:
            if isOk24(tcand):
                seed_queue.put(tc)
                seeds_generated += 1
                if seeds_generated % 1000 == 0:
                    logger.info(f"{seeds_generated} valid seeds generated")


actual = 0
save = False
written = 0


# taking each permutated element, check if the element is a tuple, if so, permute the tuple and check
def candTester(cand):
    global actual, save, written
    noTup = True
    for x, c in enumerate(cand):
        # if element in candidate is tuple, iterate over subelements and resurse
        if isinstance(c, tuple):
            noTup = False
            for d in c:
                candcopy = cand.copy()
                candcopy[x] = d
                candTester(candcopy)
            # stop in the first occurance of tuple element
            break
    if noTup:
        # print("cand: "+str(cand))
        # if (cand == control):
        #     print("Matched with control: "+str(cand))
        # exit()
        # Don't test cand, save to file.
        actual += 1
        if actual % 1000 == 0:
            print("written " + str(actual) + " actuals")
        # if (isOk(cand)):
        w1.write(" ".join(cand))
        w1.write("\n")
        written += 1
        # testCand(cand)


def menmonicRecursive(mnem, cand=[]):
    global actual, save
    length = len(mnem)
    if length > 0:
        perms = permutations(mnem[0])
        for k in perms:
            menmonicRecursive(mnem[1:], tuple(cand) + k)
    else:
        # print("mnem: "+str(mnem)+"\ncand: "+str(cand))
        # actual += 1
        candTester(list(cand))
        # if (actual >= 115000000):
        #     save = True
        # elif(actual > 460000000):
        #     print("reached 222,000,000 limit, quitting")
        #     quit()


start = time.perf_counter()




t_mneN = list(config['SEEDPERM'].split(","))
mneN = []

for i in t_mneN:
    # wrap string in list
    mneN.append(tuple(t_mneN))


SEEDLENGTH = 12
print(mneN)

# fillAddons(mneN)
# # mneN = fillAddons(mneN)
# # anotherPermuteFunc(mneN)
# # menmonicRecursive(mneN)
# w1.close()
# stop = time.perf_counter()
# runtime = stop - start
# hashpm = (hashesComputed / runtime) * 60
# print(f"Ran for {runtime:0.4f} seconds; Average: " + str(hashpm) + "H/m")
# print("Written: " + str(ggCount) + " seeds")
# print("Iterated: " + str(dCount) + " seed candidates")

async def check_balance(client, seed_phrase):
    global bals_checked
    try:
        sol_keys = seed_to_keys_and_address(seed_phrase, 'SOL')
        balance = await check_solana_balance(client, sol_keys['keypair'])
        bals_checked += 1
        if bals_checked % 100 == 0:
            logger.info(f"Checked {bals_checked} balances")
        if balance and balance.value > 0:
            logger.info(f"Found wallet with positive balance: {seed_phrase}")
            logger.info(f"Balance: {balance.value / 1e9} SOL")
            found_wallet.set()
        await asyncio.sleep(0.2)
    except Exception as e:
        logger.error(f"Error checking balance: {e}")

async def balance_checker():
    async with AsyncClient("https://mainnet.helius-rpc.com/?api-key=3c6f9703-9af7-4fd0-9588-452400129f84") as client:
        while not found_wallet.is_set():
            try:
                seed_phrase = seed_queue.get(timeout=1)
                await check_balance(client, seed_phrase)
            except queue.Empty:
                await asyncio.sleep(0.1)
            except Exception as e:
                logger.error(f"Unexpected error in balance checker: {e}")

def balance_checker_wrapper():
    asyncio.run(balance_checker())

def seed_generator():
    fillAddons(mneN)
    logger.info("Seed generation completed")

async def main():
    logger.info("Starting wallet search...")

    # Start the seed generation thread
    seed_gen_thread = threading.Thread(target=seed_generator, name="SeedGenerator")
    seed_gen_thread.start()

    # Start the balance checking thread
    balance_check_thread = threading.Thread(target=balance_checker_wrapper, name="BalanceChecker")
    balance_check_thread.start()

    # Wait for the threads to complete
    while not found_wallet.is_set():
        if not seed_gen_thread.is_alive() and seed_queue.empty():
            logger.info("All seeds checked, no positive balance found")
            break
        await asyncio.sleep(1)

    found_wallet.set()  # Ensure we signal the threads to stop
    seed_gen_thread.join()
    balance_check_thread.join()

    logger.info("Wallet search completed.")

if __name__ == "__main__":
    asyncio.run(main())