refactor: fixes, improvements

This commit is contained in:
Artur
2025-03-06 14:44:47 -03:00
parent 56da51373d
commit 8554f88e7f
9 changed files with 116 additions and 81 deletions

View File

@@ -1,16 +1,25 @@
BTC_ELECTRUM_RPC_URL=""
LTC_ELECTRUM_RPC_URL=""
ELECTRUM_RPC_USERNAME=""
ELECTRUM_RPC_PASSWORD=""
BITCOIN_ELECTRUM_RPC_URL=""
BITCOIN_ELECTRUM_RPC_USERNAME="user"
BITCOIN_ELECTRUM_RPC_PASSWORD=""
LITECOIN_ELECTRUM_RPC_URL=""
LITECOIN_ELECTRUM_RPC_USERNAME="user"
LITECOIN_ELECTRUM_RPC_PASSWORD=""
BITCOIN_ELECTRUM_SERVER_ADDRESS=""
LITECOIN_ELECTRUM_SERVER_ADDRESS=""
BITCOIN_WALLET_SEED=""
LITECOIN_WALLET_SEED=""
MONERO_DAEMON_ADDRESS=""
MONERO_RPC_URL=""
MONERO_RPC_USERNAME=""
MONERO_RPC_USERNAME="user"
MONERO_RPC_PASSWORD=""
MONERO_WALLET_SEED=""
MONERO_WALLET_PASSWORD=""
MONERO_WALLET_HEIGHT=""
MONERO_DAEMON_ADDRESS=""
NBXPLORER_URL=""
NBXPLORER_USERNAME=""
NBXPLORER_PASSWORD=""
KRAKEN_API_KEY=""
KRAKEN_API_SECRET=""

View File

@@ -4,13 +4,13 @@ set -ex
trap 'pkill -TERM -P1; electrum stop; exit 0' SIGTERM
rm -f .electrum/daemon
electrum --offline setconfig rpcuser ${BTC_ELECTRUM_USER}
electrum --offline setconfig rpcpassword ${BTC_ELECTRUM_PASSWORD}
electrum --offline setconfig rpcuser ${BITCOIN_ELECTRUM_USER}
electrum --offline setconfig rpcpassword ${BITCOIN_ELECTRUM_PASSWORD}
electrum --offline setconfig rpchost 0.0.0.0
electrum --offline setconfig rpcport 7000
if [ -n "${BTC_ELECTRUM_SERVER_ADDRESS}" ]; then
electrum daemon -1 -s "${BTC_ELECTRUM_SERVER_ADDRESS}" "$@"
if [ -n "${BITCOIN_ELECTRUM_SERVER_ADDRESS}" ]; then
electrum daemon -1 -s "${BITCOIN_ELECTRUM_SERVER_ADDRESS}" "$@"
else
electrum daemon "$@"
fi

View File

