Чекер Bitcoin с автовыводом и глубокой деривацией

D2

D2

Администратор
Регистрация
19 Фев 2025
Сообщения
4,380
Реакции
0

Введение​

В данной статье будет реализован чекер мнемонических фраз на баланс Bitcoin, а также вывод монет на свой адрес.

Как будет работать чекер​

Баланс будет проверяться с использованием стороннего сервиса, а не ноды. Также поиск адресов с балансом будет происходить не по одному адресу от мнемонической фразы, а с глубокой деривацией вплоть до сотого адреса или больше по желанию. Формат адресов будет как Legacy, так и Segwit.

Почему не будет использоваться публичная нода по типу getblock.io или nownodes.io?​

Дело в том, что Bitcoin нода сама по себе не имеет функции для получения баланса кошелька по адресу. Сразу же приходит мысль посчитать все UTXO (непотраченные выходы), но с этим тоже возникают проблемы, т.к. команды, которые могут с этим помочь, заблокированы в публичных нодах.

Какие есть варианты что бы посчитать баланс кошелька по адресу через ноду?​

  1. scantxoutset. Сканирует все непотраченные выходы по адресу (не смог найти ни одной ноды, где эта функция не заблокирована).
  2. listunspent. Возвращает все непотраченные выходы по адресу (работает только с кошельками, чьи адреса импортированы локально в ноду, также запрещено в публичных нодах).

В связи с сложившейся ситуацией, единственным, по моему мнению, вариантом остается использование сервиса типа Blockchain.com или blockcypher.com.

Как будет работать вывод монет?​

Получение всех неистраченных выходов будет происходить через сторонний сервис, а создание и подписание транзакций — через библиотеку bitcoinlib. Отправка транзакции в обработку будет осуществляться с использованием публичной ноды.

Необходимые модули​

Перед началом написания проекта потребуется установить необходимый модуль C++ для корректной установки необходимых в будущем библиотек Python: Microsoft C++ Build Tools - Microsoft C++ Build Tools - Visual Studio
1736542199764.png



После установки Microsoft C++ Build Tools и необходимых компонентов можно приступать к написанию самого софта.

Получение приватных ключей и адресов​

Первое, что будет реализовано, — это получение адресов из мнемонических фраз, а также приватных ключей.

Принцип получения из мнемонической фразы приватного ключа​

Для начала рассмотрим принцип получения из мнемонической фразы приватного ключа и адреса.
  1. Получаем корневой ключ из мнемонической фразы (приватный ключ, но не от конкретного адреса, а от самой мнемонической фразы).
  2. Настраиваем деривационный путь с указанием монеты, типом BIP и индексом адреса.
  3. Используя деривационный путь, получаем публичный ключ.
  4. Из публичного ключа получаем адрес.
  5. Используя деривационный путь, также получаем приватный ключ.
  6. Конвертируем приватный ключ в формат WIF (приватный ключ в привычном для людей виде).

Чтение мнемонических фраз​

Теперь можно приступить к написанию кода, и первое, что будет сделано, — это создание необходимых переменных и чтение мнемонических фраз из текстового файла.
Python: Скопировать в буфер обмена
Код: 
import asyncio
import aiofiles
from bip_utils import Bip39SeedGenerator, Bip44, Bip49, Bip84, Bip44Coins, Bip49Coins, Bip84Coins, Bip44Changes
import hashlib
import base58


async def process_mnemonics():
   # Считывает каждую строку из текстового файла
   async with aiofiles.open(file_path, "r", encoding="utf-8") as file:
       mnemonics = await file.readlines()


   # Создаем задачи для всех мнемонических фраз
   tasks = [process_mnemonic(mnemonic) for mnemonic in mnemonics]
   await asyncio.gather(*tasks)


if __name__ == "__main__":
   file_path = "mnemonics.txt"  # Путь к файлу с мнемоническими фразами
   output_file_path = "output.txt"  # Путь к файлу для записи результата
   depth = 5  # Количество генерируемых адресов для каждой фразы

   asyncio.run(process_mnemonics())
Как видно, код будет асинхронным. По моим тестам, в данном случае скорость гораздо выше, чем при работе с многопоточностью (было проведено сравнение). Также, как видно из импортов, основной библиотекой для работы с мнемоническими фразами, адресами и т.д. будет использоваться библиотека bip_utils (именно для нее и был необходим Microsoft C++ Build Tools).

Рассмотрим одну строку более подробно:
Python: Скопировать в буфер обмена
tasks = [process_mnemonic(mnemonic) for mnemonic in mnemonics]
В данной строке в цикле перебираются все мнемонические фразы, записываются в переменную mnemonic и передаются в функцию process_mnemonic (которой пока что нет).

Вызов функции для получения адресов, приватных ключей и обработка полученных данных​

Теперь рассмотрим саму функцию process_mnemonic.
Данная функция нужна лишь для того, чтобы вызвать основную функцию (mnemonic_to_wallet) для генерации адресов и ключей из мнемонической фразы, а затем полученные данные разбить на несколько переменных и вывести в консоль.
Python: Скопировать в буфер обмена
Код: 
async def process_mnemonic(mnemonic_phrase):
   async with aiofiles.open(output_file_path, "a", encoding="utf-8") as output_file:
       # Убирает пробелы в начале и в конце мнемонической фразы если они есть
       mnemonic_phrase = mnemonic_phrase.strip()
       # Если мнемоническая фраза есть
       if mnemonic_phrase:
           # Вызываем функцию для генерации приватных ключей и адресов передавая в функцию мнемоническую фразу
           # В переменную записываются данные из функции генерации приватных ключей и адресов
           wallets = await mnemonic_to_wallet(mnemonic_phrase)

           # Если переменная не пустая
           if wallets:
               for wallet in wallets:
                   # Данные из переменной разбиваются на отдельные переменные для адресов, приватных ключей и т.д
                   address, private_key_wif, mnemonic = wallet
                   # Данные из двух переменных записываются в переменную result.
                   result = f"{mnemonic}:{private_key_wif}:{address}\n"
                   # Вывод данных в консоль
                   print(result)
           else:
               print(f"Не удалось обработать фразу: {mnemonic_phrase}")

Функция генерации приватных ключей и адресов​

Теперь рассмотрим функцию mnemonic_to_wallet. Именно в ней будет происходить вся логика работы с мнемонической фразой.
Python: Скопировать в буфер обмена
Код: 
async def mnemonic_to_wallet(mnemonic):
   # Создаем пустой массив для хранения всех данных, полученных из мнемонической фразы
   wallets = []
   try:
       # Конвертация мнемонической фразы в её байтовое представление (seed)
       seed_bytes = Bip39SeedGenerator(mnemonic).Generate("")

       # Цикл, выполняющийся столько раз, сколько указано в depth
       for i in range(depth):
           # BIP44: Генерация внешних и внутренних адресов
           bip44 = Bip44.FromSeed(seed_bytes, Bip44Coins.BITCOIN)

           # Внешние адреса (для приема монет)
           bip44_ctx_ext = bip44.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT).AddressIndex(i)
           bip44_address_ext = bip44_ctx_ext.PublicKey().ToAddress()
           bip44_private_key_ext = await private_key_to_wif(bip44_ctx_ext.PrivateKey().Raw().ToHex())

           # Внутренние адреса (для сдачи монет)
           bip44_ctx_int = bip44.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_INT).AddressIndex(i)
           bip44_address_int = bip44_ctx_int.PublicKey().ToAddress()
           bip44_private_key_int = await private_key_to_wif(bip44_ctx_int.PrivateKey().Raw().ToHex())

           # Добавляем оба типа адресов в список
           wallets.append((bip44_address_ext, bip44_private_key_ext, mnemonic))
           wallets.append((bip44_address_int, bip44_private_key_int, mnemonic))

           # BIP49: Генерация внешних и внутренних адресов
           bip49 = Bip49.FromSeed(seed_bytes, Bip49Coins.BITCOIN)

           # Внешние адреса (для приема монет)
           bip49_ctx_ext = bip49.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT).AddressIndex(i)
           bip49_address_ext = bip49_ctx_ext.PublicKey().ToAddress()
           bip49_private_key_ext = await private_key_to_wif(bip49_ctx_ext.PrivateKey().Raw().ToHex())

           # Внутренние адреса (для сдачи монет)
           bip49_ctx_int = bip49.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_INT).AddressIndex(i)
           bip49_address_int = bip49_ctx_int.PublicKey().ToAddress()
           bip49_private_key_int = await private_key_to_wif(bip49_ctx_int.PrivateKey().Raw().ToHex())

           # Добавляем оба типа адресов в список
           wallets.append((bip49_address_ext, bip49_private_key_ext, mnemonic))
           wallets.append((bip49_address_int, bip49_private_key_int, mnemonic))

           # BIP84: Генерация внешних и внутренних адресов
           bip84 = Bip84.FromSeed(seed_bytes, Bip84Coins.BITCOIN)

           # Внешние адреса (для приема монет)
           bip84_ctx_ext = bip84.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT).AddressIndex(i)
           bip84_address_ext = bip84_ctx_ext.PublicKey().ToAddress()
           bip84_private_key_ext = await private_key_to_wif(bip84_ctx_ext.PrivateKey().Raw().ToHex())

           # Внутренние адреса (для сдачи монет)
           bip84_ctx_int = bip84.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_INT).AddressIndex(i)
           bip84_address_int = bip84_ctx_int.PublicKey().ToAddress()
           bip84_private_key_int = await private_key_to_wif(bip84_ctx_int.PrivateKey().Raw().ToHex())

           # Добавляем оба типа адресов в список
           wallets.append((bip84_address_ext, bip84_private_key_ext, mnemonic))
           wallets.append((bip84_address_int, bip84_private_key_int, mnemonic))

       return wallets
   except Exception as e:
       print(f"Ошибка при генерации адресов: {e}")
       return []

