Module libra.txnmetadata

LIP-4 Transaction Metadata Utilities

This module implements utility functions for application to create transaction metadata and metadata signature. See https://lip.libra.org/lip-4 for more details

Expand source code
# Copyright (c) The Libra Core Contributors
# SPDX-License-Identifier: Apache-2.0


"""LIP-4 Transaction Metadata Utilities

This module implements utility functions for application to create transaction metadata and metadata signature.
See https://lip.libra.org/lip-4 for more details
"""


from dataclasses import dataclass
import typing

from . import libra_types, serde_types, lcs, jsonrpc, utils


class InvalidEventMetadataForRefundError(Exception):
    pass


@dataclass
class Attest:

    metadata: libra_types.Metadata
    sender_address: libra_types.AccountAddress
    amount: serde_types.uint64  # pyre-ignore

    def lcs_serialize(self) -> bytes:
        return lcs.serialize(self, Attest)


def travel_rule(
    off_chain_reference_id: str, sender_address: libra_types.AccountAddress, amount: int
) -> typing.Tuple[bytes, bytes]:
    """Create travel rule metadata bytes and signature message bytes.

    This is used for peer to peer transfer between 2 custodial accounts.
    """

    metadata = libra_types.Metadata__TravelRuleMetadata(
        value=libra_types.TravelRuleMetadata__TravelRuleMetadataVersion0(
            value=libra_types.TravelRuleMetadataV0(off_chain_reference_id=off_chain_reference_id)
        )
    )

    # receiver_lcs_data = lcs(metadata, sender_address, amount) + "@@$$LIBRA_ATTEST$$@@" /*ASCII-encoded string*/
    attest = Attest(metadata=metadata, sender_address=sender_address, amount=serde_types.uint64(amount))  # pyre-ignore
    signing_msg = attest.lcs_serialize() + b"@@$$LIBRA_ATTEST$$@@"

    return (metadata.lcs_serialize(), signing_msg)


def general_metadata(
    from_subaddress: typing.Optional[bytes] = None,
    to_subaddress: typing.Optional[bytes] = None,
    referenced_event: typing.Optional[int] = None,
) -> bytes:
    """Create general metadata for peer to peer transaction script

    Use this function to create metadata with from and to sub-addresses for peer to peer transfer
    from custodial account to custodial account under travel rule threshold.

    Give from_subaddress None for the case transferring from non-custodial to custodial account.
    Give to_subaddress None for the case transferring from custodial to non-custodial account.

    Returns empty bytes array if from_subaddress and to_subaddress both are None.
    """

    if from_subaddress is None and to_subaddress is None:
        return b""

    metadata = libra_types.Metadata__GeneralMetadata(
        value=libra_types.GeneralMetadata__GeneralMetadataVersion0(
            value=libra_types.GeneralMetadataV0(  # pyre-ignore
                from_subaddress=from_subaddress,
                to_subaddress=to_subaddress,
                referenced_event=serde_types.uint64(referenced_event) if referenced_event else None,
            )
        )
    )
    return metadata.lcs_serialize()


def find_refund_reference_event(
    txn: typing.Optional[jsonrpc.Transaction], receiver: typing.Union[libra_types.AccountAddress, str]
) -> typing.Optional[jsonrpc.Event]:
    """Find refund reference event from given transaction

    The event can be used as reference is the "receivedpayment" event.
    We also only return event that receiver address matches given reciever address, because
    it is possible we may have mutliple receivers for one transaction in the future.

    Returns None if given transaction is None or the event not found.
    If this function returns an event, then you may call `refund_metadata_from_event` function
    to create refund metadata for the refund transaction.
    """

    if txn is None:
        return None

    address = utils.account_address_hex(receiver)
    for event in txn.events:
        if event.data.type == "receivedpayment" and event.data.receiver == address:
            return event

    return None


