diff --git a/pybitblock/SPV/spvblock.py b/pybitblock/SPV/spvblock.py index 988bf27..b5c24a6 100644 --- a/pybitblock/SPV/spvblock.py +++ b/pybitblock/SPV/spvblock.py @@ -393,7 +393,7 @@ def opreturnOnchainONLY(): invoiceN = b invoice = invoiceN.lower() lncli = " payinvoice " - lsd = subprocess.run(shlex.split(lndconnectload["ln"]) + ["decodepayreq", invoice], capture_output=True, text=True).stdout + lsd = subprocess.run(shlex.split(lndconnectload["ln"]) + ["decodepayreq", invoice], capture_output=True, text=True).stdout # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit lsd0 = str(lsd) d = json.loads(lsd0) url = f"https://opreturnbot.com/api/status/{d['payment_hash']}" @@ -458,7 +458,7 @@ def opreturn(): invoiceN = b invoice = invoiceN.lower() lncli = " payinvoice " - lsd = subprocess.run(shlex.split(lndconnectload["ln"]) + ["decodepayreq", invoice], capture_output=True, text=True).stdout + lsd = subprocess.run(shlex.split(lndconnectload["ln"]) + ["decodepayreq", invoice], capture_output=True, text=True).stdout # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit lsd0 = str(lsd) d = json.loads(lsd0) url = f"https://opreturnbot.com/api/status/{d['payment_hash']}" @@ -742,7 +742,7 @@ def callPhoenix(): subprocess.run(["./phoenix-cli", "--help"], cwd="phoenixwallet") for _ in range(10): responseC = input("\a\nType a command of the list: ") - subprocess.run(["./phoenix-cli"] + shlex.split(responseC), cwd="phoenixwallet") + subprocess.run(["./phoenix-cli"] + shlex.split(responseC), cwd="phoenixwallet") # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit input("\a\nContinue...") except Exception as e: show_error(str(e)) @@ -1035,7 +1035,7 @@ def luxorstats(): subprocess.run(["python3", "luxor.py", "--help"], cwd=luxor_cwd) for _ in range(10): responseC = input("\a\nType a command of the list: ") - subprocess.run(["python3", "luxor.py"] + shlex.split(responseC), cwd=luxor_cwd) + subprocess.run(["python3", "luxor.py"] + shlex.split(responseC), cwd=luxor_cwd) # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit input("\a\nContinue...") except Exception as e: show_error(str(e)) @@ -2907,7 +2907,7 @@ def bip39convert(): blogo() print(output) responseC = input("Words to Tiny Seed: ") - subprocess.run(["python3", "TinySeed.py"] + shlex.split(responseC), cwd="TinySeed") + subprocess.run(["python3", "TinySeed.py"] + shlex.split(responseC), cwd="TinySeed") # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit input("\a\nContinue...") except Exception as e: show_error(str(e)) @@ -4494,7 +4494,7 @@ def callGitNostrSeedTerminal(): blogo() print(output) responseC = input("Hex to BIP39 & BIP39 to Hex: ") - subprocess.run(["python3", "nostr_seed.py"] + shlex.split(responseC), cwd="nostr_seed") + subprocess.run(["python3", "nostr_seed.py"] + shlex.split(responseC), cwd="nostr_seed") # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit input("\a\nContinue...") except Exception as e: show_error(str(e)) @@ -4517,7 +4517,7 @@ def callGitNostrQRSeedTerminal(): blogo() print(output) responseC = input("Hex to BIP39 QR & BIP39 to Hex QR: ") - subprocess.run(["python3", "nostr_c_seed_qr.py"] + shlex.split(responseC), cwd="nostr_QRseed") + subprocess.run(["python3", "nostr_c_seed_qr.py"] + shlex.split(responseC), cwd="nostr_QRseed") # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit input("\a\nContinue...") except Exception as e: show_error(str(e)) diff --git a/pybitblock/clock/data.py b/pybitblock/clock/data.py index 5ce52f0..54499d6 100644 --- a/pybitblock/clock/data.py +++ b/pybitblock/clock/data.py @@ -78,6 +78,7 @@ class ClockData: def _cli(self, command): """Run bitcoin-cli command, return stdout string.""" cmd = shlex.split(self.path["bitcoincli"]) + shlex.split(command) + # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit result = subprocess.run(cmd, capture_output=True, text=True) return result.stdout.strip() diff --git a/pybitblock/clock/renderer.py b/pybitblock/clock/renderer.py index daec2ce..5bfb60e 100644 --- a/pybitblock/clock/renderer.py +++ b/pybitblock/clock/renderer.py @@ -242,8 +242,6 @@ class Layout: lines = output.rstrip('\n').split('\n') start_row = 2 - if self._is_zen(): - start_row = max(1, (self.term_height - len(lines)) // 2) # Apply dim on odd steps wrapper = _dim if self._heartbeat_step % 2 == 1 else lambda x: x diff --git a/pybitblock/nodeconnection.py b/pybitblock/nodeconnection.py index a8eb925..c229807 100644 --- a/pybitblock/nodeconnection.py +++ b/pybitblock/nodeconnection.py @@ -23,6 +23,15 @@ lndconnectload = {"ip_port":"", "tls":"", "macaroon":"", "ln":""} settingsClock = {"gradient":"", "design":"", "colorA":"", "colorB":""} +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(): @@ -155,7 +164,7 @@ def consoleLN(): # get into the console from bitcoin-cli 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 = subprocess.run([lndconnectload['ln']] + shlex.split(cle), capture_output=True, text=True) + lsd = _run_ln(*shlex.split(cle)) lsd1 = str(lsd.stdout) print(lsd1) @@ -176,7 +185,7 @@ def locallistpeersQQ(): blogo() print("\033[0;37;40m") print("<<< Back to the Main Menu Press Control + C.\n\n") - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli), capture_output=True, text=True).stdout + lsd = _run_ln(*shlex.split(lncli)).stdout lsd0 = str(lsd) d = json.loads(lsd0) n = d['peers'] @@ -257,7 +266,7 @@ def locallistpeersQQ(): pp = input("\nDo you want to disconnect? Y/n: ") if pp in ["Y", "y"]: - lsd = subprocess.run([lndconnectload['ln'], "disconnect", nd], capture_output=True, text=True).stdout + lsd = _run_ln("disconnect", nd).stdout lsd0 = str(lsd) d = json.loads(lsd0) print("\n\tDisconnected from peer " + nd) @@ -280,7 +289,7 @@ def localconnectpeer(): print("\n\tCONNECT TO NEW PEER\n") a = input("Insert PeerID@IP:PORT: ") lncli = " connect " - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli) + [a], capture_output=True, text=True).stdout + lsd = _run_ln(*shlex.split(lncli), a).stdout lsd0 = str(lsd) print(lsd0) input("\nContinue... ") @@ -298,7 +307,7 @@ def locallistchaintxns(): border=4, ) lncli = " listchaintxns" - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli), capture_output=True, text=True).stdout + lsd = _run_ln(*shlex.split(lncli)).stdout lsd0 = str(lsd) d = json.loads(lsd0) n = d['transactions'] @@ -354,7 +363,7 @@ def locallistinvoices(): border=4, ) lncli = " listinvoices" - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli), capture_output=True, text=True).stdout + lsd = _run_ln(*shlex.split(lncli)).stdout lsd0 = str(lsd) d = json.loads(lsd0) n = d['invoices'] @@ -401,7 +410,7 @@ def locallistchannels(): lndconnectData = json.load(f) # Load the file 'bclock.conf' lndconnectload = lndconnectData # Copy the variable pathv to 'path' lncli = " listchannels" - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli), capture_output=True, text=True).stdout + lsd = _run_ln(*shlex.split(lncli)).stdout lsd0 = str(lsd) d = json.loads(lsd0) n = d['channels'] @@ -500,7 +509,7 @@ def localgetinfo(): border=4, ) lncli = " getinfo" - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli), capture_output=True, text=True).stdout + lsd = _run_ln(*shlex.split(lncli)).stdout lsd0 = str(lsd) d = json.loads(lsd0) hash = d['identity_pubkey'] @@ -560,7 +569,7 @@ def localaddinvoice(): lndconnectData = json.load(f) # Load the file 'bclock.conf' lndconnectload = lndconnectData # Copy the variable pathv to 'path' lncli = " addinvoice" - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli), capture_output=True, text=True).stdout + lsd = _run_ln(*shlex.split(lncli)).stdout lsd0 = str(lsd) d = json.loads(lsd0) qr = qrcode.QRCode( @@ -573,7 +582,7 @@ def localaddinvoice(): amount = input("Amount in sats: ") mem = input("Memo: ") memo = mem.replace(" ","_") - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli) + ["--memo", "{}-PyBLOCK".format(memo), "--amt", amount], capture_output=True, text=True).stdout + 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") @@ -584,11 +593,11 @@ def localaddinvoice(): print("Lightning Invoice: " + d['payment_request']) b = str(d['payment_request']) while True: - lsd = subprocess.run([lndconnectload['ln'], "decodepayreq", b], capture_output=True, text=True).stdout + lsd = _run_ln("decodepayreq", b).stdout lsd0 = str(lsd) d = json.loads(lsd0) r = d['payment_hash'] - lsdn = subprocess.run([lndconnectload['ln'], "lookupinvoice", r], capture_output=True, text=True).stdout + lsdn = _run_ln("lookupinvoice", r).stdout lsdn0 = str(lsdn) n = json.loads(lsdn0) if n['state'] == 'SETTLED': @@ -618,15 +627,15 @@ def localpayinvoice(): invoiceN = input("Insert the invoice to pay: ") invoice = invoiceN.lower() lncli = " payinvoice " - lsd = subprocess.run([lndconnectload['ln'], "decodepayreq", invoice], capture_output=True, text=True).stdout + 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: ") - subprocess.run([lndconnectload['ln']] + shlex.split(lncli) + [invoice] + shlex.split(amt) + [amount]) + _run_ln(*shlex.split(lncli), invoice, *shlex.split(amt), amount) else: - subprocess.run([lndconnectload['ln']] + shlex.split(lncli) + [invoice]) + _run_ln(*shlex.split(lncli), invoice) t.sleep(2) except Exception as e: # Catch specific exceptions pass @@ -636,7 +645,7 @@ def localgetnetworkinfo(): lndconnectData = json.load(f) # Load the file 'bclock.conf' lndconnectload = lndconnectData # Copy the variable pathv to 'path' lncli = " getnetworkinfo" - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli), capture_output=True, text=True).stdout + lsd = _run_ln(*shlex.split(lncli)).stdout lsd0 = str(lsd) d = json.loads(lsd0) print("\n----------------------------------------------------------------------------------------------------") @@ -874,7 +883,7 @@ def localchannelbalance(): lndconnectData = json.load(f) # Load the file 'bclock.conf' lndconnectload = lndconnectData # Copy the variable pathv to 'path' lncli = " channelbalance" - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli), capture_output=True, text=True).stdout + lsd = _run_ln(*shlex.split(lncli)).stdout lsd0 = str(lsd) d = json.loads(lsd0) print(""" @@ -894,7 +903,7 @@ def localnewaddress(): lndconnectData = json.load(f) # Load the file 'bclock.conf' lndconnectload = lndconnectData # Copy the variable pathv to 'path' lncli = " newaddress p2wkh" - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli), capture_output=True, text=True).stdout + lsd = _run_ln(*shlex.split(lncli)).stdout lsd0 = str(lsd) d = json.loads(lsd0) qr = qrcode.QRCode( @@ -916,7 +925,7 @@ def localbalanceOC(): lndconnectData = json.load(f) # Load the file 'bclock.conf' lndconnectload = lndconnectData # Copy the variable pathv to 'path' lncli = " walletbalance" - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli), capture_output=True, text=True).stdout + lsd = _run_ln(*shlex.split(lncli)).stdout lsd0 = str(lsd) d = json.loads(lsd0) print("\n----------------------------------------------------------------------------------------------------") @@ -934,7 +943,7 @@ def localrebalancelnd(): lndconnectload = lndconnectData # Copy the variable pathv to 'path' lncli = " listchannels" while True: - lsd = subprocess.run([lndconnectload['ln']] + shlex.split(lncli), capture_output=True, text=True).stdout + lsd = _run_ln(*shlex.split(lncli)).stdout lsd0 = str(lsd) d = json.loads(lsd0) n = d['channels']