Разбор функции по частям​

Python: Скопировать в буфер обмена
wallets = []
Создание пустого массива, в который впоследствии будут добавляться приватный ключ и адрес.

Python: Скопировать в буфер обмена
seed_bytes = Bip39SeedGenerator(mnemonic).Generate("")
Конвертация мнемонической фразы в её байтовое представление (seed).
P.S. Стоит упомянуть, что seed-фраза — это байтовое представление мнемонической фразы.

Python: Скопировать в буфер обмена
for i in range(depth):
Цикл, срабатывающий столько раз, сколько указано в depth. Значение записывается в переменную i. i используется для индексации адреса при составлении деривационного пути. Чуть позже будет рассказано, для чего нужна индексация адресов.

Теперь рассмотрим саму работу с мнемонической фразой на примере формата BIP44.
Python: Скопировать в буфер обмена
Код: 
bip44 = Bip44.FromSeed(seed_bytes, Bip44Coins.BITCOIN)

# Внешние адреса (для приема монет)
bip44_ctx_ext = bip44.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT).AddressIndex(i)
bip44_address_ext = bip44_ctx_ext.PublicKey().ToAddress()
bip44_private_key_ext = await private_key_to_wif(bip44_ctx_ext.PrivateKey().Raw().ToHex())

# Внутренние адреса (для сдачи монет)
bip44_ctx_int = bip44.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_INT).AddressIndex(i)
bip44_address_int = bip44_ctx_int.PublicKey().ToAddress()
bip44_private_key_int = await private_key_to_wif(bip44_ctx_int.PrivateKey().Raw().ToHex())

# Добавляем оба типа адресов в список
wallets.append((bip44_address_ext, bip44_private_key_ext, mnemonic))
wallets.append((bip44_address_int, bip44_private_key_int, mnemonic))

Составление деривационного пути​

Для начала получаем корневой ключ, используя seed-фразу с указанием монеты. Затем составляем деривационный путь в этих строках:
Python: Скопировать в буфер обмена
Код: 
bip44 = Bip44.FromSeed(seed_bytes, Bip44Coins.BITCOIN)
bip44.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT).AddressIndex(i)
Нас интересует несколько моментов
  1. Bip44Changes.CHAIN_EXT: Означает что адрес будет внешнего типа (то есть для принятия монет)
  2. AddressIndex(i): То самое i из цикла for. Обозначает индекс адреса

Зачем нужен индекс адреса (индексация адреса)?​

Дело в том, что мнемоническая фраза — это не один адрес и даже не сотня. Адресов может быть миллионы для каждого типа, а типов тоже немало для разных валют. Допустим, Segwit и Legacy — это два разных типа, и у каждого огромное количество вариантов адресов. Поэтому просто перебирать каждый для поиска нужного для проверки баланса или любой другой информации нет смысла. Существует индексация каждого адреса, и именно поэтому все кошельки выдают адреса по порядку, начиная с нулевого. Это необходимо для того, чтобы быстро перебрать адреса по индексу и найти нужные данные конкретных адресов.

Получение адреса и приватного ключа​

После составления деривационного пути можно получить публичный ключ, из которого сразу же можно получить адрес.
Python: Скопировать в буфер обмена
bip44_address_ext = bip44_ctx_ext.PublicKey().ToAddress()

Теперь получаем приватный ключ в hex формате, после его получения передаем его в функцию (которой пока что нет) для конвертации в WIF формат.
Python: Скопировать в буфер обмена
bip44_private_key_int = await private_key_to_wif(bip44_ctx_int.PrivateKey().Raw().ToHex())

