Modified QuikNode support via Blockbook UTXOs and JSON-RPC fees/broadcast

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", ""), config.get("endpoint", ""))
 
         path = b"" if derivation == cryptnox_sdk_py.Derivation.CURRENT_KEY else BTCwallet.PATH
         pubkey = card.get_public_key(derivation, path=path)

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", ""), config.get("endpoint", ""))
         wallet = BTCwallet(pubkey, network, endpoint, None)
 
         tabulate_data = {

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 ""
 

cryptnox_cli/config.py

@@ -31,6 +31,8 @@ def get_default_configuration() -> Dict:
         "btc": {
             "derivation": "DERIVE",
             "network": "mainnet",
+            "api_key": "",
+            "endpoint": "",
         },
         "eth": {
             "api_key": "",

cryptnox_cli/wallet/btc.py

@@ -7,14 +7,15 @@
 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
 
 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 +165,12 @@ class BlkHubApi:
     BlkHubApi
     """
 
-    def __init__(self, network):
+    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
         self.js_res = []
         self.web_rsc = None
 
@@ -184,7 +188,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 +225,21 @@ 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)
+            _ALLOWED_DOMAINS = ('blkhub.net', 'blockstream.info', 'mempool.space',
+                                'quiknode.pro', 'quicknode.com')
             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")
-
+                    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:
+                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 +266,35 @@ def check_api_resp(self) -> None:
             print(" !! ERRORS :")
             raise Exception(self.js_res['errors'])
 
+    def _is_blockbook(self) -> bool:
+        hostname = urlparse(self.url).hostname or ""
+        return hostname.endswith(".quiknode.pro") or hostname.endswith(".quicknode.com")
+
+    def _json_rpc(self, method: str, params=None):
+        parsed = urlparse(self.url)
+        if parsed.scheme != "https":
+            raise ValueError("JSON-RPC endpoint must use HTTPS")
+        params = params if params is not None else []
+        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
+        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:
+            raise IOError("JSON-RPC malformed response: missing 'result' key")
+        return result["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 +310,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 +335,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 +536,13 @@ class BtcValidator:
     network = EnumValidator(BtcNetworks)
     fees = IntValidator()
     derivation = EnumValidator(Derivation)
+    api_key = AnyValidator()
+    endpoint = AnyValidator()
 
     def __init__(self, network: str = "testnet", fees: int = 2000,
-                 derivation: str = "DERIVE"):
+                 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