add multisig cases

framework/helper/ckb_cli.py

@@ -181,6 +181,54 @@ def version():
     return output
 
 
+def tx_build_multisig_address(
+    addresses,
+    threshold=None,
+    multisig_code_hash="legacy",
+    api_url="http://127.0.0.1:8114",
+):
+    """
+    Build a multisig address using ckb-cli tx build-multisig-address command.
+
+    Args:
+        addresses (list): List of sighash addresses, at least two addresses required (e.g., ["ckt1qz...", "ckt1qz..."]).
+        threshold (int, optional): Number of signatures required. Defaults to len(addresses).
+        multisig_code_hash (str, optional): Multisig code hash. Defaults to "legacy".
+        cli_path (str): Path to ckb-cli executable. Defaults to "ckb-cli".
+
+    Returns:
+        str: Output of the ckb-cli command.
+
+    Example:
+        addresses = [
+            "ckt1qzda0cr08m85hc8jlnfp3zer7xulejywt49kt2rr0vthywaa50xwsqvm4mmpqw7vp4alvjuls8lxqz0jtvd47mqg0estw",
+            "ckt1qzda0cr08m85hc8jlnfp3zer7xulejywt49kt2rr0vthywaa50xwsqtmcfut7hfzcpcjx5m2c6ylrnfkckyvldcu8d67f"
+        ]
+        output = build_multisig_address(addresses, threshold=2, multisig_code_hash="legacy")
+    """
+    if threshold is None:
+        threshold = len(addresses)
+
+    if len(addresses) < 2:
+        raise ValueError("At least two addresses are required.")
+    if threshold < 2 or threshold > len(addresses):
+        raise ValueError(f"Threshold must be between 2 and {len(addresses)}.")
+
+    cmd = f"export API_URL={api_url} && {cli_path} tx build-multisig-address"
+    for addr in addresses:
+        cmd += f" --sighash-address {addr}"
+    cmd += f" --threshold {threshold} --multisig-code-hash {multisig_code_hash}"
+    cmd += f" --output-format json"
+
+    output = run_command(cmd)
+
+    # print("\n=============== Multisig Address Output ===============")
+    # print(output.strip())
+    # print("=======================================================\n")
+
+    return json.loads(output)
+
+
 def deploy_gen_txs(
     from_address, deployment_config_path, tx_info_path, api_url="http://127.0.0.1:8114"
 ):
@@ -386,6 +434,34 @@ def tx_add_input(tx_hash, index, tx_file, api_url="http://127.0.0.1:8114"):
     return run_command(cmd)
 
 