def refund_metadata_from_event(event: jsonrpc.Event) -> typing.Optional[bytes]:
    """create refund metadat for the event

    The given event should be the reference event for the refund, it should have metadata describes
    the payment details.
    May call `find_refund_reference_event` function to find reference event from a peer to peer transfer
    transaction.

    Returns empty bytes array if given event metadata is None or empty string, this is for the case
    the peer to peer transaction is a non-custodial to non-custodial account, which does not require
    metadata, hence the refund transaction should not have metadata too.

    Raises InvalidEventMetadataForRefundError if metadata can't be decoded as
    libra_types.GeneralMetadata__GeneralMetadataVersion0 for creating the refund metadata
    """

    if not event.data.metadata:
        return b""

    try:
        metadata_bytes = bytes.fromhex(event.data.metadata)
        metadata = libra_types.Metadata.lcs_deserialize(metadata_bytes)

        if isinstance(metadata, libra_types.Metadata__GeneralMetadata):
            if isinstance(metadata.value, libra_types.GeneralMetadata__GeneralMetadataVersion0):
                gmv0 = metadata.value.value
                return general_metadata(gmv0.to_subaddress, gmv0.from_subaddress, event.sequence_number)

            raise InvalidEventMetadataForRefundError("unknown metadata type: {metadata}")

        raise InvalidEventMetadataForRefundError(f"unknown metadata type: {metadata}")
    except ValueError as e:
        raise InvalidEventMetadataForRefundError(f"invalid event metadata for refund: {e}, event: {event}")

Functions

def find_refund_reference_event(txn: Union[jsonrpc_pb2.Transaction, NoneType], receiver: Union[AccountAddress, str]) ‑> Union[jsonrpc_pb2.Event, NoneType]

Find refund reference event from given transaction

The event can be used as reference is the "receivedpayment" event. We also only return event that receiver address matches given reciever address, because it is possible we may have mutliple receivers for one transaction in the future.

Returns None if given transaction is None or the event not found. If this function returns an event, then you may call refund_metadata_from_event() function to create refund metadata for the refund transaction.

Expand source code
def find_refund_reference_event(
    txn: typing.Optional[jsonrpc.Transaction], receiver: typing.Union[libra_types.AccountAddress, str]
) -> typing.Optional[jsonrpc.Event]:
    """Find refund reference event from given transaction

    The event can be used as reference is the "receivedpayment" event.
    We also only return event that receiver address matches given reciever address, because
    it is possible we may have mutliple receivers for one transaction in the future.

    Returns None if given transaction is None or the event not found.
    If this function returns an event, then you may call `refund_metadata_from_event` function
    to create refund metadata for the refund transaction.
    """

    if txn is None:
        return None

    address = utils.account_address_hex(receiver)
    for event in txn.events:
        if event.data.type == "receivedpayment" and event.data.receiver == address:
            return event

    return None
def general_metadata(from_subaddress: Union[bytes, NoneType] = None, to_subaddress: Union[bytes, NoneType] = None, referenced_event: Union[int, NoneType] = None) ‑> bytes

Create general metadata for peer to peer transaction script

Use this function to create metadata with from and to sub-addresses for peer to peer transfer from custodial account to custodial account under travel rule threshold.

Give from_subaddress None for the case transferring from non-custodial to custodial account. Give to_subaddress None for the case transferring from custodial to non-custodial account.

Returns empty bytes array if from_subaddress and to_subaddress both are None.

Expand source code
def general_metadata(
    from_subaddress: typing.Optional[bytes] = None,
    to_subaddress: typing.Optional[bytes] = None,
    referenced_event: typing.Optional[int] = None,
) -> bytes:
    """Create general metadata for peer to peer transaction script

    Use this function to create metadata with from and to sub-addresses for peer to peer transfer
    from custodial account to custodial account under travel rule threshold.

    Give from_subaddress None for the case transferring from non-custodial to custodial account.
    Give to_subaddress None for the case transferring from custodial to non-custodial account.

    Returns empty bytes array if from_subaddress and to_subaddress both are None.
    """

    if from_subaddress is None and to_subaddress is None:
        return b""

    metadata = libra_types.Metadata__GeneralMetadata(
        value=libra_types.GeneralMetadata__GeneralMetadataVersion0(
            value=libra_types.GeneralMetadataV0(  # pyre-ignore
                from_subaddress=from_subaddress,
                to_subaddress=to_subaddress,
                referenced_event=serde_types.uint64(referenced_event) if referenced_event else None,
            )
        )
    )
    return metadata.lcs_serialize()
def refund_metadata_from_event(event: jsonrpc_pb2.Event) ‑> Union[bytes, NoneType]

create refund metadat for the event

