feat: address parsing in libwally

.gitignore

@@ -1,5 +1,6 @@
 .DS_Store
 ._*
+.idea
 
 __pycache__
 *.py[cod]

boltz_client/onchain_wally.py

@@ -7,12 +7,73 @@
 from __future__ import annotations
 
 import secrets
-from typing import Optional
+from dataclasses import dataclass
+from typing import Any, Optional
 
 from .mempool import LockupData
 
 
-def get_entropy(num_outputs_to_blind):
+@dataclass
+class Network:
+    name: str
+    lbtc_asset: bytes
+    blech32_prefix: str
+    bech32_prefix: str
+
+    def wif_net(self, wally) -> Any:
+        if self.name == "mainnet":
+            return wally.WALLY_ADDRESS_VERSION_WIF_MAINNET
+        return wally.WALLY_ADDRESS_VERSION_WIF_TESTNET
+
+    def blinded_prefix(self, wally) -> Any:
+        if self.name == "mainnet":
+            return wally.WALLY_CA_PREFIX_LIQUID
+        if self.name == "testnet":
+            return wally.WALLY_CA_PREFIX_LIQUID_TESTNET
+        return wally.WALLY_CA_PREFIX_LIQUID_REGTEST
+
+    def wally_network(self, wally) -> Any:
+        if self.name == "mainnet":
+            return wally.WALLY_NETWORK_LIQUID
+        if self.name == "testnet":
+            return wally.WALLY_NETWORK_LIQUID_TESTNET
+        return wally.WALLY_NETWORK_LIQUID_REGTEST
+
+    @staticmethod
+    def parse_asset(asset: str) -> bytes:
+        return bytes.fromhex(asset)[::-1]
+
+
+# TODO: is this type hint compatible with all support Python versions of lnbits
+NETWORKS: list[Network] = [
+    Network(
+        name="mainnet",
+        lbtc_asset=Network.parse_asset(
+            "6f0279e9ed041c3d710a9f57d0c02928416460c4b722ae3457a11eec381c526d"
+        ),
+        blech32_prefix="lq",
+        bech32_prefix="ex",
+    ),
+    Network(
+        name="testnet",
+        lbtc_asset=Network.parse_asset(
+            "144c654344aa716d6f3abcc1ca90e5641e4e2a7f633bc09fe3baf64585819a49"
+        ),
+        blech32_prefix="tlq",
+        bech32_prefix="tex",
+    ),
+    Network(
+        name="regtest",
+        lbtc_asset=Network.parse_asset(
+            "5ac9f65c0efcc4775e0baec4ec03abdde22473cd3cf33c0419ca290e0751b225"
+        ),
+        blech32_prefix="el",
+        bech32_prefix="ert",
+    ),
+]
+
+
+def get_entropy(num_outputs_to_blind: int) -> bytes:
     # For each output to blind, we need 32 bytes of entropy for each of:
     # - Output assetblinder
     # - Output amountblinder
@@ -22,6 +83,69 @@ def get_entropy(num_outputs_to_blind):
     return secrets.token_bytes(num_outputs_to_blind * 5 * 32)
 
 