+def tx_add_output_multisig(
+    address, capacity, tx_file, is_multisig=False, api_url="http://127.0.0.1:8114"
+):
+    """
+    Add output to transaction.
+    Args:
+        address: recipient CKB address
+        capacity: capacity in CKB (1 CKB = 10^8 shannons)
+        tx_file: transaction file path
+        is_multisig: whether the address is a short multisig address
+        api_url: CKB node RPC URL
+    Returns:
+        command execution result
+    """
+    address_flag = (
+        "--to-short-multisig-address" if is_multisig else "--to-sighash-address"
+    )
+
+    cmd = (
+        f"export API_URL={api_url} && "
+        f"{cli_path} tx add-output "
+        f"{address_flag} {address} "
+        f"--capacity {capacity} "
+        f"--tx-file {tx_file}"
+    )
+    return run_command(cmd)
+
+
 def tx_add_multisig_config(ckb_address, tx_file, api_url="http://127.0.0.1:8114"):
     """
     ./ckb-cli tx add-multisig-config   --multisig-code-hash legacy  --sighash-address ckt1qyqdfjzl8ju2vfwjtl4mttx6me09hayzfldq8m3a0y --tx-file tx.txt
@@ -427,6 +503,31 @@ def tx_add_multisig_config(ckb_address, tx_file, api_url="http://127.0.0.1:8114"
     return run_command(cmd)
 
 
+def tx_add_multisig_config_for_addr_list(
+    addresses,
+    tx_file,
+    threshold=None,
+    multisig_code_hash="legacy",
+    api_url="http://127.0.0.1:8114",
+):
+    if not addresses:
+        raise ValueError("At least two addresses are required.")
+
+    threshold = threshold or len(addresses)
+    sighash_addresses = " ".join(f"--sighash-address {addr}" for addr in addresses)
+
+    cmd = (
+        f"export API_URL={api_url} && "
+        f"{cli_path} tx add-multisig-config "
+        f"{sighash_addresses} "
+        f"--threshold {threshold} "
+        f"--multisig-code-hash {multisig_code_hash} "
+        f"--tx-file {tx_file}"
+    )
+
+    return run_command(cmd)
+
+
 def tx_info(tx_file_path, api_url="http://127.0.0.1:8114"):
     cmd = f"export API_URL={api_url} && {cli_path} tx info --tx-file {tx_file_path}"
     return run_command(cmd)

framework/segwit_addr.py

@@ -0,0 +1,149 @@
+# Copyright (c) 2017, 2020 Pieter Wuille
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+
+"""Reference implementation for Bech32/Bech32m and segwit addresses."""
+
+from enum import Enum
+
+
+class Encoding(Enum):
+    """Enumeration type to list the various supported encodings."""
+
+    BECH32 = 1
+    BECH32M = 2
+
+
+CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
+BECH32M_CONST = 0x2BC830A3
+
+
+def bech32_polymod(values):
+    """Internal function that computes the Bech32 checksum."""
+    generator = [0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3]
+    chk = 1
+    for value in values:
+        top = chk >> 25
+        chk = (chk & 0x1FFFFFF) << 5 ^ value
+        for i in range(5):
+            chk ^= generator[i] if ((top >> i) & 1) else 0
+    return chk
+
+
+def bech32_hrp_expand(hrp):
+    """Expand the HRP into values for checksum computation."""
+    return [ord(x) >> 5 for x in hrp] + [0] + [ord(x) & 31 for x in hrp]
+
+
+def bech32_verify_checksum(hrp, data):
+    """Verify a checksum given HRP and converted data characters."""
+    const = bech32_polymod(bech32_hrp_expand(hrp) + data)
+    if const == 1:
+        return Encoding.BECH32
+    if const == BECH32M_CONST:
+        return Encoding.BECH32M
+    return None
+
+
+def bech32_create_checksum(hrp, data, spec):
+    """Compute the checksum values given HRP and data."""
+    values = bech32_hrp_expand(hrp) + data
+    const = BECH32M_CONST if spec == Encoding.BECH32M else 1
+    polymod = bech32_polymod(values + [0, 0, 0, 0, 0, 0]) ^ const
+    return [(polymod >> 5 * (5 - i)) & 31 for i in range(6)]
+
+
+def bech32_encode(hrp, data, spec):
+    """Compute a Bech32 string given HRP and data values."""
+    combined = data + bech32_create_checksum(hrp, data, spec)
+    return hrp + "1" + "".join([CHARSET[d] for d in combined])
+
+
+def bech32_decode(bech):
+    """Validate a Bech32/Bech32m string, and determine HRP and data."""
+    if (any(ord(x) < 33 or ord(x) > 126 for x in bech)) or (
+        bech.lower() != bech and bech.upper() != bech
+    ):
+        return (None, None, None)
+    bech = bech.lower()
+    pos = bech.rfind("1")
+    if pos < 1 or pos + 7 > len(bech):
+        return (None, None, None)
+    if not all(x in CHARSET for x in bech[pos + 1 :]):
+        return (None, None, None)
+    hrp = bech[:pos]
+    data = [CHARSET.find(x) for x in bech[pos + 1 :]]
+    spec = bech32_verify_checksum(hrp, data)
+    if spec is None:
+        return (None, None, None)
+    return (hrp, data[:-6], spec)
+
+
+def convertbits(data, frombits, tobits, pad=True):
+    """General power-of-2 base conversion."""
+    acc = 0
+    bits = 0
+    ret = []
+    maxv = (1 << tobits) - 1
+    max_acc = (1 << (frombits + tobits - 1)) - 1
+    for value in data:
+        if value < 0 or (value >> frombits):
+            return None
+        acc = ((acc << frombits) | value) & max_acc
+        bits += frombits
+        while bits >= tobits:
+            bits -= tobits
+            ret.append((acc >> bits) & maxv)
+    if pad:
+        if bits:
+            ret.append((acc << (tobits - bits)) & maxv)
+    elif bits >= frombits or ((acc << (tobits - bits)) & maxv):
+        return None
+    return ret
+
+
+def decode(hrp, addr):
+    """Decode a segwit address."""
+    hrpgot, data, spec = bech32_decode(addr)
+    if hrpgot != hrp:
+        return (None, None)
+    decoded = convertbits(data[1:], 5, 8, False)
+    if decoded is None or len(decoded) < 2 or len(decoded) > 40:
+        return (None, None)
+    if data[0] > 16:
+        return (None, None)
+    if data[0] == 0 and len(decoded) != 20 and len(decoded) != 32:
+        return (None, None)
+    if (
+        data[0] == 0
+        and spec != Encoding.BECH32
+        or data[0] != 0
+        and spec != Encoding.BECH32M
+    ):
+        return (None, None)
+    return (data[0], decoded)
+
+
+def encode(hrp, witver, witprog):
+    """Encode a segwit address."""
+    spec = Encoding.BECH32 if witver == 0 else Encoding.BECH32M
+    ret = bech32_encode(hrp, [witver] + convertbits(witprog, 8, 5), spec)
+    if decode(hrp, ret) == (None, None):
+        return None
+    return ret

framework/util.py

@@ -8,6 +8,8 @@
 
 import hashlib
 
+from framework import segwit_addr as sa
+
 H256_ZEROS = "0x0000000000000000000000000000000000000000000000000000000000000000"
 
 U128_MIN_COMPATIBLE = 0  # Adjust according to your definition
@@ -199,6 +201,53 @@ def generate_random_preimage():
     return hash_str
 
 
+# ref: https://github.com/nervosnetwork/rfcs/blob/master/rfcs/0021-ckb-address-format/0021-ckb-address-format.md
+FORMAT_TYPE_FULL = 0x00
+FORMAT_TYPE_SHORT = 0x01
+FORMAT_TYPE_FULL_DATA = 0x02
+FORMAT_TYPE_FULL_TYPE = 0x04
+
+CODE_INDEX_SECP256K1_SINGLE = 0x00
+CODE_INDEX_SECP256K1_MULTI = 0x01
+CODE_INDEX_ACP = 0x02
+
+BECH32_CONST = 1
+BECH32M_CONST = 0x2BC830A3
+
+
+def decodeAddress(addr, network="mainnet"):
+    hrp = {"mainnet": "ckb", "testnet": "ckt"}[network]
+    hrpgot, data, spec = sa.bech32_decode(addr)
+    if hrpgot != hrp or data == None:
+        return False
+    decoded = sa.convertbits(data, 5, 8, False)
+    if decoded == None:
+        return False
+    payload = bytes(decoded)
+    format_type = payload[0]
+    if format_type == FORMAT_TYPE_FULL:
+        ptr = 1
+        code_hash = "0x" + payload[ptr : ptr + 32].hex()
+        ptr += 32
+        hash_type = payload[ptr : ptr + 1].hex()
+        ptr += 1
+        args = "0x" + payload[ptr:].hex()
+        return ("full", code_hash, hash_type, args)
+    elif format_type == FORMAT_TYPE_SHORT:
+        code_index = payload[1]
+        pk = "0x" + payload[2:].hex()
+        return ("short", code_index, pk)
+    elif format_type == FORMAT_TYPE_FULL_DATA or format_type == FORMAT_TYPE_FULL_TYPE:
+        full_type = {FORMAT_TYPE_FULL_DATA: "Data", FORMAT_TYPE_FULL_TYPE: "Type"}[
+            format_type
+        ]
+        ptr = 1
+        code_hash = payload[ptr : ptr + 32].hex()
+        ptr += 32
+        args = payload[ptr:].hex()
+        return ("deprecated full", full_type, code_hash, args)
+
+
 if __name__ == "__main__":
     ret = to_big_uint128_le_compatible(100000)
     ret1 = to_int_from_big_uint128_le(ret)