Accept Taproot (bech32m) BTC addresses in address validation

allways/chain_providers/bitcoin.py

@@ -4,10 +4,11 @@
 from urllib.parse import urlparse
 
 import base58
-import bech32
 import bittensor as bt
 import requests
 from bitcoin_message_tool.bmt import sign_message, verify_message
+from embit.networks import NETWORKS
+from embit.script import address_to_scriptpubkey
 
 from allways.chain_providers.base import ChainProvider, ProviderUnreachableError, TransactionInfo
 from allways.chains import CHAIN_BTC, ChainDefinition
@@ -53,20 +54,15 @@ def to_mainnet_wif(wif: str) -> str:
 
 
 def to_mainnet_address(address: str) -> str:
-    """Convert a testnet/regtest address to mainnet equivalent for verification."""
-    if address.startswith('bcrt1') or address.startswith('tb1'):
-        hrp, data = bech32.bech32_decode(address)
-        if data is not None:
-            return bech32.bech32_encode('bc', data)
-    if address.startswith(('m', 'n')):
-        decoded = base58.b58decode_check(address)
-        if decoded[0] == 0x6F:
-            return base58.b58encode_check(bytes([0x00]) + decoded[1:]).decode()
-    if address.startswith('2'):
-        decoded = base58.b58decode_check(address)
-        if decoded[0] == 0xC4:
-            return base58.b58encode_check(bytes([0x05]) + decoded[1:]).decode()
-    return address
+    """Convert a testnet/regtest address to mainnet equivalent for verification.
+
+    Re-encodes the scriptPubKey under the mainnet network (handles legacy, segwit
+    v0, and Taproot); returns the address unchanged if it can't be parsed.
+    """
+    try:
+        return address_to_scriptpubkey(address).address(NETWORKS['main'])
+    except Exception:
+        return address
 
 
 def parse_esplora_urls(raw: str, auth_header: str = 'Authorization') -> list[tuple[str, Optional[dict]]]:
@@ -530,15 +526,11 @@ def api_get_balance(self, address: str) -> int:
             return 0
 
     def is_valid_address(self, address: str) -> bool:
-        """Validate BTC address format without RPC (bech32/base58 decode)."""
+        """Validate BTC address format without RPC (embit decode; all types incl. Taproot)."""
         if not address or not isinstance(address, str):
             return False
         try:
-            if address.lower().startswith(('bc1', 'tb1', 'bcrt1')):
-                hrp, data = bech32.bech32_decode(address)
-                return data is not None
-            decoded = base58.b58decode_check(address)
-            return len(decoded) == 21 and decoded[0] in (0x00, 0x05, 0x6F, 0xC4)
+            return address_to_scriptpubkey(address) is not None
         except Exception:
             return False
 

pyproject.toml

@@ -33,7 +33,6 @@ dependencies = [
     "rich",
     "embit",
     "base58",
-    "bech32",
     "pycryptodome",
     "numpy",
     "psycopg[binary]==3.3.4",

tests/test_bitcoin_signing.py

@@ -13,6 +13,7 @@
     ADDR_TYPE_P2WPKH,
     BitcoinProvider,
     detect_address_type,
+    to_mainnet_address,
 )
 
 # Known test WIF (compressed)
@@ -271,3 +272,84 @@ def test_fresh_provider_starts_with_empty_set(self):
         consumed tx hash can't leak across processes."""
         provider = make_lightweight_provider()
         assert provider.broadcasted_txids == set()
