mirror of
https://github.com/curly60e/pyblock.git
synced 2026-08-13 12:33:15 +02:00
893aabc85d
- Replace all shell=True subprocess calls with Python-native processing (nodeconnection.py, SPV/nodeconnection.py, SPV/ppi.py) - Mask sensitive inputs (private keys, passwords, tokens) with getpass - Add threading.Lock to block_explorer.py shared state - Use json.loads() instead of fragile string splitting in apisnd.py - Add path validation before file open in apisnd.py - Replace random.randint with secrets.randbelow for mining nonces - Fix destructive exception handlers in clone.py and feed.py - Replace bare except clauses with specific exceptions + logging - Remove unused imports (psutil, xmltodict, block_visualizer, base64, say) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1486 lines
53 KiB
Python
1486 lines
53 KiB
Python
#Developer: Curly60e
|
||
#Tester: __B__T__C__
|
||
#ℙ𝕪𝔹𝕃𝕆ℂ𝕂 𝕚𝕥𝕤 𝕒 𝔹𝕚𝕥𝕔𝕠𝕚𝕟 𝔻𝕒𝕤𝕙𝕓𝕠𝕒𝕣𝕕 𝕨𝕚𝕥𝕙 ℂ𝕪𝕡𝕙𝕖𝕣𝕡𝕦𝕟𝕜 𝕒𝕖𝕤𝕥𝕙𝕖𝕥𝕚𝕔.
|
||
|
||
|
||
import codecs, requests
|
||
import logging
|
||
import shlex
|
||
import subprocess
|
||
import os
|
||
import os.path
|
||
import qrcode
|
||
import sys
|
||
try:
|
||
import simplejson as json
|
||
except ImportError:
|
||
import json
|
||
import time as t
|
||
import numpy as np
|
||
from cfonts import render
|
||
from pblogo import blogo
|
||
from PIL import Image
|
||
from robohash import Robohash
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
lndconnectload = {"ip_port":"", "tls":"", "macaroon":"", "ln":""}
|
||
settingsClock = {"gradient":"", "design":"", "colorA":"", "colorB":""}
|
||
|
||
|
||
def _load_lnd_config():
|
||
"""Load LND connection configuration."""
|
||
with open("config/blndconnect.conf", "r") as f:
|
||
return json.load(f)
|
||
|
||
|
||
def _run_ln(*args):
|
||
"""Run lightning CLI safely."""
|
||
# nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit
|
||
return subprocess.run(
|
||
[lndconnectload['ln']] + list(args),
|
||
capture_output=True, text=True
|
||
)
|
||
|
||
|
||
def clear(): # clear the screen
|
||
subprocess.run(['clear'] if os.name != 'nt' else ['cls'], shell=(os.name == 'nt'))
|
||
def closed():
|
||
print("<<< Back Control + C.\n\n")
|
||
|
||
#-------------------------RPC BITCOIN NODE CONNECTION
|
||
|
||
def rpc(method, params=None):
|
||
if params is None:
|
||
params = []
|
||
payload = json.dumps({
|
||
"jsonrpc": "2.0",
|
||
"id": "minebet",
|
||
"method": method,
|
||
"params": params
|
||
})
|
||
path = {"ip_port":"", "rpcuser":"", "rpcpass":"", "bitcoincli":""}
|
||
if os.path.isfile('bclock.conf'): # Check if the file 'bclock.conf' is in the same folder
|
||
with open("bclock.conf", "r") as f:
|
||
pathv = json.load(f) # Load the file 'bclock.conf'
|
||
path = pathv # Copy the variable pathv to 'path'
|
||
return requests.post(path['ip_port'], auth=(path['rpcuser'], path['rpcpass']), data=payload, timeout=10).json()['result']
|
||
|
||
def remoteHalving():
|
||
|
||
b = rpc('getblockcount')
|
||
c = str(b)
|
||
oneh = 0 - int(c) + 210000
|
||
twoh = 210000 - int(c) + 210000
|
||
thrh = 420000 - int(c) + 210000
|
||
forh = 630000 - int(c) + 210000
|
||
fifh = 840000 - int(c) + 210000
|
||
sixh = 1050000 - int(c) + 210000
|
||
sevh = 1260000 - int(c) + 210000
|
||
eith = 1470000 - int(c) + 210000
|
||
ninh = 1680000 - int(c) + 210000
|
||
tenh = 1890000 - int(c) + 210000
|
||
|
||
q = """
|
||
\033[0;37;40m------------------- HALVING HISTORY -------------------
|
||
|
||
1st Halving: in {} Blocks {}
|
||
2nd Halving: in {} Blocks {}
|
||
3rd Halving: in {} Blocks {}
|
||
4th Halving: in {} Blocks {}
|
||
5th Halving: in {} Blocks {}
|
||
6th Halving: in {} Blocks {}
|
||
7th Halving: in {} Blocks {}
|
||
8th Halving: in {} Blocks {}
|
||
9th Halving: in {} Blocks {}
|
||
10th Halving: in {} Blocks {}
|
||
|
||
-------------------------------------------------------
|
||
""".format("0" if int(c) == 210000 else oneh,"\033[1;32;40mCOMPLETE\033[0;37;40m","0" if int(c) == 420000 else twoh,"\033[1;32;40mCOMPLETE\033[0;37;40m", "0" if int(c) == 630000 else thrh,"\033[1;32;40mCOMPLETE\033[0;37;40m","0" if int(c) == 840000 else forh,"\033[1;32;40mCOMPLETE\033[0;37;40m" if int(c) >= 840000 else "\033[1;35;40mPENDING\033[0;37;40m", "0" if int(c) >= 1050000 else fifh , "\033[1;32;40mCOMPLETE\033[0;37;40m" if int(c) >= 1050000 else "\033[1;35;40mPENDING\033[0;37;40m", sixh, "\033[1;32;40mCOMPLETE\033[0;37;40m" if int(c) >= 1260000 else "\033[1;35;40mPENDING\033[0;37;40m", sevh,"\033[1;32;40mCOMPLETE\033[0;37;40m" if int(c) >= 1470000 else "\033[1;35;40mPENDING\033[0;37;40m", eith,"\033[1;32;40mCOMPLETE\033[0;37;40m" if int(c) >= 1680000 else "\033[1;35;40mPENDING\033[0;37;40m", ninh, "\033[1;32;40mCOMPLETE\033[0;37;40m" if int(c) >= 1890000 else "\033[1;35;40mPENDING\033[0;37;40m", tenh, "\033[1;32;40mCOMPLETE\033[0;37;40m" if int(c) >= 1890000 else "\033[1;35;40mPENDING\033[0;37;40m")
|
||
print(q)
|
||
input("\nContinue...")
|
||
|
||
|
||
def remotegetblock():
|
||
if os.path.isfile('pyblocksettingsClock.conf') or os.path.isfile('pyblocksettingsClock.conf'): # Check if the file 'bclock.conf' is in the same folder
|
||
with open("pyblocksettingsClock.conf", "r") as f:
|
||
settingsv = json.load(f) # Load the file 'bclock.conf'
|
||
settingsClock = settingsv # Copy the variable pathv to 'path'
|
||
else:
|
||
settingsClock = {"gradient":"", "design":"block", "colorA":"green", "colorB":"yellow"}
|
||
with open("pyblocksettingsClock.conf", "w") as f:
|
||
json.dump(settingsClock, f, indent=2)
|
||
b = rpc('getblockcount')
|
||
c = str(b)
|
||
a = c
|
||
output = render(str(c), colors=[settingsClock['colorA'], settingsClock['colorB']], align='center')
|
||
print("\x1b[?25l" + output)
|
||
while True:
|
||
x = a
|
||
b = rpc('getblockcount')
|
||
c = str(b)
|
||
if c > a:
|
||
clear()
|
||
closed()
|
||
output = render(str(c), colors=[settingsClock['colorA'], settingsClock['colorB']], align='center')
|
||
print("\a\x1b[?25l" + output)
|
||
a = c
|
||
|
||
def remotegetblockcount(): # get access to bitcoin-cli with the command getblockcount
|
||
while True:
|
||
try:
|
||
a = rpc('getblockchaininfo')
|
||
d = a
|
||
clear()
|
||
blogo()
|
||
closed()
|
||
print("""
|
||
----------------------------------------------------------------------------
|
||
\tGET BLOCKCHAIN INFORMATION
|
||
Chain: {}
|
||
Blocks: {}
|
||
Best BlockHash: {}
|
||
Difficulty: {}
|
||
Verification Progress: {}
|
||
Size on Disk: {}
|
||
Pruned: {}
|
||
----------------------------------------------------------------------------
|
||
""".format(d['chain'], d['blocks'], d['bestblockhash'], d['difficulty'], d['verificationprogress'], d['size_on_disk'], d['pruned']))
|
||
t.sleep(2)
|
||
except Exception as e: # Catch specific exceptions
|
||
break
|
||
|
||
def remoteconsole(): # get into the console from bitcoin-cli
|
||
print("\t\033[0;37;40mThis is \033[1;33;40mBitcoin-cli's \033[0;37;40mconsole. Type your respective commands you want to display.\n\n")
|
||
while True:
|
||
cle = input("\033[1;32;40mconsole $>: \033[0;37;40m")
|
||
a = rpc(cle)
|
||
print(a)
|
||
|
||
def runthenumbersConn():
|
||
try:
|
||
b = rpc('gettxoutsetinfo')
|
||
c = str(b)
|
||
print(c)
|
||
input("\nContinue...")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
#-------------------------END RPC BITCOIN NODE CONNECTION
|
||
|
||
def consoleLN(): # get into the console from bitcoin-cli
|
||
lndconnectload = _load_lnd_config()
|
||
print("\t\033[0;37;40mThis is \033[1;33;40mLncli's \033[0;37;40mconsole. Type your respective commands you want to display.\n\n")
|
||
while True:
|
||
cle = input("\033[1;32;40mconsole $>: \033[0;37;40m")
|
||
lsd = _run_ln(*shlex.split(cle))
|
||
lsd1 = str(lsd.stdout)
|
||
print(lsd1)
|
||
|
||
def locallistpeersQQ():
|
||
lndconnectload = _load_lnd_config()
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
lncli = " listpeers"
|
||
while True:
|
||
clear()
|
||
print("\033[1;32;40m")
|
||
blogo()
|
||
print("\033[0;37;40m")
|
||
print("<<< Back to the Main Menu Press Control + C.\n\n")
|
||
lsd = _run_ln(*shlex.split(lncli)).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
n = d['peers']
|
||
try:
|
||
print("\n\tLIST PEERS\n")
|
||
for item_ in n:
|
||
s = item_
|
||
hash = s['pub_key']
|
||
rh = Robohash(hash)
|
||
rh.assemble(roboset='set1')
|
||
if not os.path.isfile(str(f'{hash}.png')):
|
||
with open(f'{hash}.png', "wb") as f:
|
||
rh.img.save(f, format="png")
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 1
|
||
w = int((img.width / img.height) * 5)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 1
|
||
w = int((img.width / img.height) * 5)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
for x in range(h):
|
||
for y in range(w):
|
||
pix = img_arr[x][y]
|
||
print(get_color(pix[0], pix[1], pix[2]), sep='', end='')
|
||
print()
|
||
print("PubKey: " + s['pub_key'] + " @" + s['address'])
|
||
|
||
nd = input("\nSelect PubKey: ")
|
||
for item in n:
|
||
s = item
|
||
nn = s['pub_key']
|
||
if nd == nn:
|
||
hash = s['pub_key']
|
||
rh = Robohash(hash)
|
||
rh.assemble(roboset='set1')
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 20
|
||
w = int((img.width / img.height) * 50)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
for x in range(h):
|
||
for y in range(w):
|
||
pix = img_arr[x][y]
|
||
print(get_color(pix[0], pix[1], pix[2]), sep='', end='')
|
||
print()
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("""
|
||
PEER DECODED\n
|
||
Bytes Sent: {}
|
||
Bytes Recv: {}
|
||
Sat Sent: {} sats
|
||
Sat Recv: {} sats
|
||
""".format(s['bytes_sent'], s['bytes_recv'], s['sat_sent'], s['sat_recv']))
|
||
print("-----------------------------------------------------------------------------------------------------\n")
|
||
print("\n\tPeer: " + nd)
|
||
print("\033[1;30;47m")
|
||
qr.add_data(s['pub_key'])
|
||
qr.print_ascii()
|
||
print("\033[0;37;40m")
|
||
qr.clear()
|
||
|
||
pp = input("\nDo you want to disconnect? Y/n: ")
|
||
if pp in ["Y", "y"]:
|
||
lsd = _run_ln("disconnect", nd).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
print("\n\tDisconnected from peer " + nd)
|
||
input("\nContinue... ")
|
||
elif pp in ["N", "n"]:
|
||
input("\nContinue... ")
|
||
except Exception as e: # Catch specific exceptions
|
||
break
|
||
|
||
def localconnectpeer():
|
||
lndconnectload = _load_lnd_config()
|
||
try:
|
||
clear()
|
||
print("\033[1;32;40m")
|
||
blogo()
|
||
print("\033[0;37;40m")
|
||
print("<<< Back to the Main Menu Press Control + C.\n\n")
|
||
print("\n\tCONNECT TO NEW PEER\n")
|
||
a = input("Insert PeerID@IP:PORT: ")
|
||
lncli = " connect "
|
||
lsd = _run_ln(*shlex.split(lncli), a).stdout
|
||
lsd0 = str(lsd)
|
||
print(lsd0)
|
||
input("\nContinue... ")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def locallistchaintxns():
|
||
lndconnectload = _load_lnd_config()
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
lncli = " listchaintxns"
|
||
lsd = _run_ln(*shlex.split(lncli)).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
n = d['transactions']
|
||
while True:
|
||
clear()
|
||
print("\033[1;32;40m")
|
||
blogo()
|
||
print("\033[0;37;40m")
|
||
print("<<< Back to the Main Menu Press Control + C.\n\n")
|
||
print("\t\nTransactions\n")
|
||
try:
|
||
print("\n\tLIST ONCHAIN TRANSACTIONS\n")
|
||
for item_ in n:
|
||
s = item_
|
||
|
||
print("Transaction Hash: " + s['tx_hash'] + " " + s['amount'] + " sats")
|
||
nd = input("\nSelect RHash: ")
|
||
|
||
for item in n:
|
||
s = item
|
||
nn = s['tx_hash']
|
||
trx = s['dest_addresses']
|
||
if nd == nn:
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("""
|
||
\nONCHAIN TRANSACTION DECODED
|
||
Amount: {} sats
|
||
Tx Hash: {}
|
||
Block Hash: {}
|
||
Block Height: {}
|
||
Confirmations: {}
|
||
Destination: {}
|
||
""".format(s['amount'], s['tx_hash'], s['block_hash'], s['block_height'], s['num_confirmations'], trx))
|
||
print("-----------------------------------------------------------------------------------------------------\n")
|
||
print("\nTransaction Hash")
|
||
print("\033[1;30;47m")
|
||
qr.add_data(s['tx_hash'])
|
||
qr.print_ascii()
|
||
print("\033[0;37;40m")
|
||
qr.clear()
|
||
input("\nContinue... ")
|
||
except Exception as e: # Catch specific exceptions
|
||
break
|
||
|
||
def locallistinvoices():
|
||
lndconnectload = _load_lnd_config()
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
lncli = " listinvoices"
|
||
lsd = _run_ln(*shlex.split(lncli)).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
n = d['invoices']
|
||
while True:
|
||
clear()
|
||
print("\033[1;32;40m")
|
||
blogo()
|
||
print("\033[0;37;40m")
|
||
print("<<< Back to the Main Menu Press Control + C.\n\n")
|
||
print("\tInvoices\n")
|
||
try:
|
||
print("\n\tLIST INVOICES\n")
|
||
for item_ in n:
|
||
s = item_
|
||
|
||
print("Invoice: " + s['r_hash'] + " " + s['state'])
|
||
|
||
nd = input("\nSelect RHash: ")
|
||
|
||
for item in n:
|
||
s = item
|
||
nn = s['r_hash']
|
||
if nd == nn:
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("""
|
||
\nINVOICE DECODED
|
||
Memo: {}
|
||
Invoice: {}
|
||
Amount: {} sats
|
||
State: {}
|
||
""".format(s['memo'], s['payment_request'], s['amt_paid_sat'], s['state']))
|
||
print("----------------------------------------------------------------------------------------------------\n")
|
||
print("\033[1;30;47m")
|
||
qr.add_data(s['payment_request'])
|
||
qr.print_ascii()
|
||
print("\033[0;37;40m")
|
||
qr.clear()
|
||
input("\nContinue... ")
|
||
except Exception as e: # Catch specific exceptions
|
||
break
|
||
|
||
def locallistchannels():
|
||
lndconnectload = _load_lnd_config()
|
||
lncli = " listchannels"
|
||
lsd = _run_ln(*shlex.split(lncli)).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
n = d['channels']
|
||
while True:
|
||
clear()
|
||
print("\033[1;32;40m")
|
||
blogo()
|
||
print("\033[0;37;40m")
|
||
print("<<< Back to the Main Menu Press Control + C.\n\n")
|
||
print("\t\nChannels\n")
|
||
try:
|
||
print("\n\tLIST CHANNELS\n")
|
||
for item_ in n:
|
||
s = item_
|
||
hash = s['remote_pubkey']
|
||
rh = Robohash(hash)
|
||
rh.assemble(roboset='set1')
|
||
if not os.path.isfile(str(f'{hash}.png')):
|
||
with open(f'{hash}.png', "wb") as f:
|
||
rh.img.save(f, format="png")
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 1
|
||
w = int((img.width / img.height) * 5)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 1
|
||
w = int((img.width / img.height) * 5)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
for x in range(h):
|
||
for y in range(w):
|
||
pix = img_arr[x][y]
|
||
print(get_color(pix[0], pix[1], pix[2]), sep='', end='')
|
||
print()
|
||
print("Node ID: " + s['remote_pubkey'])
|
||
|
||
nd = input("\nSelect a Node ID: ")
|
||
for item in n:
|
||
s = item
|
||
nn = s['remote_pubkey']
|
||
if nd == nn:
|
||
hash = s['remote_pubkey']
|
||
rh = Robohash(hash)
|
||
rh.assemble(roboset='set1')
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 20
|
||
w = int((img.width / img.height) * 50)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
for x in range(h):
|
||
for y in range(w):
|
||
pix = img_arr[x][y]
|
||
print(get_color(pix[0], pix[1], pix[2]), sep='', end='')
|
||
print()
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("""
|
||
\tCHANNEL DECODED
|
||
Active: {}
|
||
Node ID: {}
|
||
Channel Point: {}
|
||
Channel Capacity: {} sats
|
||
Local Balance: {} sats
|
||
Remote Balance: {} sats
|
||
Total Sent: {} sats
|
||
Total Received: {} sats
|
||
""".format(s['active'], s['remote_pubkey'], s['channel_point'], s['capacity'], s['local_balance'], s['remote_balance'], s['total_satoshis_sent'], s['total_satoshis_received']))
|
||
print("----------------------------------------------------------------------------------------------------\n")
|
||
|
||
input("\nContinue... ")
|
||
except Exception as e: # Catch specific exceptions
|
||
break
|
||
|
||
def localgetinfo():
|
||
lndconnectload = _load_lnd_config()
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
lncli = " getinfo"
|
||
lsd = _run_ln(*shlex.split(lncli)).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
hash = d['identity_pubkey']
|
||
rh = Robohash(hash)
|
||
rh.assemble(roboset='set1')
|
||
if not os.path.isfile(str(f'{hash}.png')):
|
||
with open(f'{hash}.png', "wb") as f:
|
||
rh.img.save(f, format="png")
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 20
|
||
w = int((img.width / img.height) * 50)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 20
|
||
w = int((img.width / img.height) * 50)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
for x in range(h):
|
||
for y in range(w):
|
||
pix = img_arr[x][y]
|
||
print(get_color(pix[0], pix[1], pix[2]), sep='', end='')
|
||
print()
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("""
|
||
\tNODE INFORMATION
|
||
|
||
Version: {}
|
||
Node ID: {}
|
||
Alias: {}
|
||
Color: {}
|
||
Pending Channels: {}
|
||
Active Channels: {}
|
||
Inactive Channels: {}
|
||
Peers: {}
|
||
URLS: {}
|
||
""".format(d['version'], d['identity_pubkey'], d['alias'], d['color'], d['num_pending_channels'], d['num_active_channels'], d['num_inactive_channels'], d['num_peers'], d['uris']))
|
||
print("\033[1;30;47m")
|
||
qr.add_data(d['identity_pubkey'])
|
||
qr.print_ascii()
|
||
print("\033[0;37;40m")
|
||
qr.clear()
|
||
print("----------------------------------------------------------------------------------------------------\n")
|
||
input("\nContinue... ")
|
||
|
||
def localaddinvoice():
|
||
lndconnectload = _load_lnd_config()
|
||
lncli = " addinvoice"
|
||
lsd = _run_ln(*shlex.split(lncli)).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
try:
|
||
amount = input("Amount in sats: ")
|
||
mem = input("Memo: ")
|
||
memo = mem.replace(" ","_")
|
||
lsd = _run_ln(*shlex.split(lncli), "--memo", "{}-PyBLOCK".format(memo), "--amt", amount).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
print("\033[1;30;47m")
|
||
qr.add_data(d['payment_request'])
|
||
qr.print_ascii()
|
||
print("\033[0;37;40m")
|
||
qr.clear()
|
||
print("Lightning Invoice: " + d['payment_request'])
|
||
b = str(d['payment_request'])
|
||
while True:
|
||
lsd = _run_ln("decodepayreq", b).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
r = d['payment_hash']
|
||
lsdn = _run_ln("lookupinvoice", r).stdout
|
||
lsdn0 = str(lsdn)
|
||
n = json.loads(lsdn0)
|
||
if n['state'] == 'SETTLED':
|
||
print("\033[1;32;40m")
|
||
clear()
|
||
blogo()
|
||
tick()
|
||
print("\033[0;37;40m")
|
||
t.sleep(2)
|
||
break
|
||
elif n['state'] == 'CANCELED':
|
||
print("\033[1;31;40m")
|
||
clear()
|
||
blogo()
|
||
canceled()
|
||
print("\033[0;37;40m")
|
||
t.sleep(2)
|
||
break
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def localpayinvoice():
|
||
lndconnectload = _load_lnd_config()
|
||
try:
|
||
invoiceN = input("Insert the invoice to pay: ")
|
||
invoice = invoiceN.lower()
|
||
lncli = " payinvoice "
|
||
lsd = _run_ln("decodepayreq", invoice).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
if d['num_satoshis'] == "0":
|
||
amt = " --amt "
|
||
amount = input("Amount in satoshis: ")
|
||
_run_ln(*shlex.split(lncli), invoice, *shlex.split(amt), amount)
|
||
else:
|
||
_run_ln(*shlex.split(lncli), invoice)
|
||
t.sleep(2)
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def localgetnetworkinfo():
|
||
lndconnectload = _load_lnd_config()
|
||
lncli = " getnetworkinfo"
|
||
lsd = _run_ln(*shlex.split(lncli)).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("""
|
||
\tLIGHTNING NETWORK INFORMATION
|
||
Numbers of Nodes: {}
|
||
Numbers of Channels: {}
|
||
Total Network Capacity: {} sats
|
||
Average Channel Size: {}
|
||
Minimum Channel Size: {}
|
||
Maximum Channel Size: {}
|
||
Median Channel Size: {} sats
|
||
Zombie channels: {}
|
||
""".format(d['num_nodes'], d['num_channels'], d['total_network_capacity'], d['avg_channel_size'], d['min_channel_size'], d['max_channel_size'], d['median_channel_size_sat'], d['num_zombie_chans']))
|
||
print("----------------------------------------------------------------------------------------------------\n")
|
||
input("\nContinue... ")
|
||
|
||
def _process_lncli_output(command, grep_pattern, sed_from, sed_to):
|
||
"""Run an lncli command and process output in pure Python.
|
||
|
||
Replaces shell pipe chains (grep | tr | sed | xxd -r -p) with safe
|
||
Python-native equivalents. No shell=True is used.
|
||
|
||
Args:
|
||
command: lncli sub-command, e.g. "listinvoices" or "listpayments".
|
||
grep_pattern: string to filter lines on (equivalent to grep).
|
||
sed_from: string to replace in matching lines (equivalent to sed 's/…').
|
||
sed_to: replacement hex string (equivalent to sed '…/…/g').
|
||
|
||
Returns:
|
||
Decoded text produced by the pipeline.
|
||
"""
|
||
result = subprocess.run(
|
||
["lncli", command], capture_output=True, text=True
|
||
)
|
||
lines = result.stdout.splitlines()
|
||
filtered = [line for line in lines if grep_pattern in line]
|
||
processed = []
|
||
for line in filtered:
|
||
# tr -d '"' | tr -d ','
|
||
line = line.replace('"', '').replace(',', '')
|
||
# sed replacement
|
||
line = line.replace(sed_from, sed_to)
|
||
# strip whitespace (html2text-like cleanup) then hex-decode
|
||
line = line.strip()
|
||
try:
|
||
decoded = bytes.fromhex(line).decode('utf-8', errors='replace')
|
||
processed.append(decoded)
|
||
except ValueError:
|
||
# If the line isn't valid hex after processing, keep it as-is
|
||
processed.append(line)
|
||
return "\n".join(processed)
|
||
|
||
|
||
def localFullProtocol():
|
||
lndconnectload = _load_lnd_config()
|
||
|
||
p1 = _process_lncli_output("listinvoices", "34349334", "34349334",
|
||
"0a0a2d5079424c4f434b204d6573736167652052656365697665643a200a")
|
||
p2 = _process_lncli_output("listinvoices", "7629171", "7629171",
|
||
"0a0a2d5079424c4f434b204d6573736167652052656365697665643a200a")
|
||
p3 = _process_lncli_output("listinvoices", "34343434", "34343434",
|
||
"0a0a2d5079424c4f434b204d6573736167652052656365697665643a200a")
|
||
|
||
p1 = _process_lncli_output("listpayments", "34349334", "34349334",
|
||
"0a0a202d5079424c4f434b204d6573736167653a200a")
|
||
p2 = _process_lncli_output("listpayments", "7629171", "7629171",
|
||
"0a0a202d5079424c4f434b204d6573736167653a200a")
|
||
p3 = _process_lncli_output("listpayments", "34343434", "34343434",
|
||
"0a0a202d5079424c4f434b204d6573736167653a200a")
|
||
|
||
|
||
|
||
def localkeysend():
|
||
lndconnectload = _load_lnd_config()
|
||
try:
|
||
closed()
|
||
print("\n\tYou are going to send a payment using KeySend - Note: You don't need any invoice, just your peer ID.\n")
|
||
lncli = " sendpayment "
|
||
node = input("Send to NodeID: ")
|
||
amount = input("Amount in sats: ")
|
||
while True:
|
||
if amount in ["", "0"]:
|
||
amount = input("\nAmount in sats: ")
|
||
else:
|
||
break
|
||
subprocess.run(
|
||
["lncli", "sendpayment", "--keysend", f"--d={node}", f"--amt={amount}",
|
||
"--final_cltv_delta=40"]
|
||
)
|
||
|
||
input("\nContinue...")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def localchatsendA():
|
||
lndconnectload = _load_lnd_config()
|
||
try:
|
||
closed()
|
||
print("\n\tWrite.\n")
|
||
lncli = " sendpayment "
|
||
node = input("Send to NodeID: ")
|
||
amount = input("Amount in sats: ")
|
||
message = input("Message: ")
|
||
encoded_message = message.encode('utf-8')
|
||
hex_encoded_message = encoded_message.hex()
|
||
input("\nContinue...")
|
||
|
||
while True:
|
||
if amount in ["", "0"]:
|
||
amount = input("\nAmount in sats: ")
|
||
else:
|
||
break
|
||
subprocess.run(
|
||
["lncli", "sendpayment", "--keysend", f"--d={node}", f"--amt={amount}",
|
||
"--data", "34349334=" + hex_encoded_message]
|
||
)
|
||
|
||
input("\nContinue...")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def localchatnewA():
|
||
lndconnectload = _load_lnd_config()
|
||
try:
|
||
closed()
|
||
print("\n\tRead.\n")
|
||
print(_process_lncli_output("listinvoices", "34349334", "34349334",
|
||
"0a0a2d5079424c4f434b204d6573736167652052656365697665643a200a"))
|
||
input("\nContinue...")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def localchatlistA():
|
||
lndconnectload = _load_lnd_config()
|
||
try:
|
||
closed()
|
||
print("\n\tList.\n")
|
||
print(_process_lncli_output("listpayments", "34349334", "34349334",
|
||
"0a0a202d5079424c4f434b204d6573736167653a200a"))
|
||
input("\nContinue...")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def localchatsendB():
|
||
lndconnectload = _load_lnd_config()
|
||
try:
|
||
closed()
|
||
print("\n\tWrite.\n")
|
||
lncli = " sendpayment "
|
||
node = input("Send to NodeID: ")
|
||
amount = input("Amount in sats: ")
|
||
message = input("Message: ")
|
||
encoded_message = message.encode('utf-8')
|
||
hex_encoded_message = encoded_message.hex()
|
||
print(encoded_message.hex())
|
||
input("\nContinue...")
|
||
|
||
while True:
|
||
if amount in ["", "0"]:
|
||
amount = input("\nAmount in sats: ")
|
||
else:
|
||
break
|
||
subprocess.run(
|
||
["lncli", "sendpayment", "--keysend", f"--d={node}", f"--amt={amount}",
|
||
"--data", "7629171=" + hex_encoded_message]
|
||
)
|
||
|
||
input("\nContinue...")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def localchatnewB():
|
||
lndconnectload = _load_lnd_config()
|
||
try:
|
||
closed()
|
||
print("\n\tRead.\n")
|
||
print(_process_lncli_output("listinvoices", "7629171", "7629171",
|
||
"0a0a2d5079424c4f434b204d6573736167652052656365697665643a200a"))
|
||
input("\nContinue...")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def localchatlistB():
|
||
lndconnectload = _load_lnd_config()
|
||
try:
|
||
closed()
|
||
print("\n\tList.\n")
|
||
print(_process_lncli_output("listpayments", "7629171", "7629171",
|
||
"0a0a202d5079424c4f434b204d6573736167653a200a"))
|
||
input("\nContinue...")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def localchatsendC():
|
||
lndconnectload = _load_lnd_config()
|
||
try:
|
||
closed()
|
||
print("\n\tWrite.\n")
|
||
lncli = " sendpayment "
|
||
node = input("Send to NodeID: ")
|
||
amount = input("Amount in sats: ")
|
||
message = input("Message: ")
|
||
encoded_message = message.encode('utf-8')
|
||
hex_encoded_message = encoded_message.hex()
|
||
print(encoded_message.hex())
|
||
input("\nContinue...")
|
||
|
||
while True:
|
||
if amount in ["", "0"]:
|
||
amount = input("\nAmount in sats: ")
|
||
else:
|
||
break
|
||
subprocess.run(
|
||
["lncli", "sendpayment", "--keysend", f"--d={node}", f"--amt={amount}",
|
||
"--data", "34343434=" + hex_encoded_message]
|
||
)
|
||
|
||
input("\nContinue...")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def localchatnewC():
|
||
lndconnectload = _load_lnd_config()
|
||
try:
|
||
closed()
|
||
print("\n\tRead.\n")
|
||
print(_process_lncli_output("listinvoices", "34343434", "34343434",
|
||
"0a0a2d5079424c4f434b204d6573736167652052656365697665643a200a"))
|
||
input("\nContinue...")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def localchatlistC():
|
||
lndconnectload = _load_lnd_config()
|
||
try:
|
||
closed()
|
||
print("\n\tList.\n")
|
||
print(_process_lncli_output("listpayments", "34343434", "34343434",
|
||
"0a0a202d5079424c4f434b204d6573736167653a200a"))
|
||
input("\nContinue...")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def localchannelbalance():
|
||
lndconnectload = _load_lnd_config()
|
||
lncli = " channelbalance"
|
||
lsd = _run_ln(*shlex.split(lncli)).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
print("""
|
||
---------------------------------------------------------
|
||
|
||
\tLOCAL CHANNEL BALANCE
|
||
|
||
Balance: {} sats
|
||
Pending Channels: {} sats
|
||
|
||
---------------------------------------------------------
|
||
""".format(d['balance'], d['pending_open_balance']))
|
||
input("\nContinue... ")
|
||
|
||
def localnewaddress():
|
||
lndconnectload = _load_lnd_config()
|
||
lncli = " newaddress p2wkh"
|
||
lsd = _run_ln(*shlex.split(lncli)).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
print("\033[1;30;47m")
|
||
qr.add_data(d['address'])
|
||
qr.print_ascii()
|
||
print("\033[0;37;40m")
|
||
qr.clear()
|
||
print("Bitcoin Address: " + d['address'])
|
||
input("\nContinue... ")
|
||
|
||
def localbalanceOC():
|
||
lndconnectload = _load_lnd_config()
|
||
lncli = " walletbalance"
|
||
lsd = _run_ln(*shlex.split(lncli)).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("\n\tLOCAL ONCHAIN BALANCE\n")
|
||
print("Total Balance: " + d['total_balance'] + " sats")
|
||
print("Confirmed Balance: " + d['confirmed_balance'] + " sats")
|
||
print("Unconfirmed Balance: " + d['unconfirmed_balance'] + " sats")
|
||
print("----------------------------------------------------------------------------------------------------\n")
|
||
input("\nContinue... ")
|
||
|
||
|
||
def localrebalancelnd():
|
||
lndconnectload = _load_lnd_config()
|
||
lncli = " listchannels"
|
||
while True:
|
||
lsd = _run_ln(*shlex.split(lncli)).stdout
|
||
lsd0 = str(lsd)
|
||
d = json.loads(lsd0)
|
||
n = d['channels']
|
||
clear()
|
||
print("\033[1;32;40m")
|
||
blogo()
|
||
print("\033[0;37;40m")
|
||
print("<<< Back to the Main Menu Press Control + C.\n\n")
|
||
print("\t\nChannels\n")
|
||
try:
|
||
print("""\n\tLIST CHANNELS TO REBALANCE\n
|
||
\t\033[1;32;40mLOCAL\033[0;37;40m BALANCE \t\033[1;31;40mREMOTE\033[0;37;40m BALANCE
|
||
""")
|
||
|
||
for item in n:
|
||
s = item
|
||
if int(s['local_balance']) >= int(s['remote_balance']):
|
||
total = int(s['local_balance']) - int(s['remote_balance'])
|
||
elif int(s['local_balance']) <= int(s['remote_balance']):
|
||
total = int(s['remote_balance']) - int(s['local_balance'])
|
||
print("Node ID: " + str(s['chan_id']) + "\t\033[1;32;40m " + str(s['local_balance']) + "\033[0;37;40m sats \033[1;31;40m\t" + str(s['remote_balance']) + "\033[0;37;40m sats \033[3;33;40m" + "\tDIFFERENCE: {}\033[0;37;40m sats".format(str(total)) )
|
||
fromnode = input("\nSelect FROM a Node ID : ")
|
||
tonode = input("\nSelect TO a Node ID : ")
|
||
amt = input("\nAmount in sats: ")
|
||
fee = input("\nMax Fee factor in sats: ")
|
||
fromtonode = "python3 rebalance.py -f {} -t {} -a {} --max-fee-factor {}".format(fromnode,tonode,amt,fee)
|
||
subprocess.run(["python3", "rebalance.py", "-f", fromnode, "-t", tonode, "-a", amt, "--max-fee-factor", fee])
|
||
input("Continue...")
|
||
except Exception as e: # Catch specific exceptions
|
||
break
|
||
|
||
# Remote connection with rest -------------------------------------
|
||
|
||
def getnewinvoice():
|
||
lndconnectload = _load_lnd_config()
|
||
cert_path = lndconnectload["tls"]
|
||
with open(lndconnectload["macaroon"], 'rb') as f:
|
||
macaroon = codecs.encode(f.read(), 'hex')
|
||
headers = {'Grpc-Metadata-macaroon': macaroon}
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
try:
|
||
amount = input("Amount in sats: ")
|
||
memo = input("Memo: ")
|
||
url = 'https://{}/v1/invoices'.format(lndconnectload["ip_port"])
|
||
data = {
|
||
|
||
}
|
||
if amount == "":
|
||
r = requests.post(
|
||
url,
|
||
headers=headers,
|
||
verify=cert_path,
|
||
json={"memo": f'{memo} -PyBLOCK'},
|
||
timeout=10,
|
||
)
|
||
|
||
else:
|
||
r = requests.post(
|
||
url,
|
||
headers=headers,
|
||
verify=cert_path,
|
||
json={"value": amount, "memo": f'{memo} -PyBLOCK'},
|
||
timeout=10,
|
||
)
|
||
|
||
|
||
a = r.json()
|
||
print("\033[1;30;47m")
|
||
qr.add_data(a['payment_request'])
|
||
qr.print_ascii()
|
||
print("\033[0;37;40m")
|
||
qr.clear()
|
||
print("Lightning Invoice: " + a['payment_request'])
|
||
b = str(a['payment_request'])
|
||
while True:
|
||
url = 'https://{}/v1/payreq/{}'.format(lndconnectload["ip_port"], b)
|
||
r = requests.get(url, headers=headers, verify=cert_path, timeout=10)
|
||
a = r.json()
|
||
url = 'https://{}/v1/invoice/{}'.format(lndconnectload["ip_port"],a['payment_hash'])
|
||
rr = requests.get(url, headers=headers, verify=cert_path, timeout=10)
|
||
m = rr.json()
|
||
if m['state'] == 'SETTLED':
|
||
print("\033[1;32;40m")
|
||
clear()
|
||
blogo()
|
||
tick()
|
||
print("\033[0;37;40m")
|
||
t.sleep(2)
|
||
break
|
||
elif m['state'] == 'CANCELED':
|
||
print("\033[1;31;40m")
|
||
clear()
|
||
blogo()
|
||
canceled()
|
||
print("\033[0;37;40m")
|
||
t.sleep(2)
|
||
break
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def payinvoice():
|
||
lndconnectload = _load_lnd_config()
|
||
cert_path = lndconnectload["tls"]
|
||
with open(lndconnectload["macaroon"], 'rb') as f:
|
||
macaroon = codecs.encode(f.read(), 'hex')
|
||
headers = {'Grpc-Metadata-macaroon': macaroon}
|
||
try:
|
||
while True:
|
||
bolt11N = input("Insert the invoice to pay: ")
|
||
url = 'https://{}/v1/payreq/{}'.format(lndconnectload["ip_port"],bolt11N)
|
||
r = requests.get(url, headers=headers, verify=cert_path, timeout=10)
|
||
s = r.json()
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("""
|
||
\tINVOICE DECODED
|
||
Destination: {}
|
||
Payment Hash: {}
|
||
Amount: {} sats
|
||
Description: {}
|
||
""".format(s['destination'], s['payment_hash'], s['num_satoshis'], s['description']))
|
||
print("----------------------------------------------------------------------------------------------------\n")
|
||
print("<<< Cancel Control + C")
|
||
input("\nEnter to Continue... ")
|
||
bolt11 = bolt11N.lower()
|
||
r = requests.post(
|
||
url='https://{}/v1/channels/transactions'.format(lndconnectload["ip_port"]), headers=headers, verify=cert_path, json={"payment_request": bolt11}, timeout=10
|
||
)
|
||
try:
|
||
r.json()['error']
|
||
print("\nThe Invoice don't have an amount. Please insert an Invoice with amount. \n")
|
||
continue
|
||
except Exception as e: # Catch specific exceptions
|
||
break
|
||
ok, checking_id, fee_msat, error_message = r.ok, None, 0, None
|
||
r = requests.get(url='https://{}/v1/payreq/{}'.format(lndconnectload["ip_port"],bolt11), headers=headers, verify=cert_path, timeout=10)
|
||
t.sleep(5)
|
||
if r.ok:
|
||
checking_id = r.json()["payment_hash"]
|
||
print("\033[1;32;40m")
|
||
clear()
|
||
blogo()
|
||
tick()
|
||
else:
|
||
error_message = r.json()["error"]
|
||
print("\033[1;31;40m")
|
||
clear()
|
||
blogo()
|
||
canceled()
|
||
print("\033[0;37;40m")
|
||
t.sleep(2)
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def getnewaddress():
|
||
lndconnectload = _load_lnd_config()
|
||
cert_path = lndconnectload["tls"]
|
||
with open(lndconnectload["macaroon"], 'rb') as f:
|
||
macaroon = codecs.encode(f.read(), 'hex')
|
||
headers = {'Grpc-Metadata-macaroon': macaroon}
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
try:
|
||
url = 'https://{}/v1/newaddress'.format(lndconnectload["ip_port"])
|
||
r = requests.get(url, headers=headers, verify=cert_path, timeout=10)
|
||
addr = r.json()
|
||
print("\033[1;30;47m")
|
||
qr.add_data(addr['address'])
|
||
qr.print_ascii()
|
||
print("\033[0;37;40m")
|
||
print("Bitcoin Address: " + addr['address'])
|
||
qr.clear()
|
||
input("\nContinue... ")
|
||
except (KeyboardInterrupt, EOFError):
|
||
pass
|
||
except Exception as e:
|
||
logger.debug("nodeconnection: %s", e)
|
||
|
||
def listinvoice():
|
||
lndconnectload = _load_lnd_config()
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
cert_path = lndconnectload["tls"]
|
||
with open(lndconnectload["macaroon"], 'rb') as f:
|
||
macaroon = codecs.encode(f.read(), 'hex')
|
||
headers = {'Grpc-Metadata-macaroon': macaroon}
|
||
url = 'https://{}/v1/invoices'.format(lndconnectload["ip_port"])
|
||
r = requests.get(url, headers=headers, verify=cert_path, timeout=10)
|
||
a = r.json()
|
||
n = a['invoices']
|
||
while True:
|
||
clear()
|
||
print("\033[1;32;40m")
|
||
blogo()
|
||
print("\033[0;37;40m")
|
||
print("<<< Back to the Main Menu Press Control + C.\n\n")
|
||
print("\tInvoices\n")
|
||
try:
|
||
print("\n\tLIST INVOICES\n")
|
||
for r in range(len(n)):
|
||
s = n[r]
|
||
print("Invoice: " + s['r_hash'] + " " + s['state'])
|
||
|
||
nd = input("\nSelect RHash: ")
|
||
for item in n:
|
||
s = item
|
||
nn = s['r_hash']
|
||
if nd == nn:
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("""
|
||
\tINVOICE DECODED
|
||
Memo: {}
|
||
Invoice: {}
|
||
Amount: {} sats
|
||
State: {}
|
||
""".format(s['memo'], s['payment_request'], s['amt_paid_sat'], s['state']))
|
||
print("----------------------------------------------------------------------------------------------------\n")
|
||
print("\033[1;30;47m")
|
||
qr.add_data(s['payment_request'])
|
||
qr.print_ascii()
|
||
print("\033[0;37;40m")
|
||
qr.clear()
|
||
input("\nContinue... ")
|
||
except Exception as e: # Catch specific exceptions
|
||
break
|
||
input("\nContinue... ")
|
||
|
||
def getinfo():
|
||
lndconnectload = _load_lnd_config()
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
cert_path = lndconnectload["tls"]
|
||
with open(lndconnectload["macaroon"], 'rb') as f:
|
||
macaroon = codecs.encode(f.read(), 'hex')
|
||
headers = {'Grpc-Metadata-macaroon': macaroon}
|
||
url = 'https://{}/v1/getinfo'.format(lndconnectload["ip_port"])
|
||
r = requests.get(url, headers=headers, verify=cert_path, timeout=10)
|
||
a = r.json()
|
||
hash = a['identity_pubkey']
|
||
rh = Robohash(hash)
|
||
rh.assemble(roboset='set1')
|
||
if not os.path.isfile(str(f'{hash}.png')):
|
||
with open(f'{hash}.png', "wb") as f:
|
||
rh.img.save(f, format="png")
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 20
|
||
w = int((img.width / img.height) * 50)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 20
|
||
w = int((img.width / img.height) * 50)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
for x in range(h):
|
||
for y in range(w):
|
||
pix = img_arr[x][y]
|
||
print(get_color(pix[0], pix[1], pix[2]), sep='', end='')
|
||
print()
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("""
|
||
\t NODE INFORMATION
|
||
Version: {}
|
||
Node ID: {}
|
||
Alias: {}
|
||
Color: {}
|
||
Pending Channels: {}
|
||
Active Channels: {}
|
||
Inactive Channels: {}
|
||
Peers: {}
|
||
URLS: {}
|
||
""".format(a['version'], a['identity_pubkey'], a['alias'], a['color'], a['num_pending_channels'], a['num_active_channels'], a['num_inactive_channels'], a['num_peers'], a['uris']))
|
||
print("\033[1;30;47m")
|
||
qr.add_data(a['identity_pubkey'])
|
||
qr.print_ascii()
|
||
print("\033[0;37;40m")
|
||
qr.clear()
|
||
print("----------------------------------------------------------------------------------------------------\n")
|
||
input("\nContinue... ")
|
||
|
||
#--------------------------------- NYMs -----------------------------------
|
||
|
||
def get_ansi_color_code(r, g, b):
|
||
if r == g == b:
|
||
if r < 8:
|
||
return 16
|
||
if r > 248:
|
||
return 231
|
||
return round(((r - 8) / 247) * 24) + 232
|
||
return 16 + (36 * round(r / 255 * 5)) + (6 * round(g / 255 * 5)) + round(b / 255 * 5)
|
||
|
||
|
||
def get_color(r, g, b):
|
||
return "\x1b[48;5;{}m \x1b[0m".format(int(get_ansi_color_code(r,g,b)))
|
||
|
||
def channels():
|
||
lndconnectload = _load_lnd_config()
|
||
cert_path = lndconnectload["tls"]
|
||
with open(lndconnectload["macaroon"], 'rb') as f:
|
||
macaroon = codecs.encode(f.read(), 'hex')
|
||
headers = {'Grpc-Metadata-macaroon': macaroon}
|
||
url = 'https://{}/v1/channels'.format(lndconnectload["ip_port"])
|
||
r = requests.get(url, headers=headers, verify=cert_path, timeout=10)
|
||
a = r.json()
|
||
n = a['channels']
|
||
while True:
|
||
clear()
|
||
print("\033[1;32;40m")
|
||
blogo()
|
||
print("\033[0;37;40m")
|
||
print("<<< Back to the Main Menu Press Control + C.\n\n")
|
||
print("\t\nChannels\n")
|
||
try:
|
||
print("\n\tLIST CHANNELS\n")
|
||
for r in range(len(n)):
|
||
s = n[r]
|
||
hash = s['remote_pubkey']
|
||
rh = Robohash(hash)
|
||
rh.assemble(roboset='set1')
|
||
if not os.path.isfile(str(f'{hash}.png')):
|
||
with open(f'{hash}.png', "wb") as f:
|
||
rh.img.save(f, format="png")
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 1
|
||
w = int((img.width / img.height) * 5)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 1
|
||
w = int((img.width / img.height) * 5)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
for x in range(h):
|
||
for y in range(w):
|
||
pix = img_arr[x][y]
|
||
print(get_color(pix[0], pix[1], pix[2]), sep='', end='')
|
||
print()
|
||
print("Node ID: " + s['remote_pubkey'])
|
||
|
||
nd = input("\nSelect a Node ID: ")
|
||
for item in n:
|
||
s = item
|
||
nn = s['remote_pubkey']
|
||
if nd == nn:
|
||
hash = s['remote_pubkey']
|
||
rh = Robohash(hash)
|
||
rh.assemble(roboset='set1')
|
||
|
||
img = Image.open(f'{hash}.png')
|
||
|
||
h = 20
|
||
w = int((img.width / img.height) * 50)
|
||
|
||
img = img.resize((w,h), Image.ANTIALIAS)
|
||
img_arr = np.asarray(img)
|
||
h,w,c = img_arr.shape
|
||
|
||
for x in range(h):
|
||
for y in range(w):
|
||
pix = img_arr[x][y]
|
||
print(get_color(pix[0], pix[1], pix[2]), sep='', end='')
|
||
print()
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("""
|
||
\tCHANNEL DECODED
|
||
Active: {}
|
||
Node ID: {}
|
||
Channel Point: {}
|
||
Channel Capacity: {} sats
|
||
Local Balance: {} sats
|
||
Remote Balance: {} sats
|
||
Total Sent: {} sats
|
||
Total Received: {} sats
|
||
""".format(s['active'], s['remote_pubkey'], s['channel_point'], s['capacity'], s['local_balance'], s['remote_balance'], s['total_satoshis_sent'], s['total_satoshis_received']))
|
||
print("----------------------------------------------------------------------------------------------------\n")
|
||
|
||
input("\nContinue... ")
|
||
except Exception as e: # Catch specific exceptions
|
||
break
|
||
|
||
def channelbalance():
|
||
lndconnectload = _load_lnd_config()
|
||
cert_path = lndconnectload["tls"]
|
||
with open(lndconnectload["macaroon"], 'rb') as f:
|
||
macaroon = codecs.encode(f.read(), 'hex')
|
||
headers = {'Grpc-Metadata-macaroon': macaroon}
|
||
url = 'https://{}/v1/balance/channels'.format(lndconnectload["ip_port"])
|
||
r = requests.get(url, headers=headers, verify=cert_path, timeout=10)
|
||
a = r.json()
|
||
print("""
|
||
---------------------------------------------------------
|
||
|
||
\tLOCAL CHANNEL BALANCE
|
||
|
||
Balance: {} sats
|
||
Pending Channels: {} sats
|
||
|
||
---------------------------------------------------------
|
||
""".format(a['balance'], a['pending_open_balance']))
|
||
input("\nContinue... ")
|
||
|
||
def listonchaintxs():
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
cert_path = lndconnectload["tls"]
|
||
with open(lndconnectload["macaroon"], 'rb') as f:
|
||
macaroon = codecs.encode(f.read(), 'hex')
|
||
headers = {'Grpc-Metadata-macaroon': macaroon}
|
||
url = 'https://{}/v1/transactions'.format(lndconnectload["ip_port"])
|
||
r = requests.get(url, headers=headers, verify=cert_path, timeout=10)
|
||
a = r.json()
|
||
n = a['transactions']
|
||
while True:
|
||
clear()
|
||
print("\033[1;32;40m")
|
||
blogo()
|
||
print("\033[0;37;40m")
|
||
print("<<< Back to the Main Menu Press Control + C.\n\n")
|
||
print("\t\nTransactions\n")
|
||
try:
|
||
print("\n\tLIST ONCHAIN TRANSACTIONS\n")
|
||
for r in range(len(n)):
|
||
s = n[r]
|
||
print("Transaction Hash: " + " " + s['tx_hash'] + " sats")
|
||
nd = input("\nSelect RHash: ")
|
||
|
||
for item in n:
|
||
s = item
|
||
nn = s['tx_hash']
|
||
trx = s['dest_addresses']
|
||
if nd == nn:
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("""
|
||
\tONCHAIN TRANSACTION DECODED
|
||
Amount: {} sats
|
||
Tx Hash: {}
|
||
Block Hash: {}
|
||
Block Height: {}
|
||
Confirmations: {}
|
||
Destination: {}
|
||
""".format(s['amount'], s['tx_hash'], s['block_hash'], s['block_height'], s['num_confirmations'], trx))
|
||
print("----------------------------------------------------------------------------------------------------\n")
|
||
print("\nTransaction Hash")
|
||
print("\033[1;30;47m")
|
||
qr.add_data(s['tx_hash'])
|
||
qr.print_ascii()
|
||
print("\033[0;37;40m")
|
||
qr.clear()
|
||
input("\nContinue... ")
|
||
except Exception as e: # Catch specific exceptions
|
||
break
|
||
|
||
def balanceOC():
|
||
lndconnectload = _load_lnd_config()
|
||
cert_path = lndconnectload["tls"]
|
||
with open(lndconnectload["macaroon"], 'rb') as f:
|
||
macaroon = codecs.encode(f.read(), 'hex')
|
||
headers = {'Grpc-Metadata-macaroon': macaroon}
|
||
url = 'https://{}/v1/balance/blockchain'.format(lndconnectload["ip_port"])
|
||
r = requests.get(url, headers=headers, verify=cert_path, timeout=10)
|
||
a = r.json()
|
||
print("\n----------------------------------------------------------------------------------------------------")
|
||
print("\n\tLOCAL ONCHAIN BALANCE\n")
|
||
print("Total Balance: " + a['total_balance'] + " sats")
|
||
print("Confirmed Balance: " + a['confirmed_balance'] + " sats")
|
||
print("Unconfirmed Balance: " + a['unconfirmed_balance'] + " sats")
|
||
print("----------------------------------------------------------------------------------------------------\n")
|
||
input("\nContinue... ")
|
||
|
||
# END Remote connection with rest -------------------------------------
|
||
#---------------------------------OPENDIME-----------------------------
|
||
|
||
def ADDRbalance():
|
||
import os, sys; sys.path.insert(0, os.path.normpath(__file__ + '/support/pycode.zip'))
|
||
import support.pycode.od_wallet; support.pycode.od_wallet.main()
|