Сейчас была рассмотрена логика для получения внешних адресов BIP 44 формата. Для других форматов всё аналогично, как для внешних, так и для внутренних адресов.

Конвертация приватного ключа в WIF формат​

Теперь рассмотрим функцию для конвертации приватного ключа в WIF формат путём добавления префиксов и несколькими этапами хеширования.

Почему получение приватного ключа в WIF формате сделать сложнее чем получение адреса или публичного ключа?​

На самом деле, получение публичного ключа и адреса должно происходить таким же образом, как и WIF приватного ключа, просто библиотека bip_utils делает все эти расчеты за нас и может делать это для многих монет.

Для чего несколько этапов хеширования?​

По сути, хеширование задумывалось как улучшенная защита данных и уникализация. Но на практике, в данном случае, оно нужно, чтобы получить приватный ключ в WIF формате.

Этапы получения приватного ключа в WIF формате​

  1. Конвертация ключа из hex в байты.
  2. В конец полученного байтового ключа добавляется префикс \x01, обозначающий, что ключ будет компрессированным.
  3. В начале получившегося ключа с префиксом компрессии добавляется префикс, обозначающий, что ключ предназначен для монеты Bitcoin.
  4. Полученный ключ с двумя префиксами хешируется дважды с помощью SHA-256.
  5. Берём первые 4 байта от получившегося хеша — это контрольная сумма.
  6. Берём ключ, получившийся в пункте 3, и добавляем в конце контрольную сумму из пункта 5.
  7. Кодируем в Base58.
  8. Декодируем в строку.
  9. Возвращаем в функцию mnemonic_to_wallet.
Python: Скопировать в буфер обмена
Код: 
async def private_key_to_wif(private_key_hex):
   prefix = b'\x80'
   key_bytes = bytes.fromhex(private_key_hex)
   key_bytes += b'\x01'
   extended_key = prefix + key_bytes
   first_hash = hashlib.sha256(extended_key).digest()
   second_hash = hashlib.sha256(first_hash).digest()
   checksum = second_hash[:4]
   extended_key_with_checksum = extended_key + checksum
   encoded_key = base58.b58encode(extended_key_with_checksum)
   wif_key = encoded_key.decode()

   return wif_key

Добавление сгенерированных данных в массив​

Т.к. приватный ключ WIF формата передается обратно в mnemonic_to_wallet, то рассмотрим, что с этим ключом в дальнейшем происходит в функции mnemonic_to_wallet.
Опять же рассмотрим на примере BIP44.
Python: Скопировать в буфер обмена
Код: 
wallets.append((bip44_address_ext, bip44_private_key_ext, mnemonic))
wallets.append((bip44_address_int, bip44_private_key_int, mnemonic))
Как видно, приватный ключ WIF формата, а также адрес и мнемоническая фраза добавляются в массив, который в конце функции возвращается в функцию process_mnemonic, чтобы полученные данные разбить на несколько переменных и вывести в консоль.

Проверка баланса​

На этом с получением приватных ключей и адресов закончено, теперь нужно приступать к работе над самим чекером. Как было сказано в начале статьи, работа будет проходить именно с сторонним сервисом.

Выбор сервиса для проверки баланса​

При выборе сервиса было решено использовать blockcypher.com, а не Blockchain.com, т.к. у blockcypher запросы кончаются заметно медленнее, да и данные у них обновляются значительно быстрее, чем у Blockchain.

Как обойти ограничение количества запросов?​

Стоит понимать, что у подобных сервисов есть ограничение на количество запросов, поэтому его нужно обойти. Для этого у меня было два варианта:
  1. Использовать Tor как прокси и при каждом запросе перезапускать соединение для смены IP.
  2. Использовать нормальные прокси, но платные.
Изначально я хотел воспользоваться Tor, но скорость его работы оставляет желать лучшего, и к тому же он не всегда запускается, даже с мостами. Из-за этих важных причин были выбраны обычные прокси https.

Принцип работы с прокси​

Раз было выяснено, как обойти ограничение, то теперь нужно для этого написать логику.
Как будет устроена работа с прокси:
  1. Считываем текстовый файл со списком прокси.
  2. Берём рандомную строку.
  3. Отправляем запрос на сервис, используя этот рандомный прокси.

Обновление process_mnemonic​

