diff --git a/cryptnox_cli/command/btc.py b/cryptnox_cli/command/btc.py index 1769997..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) + 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 3b7fb47..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) + 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 6997653..dadce5a 100644 --- a/cryptnox_cli/config.py +++ b/cryptnox_cli/config.py @@ -31,6 +31,8 @@ def get_default_configuration() -> Dict: "btc": { "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 8d97e03..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 @@ -14,7 +15,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 +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