wallet_w5.py

from pytoniq import LiteClientLike, Slice, BaseWallet, WalletError, LiteBalancer
from pytoniq_core.crypto.keys import mnemonic_is_valid, mnemonic_to_private_key
from pytoniq_core.crypto.signature import sign_message
from pytoniq_core.boc import Cell, Builder
from pytoniq_core.tlb.custom.wallet import WalletV4Data, WalletMessage, TlbScheme
import typing
import time

class WalletV5Data(TlbScheme):
    """
    wallet_v5_data#_ is_signature_allowed:uint1 seqno:uint32 wallet_id:uint32 public_key:bits256 extensions_dict:(Maybe ^Cell) = WalletV5Data;
    """
    def __init__(self,
                 is_signature_allowed: bool = True,
                 seqno: typing.Optional[int] = 0,
                 wallet_id: typing.Optional[int] = None,
                 public_key: typing.Optional[bytes] = None,
                 extensions_dict: typing.Optional[Cell] = None
                 ):
        self.is_signature_allowed = is_signature_allowed
        self.seqno = seqno
        if wallet_id is None:
            wallet_id = 698983191
        self.wallet_id = wallet_id
        if public_key is None:
            raise Exception('Public Key required for Wallet!')
        self.public_key = public_key
        self.extensions_dict = extensions_dict

    def serialize(self) -> Cell:
        builder = Builder()
        builder \
            .store_uint(int(self.is_signature_allowed), 1) \
            .store_uint(self.seqno, 32) \
            .store_uint(self.wallet_id, 32) \
            .store_bytes(self.public_key) \
            .store_dict(self.extensions_dict)
        return builder.end_cell()

    @classmethod
    def deserialize(cls, cell_slice: Slice):
        return cls(is_signature_allowed=bool(cell_slice.load_uint(1)), seqno=cell_slice.load_uint(32), wallet_id=cell_slice.load_uint(32),
                   public_key=cell_slice.load_bytes(32), extensions_dict=cell_slice.load_maybe_ref())



WALLET_V5_CODE = Cell.one_from_boc(
    "b5ee9c7201021401000281000114ff00f4a413f4bcf2c80b01020120020302014804050102f20e02dcd020d749c120915b8f6320d70b1f2082106578746ebd21821073696e74bdb0925f03e082106578746eba8eb48020d72101d074d721fa4030fa44f828fa443058bd915be0ed44d0810141d721f4058307f40e6fa1319130e18040d721707fdb3ce03120d749810280b99130e070e2100f020120060702012008090019be5f0f6a2684080a0eb90fa02c02016e0a0b0201480c0d0019adce76a2684020eb90eb85ffc00019af1df6a2684010eb90eb858fc00017b325fb51341c75c875c2c7e00011b262fb513435c28020011e20d70b1f82107369676ebaf2e08a7f0f01e68ef0eda2edfb218308d722028308d723208020d721d31fd31fd31fed44d0d200d31f20d31fd3ffd70a000af90140ccf9109a28945f0adb31e1f2c087df02b35007b0f2d0845125baf2e0855036baf2e086f823bbf2d0882292f800de01a47fc8ca00cb1f01cf16c9ed542092f80fde70db3cd81003f6eda2edfb02f404216e926c218e4c0221d73930709421c700b38e2d01d72820761e436c20d749c008f2e09320d74ac002f2e09320d71d06c712c2005230b0f2d089d74cd7393001a4e86c128407bbf2e093d74ac000f2e093ed55e2d20001c000915be0ebd72c08142091709601d72c081c12e25210b1e30f20d74a111213009601fa4001fa44f828fa443058baf2e091ed44d0810141d718f405049d7fc8ca0040048307f453f2e08b8e14038307f45bf2e08c22d70a00216e01b3b0f2d090e2c85003cf1612f400c9ed54007230d72c08248e2d21f2e092d200ed44d0d2005113baf2d08f54503091319c01810140d721d70a00f2e08ee2c8ca0058cf16c9ed5493f2c08de20010935bdb31e1d74cd0"
)


def generate_wallet_id(
        wallet_id: int,
        wc: int = 0,
        wallet_version: int = 0,
        network_global_id: int = -239,
) -> int:
    """
    Generates a wallet ID based on global ID, workchain, wallet version, and wallet id.
    :param wallet_id: The subwallet ID (16-bit unsigned integer).
    :param wc: The workchain value (8-bit signed integer).
    :param wallet_version: The wallet version (8-bit unsigned integer).
    :param network_global_id: The network global ID (32-bit signed integer).
    """
    ctx = 0
    ctx |= 1 << 31
    ctx |= (wc & 0xFF) << 23
    ctx |= (wallet_version & 0xFF) << 15
    ctx |= (wallet_id & 0xFFFF)

    return ctx ^ (network_global_id & 0xFFFFFFFF)