Теперь можно приступить к написанию кода.
Первое, что будет сделано, — это обновление функции process_mnemonic. В ней нужно вызвать будущую функцию для проверки баланса внутри цикла for.
Python: Скопировать в буфер обмена
Код: 
for wallet in wallets:
   # Данные из переменной разбиваются на отдельные переменные для адресов, приватных ключей и т.д
   address, private_key_wif, mnemonic = wallet
   balance = await check_balance(address)
   # Данные из двух переменных записываются в переменную result.
   result = f"{mnemonic}:{private_key_wif}:{address}:{balance}\n"
   # Данные из переменной result записываются в текстовый файл
   await output_file.write(result)
   # Вывод данных в консоль
   print(result)

Чтение списка прокси и выбор рандомного из списка​

Теперь рассмотрим саму функцию для проверки баланса check_balance по частям.
Первое, что нужно в ней сделать, — это вызвать функции для получения списка прокси и выбора из него рандомного прокси.
Python: Скопировать в буфер обмена
Код: 
proxies = await load_proxies("proxy.txt")
rnd_proxy = await get_next_proxy(proxies)

Функции по своей сути не сложные, так что комментировать их особо нет смысла. Поэтому просто предоставлю код.
Python: Скопировать в буфер обмена
Код: 
async def load_proxies(proxy_file_path):
   try:
       # Используем асинхронное чтение файла
       async with aiofiles.open(proxy_file_path, mode="r") as file:
           proxies = [line.strip() async for line in file if line.strip()]
       return proxies
   except Exception as e:
       print(f"Ошибка загрузки прокси: {e}")
       return []


async def get_next_proxy(proxies):
   # Если список пустой
   if not proxies:
       print("Нет доступных прокси.")
       return None

   # Берем случайную прокси
   proxy_info = random.choice(proxies)
   # Разделяем прокси на части по знаку ":"
   parts = proxy_info.split(":")


   # Если частей не 4
   if len(parts) != 4:
       print(f"Неверный формат прокси: {proxy_info}")
       return None

   # Записываем каждую часть в отдельную переменную
   ip, port, username, password = parts

   # Собираем первые две части в виде ссылки
   proxy = f"http://{username}:{password}@{ip}:{port}"
   return proxy
Хочу лишь упомянуть разбиение на части в load_proxies. Проверка происходит именно на 4 части, потому что прокси должны быть в таком формате:
194.226.232.141:9067:sRNGPB:E9b1pa

Проверка работоспособности прокси​

С функциями получения рандомного прокси закончено, и теперь перейдем обратно в функцию check_balance, где они вызывались.
Python: Скопировать в буфер обмена
Код: 
if not rnd_proxy:
   print("Нет доступных прокси.")
   return None

# Перед проверкой баланса проверим IP
ip = await check_ip(rnd_proxy)

if ip:
   print(f"Используем прокси с IP: {ip}")
else:
   print("Не удалось проверить IP. Прокси может быть недоступен.")
   return None
В данном блоке кода присутствует вызов функции check_ip для проверки валидности прокси. Рассмотрим же эту функцию.
Python: Скопировать в буфер обмена
Код: 
async def check_ip(proxy):
   url = "http://httpbin.org/ip"
   try:
       async with aiohttp.ClientSession() as session:
           # Делаем запрос через прокси
           async with session.get(url, proxy=proxy) as response:
               if response.status == 200:
                   data = await response.json()
                   ip = data.get("origin")
                   print(f"Прокси IP: {ip}")
                   return ip
               else:
                   print(f"Ошибка при проверке IP с прокси: {response.status}")
                   return None
   except Exception as e:
       print(f"Ошибка при проверке IP с прокси: {e}")
       return None

Отправка запроса на получение баланса через прокси​

Теперь возвращаемся в функцию check_balance.
Python: Скопировать в буфер обмена
Код: 
url = f"https://api.blockcypher.com/v1/btc/main/addrs/{address}/balance"
try:
   async with aiohttp.ClientSession() as session:
       async with session.get(url, proxy=rnd_proxy) as response:
           if response.status == 200:
               data = await response.json()
               balance = data.get('final_balance', 0)
               return balance
           else:
               print(f"Ошибка при получении баланса для {address}: {response.status}")
               return None
except Exception as e:
   print(f"Ошибка при запросе баланса для {address}: {e}")
   return None
Как видим, это базовый код для отправки запросов. Хочу лишь обратить внимание на эту строчку:
Python: Скопировать в буфер обмена
session.get(url, proxy=rnd_proxy)
Именно в ней указывается прокси. rnd_proxy представляет собой ссылку такого формата:
Python: Скопировать в буфер обмена
f"http://{username}:{password}@{ip}:{port}"
То есть, если вам нужен лишь один прокси, можно просто указать его здесь и не писать функции для чтения файла с прокси и выбора рандомной прокси из списка.

