Multi-Signature (Multisig) Schemes

13-of-5 multisig wallet
2Signers: CEO, CFO, CTO, COO, Board Member
3Any 3 must approve to spend funds

12-of-3 multisig
2Parties: Buyer, Seller, Arbiter
3- Normal case: Buyer + Seller sign (2 of 3)
4- Dispute: Arbiter + Winner sign (2 of 3)

12-of-3 multisig
2Keys: Desktop, Mobile, Hardware Wallet
3- Normal: Desktop + Mobile
4- Lost device: Any 2 remaining keys

12-of-3 multisig
2Keys: Owner (2 keys), Heir (1 key)
3- Owner alive: Uses 2 keys
4- Owner deceased: Heir + Lawyer/Executor

1OP_2                    # Require 2 signatures
2<pubkey1>               # First public key
3<pubkey2>               # Second public key
4<pubkey3>               # Third public key
5OP_3                    # Total of 3 keys
6OP_CHECKMULTISIG        # Verify signatures

 1from hashlib import sha256
 2import ecdsa
 3
 4class MultisigWallet:
 5    def __init__(self, m, public_keys):
 6        self.m = m  # Required signatures
 7        self.n = len(public_keys)  # Total keys
 8        self.public_keys = public_keys
 9        
10    def create_address(self):
11        """Create multisig address from public keys"""
12        # Build redeem script
13        script = f"OP_{self.m} "
14        for pk in self.public_keys:
15            script += f"{pk.hex()} "
16        script += f"OP_{self.n} OP_CHECKMULTISIG"
17        
18        # Hash script to create address
19        script_hash = sha256(script.encode()).digest()
20        # Add version byte and checksum (simplified)
21        address = "3" + script_hash[:20].hex()
22        return address, script
23    
24    def sign_transaction(self, tx_data, private_key):
25        """Sign transaction with one key"""
26        sk = ecdsa.SigningKey.from_string(private_key, curve=ecdsa.SECP256k1)
27        signature = sk.sign(tx_data)
28        return signature
29    
30    def verify_multisig(self, tx_data, signatures):
31        """Verify m-of-n signatures are valid"""
32        valid_sigs = 0
33        for sig, pubkey in zip(signatures, self.public_keys):
34            vk = ecdsa.VerifyingKey.from_string(pubkey, curve=ecdsa.SECP256k1)
35            try:
36                if vk.verify(sig, tx_data):
37                    valid_sigs += 1
38            except:
39                continue
40        
41        return valid_sigs >= self.m
42
43# Example usage
44pubkeys = [generate_pubkey() for _ in range(3)]
45wallet = MultisigWallet(m=2, public_keys=pubkeys)
46address, redeem_script = wallet.create_address()
47
48# Sign with 2 keys
49tx = b"transaction_data"
50sig1 = wallet.sign_transaction(tx, private_key_1)
51sig2 = wallet.sign_transaction(tx, private_key_2)
52
53# Verify
54is_valid = wallet.verify_multisig(tx, [sig1, sig2])

 1// SPDX-License-Identifier: MIT
 2pragma solidity ^0.8.0;
 3
 4contract MultisigWallet {
 5    address[] public owners;
 6    uint256 public required;
 7    
 8    struct Transaction {
 9        address to;
10        uint256 value;
11        bytes data;
12        bool executed;
13        uint256 confirmations;
14    }
15    
16    Transaction[] public transactions;
17    mapping(uint256 => mapping(address => bool)) public confirmations;
18    
19    modifier onlyOwner() {
20        require(isOwner(msg.sender), "Not owner");
21        _;
22    }
23    
24    constructor(address[] memory _owners, uint256 _required) {
25        require(_owners.length > 0, "Owners required");
26        require(_required > 0 && _required <= _owners.length, "Invalid required");
27        
28        owners = _owners;
29        required = _required;
30    }
31    
32    function submitTransaction(address _to, uint256 _value, bytes memory _data)
33        public
34        onlyOwner
35        returns (uint256)
36    {
37        uint256 txId = transactions.length;
38        transactions.push(Transaction({
39            to: _to,
40            value: _value,
41            data: _data,
42            executed: false,
43            confirmations: 0
44        }));
45        
46        return txId;
47    }
48    
49    function confirmTransaction(uint256 _txId) public onlyOwner {
50        require(!confirmations[_txId][msg.sender], "Already confirmed");
51        
52        confirmations[_txId][msg.sender] = true;
53        transactions[_txId].confirmations++;
54        
55        if (transactions[_txId].confirmations >= required) {
56            executeTransaction(_txId);
57        }
58    }
59    
60    function executeTransaction(uint256 _txId) internal {
61        Transaction storage txn = transactions[_txId];
62        require(!txn.executed, "Already executed");
63        require(txn.confirmations >= required, "Not enough confirmations");
64        
65        txn.executed = true;
66        (bool success, ) = txn.to.call{value: txn.value}(txn.data);
67        require(success, "Transaction failed");
68    }
69    
70    function isOwner(address _addr) public view returns (bool) {
71        for (uint256 i = 0; i < owners.length; i++) {
72            if (owners[i] == _addr) return true;
73        }
74        return false;
75    }
76}

 1from hashlib import sha256
 2import secrets
 3
 4class MuSig:
 5    def __init__(self, G, n):
 6        self.G = G  # Generator point
 7        self.n = n  # Curve order
 8    
 9    def aggregate_pubkeys(self, pubkeys):
