mirror of
https://github.com/curly60e/pyblock.git
synced 2026-08-13 12:33:15 +02:00
d79977ce00
Security (Critical): - Eliminate all shell=True command injection vectors (~95 instances in ppi.py, spvblock.py) - Replace subprocess curl calls with requests library - Add input validation (fiat code allowlist, IP address validation) - Replace weak random.randint/choice with secrets module for crypto ops - Remove token/credential exposure from print statements - Add path traversal prevention in config.py - Create .conf.example templates, scrub local credentials Stability: - Replace 63 bare except clauses with specific exceptions + logging - Fix file handle leaks with context managers (lnd.py, apisnd.py) - Add threading.Lock for race conditions in clock/data.py - Cap unbounded list growth (MAX_HISTORY_LEN=50) - Add timeout=10 to ~50 requests calls missing timeouts Maintainability: - Extract _load_macaroon() helper (dedup 69 instances in PyBlock.py) - Extract _load_lnd_config() helper (dedup 33 instances in nodeconnection.py) - Normalize json import (simplejson with stdlib fallback) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
import base64
|
|
import os
|
|
from os.path import expanduser
|
|
import codecs
|
|
import grpc
|
|
import sys
|
|
|
|
from grpc_generated import router_pb2_grpc as lnrouterrpc, router_pb2 as lnrouter, rpc_pb2_grpc as lnrpc, rpc_pb2 as ln
|
|
|
|
MESSAGE_SIZE_MB = 50 * 1024 * 1024
|
|
|
|
|
|
def debug(message):
|
|
sys.stderr.write(message + "\n")
|
|
|
|
|
|
class Lnd:
|
|
def __init__(self, lnd_dir, server):
|
|
os.environ['GRPC_SSL_CIPHER_SUITES'] = 'HIGH+ECDSA'
|
|
lnd_dir = expanduser(lnd_dir)
|
|
combined_credentials = self.get_credentials(lnd_dir)
|
|
channel_options = [
|
|
('grpc.max_message_length', MESSAGE_SIZE_MB),
|
|
('grpc.max_receive_message_length', MESSAGE_SIZE_MB)
|
|
]
|
|
grpc_channel = grpc.secure_channel(server, combined_credentials, channel_options)
|
|
self.stub = lnrpc.LightningStub(grpc_channel)
|
|
self.router_stub = lnrouterrpc.RouterStub(grpc_channel)
|
|
self.graph = None
|
|
self.info = None
|
|
self.channels = None
|
|
|
|
@staticmethod
|
|
def get_credentials(lnd_dir):
|
|
with open(lnd_dir + '/tls.cert', 'rb') as f:
|
|
tls_certificate = f.read()
|
|
ssl_credentials = grpc.ssl_channel_credentials(tls_certificate)
|
|
with open(lnd_dir + '/data/chain/bitcoin/mainnet/admin.macaroon', 'rb') as f:
|
|
macaroon = codecs.encode(f.read(), 'hex')
|
|
auth_credentials = grpc.metadata_call_credentials(lambda _, callback: callback([('macaroon', macaroon)], None))
|
|
combined_credentials = grpc.composite_channel_credentials(ssl_credentials, auth_credentials)
|
|
return combined_credentials
|
|
|
|
def get_info(self):
|
|
if self.info is None:
|
|
self.info = self.stub.GetInfo(ln.GetInfoRequest())
|
|
return self.info
|
|
|
|
def get_graph(self):
|
|
if self.graph is None:
|
|
self.graph = self.stub.DescribeGraph(ln.ChannelGraphRequest(include_unannounced=True))
|
|
return self.graph
|
|
|
|
def get_own_pubkey(self):
|
|
return self.get_info().identity_pubkey
|
|
|
|
def get_edges(self):
|
|
return self.get_graph().edges
|
|
|
|
def generate_invoice(self, memo, amount):
|
|
invoice_request = ln.Invoice(
|
|
memo=memo,
|
|
value=amount,
|
|
)
|
|
add_invoice_response = self.stub.AddInvoice(invoice_request)
|
|
return self.decode_payment_request(add_invoice_response.payment_request)
|
|
|
|
def decode_payment_request(self, payment_request):
|
|
request = ln.PayReqString(
|
|
pay_req=payment_request,
|
|
)
|
|
return self.stub.DecodePayReq(request)
|
|
|
|
def get_channels(self):
|
|
if self.channels is None:
|
|
request = ln.ListChannelsRequest(
|
|
active_only=True,
|
|
)
|
|
self.channels = self.stub.ListChannels(request).channels
|
|
return self.channels
|
|
|
|
def get_route(self, pub_key, amount, ignored_edges, ignored_nodes, first_hop_channel_id):
|
|
if pub_key:
|
|
last_hop_pubkey = base64.b16decode(pub_key, True)
|
|
else:
|
|
last_hop_pubkey = None
|
|
request = ln.QueryRoutesRequest(
|
|
pub_key=self.get_own_pubkey(),
|
|
last_hop_pubkey=last_hop_pubkey,
|
|
amt=amount,
|
|
ignored_edges=ignored_edges,
|
|
ignored_nodes=ignored_nodes,
|
|
use_mission_control=True,
|
|
outgoing_chan_id=first_hop_channel_id,
|
|
)
|
|
try:
|
|
response = self.stub.QueryRoutes(request)
|
|
return response.routes
|
|
except Exception:
|
|
return None
|
|
|
|
def send_payment(self, payment_request, route):
|
|
request = lnrouter.SendToRouteRequest(route=route)
|
|
request.payment_hash = self.hex_string_to_bytes(payment_request.payment_hash)
|
|
return self.router_stub.SendToRoute(request)
|
|
|
|
@staticmethod
|
|
def hex_string_to_bytes(hex_string):
|
|
decode_hex = codecs.getdecoder("hex_codec")
|
|
return decode_hex(hex_string)[0]
|