Автовывод монет​

С чекером баланса закончено, теперь можно приступать к написанию логов для вывода монет, но перед этим нужно сначала зарегистрироваться в сервисе для получения доступа к публичной ноде.

Выбор публичной ноды​

Публичных нод достаточно много, но была выбрана именно эта — getblock.io.

Почему именно эта нода?​

Выбрана она была потому, что регистрацию на ней можно пройти, используя временную почту, что позволяет легко избегать лимитов. Из сервисов, предоставляющих временную почту, можно выбрать этот — inboxes.com.

Как получить доступ​

После регистрации в сервисе нужно получить ссылку, по которой нужно делать запросы к ноде.
1736543094693.png


Логика вывода монет​

  1. Если на адресе более 500 сатоши, то вызвать функцию для получения всех неистраченных входов.
  2. Все неистраченные входы отправить в функцию для создания транзакции.
  3. В функции для создания транзакции использовать неистраченные входы, а также указать сумму вывода, комиссию, адрес, куда выводить, и адрес, откуда выводить.
  4. Отправить получившуюся транзакцию на обработку в сеть, используя функцию для отправки запросов на публичную ноду.

Обновление process_mnemonic​

Теперь можно приступить к написанию самого кода, и первое, что будет сделано, — это обновление функции process_mnemonic.
Python: Скопировать в буфер обмена
Код: 
# Асинхронная обработка одной мнемонической фразы
async def process_mnemonic(mnemonic_phrase):
   async with aiofiles.open(output_file_path, "a", encoding="utf-8") as output_file:
       # Убирает пробелы в начале и в конце мнемонической фразы если они есть
       mnemonic_phrase = mnemonic_phrase.strip()
       # Если мнемоническая фраза есть
       if mnemonic_phrase:
           # Вызываем функцию для генерации приватных ключей и адресов передавая в функцию мнемоническую фразу
           # В переменную записываются данные из функции генерации приватных ключей и адресов
           wallets = await mnemonic_to_wallet(mnemonic_phrase)

           # Если переменная не пустая
           if wallets:
               for wallet in wallets:
                   # Данные из переменной разбиваются на отдельные переменные для адресов, приватных ключей и т.д
                   address, private_key_wif, mnemonic = wallet
                   balance = await check_balance(address)
                   if balance > 500:

                       # Вызов функции для получения utxo
                       transaction_info = await get_utxos(address, private_key_wif)


                       # Добавляем данные транзакции в результат
                       result = (
                           f"=========================================================\n"
                           f"{mnemonic}:{private_key_wif}:{address}:{balance} Satoshi\n"
                           f"Сумма отправки: {transaction_info['amount_sent']} Satoshi\n"
                           f"Комиссия: {transaction_info['total_fee']} Satoshi\n"
                           f"=========================================================\n\n"
                       )
                   else:
                       # Если баланс меньше 500, возвращаем только базовые данные
                       result = (
                           f"=========================================================\n"
                           f"{mnemonic}:{private_key_wif}:{address}:{balance} Satoshi\n"
                           f"=========================================================\n\n"
                       )
                   # Данные из переменной result записываются в текстовый файл
                   await output_file.write(result)
                   # Вывод данных в консоль
                   print(result)
           else:
               print(f"Не удалось обработать фразу: {mnemonic_phrase}")
В данной функции был изменён этот блок кода:
Python: Скопировать в буфер обмена
Код: 
for wallet in wallets:
   # Данные из переменной разбиваются на отдельные переменные для адресов, приватных ключей и т.д
   address, private_key_wif, mnemonic = wallet
   balance = await check_balance(address)
   if balance > 500:

       # Вызов функции для получения utxo
       transaction_info = await get_utxos(address, private_key_wif)

       # Добавляем данные транзакции в результат
       result = (
           f"=========================================================\n"
           f"{mnemonic}:{private_key_wif}:{address}:{balance} Satoshi\n"
           f"Сумма отправки: {transaction_info['amount_sent']} Satoshi\n"
           f"Комиссия: {transaction_info['total_fee']} Satoshi\n"
           f"=========================================================\n\n"
       )
   else:
       # Если баланс меньше 500, возвращаем только базовые данные
       result = (
           f"=========================================================\n"
           f"{mnemonic}:{private_key_wif}:{address}:{balance} Satoshi\n"
           f"=========================================================\n\n"
       )
   # Данные из переменной result записываются в текстовый файл
   await output_file.write(result)
   # Вывод данных в консоль
   print(result)