+def get_address_network(wally, address: str) -> Network:
+    def address_has_network_prefix(n: Network) -> bool:
+        # If address decoding doesn't fail -> correct network
+        try:
+            decode_address(wally, n, address)
+            return True
+        except Exception:
+            return False
+
+    network = next(
+        (network for network in NETWORKS if address_has_network_prefix(network)),
+        None,
+    )
+
+    if network is None:
+        raise ValueError("Unknown network of address")
+
+    return network
+
+
+def is_possible_confidential_address(wally, address) -> bool:
+    expected_len = (
+        2 + wally.EC_PUBLIC_KEY_LEN + wally.HASH160_LEN + wally.BASE58_CHECKSUM_LEN
+    )
+    try:
+        return wally.base58_n_get_length(address, len(address)) == expected_len
+    except ValueError:
+        return False
+
+
+# TODO: is this type hint compatible with all support Python versions of lnbits
+def decode_address(
+    wally, network: Network, address: str
+) -> tuple[bytearray, bytearray]:
+    if address.lower().startswith(network.blech32_prefix):
+        blinding_key = wally.confidential_addr_segwit_to_ec_public_key(
+            address, network.blech32_prefix
+        )
+        unconfidential_address = wally.confidential_addr_to_addr_segwit(
+            address, network.blech32_prefix, network.bech32_prefix
+        )
+
+        return blinding_key, wally.addr_segwit_to_bytes(
+            unconfidential_address, network.bech32_prefix, 0
+        )
+
+    if is_possible_confidential_address(wally, address):
+        unconfidential_address = wally.confidential_addr_to_addr(
+            address, network.blinded_prefix(wally)
+        )
+
+        blinding_key = wally.confidential_addr_to_ec_public_key(
+            address,
+            network.blinded_prefix(wally),
+        )
+
+        return blinding_key, wally.address_to_scriptpubkey(
+            unconfidential_address, network.wally_network(wally)
+        )
+
+    raise ValueError("only confidential addresses are supported")
+
+
 def create_liquid_tx(
     lockup_tx: LockupData,
     receive_address: str,
@@ -33,39 +157,20 @@ def create_liquid_tx(
     preimage_hex: str = "",
     blinding_key: Optional[str] = None,
 ) -> str:
-
     try:
         import wallycore as wally
     except ImportError as exc:
         raise ImportError(
             "`wallycore` is not installed, but required for liquid support."
         ) from exc
 
-    if receive_address.startswith("ert") or receive_address.startswith("el"):
-        lasset_hex = "5ac9f65c0efcc4775e0baec4ec03abdde22473cd3cf33c0419ca290e0751b225"
-        confidential_addr_prefix = "ert"
-        confidential_addr_family = "el"
-        wif_net = wally.WALLY_ADDRESS_VERSION_WIF_TESTNET
-    elif receive_address.startswith("tex") or receive_address.startswith("tlq"):
-        lasset_hex = "144c654344aa716d6f3abcc1ca90e5641e4e2a7f633bc09fe3baf64585819a49"
-        confidential_addr_prefix = "tex"
-        confidential_addr_family = "tlq"
-        wif_net = wally.WALLY_ADDRESS_VERSION_WIF_TESTNET
-    elif receive_address.startswith("ex") or receive_address.startswith("lq"):
-        lasset_hex = "6f0279e9ed041c3d710a9f57d0c02928416460c4b722ae3457a11eec381c526d"
-        confidential_addr_prefix = "ex"
-        confidential_addr_family = "lq"
-        wif_net = wally.WALLY_ADDRESS_VERSION_WIF_MAINNET
-    else:
-        raise ValueError(f"Unknown prefix: {receive_address[:3]}")
-
-    LASSET = bytes.fromhex(lasset_hex)[::-1]
+    network = get_address_network(wally, receive_address)
 
     redeem_script = bytes.fromhex(redeem_script_hex)
     preimage = bytes.fromhex(preimage_hex)
     private_key = wally.wif_to_bytes(
         privkey_wif,
-        wif_net,
+        network.wif_net(wally),
         wally.WALLY_WIF_FLAG_COMPRESSED,
     )  # type: ignore
 
@@ -75,15 +180,9 @@ def create_liquid_tx(
     except ValueError as exc:
         raise ValueError("blinding_key must be hex encoded") from exc
 
-    receive_blinding_pubkey = wally.confidential_addr_segwit_to_ec_public_key(
-        receive_address, confidential_addr_family
-    )  # type: ignore
-    receive_unconfidential_address = wally.confidential_addr_to_addr_segwit(
-        receive_address, confidential_addr_family, confidential_addr_prefix
-    )  # type: ignore
-    receive_script_pubkey = wally.addr_segwit_to_bytes(
-        receive_unconfidential_address, confidential_addr_prefix, 0
-    )  # type: ignore
+    receive_blinding_pubkey, receive_script_pubkey = decode_address(
+        wally, network, receive_address
+    )
 
     # parse lockup tx
     lockup_transaction = wally.tx_from_hex(
@@ -92,7 +191,9 @@ def create_liquid_tx(
     vout_n: Optional[int] = None
     for vout in range(wally.tx_get_num_outputs(lockup_transaction)):
         script_out = wally.tx_get_output_script(lockup_transaction, vout)  # type: ignore
-        pub_key = wally.addr_segwit_from_bytes(script_out, confidential_addr_prefix, 0)
+
+        # Lockup addresses on liquid are always bech32
+        pub_key = wally.addr_segwit_from_bytes(script_out, network.bech32_prefix, 0)
         if pub_key == lockup_tx.script_pub_key:
             vout_n = vout
             break
@@ -116,7 +217,7 @@ def create_liquid_tx(
         lockup_asset_commitment,
     )  # type: ignore
 
-    assert unblinded_asset == LASSET, "Wrong asset"
+    assert unblinded_asset == network.lbtc_asset, "Wrong asset"
 
     # INITIALIZE PSBT (PSET)
     num_vin = 1

pyproject.toml

@@ -63,4 +63,5 @@ disable = [
   "too-many-instance-attributes",
   "too-many-locals",
   "too-many-statements",
+  "broad-exception-caught",
 ]

tests/test_onchain_wally.py

@@ -0,0 +1,69 @@
+import pytest
+import wallycore as wally
+
+from boltz_client.onchain_wally import NETWORKS, get_address_network, Network, is_possible_confidential_address, \
+    decode_address
+
+
+@pytest.mark.parametrize(
+    "address, expected_network",
+    [
+        (
+            "LQ1QQ2C8P2DV7CWH4PJW4YNL4UWLUCVXEAWYPPEYUPYYVFUUUTAAQSL87F966UEZGXKKGDFHZZWEPNDAWL2JZU3UC6HMAPNHY6XG4",
+            NETWORKS[0]
+        ),
+        (
+            "lq1qq2c8p2dv7cwh4pjw4ynl4uwlucvxeawyppeyupyyvfuuutaaqsl87f966uezgxkkgdfhzzwepndawl2jzu3uc6hmapnhy6xg4",
+            NETWORKS[0]
+        ),
+        ("VJLApqRQPjHdBtTUQbWkePvmhU3p4SYfcRqX3BbxUpiKG3jfSW19oFULizTF7SPkcZp4uBf8TFMyu369", NETWORKS[0]),
+        ("VTq4Wry5xzh7PzGsF3tqDDA3NveNzZjUEUZM4T5tUVgCBvnebMCiW8jYuZmb858dSU2bYVmZEZQu5jPG", NETWORKS[0]),
+
+        (
+            "tlq1qq2yycz9ms8y3nwj8jyxph3y0y7q54murxlf6jzc0ms35r0km09qcd6teckcu4pk2j07nsvyxk5rf030penz5svrj4zjcp3jcv",
+            NETWORKS[1]
+        ),
+        ("vjTudrrmqMSaEiMthDRwQbcPiGqUU7HwwVr243mgfeuRFJ8SoUkXDnTvHRtUPg3yaReoFSeHSyCq7XwW", NETWORKS[1]),
+        ("vtSLD2eP5njy5Fkh7q58H9s2wGsRytGLLfWmzfnbRpVYvQDYhnxDEtdZZDBzuTkj72Fz16YXCQGSyFE8", NETWORKS[1]),
+        (
+            "el1qqgdry554u64x9uaj2egy9tlwqm68sqa024uvhmfn8kms8gzc6eg632lcuhh8fdq4adraffx6u9fjyz6zx8nas0txfae24mlzr",
+            NETWORKS[2]
+        ),
+        ("AzpmmwFSgXmWxhSAGtWyhfRiMTkBfy1dXJZvmaepbcCsc3RUmKxKT9uK9emXPMJnJQ67xJh3W4w8oTgb", NETWORKS[2]),
+        ("CTEk5KDeMivFF9WqDCUcPNh94AcQfF6hpUYzprJmxCbXheeMxahgry5qcwVJ8k2mw1ECYs8KokTbj77R", NETWORKS[2]),
+    ],
+)
+def test_get_address_network(address: str, expected_network: Network) -> None:
+    assert get_address_network(wally, address) == expected_network
+
+
+@pytest.mark.parametrize(
+    "address, expected",
+    [
+        ("el1qqgdry554u64x9uaj2egy9tlwqm68sqa024uvhmfn8kms8gzc6eg632lcuhh8fdq4adraffx6u9fjyz6zx8nas0txfae24mlzr", False),
+        ("AzpmmwFSgXmWxhSAGtWyhfRiMTkBfy1dXJZvmaepbcCsc3RUmKxKT9uK9emXPMJnJQ67xJh3W4w8oTgb", True),
+        ("CTEk5KDeMivFF9WqDCUcPNh94AcQfF6hpUYzprJmxCbXheeMxahgry5qcwVJ8k2mw1ECYs8KokTbj77R", True),
+    ],
+)
+def test_is_possible_confidential_address(address: str, expected: bool) -> None:
+    assert is_possible_confidential_address(wally, address) == expected
+
+
+@pytest.mark.parametrize(
+    "address, blinding_pubkey, script_pubkey",
+    [
+        ("el1qqgdry554u64x9uaj2egy9tlwqm68sqa024uvhmfn8kms8gzc6eg632lcuhh8fdq4adraffx6u9fjyz6zx8nas0txfae24mlzr",
+         "021a325295e6aa62f3b2565042afee06f47803af5578cbed333db703a058d651a8",
+         "0014abf8e5ee74b415eb47d4a4dae153220b4231e7d8"),
+        ("AzpmmwFSgXmWxhSAGtWyhfRiMTkBfy1dXJZvmaepbcCsc3RUmKxKT9uK9emXPMJnJQ67xJh3W4w8oTgb",
+         "026792e6d7da21666c305fc7ab46fda31cf621b155914f5836a99b43ea5fca41d8",
+         "a91423ec739a84326cc9a0dd1682457ad210d31bc43787"),
+        ("CTEk5KDeMivFF9WqDCUcPNh94AcQfF6hpUYzprJmxCbXheeMxahgry5qcwVJ8k2mw1ECYs8KokTbj77R",
+         "020e6c14fa10b893fb8c6fa2b378907e6e21d68c98591393f81b6078b636ba01d9",
+         "76a91469c4e8147887c27542472b021cbd34458a714c5388ac"),
+    ],
+)
+def test_decode_address(address: str, blinding_pubkey: str, script_pubkey: str) -> None:
+    blinding, script = decode_address(wally, NETWORKS[2], address)
+    assert blinding.hex() == blinding_pubkey
+    assert script.hex() == script_pubkey