10        """Aggregate public keys with coefficients"""
11        L = sha256(b''.join(sorted([pk.to_bytes() for pk in pubkeys]))).digest()
12        
13        agg_key = None
14        for pk in pubkeys:
15            coef = int.from_bytes(sha256(L + pk.to_bytes()).digest(), 'big') % self.n
16            weighted_pk = pk * coef
17            agg_key = weighted_pk if agg_key is None else agg_key + weighted_pk
18        
19        return agg_key
20    
21    def sign_partial(self, private_key, nonce, R_agg, P_agg, message):
22        """Generate partial signature"""
23        e = int.from_bytes(sha256(
24            R_agg.to_bytes() + P_agg.to_bytes() + message
25        ).digest(), 'big') % self.n
26        
27        # Compute coefficient for this key
28        L = sha256(b''.join(sorted([...]))).digest()
29        a = int.from_bytes(sha256(L + (private_key * self.G).to_bytes()).digest(), 'big') % self.n
30        
31        s = (nonce + e * private_key * a) % self.n
32        return s
33    
34    def aggregate_signatures(self, partial_sigs):
35        """Combine partial signatures"""
36        return sum(partial_sigs) % self.n
37    
38    def verify(self, signature, pubkey_agg, message):
39        """Verify aggregated signature"""
40        R, s = signature
41        e = int.from_bytes(sha256(
42            R.to_bytes() + pubkey_agg.to_bytes() + message
43        ).digest(), 'big') % self.n
44        
45        # Check: s·G = R + e·P
46        return s * self.G == R + e * pubkey_agg

 1✅ DO:
 2- Store keys in separate secure locations
 3- Use hardware wallets for high-value keys
 4- Document key holders and recovery procedures
 5- Regular key rotation for organizational multisig
 6
 7❌ DON'T:
 8- Store all keys in one location
 9- Share private keys via insecure channels
10- Use predictable key generation
11- Forget backup procedures

1P2SH Address (starts with 3):
23J98t1WpEZ73CNmYviecrnyiWrnqRhWNLy
3
4Redeem Script:
5OP_2 <pubkey1> <pubkey2> <pubkey3> OP_3 OP_CHECKMULTISIG

1Single signature on-chain:
2- Could be 1-of-1
3- Could be 2-of-2 MuSig
4- Could be 3-of-5 threshold
5→ All look identical (privacy!)

1✅ Choose appropriate m-of-n threshold
2✅ Generate keys on secure, offline devices
3✅ Verify all public keys before creating address
4✅ Test with small amounts first
5✅ Document key locations and holders
6✅ Create recovery procedures
7✅ Regular security audits
8✅ Plan for key holder unavailability

 1# Example: Secure multisig workflow
 2class SecureMultisigWorkflow:
 3    def __init__(self, threshold, total_keys):
 4        self.threshold = threshold
 5        self.total_keys = total_keys
 6        self.signatures = []
 7    
 8    def request_signature(self, tx_data, signer_id):
 9        """Request signature from specific signer"""
10        # 1. Display transaction details clearly
11        self.display_tx_details(tx_data)
12        
13        # 2. Require explicit confirmation
14        confirmed = self.require_confirmation(signer_id)
15        if not confirmed:
16            return None
17        
18        # 3. Sign on secure device
19        signature = self.sign_on_hardware_wallet(tx_data, signer_id)
20        
21        # 4. Verify signature immediately
22        if self.verify_signature(signature, tx_data, signer_id):
23            self.signatures.append(signature)
24            return signature
25        
26        return None
27    
28    def can_execute(self):
29        """Check if threshold met"""
30        return len(self.signatures) >= self.threshold

  1from hashlib import sha256
  2import hashlib
  3
  4class CosmosMultisig:
  5    def __init__(self, threshold, pubkeys):
  6        """
  7        Create Cosmos multisig account
  8        
  9        Args:
 10            threshold: Minimum signatures required
 11            pubkeys: List of secp256k1 public keys (33 bytes compressed)
 12        """
 13        self.threshold = threshold
 14        self.pubkeys = sorted(pubkeys)  # Sort for deterministic ordering
 15    
 16    def create_multisig_pubkey(self):
 17        """
 18        Create LegacyAminoPubKey structure
 19        
 20        Structure:
 21        {
 22          "@type": "/cosmos.crypto.multisig.LegacyAminoPubKey",
 23          "threshold": 2,
 24          "public_keys": [
 25            {"@type": "/cosmos.crypto.secp256k1.PubKey", "key": "base64..."},
 26            {"@type": "/cosmos.crypto.secp256k1.PubKey", "key": "base64..."},
 27            {"@type": "/cosmos.crypto.secp256k1.PubKey", "key": "base64..."}
 28          ]
 29        }
 30        """
 31        # Amino encoding for multisig pubkey
 32        amino_prefix = bytes([0x22, 0xC1, 0xF7, 0xE2])  # LegacyAminoPubKey prefix
 33        
 34        # Encode threshold (varint)
 35        threshold_bytes = self._encode_varint(self.threshold)
 36        
 37        # Encode each public key
 38        encoded_keys = []
 39        for pubkey in self.pubkeys:
 40            # secp256k1 pubkey amino prefix
 41            key_prefix = bytes([0xEB, 0x5A, 0xE9, 0x87, 0x21])
 42            # Length prefix + pubkey
 43            key_length = len(pubkey)
 44            encoded_key = key_prefix + bytes([key_length]) + pubkey
 45            encoded_keys.append(encoded_key)
 46        
 47        # Combine all
 48        multisig_data = amino_prefix + threshold_bytes
 49        for key in encoded_keys:
 50            multisig_data += key
 51        
 52        return multisig_data
 53    
 54    def derive_address(self, hrp="cosmos"):
 55        """
 56        Derive Cosmos multisig address
 57        
 58        Process:
 59        1. Create LegacyAminoPubKey
 60        2. SHA256 hash
 61        3. RIPEMD160 hash
 62        4. Bech32 encode with prefix
 63        
 64        Args:
 65            hrp: Human-readable part (e.g., "cosmos", "osmo", "juno")
 66        """
 67        # Get multisig pubkey
 68        multisig_pubkey = self.create_multisig_pubkey()
 69        
 70        # Hash: SHA256 then RIPEMD160
 71        sha_hash = sha256(multisig_pubkey).digest()
 72        ripemd = hashlib.new('ripemd160')
 73        ripemd.update(sha_hash)
 74        address_bytes = ripemd.digest()  # 20 bytes
 75        
 76        # Bech32 encode
 77        address = self._bech32_encode(hrp, address_bytes)
 78        
 79        return address
 80    
 81    def _encode_varint(self, n):
 82        """Encode integer as varint"""
 83        result = []
 84        while n > 0x7F:
 85            result.append((n & 0x7F) | 0x80)
 86            n >>= 7
 87        result.append(n & 0x7F)
 88        return bytes(result)
 89    
 90    def _bech32_encode(self, hrp, data):
 91        """Bech32 encoding (simplified)"""
 92        # In production, use proper bech32 library
 93        import bech32
 94        converted = bech32.convertbits(data, 8, 5)
 95        return bech32.bech32_encode(hrp, converted)
 96
 97# Example usage
 98pubkeys = [
 99    bytes.fromhex("02a1633cafcc01ebfb6d78e39f687a1f0995c62fc95f51ead10a02ee0be551b5dc"),
100    bytes.fromhex("03d31479e789014a96ba6dd60d50210045aa8292fe693f293d44615929f04cf57a"),
101    bytes.fromhex("0223a10d3e55c2b8c9b7e3e4c9f8e7d6c5b4a39281706f5e4d3c2b1a09f8e7d6c5")
102]
103
104multisig = CosmosMultisig(threshold=2, pubkeys=pubkeys)
105address = multisig.derive_address(hrp="cosmos")
106print(f"Multisig address: {address}")

 1# Create multisig account
 2gaiad keys add multisig_account \
 3    --multisig=alice,bob,charlie \
 4    --multisig-threshold=2
 5
 6# Output:
 7# - address: cosmos1abc...xyz
 8# - pubkey: '{"@type":"/cosmos.crypto.multisig.LegacyAminoPubKey",...}'
 9# - type: multi
10
11# Generate transaction
12gaiad tx bank send multisig_account cosmos1recipient... 1000uatom \
13    --from=multisig_account \
14    --generate-only > unsigned.json
15
16# Sign by first party
17gaiad tx sign unsigned.json \
18    --from=alice \
19    --multisig=cosmos1abc...xyz \
20    --sign-mode=amino-json \
21    --output-document=alice_sig.json
22
23# Sign by second party
24gaiad tx sign unsigned.json \
25    --from=bob \
26    --multisig=cosmos1abc...xyz \
27    --sign-mode=amino-json \
28    --output-document=bob_sig.json
29
30# Combine signatures
31gaiad tx multisign unsigned.json multisig_account \
32    alice_sig.json bob_sig.json > signed.json
33
34# Broadcast
35gaiad tx broadcast signed.json

 1class CosmosMultisigVerifier:
 2    def __init__(self, multisig_pubkey, threshold):
 3        self.multisig_pubkey = multisig_pubkey
 4        self.threshold = threshold
 5    
 6    def verify_transaction(self, tx_bytes, signatures):
 7        """
 8        Verify multisig transaction
 9        
10        Args:
11            tx_bytes: Transaction bytes to verify
12            signatures: List of (pubkey_index, signature) tuples
13        
14        Returns:
15            bool: True if valid
16        """
17        if len(signatures) < self.threshold:
18            return False
19        
20        # Verify each signature
21        valid_sigs = 0
22        seen_indices = set()
23        
24        for pubkey_index, signature in signatures:
25            # Check no duplicate signers
26            if pubkey_index in seen_indices:
27                return False
28            seen_indices.add(pubkey_index)
29            
30            # Get public key
31            pubkey = self.multisig_pubkey.public_keys[pubkey_index]
32            
33            # Verify signature
34            if self._verify_secp256k1(tx_bytes, signature, pubkey):
35                valid_sigs += 1
36        
37        return valid_sigs >= self.threshold
38    
39    def _verify_secp256k1(self, message, signature, pubkey):
40        """Verify secp256k1 signature"""
41        import ecdsa
42        
43        # Parse signature (r, s)
44        r = int.from_bytes(signature[:32], 'big')
45        s = int.from_bytes(signature[32:64], 'big')
46        
47        # Verify
48        vk = ecdsa.VerifyingKey.from_string(
49            pubkey, 
50            curve=ecdsa.SECP256k1,
51            hashfunc=sha256
52        )
53        
54        try:
55            return vk.verify_digest(
56                signature,
57                sha256(message).digest()
58            )
59        except:
60            return False

 1{
 2  "body": {
 3    "messages": [...],
 4    "memo": "",
 5    "timeout_height": "0",
 6    "extension_options": [],
 7    "non_critical_extension_options": []
 8  },
 9  "auth_info": {
10    "signer_infos": [
11      {
12        "public_key": {
13          "@type": "/cosmos.crypto.multisig.LegacyAminoPubKey",
14          "threshold": 2,
15          "public_keys": [
16            {"@type": "/cosmos.crypto.secp256k1.PubKey", "key": "..."},
17            {"@type": "/cosmos.crypto.secp256k1.PubKey", "key": "..."},
18            {"@type": "/cosmos.crypto.secp256k1.PubKey", "key": "..."}
19          ]
20        },
21        "mode_info": {
22          "multi": {
23            "bitarray": {
24              "extra_bits_stored": 3,
25              "elems": "Bg=="  // Bitarray indicating which keys signed
26            },
27            "mode_infos": [
28              {"single": {"mode": "SIGN_MODE_LEGACY_AMINO_JSON"}},
29              {"single": {"mode": "SIGN_MODE_LEGACY_AMINO_JSON"}}
30            ]
31          }
32        },
33        "sequence": "0"
34      }
35    ],
36    "fee": {...}
37  },
38  "signatures": [
39    "base64_signature_1",
40    "base64_signature_2"
41  ]
42}

 1def create_bitarray(total_keys, signing_indices):
 2    """
 3    Create compact bitarray for multisig
 4    
 5    Args:
 6        total_keys: Total number of keys in multisig
 7        signing_indices: List of indices that signed (0-based)
 8    
 9    Returns:
10        bytes: Compact bitarray
11    """
12    # Create bit array
13    bits = ['0'] * total_keys
14    for idx in signing_indices:
15        bits[idx] = '1'
16    
17    # Convert to bytes (big-endian)
18    bit_string = ''.join(bits)
19    
20    # Pad to byte boundary
21    while len(bit_string) % 8 != 0:
22        bit_string += '0'
23    
24    # Convert to bytes
25    byte_array = []
26    for i in range(0, len(bit_string), 8):
27        byte = int(bit_string[i:i+8], 2)
28        byte_array.append(byte)
29    
30    return bytes(byte_array)
31
32# Example: 3 keys, indices 0 and 2 signed
33# Bitarray: 101 -> 10100000 (padded) -> 0xA0
34bitarray = create_bitarray(3, [0, 2])
35print(f"Bitarray: {bitarray.hex()}")  # Output: a0

 1✅ DO:
 2- Always sort public keys deterministically
 3- Verify bitarray matches actual signatures
 4- Check for duplicate signers
 5- Validate threshold before accepting
 6- Use hardware wallets for signing
 7
 8❌ DON'T:
 9- Reuse signatures across transactions
10- Trust unverified multisig addresses
11- Skip signature verification
12- Allow threshold > total keys

 1// CW3 Multisig Contract (simplified)
 2pub struct Config {
 3    pub threshold: Threshold,
 4    pub total_weight: u64,
 5    pub max_voting_period: Duration,
 6}
 7
 8pub enum Threshold {
 9    AbsoluteCount { weight: u64 },
10    AbsolutePercentage { percentage: Decimal },
11    ThresholdQuorum { threshold: Decimal, quorum: Decimal },
12}
13
14// Proposal execution
15pub fn execute_proposal(
16    deps: DepsMut,
17    info: MessageInfo,
18    proposal_id: u64,
19) -> Result<Response, ContractError> {
20    let mut prop = proposals().load(deps.storage, proposal_id)?;
21    
22    // Check if threshold met
23    if prop.votes.yes < config.threshold {
24        return Err(ContractError::NotPassed {});
25    }
26    
27    // Execute
28    prop.status = Status::Passed;
29    proposals().save(deps.storage, proposal_id, &prop)?;
30    
31    Ok(Response::new()
32        .add_messages(prop.msgs)
33        .add_attribute("action", "execute")
34        .add_attribute("proposal_id", proposal_id.to_string()))
35}