class WalletV5(BaseWallet):

    @classmethod
    async def from_code_and_data(cls, provider: LiteClientLike, code: Cell, public_key: bytes, wc: int = 0,
                                 is_signature_allowed: bool = True, global_id: int = -239, wallet_version: int = 0,
                                 wallet_id: typing.Optional[int] = 2147483409, extensions_dict: typing.Optional[Cell] = None,
                                 **kwargs):
        data = cls.create_data_cell(public_key, wallet_id, wc, is_signature_allowed, global_id, wallet_version,
                                    extensions_dict)
        return await super().from_code_and_data(provider, wc, code, data, **kwargs)

    @staticmethod
    def create_data_cell(public_key: bytes, wallet_id: typing.Optional[int] = 2147483409, wc: int = 0,
                         is_signature_allowed: bool = True, global_id: int = -239, wallet_version: int = 0,
                         extensions_dict: typing.Optional[Cell] = None) -> Cell:
        if wallet_id is None:
            wallet_id = generate_wallet_id(0, wc, wallet_version, global_id)
        return WalletV5Data(is_signature_allowed=is_signature_allowed, seqno=0, wallet_id=wallet_id,
                            public_key=public_key, extensions_dict=extensions_dict).serialize()

    @classmethod
    async def from_mnemonic(cls, provider: LiteClientLike, mnemonics: typing.Union[list, str], wc: int = 0,
                            wallet_id: typing.Optional[int] = 2147483409, version: str = 'v3r2'):
        if isinstance(mnemonics, str):
            mnemonics = mnemonics.split()
        assert mnemonic_is_valid(mnemonics), 'mnemonics are invalid!'
        public_key, _ = mnemonic_to_private_key(mnemonics)
        return await cls.from_code_and_data(provider, WALLET_V5_CODE, public_key)

    async def raw_transfer(self, msgs: typing.List[WalletMessage], seqno_from_get_meth: bool = True):
        """
        :param msgs: list of WalletMessages. to create one call create_wallet_internal_message meth
        :param seqno_from_get_meth: if True LiteClient will request seqno get method and use it, otherwise seqno from contract data will be taken
        """
        assert len(msgs) <= 255, 'for wallet v5 maximum messages amount is 255'
        if 'private_key' not in self.__dict__:
            raise WalletError('must specify wallet private key!')

        if seqno_from_get_meth:
            seqno = await self.get_seqno()
        else:
            seqno = self.seqno
        transfer_msg = self.raw_create_transfer_msg(private_key=self.private_key, seqno=seqno, wallet_id=self.wallet_id, messages=msgs)

        return await self.send_external(body=transfer_msg)

    @staticmethod
    def raw_create_transfer_msg(private_key: bytes, seqno: int, wallet_id: int, messages: typing.List[WalletMessage],
                                valid_until: typing.Optional[int] = None, op_code: int = 0x7369676e) -> Cell:
        signing_message = Builder().store_uint(op_code, 32)
        signing_message.store_uint(wallet_id, 32)
        if seqno == 0:
            signing_message.store_bits('1' * 32)  # bin(2**32 - 1)
        else:
            if valid_until is not None:
                signing_message.store_uint(valid_until, 32)
            else:
                signing_message.store_uint(int(time.time()) + 60, 32)
        signing_message.store_uint(seqno, 32)
        signing_message.store_cell(WalletV5.pack_actions(messages))
        signing_message = signing_message.end_cell()
        signature = sign_message(signing_message.hash, private_key)
        return Builder() \
            .store_cell(signing_message) \
            .store_bytes(signature) \
            .end_cell()

    @staticmethod
    def pack_actions(messages: typing.List[WalletMessage]) -> Cell:
        """
        Packs a list of wallet messages into a single Cell.
        :param messages: A list of WalletMessage instances to pack.
        :return: A Cell containing the packed messages.
        """
        list_cell = Cell.empty()

        for msg in messages:
            msg = Builder() \
                .store_uint(0x0ec3c86d, 32) \
                .store_uint(msg.send_mode, 8) \
                .store_ref(msg.message.serialize()) \
                .end_cell()
            list_cell = Builder() \
                .store_ref(list_cell) \
                .store_cell(msg) \
                .end_cell()

        return Builder() \
            .store_uint(1, 1) \
            .store_ref(list_cell) \
            .store_uint(0, 1) \
            .end_cell()

    @property
    def is_signature_allowed(self) -> int:
        """
        :return: seqno taken from contract data
        """
        return WalletV5Data.deserialize(self.state.data.begin_parse()).is_signature_allowed

    @property
    def seqno(self) -> int:
        """
        :return: seqno taken from contract data
        """
        return WalletV5Data.deserialize(self.state.data.begin_parse()).seqno

    @property
    def wallet_id(self) -> int:
        """
        :return: wallet_id taken from contract data
        """
        return WalletV5Data.deserialize(self.state.data.begin_parse()).wallet_id

    @property
    def public_key(self) -> bytes:
        """
        :return: public_key taken from contract data
        """
        return WalletV5Data.deserialize(self.state.data.begin_parse()).public_key

    @property
    def extensions_dict(self) -> Cell:
        return WalletV4Data.deserialize(self.state.data.begin_parse()).extensions_dict