The given event should be the reference event for the refund, it should have metadata describes the payment details. May call find_refund_reference_event() function to find reference event from a peer to peer transfer transaction.

Returns empty bytes array if given event metadata is None or empty string, this is for the case the peer to peer transaction is a non-custodial to non-custodial account, which does not require metadata, hence the refund transaction should not have metadata too.

Raises InvalidEventMetadataForRefundError if metadata can't be decoded as libra_types.GeneralMetadata__GeneralMetadataVersion0 for creating the refund metadata

Expand source code
def refund_metadata_from_event(event: jsonrpc.Event) -> typing.Optional[bytes]:
    """create refund metadat for the event

    The given event should be the reference event for the refund, it should have metadata describes
    the payment details.
    May call `find_refund_reference_event` function to find reference event from a peer to peer transfer
    transaction.

    Returns empty bytes array if given event metadata is None or empty string, this is for the case
    the peer to peer transaction is a non-custodial to non-custodial account, which does not require
    metadata, hence the refund transaction should not have metadata too.

    Raises InvalidEventMetadataForRefundError if metadata can't be decoded as
    libra_types.GeneralMetadata__GeneralMetadataVersion0 for creating the refund metadata
    """

    if not event.data.metadata:
        return b""

    try:
        metadata_bytes = bytes.fromhex(event.data.metadata)
        metadata = libra_types.Metadata.lcs_deserialize(metadata_bytes)

        if isinstance(metadata, libra_types.Metadata__GeneralMetadata):
            if isinstance(metadata.value, libra_types.GeneralMetadata__GeneralMetadataVersion0):
                gmv0 = metadata.value.value
                return general_metadata(gmv0.to_subaddress, gmv0.from_subaddress, event.sequence_number)

            raise InvalidEventMetadataForRefundError("unknown metadata type: {metadata}")

        raise InvalidEventMetadataForRefundError(f"unknown metadata type: {metadata}")
    except ValueError as e:
        raise InvalidEventMetadataForRefundError(f"invalid event metadata for refund: {e}, event: {event}")
def travel_rule(off_chain_reference_id: str, sender_address: AccountAddress, amount: int) ‑> Tuple[bytes, bytes]

Create travel rule metadata bytes and signature message bytes.

This is used for peer to peer transfer between 2 custodial accounts.

Expand source code
def travel_rule(
    off_chain_reference_id: str, sender_address: libra_types.AccountAddress, amount: int
) -> typing.Tuple[bytes, bytes]:
    """Create travel rule metadata bytes and signature message bytes.

    This is used for peer to peer transfer between 2 custodial accounts.
    """

    metadata = libra_types.Metadata__TravelRuleMetadata(
        value=libra_types.TravelRuleMetadata__TravelRuleMetadataVersion0(
            value=libra_types.TravelRuleMetadataV0(off_chain_reference_id=off_chain_reference_id)
        )
    )

    # receiver_lcs_data = lcs(metadata, sender_address, amount) + "@@$$LIBRA_ATTEST$$@@" /*ASCII-encoded string*/
    attest = Attest(metadata=metadata, sender_address=sender_address, amount=serde_types.uint64(amount))  # pyre-ignore
    signing_msg = attest.lcs_serialize() + b"@@$$LIBRA_ATTEST$$@@"

    return (metadata.lcs_serialize(), signing_msg)

Classes

class Attest (metadata: Metadata, sender_address: AccountAddress, amount: numpy.uint64)

Attest(metadata: libra.libra_types.Metadata, sender_address: libra.libra_types.AccountAddress, amount: numpy.uint64)

Expand source code
class Attest:

    metadata: libra_types.Metadata
    sender_address: libra_types.AccountAddress
    amount: serde_types.uint64  # pyre-ignore

    def lcs_serialize(self) -> bytes:
        return lcs.serialize(self, Attest)

Class variables

var amount : numpy.uint64
var metadataMetadata
var sender_addressAccountAddress

Methods

def lcs_serialize(self) ‑> bytes
Expand source code
def lcs_serialize(self) -> bytes:
    return lcs.serialize(self, Attest)
class InvalidEventMetadataForRefundError (*args, **kwargs)

Common base class for all non-exit exceptions.

Expand source code
class InvalidEventMetadataForRefundError(Exception):
    pass

Ancestors

  • builtins.Exception
  • builtins.BaseException