@@ -45,11 +45,16 @@ services:
container_name: seed-importer
environment:
- PYTHONUNBUFFERED=1
- BTC_ELECTRUM_RPC_URL=http://btc-electrum:7000
- LTC_ELECTRUM_RPC_URL=http://ltc-electrum:7001
- ELECTRUM_RPC_USERNAME=${ELECTRUM_RPC_USERNAME}
- ELECTRUM_RPC_PASSWORD=${ELECTRUM_RPC_PASSWORD}
- BITCOIN_ELECTRUM_RPC_URL=${BITCOIN_ELECTRUM_RPC_URL}
- BITCOIN_ELECTRUM_RPC_USERNAME=${BITCOIN_ELECTRUM_RPC_USERNAME}
- BITCOIN_ELECTRUM_RPC_PASSWORD=${BITCOIN_ELECTRUM_RPC_PASSWORD}
- LITECOIN_ELECTRUM_RPC_URL=${LITECOIN_ELECTRUM_RPC_URL}
- LITECOIN_ELECTRUM_RPC_USERNAME=${LITECOIN_ELECTRUM_RPC_USERNAME}
- LITECOIN_ELECTRUM_RPC_PASSWORD=${LITECOIN_ELECTRUM_RPC_PASSWORD}
- BITCOIN_ELECTRUM_SERVER_ADDRESS=${BITCOIN_ELECTRUM_SERVER_ADDRESS}
- LITECOIN_ELECTRUM_SERVER_ADDRESS=${LITECOIN_ELECTRUM_SERVER_ADDRESS}
- BITCOIN_WALLET_SEED=${BITCOIN_WALLET_SEED}
- LITECOIN_WALLET_SEED=${LITECOIN_WALLET_SEED}
- MONERO_RPC_URL=http://monero-wallet-rpc:18082/json_rpc
- MONERO_RPC_USERNAME=${MONERO_RPC_USERNAME}
- MONERO_RPC_PASSWORD=${MONERO_RPC_PASSWORD}
@@ -69,11 +74,16 @@ services:
restart: unless-stopped
environment:
- PYTHONUNBUFFERED=1
- BTC_ELECTRUM_RPC_URL=http://btc-electrum:7000
- LTC_ELECTRUM_RPC_URL=http://ltc-electrum:7001
- ELECTRUM_RPC_USERNAME=${ELECTRUM_RPC_USERNAME}
- ELECTRUM_RPC_PASSWORD=${ELECTRUM_RPC_PASSWORD}
- BITCOIN_ELECTRUM_RPC_URL=${BITCOIN_ELECTRUM_RPC_URL}
- BITCOIN_ELECTRUM_RPC_USERNAME=${BITCOIN_ELECTRUM_RPC_USERNAME}
- BITCOIN_ELECTRUM_RPC_PASSWORD=${BITCOIN_ELECTRUM_RPC_PASSWORD}
- LITECOIN_ELECTRUM_RPC_URL=${LITECOIN_ELECTRUM_RPC_URL}
- LITECOIN_ELECTRUM_RPC_USERNAME=${LITECOIN_ELECTRUM_RPC_USERNAME}
- LITECOIN_ELECTRUM_RPC_PASSWORD=${LITECOIN_ELECTRUM_RPC_PASSWORD}
- BITCOIN_ELECTRUM_SERVER_ADDRESS=${BITCOIN_ELECTRUM_SERVER_ADDRESS}
- LITECOIN_ELECTRUM_SERVER_ADDRESS=${LITECOIN_ELECTRUM_SERVER_ADDRESS}
- BITCOIN_WALLET_SEED=${BITCOIN_WALLET_SEED}
- LITECOIN_WALLET_SEED=${LITECOIN_WALLET_SEED}
- MONERO_RPC_URL=http://monero-wallet-rpc:18082/json_rpc
- MONERO_RPC_USERNAME=${MONERO_RPC_USERNAME}
- MONERO_RPC_PASSWORD=${MONERO_RPC_PASSWORD}

View File

@@ -4,13 +4,13 @@ set -ex
trap 'pkill -TERM -P1; electrum stop; exit 0' SIGTERM
rm -f .electrum/daemon
electrum --offline setconfig rpcuser ${LTC_ELECTRUM_USER}
electrum --offline setconfig rpcpassword ${LTC_ELECTRUM_PASSWORD}
electrum --offline setconfig rpcuser ${LITECOIN_ELECTRUM_USER}
electrum --offline setconfig rpcpassword ${LITECOIN_ELECTRUM_PASSWORD}
electrum --offline setconfig rpchost 0.0.0.0
electrum --offline setconfig rpcport 7001
if [ -n "${LTC_ELECTRUM_SERVER_ADDRESS}" ]; then
electrum daemon -1 -s "${LTC_ELECTRUM_SERVER_ADDRESS}" "$@"
if [ -n "${LITECOIN_ELECTRUM_SERVER_ADDRESS}" ]; then
electrum daemon -1 -s "${LITECOIN_ELECTRUM_SERVER_ADDRESS}" "$@"
else
electrum daemon "$@"
fi

View File

@@ -4,6 +4,7 @@ version = "0.2.0"
description = ""
authors = ["Your Name <you@example.com>"]
readme = "README.md"
package-mode = false
[tool.poetry.dependencies]
python = "^3.12"

View File

