From 6d5960a27cccebd2a2412353d0addd11cbf2fc19 Mon Sep 17 00:00:00 2001 From: YanNaingWinn Date: Tue, 7 Apr 2026 17:31:13 +0700 Subject: [PATCH 1/5] Modified QuikNode support via Blockbook UTXOs and JSON-RPC fees/broadcast --- cryptnox_cli/command/btc.py | 2 +- cryptnox_cli/command/card/info.py | 2 +- cryptnox_cli/config.py | 1 + cryptnox_cli/wallet/btc.py | 59 +++++++++++++++++++++++++------ 4 files changed, 52 insertions(+), 12 deletions(-) diff --git a/cryptnox_cli/command/btc.py b/cryptnox_cli/command/btc.py index 1769997..5fc2020 100644 --- a/cryptnox_cli/command/btc.py +++ b/cryptnox_cli/command/btc.py @@ -53,7 +53,7 @@ def _send(self, card: cryptnox_sdk_py.Card) -> None: print("Derivation is invalid") return - endpoint = BlkHubApi(network) + endpoint = BlkHubApi(network, config.get("api_key", "")) path = b"" if derivation == cryptnox_sdk_py.Derivation.CURRENT_KEY else BTCwallet.PATH pubkey = card.get_public_key(derivation, path=path) diff --git a/cryptnox_cli/command/card/info.py b/cryptnox_cli/command/card/info.py index 3b7fb47..311a8ce 100644 --- a/cryptnox_cli/command/card/info.py +++ b/cryptnox_cli/command/card/info.py @@ -115,7 +115,7 @@ def _get_btc_info(config, pubkey) -> dict: if pubkey is None: return {"name": "BTC", "address": "Bad derivation type", "network": ""} network = config.get("network", "testnet").lower() - endpoint = BlkHubApi(network) + endpoint = BlkHubApi(network, config.get("api_key", "")) wallet = BTCwallet(pubkey, network, endpoint, None) tabulate_data = { diff --git a/cryptnox_cli/config.py b/cryptnox_cli/config.py index 6997653..1154047 100644 --- a/cryptnox_cli/config.py +++ b/cryptnox_cli/config.py @@ -31,6 +31,7 @@ def get_default_configuration() -> Dict: "btc": { "derivation": "DERIVE", "network": "mainnet", + "api_key": "", }, "eth": { "api_key": "", diff --git a/cryptnox_cli/wallet/btc.py b/cryptnox_cli/wallet/btc.py index 8d97e03..693f484 100644 --- a/cryptnox_cli/wallet/btc.py +++ b/cryptnox_cli/wallet/btc.py @@ -14,7 +14,7 @@ from cryptnox_sdk_py import Derivation from tabulate import tabulate -from .validators import EnumValidator, IntValidator +from .validators import AnyValidator, EnumValidator, IntValidator try: from lib import cryptos @@ -164,9 +164,12 @@ class BlkHubApi: BlkHubApi """ - def __init__(self, network): + def __init__(self, network, api_key: str = ""): network = network.lower() self.url = BlkHubApi.get_api(network) + if not self.url.endswith("/"): + self.url += "/" + self.api_key = api_key self.js_res = [] self.web_rsc = None @@ -184,7 +187,7 @@ def get_api(network: str) -> str: if network.lower() == "testnet": return "https://blockstream.info/testnet/api/" if network.lower() == "testnet4": - return "https://mempool.space/testnet4/api/" + return "https://newest-intensive-feather.btc-testnet4.quiknode.pro/" raise Exception("Unknown BC network name") def _validate_endpoint(self, endpoint: str) -> str: @@ -221,14 +224,15 @@ def get_data(self, endpoint: str, params: Dict = None, data: bytes = None) \ # Construct full URL and validate it stays within expected domain full_url = self.url + endpoint + ("?" + params_enc if params_enc else "") parsed = urlparse(full_url) - if not parsed.hostname or not any( - domain in parsed.hostname - for domain in ['blkhub.net', 'blockstream.info', 'mempool.space']): - raise ValueError("Invalid URL: must be blkhub.net, blockstream.info, or mempool.space domain") + if not parsed.hostname or not parsed.scheme.startswith('https'): + raise ValueError("Invalid URL: must use HTTPS") + headers = {'User-Agent': 'Mozilla/5.0'} + if self.api_key: + headers['x-token'] = self.api_key req = urllib.request.Request( full_url, - headers={'User-Agent': 'Mozilla/5.0'}, + headers=headers, data=data ) self.web_rsc = urllib.request.urlopen(req, timeout=30) @@ -255,7 +259,29 @@ def check_api_resp(self) -> None: print(" !! ERRORS :") raise Exception(self.js_res['errors']) + def _is_blockbook(self) -> bool: + return "quiknode.pro" in self.url or "quicknode.com" in self.url + + def _json_rpc(self, method: str, params=None): + params = params if params is not None else [] + body = json.dumps({"jsonrpc": "2.0", "method": method, "params": params, "id": 1}).encode("utf-8") + headers = {'User-Agent': 'Mozilla/5.0', 'Content-Type': 'application/json'} + if self.api_key: + headers['x-token'] = self.api_key + req = urllib.request.Request(self.url, headers=headers, data=body) + resp = urllib.request.urlopen(req, timeout=30) + result = json.loads(resp.read()) + if "error" in result and result["error"]: + raise IOError(f"JSON-RPC error: {result['error']}") + return result.get("result") + def get_fee_estimates(self, blocks=6) -> int: + if self._is_blockbook(): + result = self._json_rpc("estimatesmartfee", [blocks]) + feerate = result.get("feerate", 0) + if feerate <= 0: + return 0 + return math.ceil(feerate * 100_000_000 / 1000) self.get_data("fee-estimates") block_entries = [int(x) for x in self.js_res.keys() if int(x) <= blocks] block_entries.sort() @@ -271,10 +297,18 @@ def get_utx_os(self, addr: str, _n_conf: int) -> List: :param int _n_conf: 0 or 1 :return: list """ + if self._is_blockbook(): + self.get_data("api/v2/utxo/" + addr) + sel_utx_os = [] + for utxo in self.js_res: + sel_utx_os.append({ + 'value': int(utxo['value']), + 'output': utxo['txid'] + ":" + str(utxo['vout']) + }) + return sel_utx_os self.get_data("address/" + addr + "/utxo") addr_utx_os = self.js_res sel_utx_os = [] - # translate inputs from blkhub to pybitcoinlib for utxo in addr_utx_os: sel_utx_os.append({ 'value': utxo['value'], @@ -288,6 +322,9 @@ def push_tx(self, tx_hex: str) -> List: :param tx_hex: str :return: List """ + if self._is_blockbook(): + result = self._json_rpc("sendrawtransaction", [tx_hex]) + return result self.get_data("tx", data=tx_hex.encode('ascii')) self.check_api_resp() return self.get_key('txid') @@ -486,9 +523,11 @@ class BtcValidator: network = EnumValidator(BtcNetworks) fees = IntValidator() derivation = EnumValidator(Derivation) + api_key = AnyValidator() def __init__(self, network: str = "testnet", fees: int = 2000, - derivation: str = "DERIVE"): + derivation: str = "DERIVE", api_key: str = ""): self.network = network self.fees = fees self.derivation = derivation + self.api_key = api_key From b0eb3db1cb78c001cafa69b5d1ca9f7b4ea22396 Mon Sep 17 00:00:00 2001 From: YanNaingWinn Date: Tue, 7 Apr 2026 19:04:46 +0700 Subject: [PATCH 2/5] Resolves CodeQL py/incomplete-url-substring-sanitization finding. --- cryptnox_cli/wallet/btc.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/cryptnox_cli/wallet/btc.py b/cryptnox_cli/wallet/btc.py index 693f484..be4289d 100644 --- a/cryptnox_cli/wallet/btc.py +++ b/cryptnox_cli/wallet/btc.py @@ -260,7 +260,8 @@ def check_api_resp(self) -> None: raise Exception(self.js_res['errors']) def _is_blockbook(self) -> bool: - return "quiknode.pro" in self.url or "quicknode.com" in self.url + hostname = urlparse(self.url).hostname or "" + return hostname.endswith(".quiknode.pro") or hostname.endswith(".quicknode.com") def _json_rpc(self, method: str, params=None): params = params if params is not None else [] From a8849283142c14864bc712686189e48901d09259 Mon Sep 17 00:00:00 2001 From: YanNaingWinn Date: Wed, 15 Apr 2026 20:36:45 +0700 Subject: [PATCH 3/5] Address PR #49 review comments: security and configurability fixes - Restore domain whitelist in BlkHubApi.get_data() with quiknode.pro/quicknode.com added, replacing the HTTPS-only check that removed the previous allowlist - Add HTTPS guard at the top of _json_rpc() to satisfy Semgrep dynamic-urllib finding - Raise IOError on missing 'result' key in _json_rpc() instead of silently returning None - Move hardcoded testnet4 QuikNode URL to configurable endpoint: add endpoint field to BtcValidator, default config, and BlkHubApi constructor so users can update it via `btc config endpoint ` without a new release Co-Authored-By: Claude Sonnet 4.6 --- cryptnox_cli/command/btc.py | 2 +- cryptnox_cli/command/card/info.py | 2 +- cryptnox_cli/command/helper/config.py | 2 +- cryptnox_cli/config.py | 1 + cryptnox_cli/wallet/btc.py | 24 ++++++++++++++++++------ 5 files changed, 22 insertions(+), 9 deletions(-) diff --git a/cryptnox_cli/command/btc.py b/cryptnox_cli/command/btc.py index 5fc2020..e04af76 100644 --- a/cryptnox_cli/command/btc.py +++ b/cryptnox_cli/command/btc.py @@ -53,7 +53,7 @@ def _send(self, card: cryptnox_sdk_py.Card) -> None: print("Derivation is invalid") return - endpoint = BlkHubApi(network, config.get("api_key", "")) + endpoint = BlkHubApi(network, config.get("api_key", ""), config.get("endpoint", "")) path = b"" if derivation == cryptnox_sdk_py.Derivation.CURRENT_KEY else BTCwallet.PATH pubkey = card.get_public_key(derivation, path=path) diff --git a/cryptnox_cli/command/card/info.py b/cryptnox_cli/command/card/info.py index 311a8ce..077de78 100644 --- a/cryptnox_cli/command/card/info.py +++ b/cryptnox_cli/command/card/info.py @@ -115,7 +115,7 @@ def _get_btc_info(config, pubkey) -> dict: if pubkey is None: return {"name": "BTC", "address": "Bad derivation type", "network": ""} network = config.get("network", "testnet").lower() - endpoint = BlkHubApi(network, config.get("api_key", "")) + endpoint = BlkHubApi(network, config.get("api_key", ""), config.get("endpoint", "")) wallet = BTCwallet(pubkey, network, endpoint, None) tabulate_data = { diff --git a/cryptnox_cli/command/helper/config.py b/cryptnox_cli/command/helper/config.py index 5bb9955..54c4ee4 100644 --- a/cryptnox_cli/command/helper/config.py +++ b/cryptnox_cli/command/helper/config.py @@ -140,7 +140,7 @@ def print_key_config(card: cryptnox_sdk_py.Card, section: str, key: str) -> int: def find_endpoint(section: str, key: str, value: str, append: str = "") -> str: if key == "network" and section == "btc": - return f"\nendpoint: {BlkHubApi.get_api(value)}{append}" + return f"\nendpoint (derived): {BlkHubApi.get_api(value)}{append}" return "" diff --git a/cryptnox_cli/config.py b/cryptnox_cli/config.py index 1154047..dadce5a 100644 --- a/cryptnox_cli/config.py +++ b/cryptnox_cli/config.py @@ -32,6 +32,7 @@ def get_default_configuration() -> Dict: "derivation": "DERIVE", "network": "mainnet", "api_key": "", + "endpoint": "", }, "eth": { "api_key": "", diff --git a/cryptnox_cli/wallet/btc.py b/cryptnox_cli/wallet/btc.py index be4289d..f520f9d 100644 --- a/cryptnox_cli/wallet/btc.py +++ b/cryptnox_cli/wallet/btc.py @@ -164,9 +164,9 @@ class BlkHubApi: BlkHubApi """ - def __init__(self, network, api_key: str = ""): + def __init__(self, network, api_key: str = "", endpoint: str = ""): network = network.lower() - self.url = BlkHubApi.get_api(network) + self.url = endpoint if endpoint else BlkHubApi.get_api(network) if not self.url.endswith("/"): self.url += "/" self.api_key = api_key @@ -224,8 +224,14 @@ def get_data(self, endpoint: str, params: Dict = None, data: bytes = None) \ # Construct full URL and validate it stays within expected domain full_url = self.url + endpoint + ("?" + params_enc if params_enc else "") parsed = urlparse(full_url) - if not parsed.hostname or not parsed.scheme.startswith('https'): - raise ValueError("Invalid URL: must use HTTPS") + _ALLOWED_DOMAINS = ('blkhub.net', 'blockstream.info', 'mempool.space', + 'quiknode.pro', 'quicknode.com') + if not parsed.hostname or not any( + parsed.hostname == d or parsed.hostname.endswith('.' + d) + for d in _ALLOWED_DOMAINS): + raise ValueError( + f"Invalid URL: must be one of {', '.join(_ALLOWED_DOMAINS)}" + ) headers = {'User-Agent': 'Mozilla/5.0'} if self.api_key: @@ -264,6 +270,8 @@ def _is_blockbook(self) -> bool: return hostname.endswith(".quiknode.pro") or hostname.endswith(".quicknode.com") def _json_rpc(self, method: str, params=None): + if not self.url.startswith("https://"): + raise ValueError("JSON-RPC endpoint must use HTTPS") params = params if params is not None else [] body = json.dumps({"jsonrpc": "2.0", "method": method, "params": params, "id": 1}).encode("utf-8") headers = {'User-Agent': 'Mozilla/5.0', 'Content-Type': 'application/json'} @@ -274,7 +282,9 @@ def _json_rpc(self, method: str, params=None): result = json.loads(resp.read()) if "error" in result and result["error"]: raise IOError(f"JSON-RPC error: {result['error']}") - return result.get("result") + if "result" not in result: + raise IOError("JSON-RPC malformed response: missing 'result' key") + return result["result"] def get_fee_estimates(self, blocks=6) -> int: if self._is_blockbook(): @@ -525,10 +535,12 @@ class BtcValidator: fees = IntValidator() derivation = EnumValidator(Derivation) api_key = AnyValidator() + endpoint = AnyValidator() def __init__(self, network: str = "testnet", fees: int = 2000, - derivation: str = "DERIVE", api_key: str = ""): + derivation: str = "DERIVE", api_key: str = "", endpoint: str = ""): self.network = network self.fees = fees self.derivation = derivation self.api_key = api_key + self.endpoint = endpoint From 25b25b9586fbc376ad07411ad2c1d0f621458b05 Mon Sep 17 00:00:00 2001 From: YanNaingWinn Date: Wed, 15 Apr 2026 22:23:17 +0700 Subject: [PATCH 4/5] Fix flake8 E127: continuation line over-indented in btc.py Co-Authored-By: Claude Sonnet 4.6 --- cryptnox_cli/wallet/btc.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cryptnox_cli/wallet/btc.py b/cryptnox_cli/wallet/btc.py index f520f9d..339170b 100644 --- a/cryptnox_cli/wallet/btc.py +++ b/cryptnox_cli/wallet/btc.py @@ -225,7 +225,7 @@ def get_data(self, endpoint: str, params: Dict = None, data: bytes = None) \ full_url = self.url + endpoint + ("?" + params_enc if params_enc else "") parsed = urlparse(full_url) _ALLOWED_DOMAINS = ('blkhub.net', 'blockstream.info', 'mempool.space', - 'quiknode.pro', 'quicknode.com') + 'quiknode.pro', 'quicknode.com') if not parsed.hostname or not any( parsed.hostname == d or parsed.hostname.endswith('.' + d) for d in _ALLOWED_DOMAINS): From ba6c761be6b22d548964afb0a3d563b9c5bb4b5a Mon Sep 17 00:00:00 2001 From: YanNaingWinn Date: Mon, 20 Apr 2026 19:57:54 +0700 Subject: [PATCH 5/5] Resolved semgrep issue --- cryptnox_cli/wallet/btc.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/cryptnox_cli/wallet/btc.py b/cryptnox_cli/wallet/btc.py index 339170b..3a060df 100644 --- a/cryptnox_cli/wallet/btc.py +++ b/cryptnox_cli/wallet/btc.py @@ -7,6 +7,7 @@ import re import urllib.parse import urllib.request +import requests from enum import Enum from typing import Union, List, Dict from urllib.parse import urlparse @@ -270,16 +271,17 @@ def _is_blockbook(self) -> bool: return hostname.endswith(".quiknode.pro") or hostname.endswith(".quicknode.com") def _json_rpc(self, method: str, params=None): - if not self.url.startswith("https://"): + parsed = urlparse(self.url) + if parsed.scheme != "https": raise ValueError("JSON-RPC endpoint must use HTTPS") params = params if params is not None else [] - body = json.dumps({"jsonrpc": "2.0", "method": method, "params": params, "id": 1}).encode("utf-8") + payload = {"jsonrpc": "2.0", "method": method, "params": params, "id": 1} headers = {'User-Agent': 'Mozilla/5.0', 'Content-Type': 'application/json'} if self.api_key: headers['x-token'] = self.api_key - req = urllib.request.Request(self.url, headers=headers, data=body) - resp = urllib.request.urlopen(req, timeout=30) - result = json.loads(resp.read()) + resp = requests.post(self.url, headers=headers, json=payload, timeout=30) + resp.raise_for_status() + result = resp.json() if "error" in result and result["error"]: raise IOError(f"JSON-RPC error: {result['error']}") if "result" not in result: