add multi-sign

Signed-off-by: linfeng-crypto <linfeng@crypto.com>

chainlibpy/amino/transaction.py

@@ -0,0 +1,96 @@
+import base64
+import hashlib
+import json
+from typing import List
+
+import ecdsa
+
+from chainlibpy.amino import StdFee, StdSignDoc, SyncMode
+from chainlibpy.amino.message import Msg
+from chainlibpy.amino.tx import Pubkey, Signature, StdTx
+from chainlibpy.wallet import Wallet
+
+class Transaction:
+ """A Cosmos transaction.
+
+ After initialization, one or more token transfers can be added by
+ calling the `add_transfer()` method. Finally, call `get_pushable()`
+ to get a signed transaction that can be pushed to the `POST /txs`
+ endpoint of the Cosmos REST API.
+ """
+
+ def __init__(
+ self,
+ wallet: Wallet,
+ account_num: int,
+ sequence: int,
+ fee: StdFee = StdFee.default(),
+ memo: str = "",
+ chain_id: str = "crypto-org-chain-mainnet-1",
+ sync_mode: SyncMode = "sync",
+ timeout_height: int = 0,
+ multi_sign_address: str = None,
+ ) -> None:
+ self._multi_sign_address = multi_sign_address
+ self._wallet = wallet
+ self._account_num = str(account_num)
+ self._sequence = str(sequence)
+ self._fee = fee
+ self._memo = memo
+ self._chain_id = chain_id
+ self._sync_mode = sync_mode
+ self._msgs: List[Msg] = []
+ self._timeout_height = str(timeout_height)
+
+ def add_msg(self, msg: Msg):
+ self._msgs.append(msg)
+
+ def get_pushable(self) -> dict:
+ """get the request post to the /txs."""
+ std_tx = StdTx(self._msgs, self._fee, self._memo, self._timeout_height, [self.signature])
+
+ pushable_tx = {
+ "tx": std_tx.to_dict(),
+ "mode": self._sync_mode,
+ }
+ return pushable_tx
+
+ @property
+ def signature(self) -> Signature:
+ pubkey = self._wallet.public_key
+ base64_pubkey = base64.b64encode(pubkey).decode("utf-8")
+ pubkey = Pubkey(value=base64_pubkey)
+ signature = Signature(self._sign(), pubkey, self._account_num, self._sequence)
+ return signature
+
+ def _sign(self) -> str:
+ sign_doc = self._get_sign_doc()
+ message_str = json.dumps(
+ sign_doc.to_dict(), separators=(",", ":"), sort_keys=True
+ )
+ message_bytes = message_str.encode("utf-8")
+
+ privkey = ecdsa.SigningKey.from_string(
+ self._wallet.private_key, curve=ecdsa.SECP256k1
+ )
+ signature_compact = privkey.sign_deterministic(
+ message_bytes,
+ hashfunc=hashlib.sha256,
+ sigencode=ecdsa.util.sigencode_string_canonize,
+ )
+
+ print(list(signature_compact))
+ signature_base64_str = base64.b64encode(signature_compact).decode("utf-8")
+ return signature_base64_str
+
+ def _get_sign_doc(self) -> StdSignDoc:
+ sign_doc = StdSignDoc(
+ account_number=self._account_num,
+ sequence=self._sequence,
+ chain_id=self._chain_id,
+ memo=self._memo,
+ fee=self._fee.to_dict(),
+ msgs=self._msgs,
+ timeout_height=self._timeout_height,
+ )
+ return sign_doc

chainlibpy/generated/cosmos/crypto/multisig/keys_pb2_grpc.py

@@ -0,0 +1,4 @@
+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+"""Client and server classes corresponding to protobuf-defined services."""
+import grpc
+

chainlibpy/grpc_client.py

@@ -7,6 +7,8 @@
 from google.protobuf.any_pb2 import Any as ProtoAny
 from grpc import ChannelCredentials, insecure_channel, secure_channel
 
+from chainlibpy.generated.cosmos.crypto.multisig.keys_pb2 import LegacyAminoPubKey
+from chainlibpy.generated.cosmos.crypto.multisig.v1beta1.multisig_pb2 import CompactBitArray, MultiSignature
 from chainlibpy.generated.cosmos.auth.v1beta1.auth_pb2 import BaseAccount
 from chainlibpy.generated.cosmos.auth.v1beta1.query_pb2 import QueryAccountRequest
 from chainlibpy.generated.cosmos.auth.v1beta1.query_pb2_grpc import (
@@ -41,17 +43,18 @@
  TxBody,
 )
 from chainlibpy.transaction import sign_transaction
-from chainlibpy.wallet import Wallet
+from chainlibpy.multisign.signature import MultiSignatureData, SingleSignatureV2
+
 
 
 class GrpcClient:
  DEFAULT_GAS_LIMIT = 200000
 
  def __init__(
  self,
- wallet: Wallet,
  chain_id: str,
  grpc_endpoint: str,
+ account_number: int = None,
  credentials: ChannelCredentials = None,
  ) -> None:
  if credentials is None:
@@ -62,10 +65,8 @@ def __init__(
  self.bank_client = BankGrpcClient(channel)
  self.tx_client = TxGrpcClient(channel)
  self.auth_client = AuthGrpcClient(channel)
- self.wallet = wallet
  self.chain_id = chain_id
- account = self.query_account_data(self.wallet.address)
- self.account_number = account.account_number
+ self.account_number = account_number
 
  def get_balance(self, address: str, denom: str) -> QueryBalanceResponse:
  res = self.bank_client.Balance(QueryBalanceRequest(address=address, denom=denom))
@@ -108,12 +109,10 @@ def generate_tx(
  tx = Tx(body=tx_body, auth_info=auth_info)
  return tx
 
- def sign_tx(self, tx: Tx):
- sign_transaction(tx, self.wallet.private_key, self.chain_id, self.account_number)
+ def sign_tx(self, private_key: bytes, tx: Tx):
+ sign_transaction(tx, private_key, self.chain_id, self.account_number)
 
- def get_packed_send_msg(
- self, from_address: str, to_address: str, amount: List[Coin]
- ) -> ProtoAny:
+ def get_packed_send_msg(self, from_address: str, to_address: str, amount: List[Coin]) -> ProtoAny:
  msg_send = MsgSend(from_address=from_address, to_address=to_address, amount=amount)
  send_msg_packed = ProtoAny()
  send_msg_packed.Pack(msg_send, type_url_prefix="/")
@@ -136,16 +135,17 @@ def broadcast_tx(self, tx: Tx, wait_time: int = 10) -> GetTxResponse:
 
  return tx_response
 
- def bank_send(self, to_address: str, amount: List[Coin]) -> GetTxResponse:
+ def bank_send(self, from_address: str, public_key: bytes, to_address: str, amount: List[Coin]) -> GetTxResponse:
+ import pdb; pdb.set_trace()
  msg = self.get_packed_send_msg(
- from_address=self.wallet.address, to_address=to_address, amount=amount
+ from_address=from_address, to_address=to_address, amount=amount
  )
 
- tx = self.generate_tx([msg], [self.wallet.address], [self.wallet.public_key])
+ tx = self.generate_tx([msg], [from_address], [public_key])
  self.sign_tx(tx)
  return self.broadcast_tx(tx)
 
- def _get_signer_info(self, from_acc: BaseAccount, pub_key: bytes) -> SignerInfo:
+ def _get_signer_info(self, from_acc: BaseAccount, pub_key: ProtoPubKey) -> SignerInfo:
  from_pub_key_packed = ProtoAny()
  from_pub_key_pb = ProtoPubKey(key=pub_key)
  from_pub_key_packed.Pack(from_pub_key_pb, type_url_prefix="/")
@@ -159,3 +159,55 @@ def _get_signer_info(self, from_acc: BaseAccount, pub_key: bytes) -> SignerInfo:
  sequence=from_acc.sequence,
  )
  return signer_info
+
+def get_muli_signer_info(sequence: int, multi_pubkey: LegacyAminoPubKey, bitarray: CompactBitArray) -> SignerInfo:
+ multi_pubkey_packed = ProtoAny()
+ multi_pubkey_packed.Pack(multi_pubkey, type_url_prefix="/")
+
+ # Prepare auth info
+ mode_infos = [ModeInfo.Single(mode=SignMode.SIGN_MODE_LEGACY_AMINO_JSON)] * multi_pubkey.threshold
+ multi = ModeInfo.Multi(bitarray, mode_infos)
+ mode_info = ModeInfo(multi=multi)
+ signer_info = SignerInfo(
+ public_key=multi_pubkey_packed,
+ mode_info=mode_info,
+ sequence=sequence,
+ )
+ return signer_info
+
+def gen_multi_tx(
+ packed_msgs: List[ProtoAny],
+ multi_pubkey: LegacyAminoPubKey,
+ signature_batch: List[SingleSignatureV2],
+ sequence: int,
+ fee: Optional[List[Coin]] = None,
+ memo: str = "",
+ gas_limit: int = 2000,
+) -> Tx:
+ def packed_proto_pubkey(pub_key: ProtoPubKey):
+ from_pub_key_packed = ProtoAny()
+ return from_pub_key_packed.Pack(pub_key, type_url_prefix="/")
+
+ all_proto_pubkeys = [p for p in multi_pubkey.public_keys]
+ n = len(signature_batch)
+ multi_sign_data = MultiSignatureData(n)
+ for sig_v2 in signature_batch:
+ multi_sign_data.add_signature_v2(sig_v2, all_proto_pubkeys)
+
+ signer_info = get_muli_signer_info(sequence, multi_pubkey, multi_sign_data.bit_array)
+ signer_infos: List[SignerInfo] = []
+ signer_infos.append(signer_info)
+ auth_info = AuthInfo(
+ signer_infos=signer_infos,
+ fee=Fee(amount=fee, gas_limit=gas_limit),
+ )
+
+ tx_body = TxBody()
+ tx_body.memo = memo
+ tx_body.messages.extend(packed_msgs)
+ multi_signature = MultiSignature(signatures=multi_sign_data.signatures)
+ signatures = list()
+ signatures.append(multi_signature)
+ packed_pubkeys = [packed_proto_pubkey(p.pub_key) for p in signature_batch]
+ tx = Tx(body=tx_body, auth_info=auth_info, public_keys=packed_pubkeys, signatures=signatures)
+ return tx

chainlibpy/multisign/bitarray.py

@@ -0,0 +1,56 @@
+import base64
+from chainlibpy.amino.basic import BasicObj
+from chainlibpy.generated.cosmos.crypto.multisig.v1beta1.multisig_pb2 import CompactBitArray as ProtoCompactBitArray
+
+
+class CompactBitArray(BasicObj):
+ MaxInt32 = 2147483647
+
+ def __init__(self, bits: int):
+ self.extra_bits_stored = 0
+ self.elems = bytearray()
+ if bits <= 0:
+ raise Exception(f"invalid bits {bits}")
+ n_elems = (bits + 7) // 8
+ if n_elems <= 0 or n_elems > self.MaxInt32:
+ raise Exception(f"invalid bits {bits}")
+ self.extra_bits_stored = bits % 8
+ self.elems = bytearray([0] * n_elems)
+
+ def __repr__(self):
+ elems = base64.b64encode(self.elems).decode('utf-8')
+ return f"extra_bits_stored:{self.extra_bits_stored}, elems:{elems}"
+
+ def bit_array(self) -> ProtoCompactBitArray:
+ return ProtoCompactBitArray(extra_bits_storede=self.extra_bits_stored, elems=self.elems)
+
+ def count(self) -> int:
+ """
+ returns the number of bits in the bitarray
+ """
+ if self.extra_bits_stored == 0:
+ return len(self.elems) * 8
+ return (len(self.elems) - 1) * 8 + int(self.extra_bits_stored)
+
+ def get_index(self, index: int) -> bool:
+ """
+ returns the bit at index i within the bit array. The behavior is undefined if i >= bA.Count()
+ """
+ if index > self.count() or index < 0:
+ return False
+ return self.elems[index >> 3] & (1 << (7 - (index % 8))) > 0
+
+ def set_index(self, i: int, v: bool) -> bool:
+ """
+ set_index sets the bit at index i within the bit array. Returns true if and only if the
+ operation succeeded. The behavior is undefined if i >= self.count()
+ """
+ if i > self.count() or i < 0:
+ return False
+
+ index = i >> 3
+ if v:
+ self.elems[index] |= (1 << int(7 - (i % 8)))
+ else:
+ self.elems[i >> 3] &= ~(1 << int(7 - (i % 8)))
+ return True