В нём была добавлена проверка на то, чтобы, если баланс больше 500 сатоши, вызывать функцию для получения UTXO. Также в этом условии изменяется result, так как теперь можно будет вывести сумму отправки и комиссию.

P.S. Сумма отправки и комиссия будут получены при вызове функции создания транзакции, которая как раз и будет вызываться в get_utxos.

Функция для получения UTXO​

Теперь перейдём к самой функции get_utxos.
Python: Скопировать в буфер обмена
Код: 
async def get_utxos(address, private_key_wif):
   proxies = await load_proxies("proxy.txt")
   rnd_proxy = await get_next_proxy(proxies)

   if not rnd_proxy:
       print("Нет доступных прокси.")
       return None

   # Перед проверкой баланса проверим IP
   ip = await check_ip(rnd_proxy)

   if ip:
       print(f"Используем прокси с IP: {ip}")
   else:
       print("Не удалось проверить IP. Прокси может быть недоступен.")
       return None

   url = f"https://api.blockcypher.com/v1/btc/main/addrs/{address}?unspentOnly=true"
   try:
       async with aiohttp.ClientSession() as session:
           async with session.get(url, proxy=rnd_proxy) as response:
               if response.status == 200:
                   data = await response.json()
                   # Пустой список для записи всех UTXO
                   utxos = []
                   # Обрабатываем каждый UTXO
                   for txref in data.get("txrefs", []) + data.get("unconfirmed_txrefs", []):
                       # Извлекаем данные из UTXO
                       utxos.append({
                           "tx_hash": txref["tx_hash"],
                           "tx_output_n": txref["tx_output_n"],
                           "value": txref["value"]
                       })

                   if utxos:
                       # Создаем транзакцию и возвращаем сумму вывода и комиссию
                       transaction_info = await create_transaction(utxos, private_key_wif, address)
                       return transaction_info
                   else:
                       print(f"Нет доступных UTXO для адреса {address}")
                       return None
               else:
                   print(f"Ошибка при получении UTXO для {address}: {response.status}")
                   return None
   except Exception as e:
       print(f"Ошибка при запросе UTXO для {address}: {e}")
       return None
Как и при получении баланса, тут вызываются функции для получения прокси, чтобы отправить запрос через него на сервис blockcypher. Из ответа от сервиса собираются данные из конкретных ключей, а именно:
  • Хеш транзакции из неистраченного входа
  • Индекс транзакции
  • Сумма транзакции
После получения всех UTXO вызывается функция для создания транзакции create_transaction, в неё передаются сам UTXO, приватный ключ от адреса и адрес, куда отправлять монеты.

Функция для создания и подписания транзакции​

Теперь рассмотрим саму функцию create_transaction.
Python: Скопировать в буфер обмена
Код: 
async def create_transaction(utxos, private_key_wif, address):
   # Определяем тип witness в зависимости от формата адреса
   if address.startswith("1"):
       witness_type = 'legacy'
   elif address.startswith("bc1"):
       witness_type = 'segwit'
   else:
       raise ValueError("Неподдерживаемый формат адреса: адрес должен быть legacy или segwit")

   tx = Transaction(network=Network('bitcoin'), replace_by_fee=False, witness_type=witness_type)

   base_fee = 170  # Фиксированная базовая комиссия
   fee_per_input = 150  # Комиссия за каждый вход
   total_fee = base_fee
   total_amount = 0

   # Указываем входы транзакции
   for utxo in utxos:
       total_amount += utxo["value"]  # Общая сумма входов
       total_fee += fee_per_input  # Увеличиваем комиссию на каждый вход
       tx.add_input(
           prev_txid=utxo["tx_hash"],
           output_n=utxo["tx_output_n"],
           value=utxo["value"],
           address=address
       )

   # Вычисляем сумму для отправки после вычета комиссии
   amount_to_send = total_amount - total_fee

   # Добавление выхода с адресом и суммой
   tx.add_output(address=service_address, value=int(round(amount_to_send)))

   # Подписываем транзакцию
   tx.sign(private_key_wif)
   signed_tx = tx.as_hex()

   # Отправляем транзакцию
   await request_node("sendrawtransaction", [str(signed_tx)]

   # Возвращаем сумму вывода и комиссию
   return {
       "amount_sent": amount_to_send,
       "total_fee": total_fee
   }
В данной функции хочу обратить внимание на несколько моментов.
Python: Скопировать в буфер обмена
Код: 
if address.startswith("1"):
   witness_type = 'legacy'
elif address.startswith("bc1"):
   witness_type = 'segwit'
else:
   raise ValueError("Неподдерживаемый формат адреса: адрес должен быть legacy или segwit")
Данный код нужен для того, чтобы указать, какого типа транзакцию в дальнейшем создавать. То есть, если адрес, с которого нужно вывести монету, начинается с единицы, то транзакция должна быть типа Legacy, как и сам адрес. Если с bc1, то транзакция должна быть типа Segwit.

Также считаю нужным объяснить эту часть кода:
Python: Скопировать в буфер обмена
Код: 
base_fee = 170  # Фиксированная базовая комиссия
fee_per_input = 150  # Комиссия за каждый вход
total_fee = base_fee
total_amount = 0

# Указываем входы транзакции
for utxo in utxos:
   total_amount += utxo["value"]  # Общая сумма входов
   total_fee += fee_per_input  # Увеличиваем комиссию на каждый вход
   tx.add_input(
       prev_txid=utxo["tx_hash"],
       output_n=utxo["tx_output_n"],
       value=utxo["value"],
       address=address
   )

# Вычисляем сумму для отправки после вычета комиссии
amount_to_send = total_amount - total_fee
В ней происходит расчет комиссии путём добавления 150 сатоши за каждый вход. Комиссию нужно рассчитывать, так как она варьируется от количества входов и выходов. Значение в 150 было выбрано примерно, а не точно равным минимальной сумме за вход или выход.
Также хочу заметить, что сумму комиссии не нужно указывать в транзакции, в комиссию идёт всё, что не пошло в выходы, то есть всё, что не указано как сумма отправки. Именно поэтому из суммы отправки вычитается сумма комиссии.

В принципе, все остальные строки подписаны, и должно быть понятно. Лишь обращу внимание на эту строку:
Python: Скопировать в буфер обмена
await request_node("sendrawtransaction", [str(signed_tx)])
В ней вызывается функция для отправки транзакции в сеть, в эту функцию передаётся команда, которая будет отправлена на ноду, и подписанная транзакция.

Отправка транзакции в сеть​

Теперь рассмотрим саму функцию для отправки транзакции в сеть.
Python: Скопировать в буфер обмена
Код: 
async def request_node(method, params):
   payload = {
       "jsonrpc": "2.0",
       "method": method,
       "params": params,
       "id": 1
   }
   try:
       async with aiohttp.ClientSession() as session:
           async with session.post(
               "https://go.getblock.io/4fa66b9bad67415e9b8f5fb5bfb0f54b",
               json=payload,
               headers={"Content-Type": "application/json"}
           ) as response:
               if response.status == 200:
                   # Если запрос удачный, возвращаем ответ с данными
                   return await response.json()
               else:
                   print(f"Ошибка: {response.status}, {await response.text()}")
                   return None
   except aiohttp.ClientError as e:
       print(f"Ошибка запроса: {e}")
       return None
В ней ничего сложного — просто отправка POST-запроса на ноду.
На этом написание вывода монет закончено, и вот что получается по итогу:
  • В process_mnemonic вызывается функция для получения всех UTXO через blockcypher, которые передаются в функцию create_transaction.
  • В create_transaction все UTXO записываются в входы для транзакции, за каждый UTXO комиссия увеличивается на 150 сатоши.
  • К входам добавляется выход с указанием адреса, куда отправить, и суммы, которая состоит из общей суммы всех входов с вычитанием 150 сатоши за каждый из них.
  • Полученная транзакция подписывается.
  • Подписанная транзакция отправляется в функцию для отправки запроса на ноду, чтобы её обработать.

Результат работы софта:​

1736543418367.png


Вывод:​

Надеюсь, статья, как и сам код, получились понятными для освоения, так как я старался написать всё максимально просто для понимания начинающими разработчиками. Насчёт кода, проведя несколько десятков тестов, могу сказать, что он полностью работоспособный и его вполне можно использовать на практике, но не исключаю непредвиденных багов. О них вы можете написать в комментариях или на GitHub.

Статья в виде документа - https://docs.google.com/document/d/1UDkvHE4fFF_JvvVJYyZ3t0ujJEyahrxnSbASueE1yCY/edit?usp=sharing

Исходный код проекта на GitHub - https://github.com/overlordgamedev/Bitcoin-Checker


Сделано OverlordGameDev специально для форума XSS.IS
 
Войдите или зарегистрируйтесь для ответа.
Сверху Снизу