+
+
+class TestIsValidAddress:
+    """BTC address format validation — must accept every standard type
+    (legacy, segwit v0, and Taproot/bech32m) and reject malformed input.
+
+    Taproot regression guard: the old bech32-only check rejected all `bc1p…`
+    addresses, blocking TAO->BTC payouts to Taproot wallets (issue #448).
+    """
+
+    # mainnet / testnet / regtest × P2PKH / P2SH / P2WPKH / P2WSH / P2TR
+    VALID = [
+        '1BvBMSEYstWetqTFn5Au4m4GFg7xJaNVN2',  # mainnet P2PKH
+        '3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy',  # mainnet P2SH
+        'bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4',  # mainnet P2WPKH
+        'bc1qrp33g0q5c5txsp9arysrx4k6zdkfs4nce4xj0gdcccefvpysxf3qccfmv3',  # mainnet P2WSH
+        'bc1p5cyxnuxmeuwuvkwfem96lqzszd02n6xdcjrs20cac6yqjjwudpxqkedrcr',  # mainnet P2TR
+        'mipcBbFg9gMiCh81Kj8tqqdgoZub1ZJRfn',  # testnet P2PKH
+        '2MzQwSSnBHWHqSAqtTVQ6v47XtaisrJa1Vc',  # testnet P2SH
+        'tb1qw508d6qejxtdg4y5r3zarvary0c5xw7kxpjzsx',  # testnet P2WPKH
+        'tb1pqqqqp399et2xygdj5xreqhjjvcmzhxw4aywxecjdzew6hylgvsesf3hn0c',  # testnet P2TR (BIP-350)
+        'bcrt1qw508d6qejxtdg4y5r3zarvary0c5xw7kygt080',  # regtest P2WPKH
+        'bcrt1p0xlxvlhemja6c4dqv22uapctqupfhlxm9h8z3k2e72q4k9hcz7vqc8gma6',  # regtest P2TR
+    ]
+
+    INVALID = [
+        '',
+        'notanaddress',
+        '0x71C7656EC7ab88b098defB751B7401B5f6d8976F',  # ETH address
+        'bc1p5cyxnuxmeuwuvkwfem96lqzszd02n6xdcjrs20cac6yqjjwudpxqkedrXXX',  # bad checksum
+        'bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3XX',  # corrupted v0
+    ]
+
+    def test_accepts_all_standard_types(self):
+        provider = make_lightweight_provider()
+        for addr in self.VALID:
+            assert provider.is_valid_address(addr), f'should accept {addr}'
+
+    def test_accepts_taproot(self):
+        """Explicit #448 guard: Taproot payout addresses must validate."""
+        provider = make_lightweight_provider()
+        assert provider.is_valid_address('bc1p5cyxnuxmeuwuvkwfem96lqzszd02n6xdcjrs20cac6yqjjwudpxqkedrcr')
+
+    def test_rejects_malformed(self):
+        provider = make_lightweight_provider()
+        for addr in self.INVALID:
+            assert not provider.is_valid_address(addr), f'should reject {addr!r}'
+
+    def test_rejects_non_string(self):
+        provider = make_lightweight_provider()
+        assert not provider.is_valid_address(None)
+
+
+class TestToMainnetAddress:
+    """testnet/regtest -> mainnet re-encoding used by the BIP-137 verify path.
+    Conversions must be byte-identical to the prior bech32/base58 behavior."""
+
+    def test_testnet_p2pkh(self):
+        assert to_mainnet_address('mipcBbFg9gMiCh81Kj8tqqdgoZub1ZJRfn') == '14JetYAhLevTRaePcAAX1vRMwaJt2QGRzp'
+
+    def test_testnet_p2sh(self):
+        assert to_mainnet_address('2MzQwSSnBHWHqSAqtTVQ6v47XtaisrJa1Vc') == '38rjNhr9g3nVEPDLnMnEJ78GgEWi6yM4He'
+
+    def test_testnet_p2wpkh(self):
+        assert (
+            to_mainnet_address('tb1qw508d6qejxtdg4y5r3zarvary0c5xw7kxpjzsx')
+            == 'bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4'
+        )
+
+    def test_regtest_p2wpkh(self):
+        assert (
+            to_mainnet_address('bcrt1qw508d6qejxtdg4y5r3zarvary0c5xw7kygt080')
+            == 'bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4'
+        )
+
+    def test_mainnet_passthrough(self):
+        addr = 'bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4'
+        assert to_mainnet_address(addr) == addr
+
+    def test_unparseable_returns_unchanged(self):
+        assert to_mainnet_address('notanaddress') == 'notanaddress'