@@ -8,20 +8,20 @@ from constants import MIN_BITCOIN_SEND_AMOUNT, MIN_LITECOIN_SEND_AMOUNT, MIN_MON
import util
import env
def get_bitcoin_fee_rate(source: str, rate: str) -> int:
def get_fee_rate(source: str, rate: str) -> int:
return requests.get(source).json()[rate]
def set_bitcoin_fee_rate(coin: str, rate: int, dynamic: bool):
def set_electrum_fee_rate(coin: Literal['btc', 'ltc'], rate: int, dynamic: bool):
if dynamic: # Fall back to the Electrum rate if there is an issue
util.request_electrum_rpc(coin, 'setconfig', ['dynamic_fees', True])
else:
util.request_electrum_rpc(coin, 'setconfig', ['dynamic_fees', False])
util.request_electrum_rpc(coin, 'setconfig', ['fee_per_kb', rate * 1000])
def get_bitcoin_balance(coin: str) -> float:
def get_electrum_balance(coin: Literal['btc', 'ltc']) -> float:
return float(util.request_electrum_rpc(coin, 'getbalance')['confirmed'])
def create_psbt(coin: str, destination_address: str) -> str:
def create_psbt(coin: Literal['btc', 'ltc'], destination_address: str) -> str:
params = {
'destination': destination_address,
'amount': '!',
@@ -30,7 +30,7 @@ def create_psbt(coin: str, destination_address: str) -> str:
return util.request_electrum_rpc(coin, 'payto', params)
def get_psbt_data(coin: str, psbt: str) -> dict:
def get_psbt_data(coin: Literal['btc', 'ltc'], psbt: str) -> dict:
return util.request_electrum_rpc(coin, 'deserialize', [psbt])
def get_total_psbt_fee(psbt_data: dict) -> float:
@@ -47,10 +47,10 @@ def get_total_psbt_fee(psbt_data: dict) -> float:
total_fee = total_fee_sats / 100000000
return total_fee
def sign_psbt(coin: str, psbt: str) -> str:
def sign_psbt(coin: Literal['btc', 'ltc'], psbt: str) -> str:
return cast(str, util.request_electrum_rpc(coin, 'signtransaction', [psbt]))
def broadcast_bitcoin_tx(coin: str, signed_tx: str):
def broadcast_electrum_tx(coin: Literal['btc', 'ltc'], signed_tx: str):
util.request_electrum_rpc(coin, 'broadcast', [signed_tx])
def get_monero_balance() -> float:
@@ -65,45 +65,48 @@ def sweep_all_monero(address: str) -> None:
util.request_monero_rpc('sweep_all', params)
def get_new_kraken_address(asset: Literal['XBT', 'XMR']) -> str:
payload = {
def get_new_kraken_address(asset: Literal['btc', 'ltc', 'xmr']) -> str:
sym_to_name = {
"btc": "Bitcoin",
"ltc": "Litecoin",
"xmr": "Monero"
}
payload: dict[str, str | bool] = {
'asset': asset,
'method': 'Bitcoin' if asset == 'XBT' else 'Monero'
'method': sym_to_name[asset]
}
result = util.kraken_request('/0/private/DepositAddresses', payload)
first_new_address = next((addr for addr in result if addr.get('new', False)), None)
if first_new_address:
return(first_new_address['address'])
return first_new_address['address']
else:
payload = {
'asset': asset,
'method': 'Bitcoin' if asset == 'XBT' else 'Monero',
'new': True
}
payload['new'] = True
result = util.kraken_request('/0/private/DepositAddresses', payload)
first_new_address = next((addr for addr in result if addr.get('new', False)), None)
return(first_new_address['address'])
if first_new_address:
return first_new_address['address']
raise Exception(f'Kraken did not return a new address: {json.dumps(result, indent=2)}')
def attempt_bitcoin_autoforward():
balance = get_bitcoin_balance(env.BTC_ELECTRUM_RPC_URL)
balance = get_electrum_balance('btc')
if balance < MIN_BITCOIN_SEND_AMOUNT:
print(util.get_time(), f'Not enough Bitcoin balance to autoforward. (Balance: {balance}, Min Send: {MIN_BITCOIN_SEND_AMOUNT})')
return
try:
fee_rate = get_bitcoin_fee_rate(env.BITCOIN_FEE_SOURCE, env.BITCOIN_FEE_RATE)
set_bitcoin_fee_rate(env.BTC_ELECTRUM_RPC_URL, fee_rate, dynamic=False)
fee_rate = get_fee_rate(env.BITCOIN_FEE_SOURCE, env.BITCOIN_FEE_RATE)
set_electrum_fee_rate('btc', fee_rate, dynamic=False)
except:
set_bitcoin_fee_rate(env.BTC_ELECTRUM_RPC_URL, fee_rate=0, dynamic=True)
address = get_new_kraken_address('XBT')
set_electrum_fee_rate('btc', rate=0, dynamic=True)
address = get_new_kraken_address('btc')
try:
psbt = create_psbt(env.BTC_ELECTRUM_RPC_URL, address)
psbt = create_psbt('btc', address)
except requests.exceptions.HTTPError as http_error:
response_json = cast(dict, http_error.response.json())
@@ -113,7 +116,7 @@ def attempt_bitcoin_autoforward():
raise http_error
psbt_data = get_psbt_data(env.BTC_ELECTRUM_RPC_URL, psbt)
psbt_data = get_psbt_data('btc', psbt)
total_fee = get_total_psbt_fee(psbt_data)
amount = balance
@@ -121,27 +124,27 @@ def attempt_bitcoin_autoforward():
print(util.get_time(), f'Not autoforwarding due to high transaction fee {total_fee} BTC.')
return
signed_tx = sign_psbt(psbt)
broadcast_bitcoin_tx(signed_tx)
signed_tx = sign_psbt('btc', psbt)
broadcast_electrum_tx('btc', signed_tx)
print(util.get_time(), f'Autoforwarded {amount} BTC to {address}!')
def attempt_litecoin_autoforward():
balance = get_bitcoin_balance(env.LTC_ELECTRUM_RPC_URL)
balance = get_electrum_balance('ltc')
if balance < MIN_LITECOIN_SEND_AMOUNT:
print(util.get_time(), f'Not enough Litecoin balance to autoforward. (Balance: {balance}, Min Send: {MIN_LITECOIN_SEND_AMOUNT})')
return
try:
fee_rate = get_bitcoin_fee_rate(env.LITECOIN_FEE_SOURCE, env.LITECOIN_FEE_RATE)
set_bitcoin_fee_rate(env.LTC_ELECTRUM_RPC_URL, fee_rate, dynamic=False)
fee_rate = get_fee_rate(env.LITECOIN_FEE_SOURCE, env.LITECOIN_FEE_RATE)
set_electrum_fee_rate('ltc', fee_rate, dynamic=False)
except:
set_bitcoin_fee_rate(env.LTC_ELECTRUM_RPC_URL, fee_rate=0, dynamic=True)
address = get_new_kraken_address('LTC')
set_electrum_fee_rate('ltc', rate=0, dynamic=True)
address = get_new_kraken_address('ltc')
try:
psbt = create_psbt(env.LTC_ELECTRUM_RPC_URL, address)
psbt = create_psbt('ltc', address)
except requests.exceptions.HTTPError as http_error:
response_json = cast(dict, http_error.response.json())
@@ -151,7 +154,7 @@ def attempt_litecoin_autoforward():
raise http_error
psbt_data = get_psbt_data(env.LTC_ELECTRUM_RPC_URL, psbt)
psbt_data = get_psbt_data('ltc', psbt)
total_fee = get_total_psbt_fee(psbt_data)
amount = balance
@@ -159,10 +162,10 @@ def attempt_litecoin_autoforward():
print(util.get_time(), f'Not autoforwarding due to high transaction fee {total_fee} LTC.')
return
signed_tx = sign_psbt(psbt)
broadcast_bitcoin_tx(signed_tx)
signed_tx = sign_psbt('ltc', psbt)
broadcast_electrum_tx('ltc', signed_tx)
print(util.get_time(), f'Autoforwarded {amount} BTC to {address}!')
print(util.get_time(), f'Autoforwarded {amount} LTC to {address}!')
def attempt_monero_autoforward():
balance = get_monero_balance()
@@ -171,7 +174,7 @@ def attempt_monero_autoforward():
print(util.get_time(), f'Not enough Monero balance to autoforward. (Balance: {balance}, Min Send: {MIN_MONERO_SEND_AMOUNT})')
return
address = get_new_kraken_address('XMR')
address = get_new_kraken_address('xmr')
sweep_all_monero(address)
print(util.get_time(), f'Autoforwarded {balance} XMR to {address}!')

View File

@@ -1,10 +1,15 @@
import os
BTC_ELECTRUM_RPC_URL = os.getenv('ELECTRUM_RPC_URL', 'http://electrs:7000')
LTC_ELECTRUM_RPC_URL = os.getenv('ELECTRUM_RPC_URL', 'http://electrs:7001')
ELECTRUM_RPC_USERNAME = os.getenv('ELECTRUM_RPC_USERNAME', '')
ELECTRUM_RPC_PASSWORD = os.getenv('ELECTRUM_RPC_PASSWORD', '')
BITCOIN_ELECTRUM_RPC_URL = os.getenv('BITCOIN_ELECTRUM_RPC_URL', 'http://electrs:7000')
LITECOIN_ELECTRUM_RPC_URL = os.getenv('LITECOIN_ELECTRUM_RPC_URL', 'http://electrs:7001')
BITCOIN_ELECTRUM_RPC_USERNAME = os.getenv('BITCOIN_ELECTRUM_RPC_USERNAME', '')
BITCOIN_ELECTRUM_RPC_PASSWORD = os.getenv('BITCOIN_ELECTRUM_RPC_PASSWORD', '')
LITECOIN_ELECTRUM_RPC_USERNAME = os.getenv('LITECOIN_ELECTRUM_RPC_USERNAME', '')
LITECOIN_ELECTRUM_RPC_PASSWORD = os.getenv('LITECOIN_ELECTRUM_RPC_PASSWORD', '')
BITCOIN_ELECTRUM_SERVER_ADDRESS = os.getenv('BITCOIN_ELECTRUM_SERVER_ADDRESS', '')
LITECOIN_ELECTRUM_SERVER_ADDRESS = os.getenv('LITECOIN_ELECTRUM_SERVER_ADDRESS', '')
BITCOIN_WALLET_SEED = os.getenv('BITCOIN_WALLET_SEED', '')
LITECOIN_WALLET_SEED = os.getenv('LITECOIN_WALLET_SEED', '')
MONERO_RPC_URL = os.getenv('MONERO_RPC_URL', 'http://monero-wallet-rpc:18082/json_rpc')
MONERO_RPC_USERNAME = os.getenv('MONERO_RPC_USERNAME', '')

View File

@@ -1,22 +1,23 @@
from typing import Literal
from bip_utils import Bip39SeedGenerator, Bip84, Bip84Coins
import traceback
import util
import env
def get_zprv_from_seed(mnemonic: str, coin: str) -> str:
def get_zprv_from_seed(coin: Literal['btc', 'ltc'], mnemonic: str ) -> str:
coin_type = Bip84Coins.BITCOIN if coin == 'btc' else Bip84Coins.LITECOIN
seed_bytes = Bip39SeedGenerator(mnemonic).Generate()
bip84_master_key = Bip84.FromSeed(seed_bytes, coin)
bip84_master_key = Bip84.FromSeed(seed_bytes, coin_type)
zprv = bip84_master_key.Purpose().Coin().Account(0).PrivateKey().ToExtended()
return zprv
def import_bitcoin_seed():
zprv = get_zprv_from_seed(env.BITCOIN_WALLET_SEED, Bip84Coins.BITCOIN)
util.request_electrum_rpc(env.BTC_ELECTRUM_RPC_URL, 'restore', [zprv])
zprv = get_zprv_from_seed('btc', env.BITCOIN_WALLET_SEED)
util.request_electrum_rpc('btc', 'restore', [zprv])
def import_litecoin_seed():
zprv = get_zprv_from_seed(env.BITCOIN_WALLET_SEED, Bip84Coins.LITECOIN)
util.request_electrum_rpc(env.LTC_ELECTRUM_RPC_URL, 'restore', [zprv])
zprv = get_zprv_from_seed('ltc', env.LITECOIN_WALLET_SEED)
util.request_electrum_rpc('ltc', 'restore', [zprv])
def import_monero_seed():
params = {
@@ -34,8 +35,8 @@ util.wait_for_rpc()
try:
import_bitcoin_seed()
util.request_btc_electrum_rpc(env.BTC_ELECTRUM_RPC_URL, 'load_wallet')
util.request_btc_electrum_rpc(env.BTC_ELECTRUM_RPC_URL, 'changegaplimit', [1000, 'iknowhatimdoing'])
util.request_electrum_rpc('btc', 'load_wallet')
util.request_electrum_rpc('btc', 'changegaplimit', [1000, 'iknowhatimdoing'])
print('Bitcoin seed has successfully been imported!')
except Exception as e:
print(util.get_time(), 'Error importing bitcoin seed:')
@@ -43,8 +44,8 @@ except Exception as e:
try:
import_litecoin_seed()
util.request_electrum_rpc(env.LTC_ELECTRUM_RPC_URL, 'load_wallet')
util.request_electrum_rpc(env.LTC_ELECTRUM_RPC_URL, 'changegaplimit', [1000, 'iknowhatimdoing'])
util.request_electrum_rpc('ltc', 'load_wallet')
util.request_electrum_rpc('ltc', 'changegaplimit', [1000, 'iknowhatimdoing'])
print('Litecoin seed has successfully been imported!')
except Exception as e:
print(util.get_time(), 'Error importing litecoin seed:')

View File

@@ -1,4 +1,5 @@
import ast
from typing import Literal
from requests.auth import HTTPDigestAuth
from datetime import datetime
import urllib.parse
@@ -14,9 +15,14 @@ import env
def get_time() -> str:
return f'[{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}]'
def request_electrum_rpc(rpc_url: str, method: str, params: list | dict = []):
def request_electrum_rpc(coin: Literal['btc', 'ltc'], method: str, params: list | dict = []):
headers = {'content-type': 'application/json'}
if coin == 'btc':
auth = (env.BITCOIN_ELECTRUM_RPC_USERNAME, env.BITCOIN_ELECTRUM_RPC_PASSWORD)
else:
auth = (env.LITECOIN_ELECTRUM_RPC_USERNAME, env.LITECOIN_ELECTRUM_RPC_PASSWORD)
data = {
'jsonrpc': '2.0',
'id': 'curltext',
@@ -25,10 +31,10 @@ def request_electrum_rpc(rpc_url: str, method: str, params: list | dict = []):
}
response = requests.post(
rpc_url,
env.BITCOIN_ELECTRUM_RPC_URL if coin == 'btc' else env.LITECOIN_ELECTRUM_RPC_URL,
headers=headers,
data=json.dumps(data),
auth=(env.ELECTRUM_RPC_USERNAME, env.ELECTRUM_RPC_PASSWORD)
auth=auth
)
response_json = response.json()
@@ -67,10 +73,10 @@ def request_monero_rpc(method: str, params: dict = {}):
return response_json['result']
def open_bitcoin_wallet():
request_electrum_rpc(env.BTC_ELECTRUM_RPC_URL, 'load_wallet')
request_electrum_rpc('btc', 'load_wallet')
def open_litecoin_wallet():
request_electrum_rpc(env.LTC_ELECTRUM_RPC_URL, 'load_wallet')
request_electrum_rpc('ltc', 'load_wallet')
def open_monero_wallet() -> None:
params = {'filename': 'foo', 'password': env.MONERO_WALLET_PASSWORD}
@@ -81,7 +87,7 @@ def wait_for_rpc():
while 1:
try:
request_electrum_rpc(env.BTC_ELECTRUM_RPC_URL, 'getinfo')
request_electrum_rpc('btc', 'getinfo')
break
except:
time.sleep(10)
@@ -90,7 +96,7 @@ def wait_for_rpc():
while 1:
try:
request_electrum_rpc(env.LTC_ELECTRUM_RPC_URL, 'getinfo')
request_electrum_rpc('ltc', 'getinfo')
break
except:
time.sleep(10)