"""
SEP: 0011
Title: Txrep: human-readable low-level representation of Stellar transactions
Author: David Mazières
Status: Active
Created: 2018-08-31
"""
import json
from decimal import Decimal
from enum import Enum
from typing import List, Union, Optional, Dict
from .. import xdr as stellar_xdr
from ..asset import Asset
from ..exceptions import ValueError as SdkValueError
from ..fee_bump_transaction import FeeBumpTransaction
from ..fee_bump_transaction_envelope import FeeBumpTransactionEnvelope
from ..memo import *
from ..operation import *
from ..price import Price
from ..signer import Signer
from ..strkey import StrKey
from ..time_bounds import TimeBounds
from ..transaction import Transaction
from ..transaction_envelope import TransactionEnvelope
__all__ = ["to_txrep", "from_txrep"]
_true = "true"
_false = "false"
class _EnvelopeType(Enum):
ENVELOPE_TYPE_TX_V0 = "ENVELOPE_TYPE_TX_V0"
ENVELOPE_TYPE_TX = "ENVELOPE_TYPE_TX"
ENVELOPE_TYPE_TX_FEE_BUMP = "ENVELOPE_TYPE_TX_FEE_BUMP"
[docs]def to_txrep(
transaction_envelope: Union[TransactionEnvelope, FeeBumpTransactionEnvelope],
) -> str:
"""Generate a human-readable format for Stellar transactions.
MuxAccount is currently not supported.
Txrep is a human-readable representation of Stellar transactions that functions like an assembly language for XDR.
See `SEP-0011 <https://github.com/stellar/stellar-protocol/blob/master/ecosystem/sep-0011.md>`_
:param transaction_envelope: Transaction envelope object.
:return: A human-readable format for Stellar transactions.
"""
is_fee_bump = isinstance(transaction_envelope, FeeBumpTransactionEnvelope)
tx_type = _EnvelopeType.ENVELOPE_TYPE_TX_FEE_BUMP.value
if not is_fee_bump:
assert isinstance(transaction_envelope, TransactionEnvelope)
if transaction_envelope.transaction.v1:
tx_type = _EnvelopeType.ENVELOPE_TYPE_TX.value
else:
tx_type = _EnvelopeType.ENVELOPE_TYPE_TX_V0.value
prefix = "feeBump.tx.innerTx.tx." if is_fee_bump else "tx."
transaction = transaction_envelope.transaction
if is_fee_bump:
assert isinstance(transaction_envelope, FeeBumpTransactionEnvelope)
fee_bump_transaction_envelope = transaction_envelope
fee_bump_transaction = fee_bump_transaction_envelope.transaction
transaction_envelope = fee_bump_transaction.inner_transaction_envelope
transaction = transaction_envelope.transaction
lines: List[str] = []
_add_line("type", tx_type, lines)
if is_fee_bump:
assert isinstance(fee_bump_transaction, FeeBumpTransaction)
assert isinstance(transaction, Transaction)
_add_line(
"feeBump.tx.feeSource", fee_bump_transaction.fee_source.public_key, lines
)
_add_line(
"feeBump.tx.fee",
fee_bump_transaction.base_fee * (len(transaction.operations) + 1),
lines,
)
_add_line(
"feeBump.tx.innerTx.type", _EnvelopeType.ENVELOPE_TYPE_TX.value, lines
)
assert isinstance(transaction, Transaction)
_add_line(f"{prefix}sourceAccount", transaction.source.public_key, lines)
_add_line(f"{prefix}fee", transaction.fee, lines)
_add_line(f"{prefix}seqNum", transaction.sequence, lines)
_add_time_bounds(transaction.time_bounds, prefix, lines)
_add_memo(transaction.memo, prefix, lines)
_add_operations(transaction.operations, prefix, lines)
_add_line(f"{prefix}ext.v", 0, lines)
_add_signatures(
transaction_envelope.signatures,
"feeBump.tx.innerTx." if is_fee_bump else "",
lines,
)
if is_fee_bump:
_add_line("feeBump.tx.ext.v", 0, lines)
_add_signatures(fee_bump_transaction_envelope.signatures, "feeBump.", lines)
return "\n".join(lines)
# Setting to ignore pass in .coveragerc will cause this function to not be counted by pytest.
[docs]def from_txrep(
txrep: str, network_passphrase: str
) -> Union[TransactionEnvelope, FeeBumpTransactionEnvelope]:
"""Parse txrep and generate transaction envelope object.
MuxAccount is currently not supported.
Txrep is a human-readable representation of Stellar transactions that functions like an assembly language for XDR.
See `SEP-0011 <https://github.com/stellar/stellar-protocol/blob/master/ecosystem/sep-0011.md>`_
:param txrep: a human-readable format for Stellar transactions.
:param network_passphrase: The network to connect, you do not need to set this value at this
time, it is reserved for future use.
:return: A human-readable format for Stellar transactions.
"""
raw_data_map = _get_raw_data_map(txrep)
tx_type = _EnvelopeType(_get_value(raw_data_map, "type"))
is_fee_bump = True if tx_type == _EnvelopeType.ENVELOPE_TYPE_TX_FEE_BUMP else False
prefix = "feeBump.tx.innerTx.tx." if is_fee_bump else "tx."
source = _get_value(raw_data_map, f"{prefix}sourceAccount")
fee = _get_int_value(raw_data_map, f"{prefix}fee")
sequence = _get_int_value(raw_data_map, f"{prefix}seqNum")
time_bounds = _get_time_bounds(raw_data_map, prefix)
memo = _get_memo(raw_data_map, prefix)
operations = _get_operations(raw_data_map, prefix)
prefix = "feeBump.tx.innerTx." if is_fee_bump else ""
transaction_signatures = _get_signatures(raw_data_map, prefix)
v1 = (
False
if not is_fee_bump and tx_type == _EnvelopeType.ENVELOPE_TYPE_TX_V0
else True
)
transaction = Transaction(
source=source,
sequence=sequence,
fee=fee,
operations=operations,
memo=memo,
time_bounds=time_bounds,
v1=v1,
)
transaction_envelope = TransactionEnvelope(
transaction=transaction,
signatures=transaction_signatures,
network_passphrase=network_passphrase,
)
if is_fee_bump:
fee_bump_fee_source = _get_value(raw_data_map, "feeBump.tx.feeSource")
fee_bump_fee = _get_int_value(raw_data_map, "feeBump.tx.fee")
fee_bump_base_fee = int(fee_bump_fee / (len(operations) + 1))
fee_bump_transaction_signatures = _get_signatures(raw_data_map, "feeBump.")
fee_bump_transaction = FeeBumpTransaction(
fee_source=fee_bump_fee_source,
base_fee=fee_bump_base_fee,
inner_transaction_envelope=transaction_envelope,
)
fee_bump_transaction_envelope = FeeBumpTransactionEnvelope(
transaction=fee_bump_transaction,
signatures=fee_bump_transaction_signatures,
network_passphrase=network_passphrase,
)
return fee_bump_transaction_envelope
return transaction_envelope
def _get_operations(raw_data_map: Dict[str, str], prefix: str) -> List[Operation]:
operations = []
operation_length = _get_int_value(raw_data_map, f"{prefix}operations.len")
for i in range(operation_length):
operation = _get_operation(i, raw_data_map, prefix)
operations.append(operation)
return operations
def _get_raw_data_map(txrep: str) -> Dict[str, str]:
lines = txrep.strip().split("\n")
raw_data_map = {}
for line in lines:
if line.startswith(":") or len(line.strip()) == 0:
# remove full-line comment and blank line
continue
parts = line.split(":", 1)
if len(parts) == 2:
key = parts[0].strip()
value = _remove_comment(parts[1])
raw_data_map[key] = value
return raw_data_map
def _get_time_bounds(raw_data_map: Dict[str, str], prefix: str) -> Optional[TimeBounds]:
time_bounds_present = _get_bool_value(raw_data_map, f"{prefix}timeBounds._present")
time_bounds = None
if time_bounds_present:
min_time = _get_int_value(raw_data_map, f"{prefix}timeBounds.minTime")
max_time = _get_int_value(raw_data_map, f"{prefix}timeBounds.maxTime")
time_bounds = TimeBounds(min_time=min_time, max_time=max_time)
return time_bounds
def _get_memo(raw_data_map: Dict[str, str], prefix: str) -> Memo:
memo_type = _get_value(raw_data_map, f"{prefix}memo.type")
if memo_type == "MEMO_TEXT":
return TextMemo(_get_bytes_value(raw_data_map, f"{prefix}memo.text"))
elif memo_type == "MEMO_ID":
return IdMemo(_get_int_value(raw_data_map, f"{prefix}memo.id"))
elif memo_type == "MEMO_HASH":
return HashMemo(_get_bytes_value(raw_data_map, f"{prefix}memo.hash"))
elif memo_type == "MEMO_RETURN":
return ReturnHashMemo(_get_bytes_value(raw_data_map, f"{prefix}memo.retHash"))
elif memo_type == "MEMO_NONE":
return NoneMemo()
else:
raise SdkValueError(
f"`{memo_type}` is not a valid memo type, expected one of `MEMO_TEXT`, `MEMO_ID`, "
f"`MEMO_HASH`, `MEMO_RETURN`, `MEMO_NONE`."
)
def _remove_comment(value: str) -> str:
value = value.strip()
if len(value) == 0:
return value
if value[0] == '"':
return _remove_string_value_comment(value)
return _remove_non_string_value_comment(value)
def _remove_non_string_value_comment(value: str) -> str:
parts = value.split(" ")
return parts[0]
def _remove_string_value_comment(value: str) -> str:
v = '"'
in_escape_sequence = False
for char in value[1:]:
if in_escape_sequence:
if char == "n":
v += "\n"
else:
v += char
in_escape_sequence = False
elif char == "\\":
in_escape_sequence = True
elif char == '"':
v += char
break
else:
v += char
return v
def _get_signature(
index: int, raw_data_map: Dict[str, str], prefix: str
) -> stellar_xdr.DecoratedSignature:
hint = _get_bytes_value(raw_data_map, f"{prefix}signatures[{index}].hint")
signature = _get_bytes_value(raw_data_map, f"{prefix}signatures[{index}].signature")
return stellar_xdr.DecoratedSignature(
stellar_xdr.SignatureHint(hint), stellar_xdr.Signature(signature)
)
def _get_signatures(
raw_data_map: Dict[str, str], prefix: str
) -> List[stellar_xdr.DecoratedSignature]:
signatures: List[stellar_xdr.DecoratedSignature] = []
signature_length = _get_int_value(raw_data_map, f"{prefix}signatures.len")
for i in range(signature_length):
signature = _get_signature(i, raw_data_map, prefix)
signatures.append(signature)
return signatures
def _get_operation(index, raw_data_map, tx_prefix):
prefix = f"{tx_prefix}operations[{index}].body."
source_account_id = None
if _get_bool_value(
raw_data_map, f"{tx_prefix}operations[{index}].sourceAccount._present"
):
source_account_id = _get_value(
raw_data_map, f"{tx_prefix}operations[{index}].sourceAccount"
)
operation_type = _get_value(raw_data_map, f"{prefix}type")
if operation_type == _to_caps_with_under(AccountMerge.__name__):
return _get_account_merge_op(source_account_id, tx_prefix, raw_data_map, index)
elif operation_type == _to_caps_with_under(AllowTrust.__name__):
operation_prefix = prefix + "allowTrustOp."
return _get_allow_trust_op(source_account_id, operation_prefix, raw_data_map)
elif operation_type == _to_caps_with_under(BumpSequence.__name__):
operation_prefix = prefix + "bumpSequenceOp."
return _get_bump_sequence_op(source_account_id, operation_prefix, raw_data_map)
elif operation_type == _to_caps_with_under(ChangeTrust.__name__):
operation_prefix = prefix + "changeTrustOp."
return _get_change_trust_op(source_account_id, operation_prefix, raw_data_map)
elif operation_type == _to_caps_with_under(CreateAccount.__name__):
operation_prefix = prefix + "createAccountOp."
return _get_create_account_op(source_account_id, operation_prefix, raw_data_map)
elif operation_type == _to_caps_with_under(CreatePassiveSellOffer.__name__):
operation_prefix = prefix + "createPassiveSellOfferOp."
return _get_create_passive_sell_offer_op(
source_account_id, operation_prefix, raw_data_map
)
elif operation_type == _to_caps_with_under(Inflation.__name__):
return _get_inflation_op(source_account_id)
elif operation_type == _to_caps_with_under(ManageBuyOffer.__name__):
operation_prefix = prefix + "manageBuyOfferOp."
return _get_manage_buy_offer_op(
source_account_id, operation_prefix, raw_data_map
)
elif operation_type == _to_caps_with_under(ManageData.__name__):
operation_prefix = prefix + "manageDataOp."
return _get_manage_data_op(source_account_id, operation_prefix, raw_data_map)
elif operation_type == _to_caps_with_under(ManageSellOffer.__name__):
operation_prefix = prefix + "manageSellOfferOp."
return _get_manage_sell_offer_op(
source_account_id, operation_prefix, raw_data_map
)
elif operation_type == _to_caps_with_under(PathPaymentStrictReceive.__name__):
operation_prefix = prefix + "pathPaymentStrictReceiveOp."
return _get_path_payment_strict_receive_op(
source_account_id, operation_prefix, raw_data_map
)
elif operation_type == _to_caps_with_under(PathPaymentStrictSend.__name__):
operation_prefix = prefix + "pathPaymentStrictSendOp."
return _get_path_payment_strict_send_op(
source_account_id, operation_prefix, raw_data_map
)
elif operation_type == _to_caps_with_under(Payment.__name__):
operation_prefix = prefix + "paymentOp."
return _get_payment_op(source_account_id, operation_prefix, raw_data_map)
elif operation_type == _to_caps_with_under(SetOptions.__name__):
operation_prefix = prefix + "setOptionsOp."
return _get_set_options_op(source_account_id, operation_prefix, raw_data_map)
else:
raise SdkValueError(
f"This operation has not been implemented yet, "
f"operation type: {operation_type}."
)
def _get_set_options_op(
source: str, operation_prefix: str, raw_data_map: Dict[str, str]
) -> SetOptions:
inflation_dest = None
clear_flags = None
set_flags = None
master_weight = None
low_threshold = None
med_threshold = None
high_threshold = None
signer = None
home_domain = None
if _get_bool_value(raw_data_map, f"{operation_prefix}inflationDest._present"):
inflation_dest = _get_value(raw_data_map, f"{operation_prefix}inflationDest")
if _get_bool_value(raw_data_map, f"{operation_prefix}clearFlags._present"):
clear_flags = _get_int_value(raw_data_map, f"{operation_prefix}clearFlags")
if _get_bool_value(raw_data_map, f"{operation_prefix}setFlags._present"):
set_flags = _get_int_value(raw_data_map, f"{operation_prefix}setFlags")
if _get_bool_value(raw_data_map, f"{operation_prefix}masterWeight._present"):
master_weight = _get_int_value(raw_data_map, f"{operation_prefix}masterWeight")
if _get_bool_value(raw_data_map, f"{operation_prefix}lowThreshold._present"):
low_threshold = _get_int_value(raw_data_map, f"{operation_prefix}lowThreshold")
if _get_bool_value(raw_data_map, f"{operation_prefix}medThreshold._present"):
med_threshold = _get_int_value(raw_data_map, f"{operation_prefix}medThreshold")
if _get_bool_value(raw_data_map, f"{operation_prefix}highThreshold._present"):
high_threshold = _get_int_value(
raw_data_map, f"{operation_prefix}highThreshold"
)
if _get_bool_value(raw_data_map, f"{operation_prefix}signer._present"):
weight = _get_int_value(raw_data_map, f"{operation_prefix}signer.weight")
key = _get_value(raw_data_map, f"{operation_prefix}signer.key")
if key.startswith("G"):
signer = Signer.ed25519_public_key(key, weight)
elif key.startswith("X"):
sha256_hash = StrKey.decode_sha256_hash(key)
signer = Signer.sha256_hash(sha256_hash, weight)
elif key.startswith("T"):
pre_auth_tx_hash = StrKey.decode_pre_auth_tx(key)
signer = Signer.pre_auth_tx(pre_auth_tx_hash, weight)
else:
raise SdkValueError("Signer key should start with `G`, `X` or `T`.")
if _get_bool_value(raw_data_map, f"{operation_prefix}homeDomain._present"):
home_domain = _get_string_value(raw_data_map, f"{operation_prefix}homeDomain")
return SetOptions(
inflation_dest=inflation_dest,
clear_flags=clear_flags,
set_flags=set_flags,
master_weight=master_weight,
low_threshold=low_threshold,
med_threshold=med_threshold,
high_threshold=high_threshold,
signer=signer,
home_domain=home_domain,
source=source,
)
def _get_path_payment_strict_receive_op(
source: str, operation_prefix: str, raw_data_map: Dict[str, str]
) -> PathPaymentStrictReceive:
send_asset = _get_asset(raw_data_map, f"{operation_prefix}sendAsset")
send_max = _get_amount_value(raw_data_map, f"{operation_prefix}sendMax")
destination = _get_value(raw_data_map, f"{operation_prefix}destination")
dest_asset = _get_asset(raw_data_map, f"{operation_prefix}destAsset")
dest_amount = _get_amount_value(raw_data_map, f"{operation_prefix}destAmount")
path_length = _get_int_value(raw_data_map, f"{operation_prefix}path.len")
path = []
for i in range(path_length):
asset = _get_asset(raw_data_map, f"{operation_prefix}path[{i}]")
path.append(asset)
return PathPaymentStrictReceive(
destination=destination,
send_asset=send_asset,
send_max=send_max,
dest_asset=dest_asset,
dest_amount=dest_amount,
path=path,
source=source,
)
def _get_path_payment_strict_send_op(
source: str, operation_prefix: str, raw_data_map: Dict[str, str]
) -> PathPaymentStrictSend:
send_asset = _get_asset(raw_data_map, f"{operation_prefix}sendAsset")
send_amount = _get_amount_value(raw_data_map, f"{operation_prefix}sendAmount")
destination = _get_value(raw_data_map, f"{operation_prefix}destination")
dest_asset = _get_asset(raw_data_map, f"{operation_prefix}destAsset")
dest_min = _get_amount_value(raw_data_map, f"{operation_prefix}destMin")
path_length = _get_int_value(raw_data_map, f"{operation_prefix}path.len")
path = []
for i in range(path_length):
asset = _get_asset(raw_data_map, f"{operation_prefix}path[{i}]")
path.append(asset)
return PathPaymentStrictSend(
destination=destination,
send_asset=send_asset,
send_amount=send_amount,
dest_asset=dest_asset,
dest_min=dest_min,
path=path,
source=source,
)
def _get_create_passive_sell_offer_op(
source: str, operation_prefix: str, raw_data_map: Dict[str, str]
) -> CreatePassiveSellOffer:
selling = _get_asset(raw_data_map, f"{operation_prefix}selling")
buying = _get_asset(raw_data_map, f"{operation_prefix}buying")
amount = _get_amount_value(raw_data_map, f"{operation_prefix}amount")
price_n = _get_int_value(raw_data_map, f"{operation_prefix}price.n")
price_d = _get_int_value(raw_data_map, f"{operation_prefix}price.d")
price = Price(n=price_n, d=price_d)
return CreatePassiveSellOffer(
selling=selling, buying=buying, amount=amount, price=price, source=source
)
def _get_manage_buy_offer_op(
source: str, operation_prefix: str, raw_data_map: Dict[str, str]
) -> ManageBuyOffer:
selling = _get_asset(raw_data_map, f"{operation_prefix}selling")
buying = _get_asset(raw_data_map, f"{operation_prefix}buying")
amount = _get_amount_value(raw_data_map, f"{operation_prefix}buyAmount")
offer_id = _get_int_value(raw_data_map, f"{operation_prefix}offerID")
price_n = _get_int_value(raw_data_map, f"{operation_prefix}price.n")
price_d = _get_int_value(raw_data_map, f"{operation_prefix}price.d")
price = Price(n=price_n, d=price_d)
return ManageBuyOffer(
selling=selling,
buying=buying,
amount=amount,
price=price,
offer_id=offer_id,
source=source,
)
def _get_manage_sell_offer_op(
source: str, operation_prefix: str, raw_data_map: Dict[str, str]
) -> ManageSellOffer:
selling = _get_asset(raw_data_map, f"{operation_prefix}selling")
buying = _get_asset(raw_data_map, f"{operation_prefix}buying")
amount = _get_amount_value(raw_data_map, f"{operation_prefix}amount")
offer_id = _get_int_value(raw_data_map, f"{operation_prefix}offerID")
price_n = _get_int_value(raw_data_map, f"{operation_prefix}price.n")
price_d = _get_int_value(raw_data_map, f"{operation_prefix}price.d")
price = Price(n=price_n, d=price_d)
return ManageSellOffer(
selling=selling,
buying=buying,
amount=amount,
price=price,
offer_id=offer_id,
source=source,
)
def _get_payment_op(
source: str, operation_prefix: str, raw_data_map: Dict[str, str]
) -> Payment:
destination = _get_value(raw_data_map, f"{operation_prefix}destination")
asset = _get_asset(raw_data_map, f"{operation_prefix}asset")
amount = _get_amount_value(raw_data_map, f"{operation_prefix}amount")
return Payment(destination=destination, asset=asset, amount=amount, source=source)
def _get_manage_data_op(
source: str, operation_prefix: str, raw_data_map: Dict[str, str]
) -> ManageData:
data_name = _get_string_value(raw_data_map, f"{operation_prefix}dataName")
data_value = None
if _get_bool_value(raw_data_map, f"{operation_prefix}dataValue._present"):
data_value = _get_bytes_value(raw_data_map, f"{operation_prefix}dataValue")
return ManageData(data_name=data_name, data_value=data_value, source=source)
def _get_bump_sequence_op(
source: str, operation_prefix: str, raw_data_map: Dict[str, str]
) -> BumpSequence:
bump_to = _get_int_value(raw_data_map, f"{operation_prefix}bumpTo")
return BumpSequence(bump_to=bump_to, source=source)
def _get_allow_trust_op(
source: str, operation_prefix: str, raw_data_map: Dict[str, str]
) -> AllowTrust:
trustor = _get_value(raw_data_map, f"{operation_prefix}trustor")
asset_code = _get_value(raw_data_map, f"{operation_prefix}asset")
authorize = _get_bool_value(raw_data_map, f"{operation_prefix}authorize")
return AllowTrust(
trustor=trustor, asset_code=asset_code, authorize=authorize, source=source
)
def _get_change_trust_op(
source: str, operation_prefix: str, raw_data_map: Dict[str, str]
) -> ChangeTrust:
line = _get_asset(raw_data_map, f"{operation_prefix}line")
limit = _get_amount_value(raw_data_map, f"{operation_prefix}limit")
return ChangeTrust(asset=line, limit=limit, source=source)
def _get_inflation_op(source: str) -> Inflation:
return Inflation(source=source)
def _get_account_merge_op(
source: str, transaction_prefix: str, raw_data_map: Dict[str, str], index: int
) -> AccountMerge:
destination = _get_value(
raw_data_map, f"{transaction_prefix}operations[{index}].body.destination"
)
return AccountMerge(destination=destination, source=source)
def _get_create_account_op(
source: str, operation_prefix: str, raw_data_map: Dict[str, str]
) -> CreateAccount:
destination = _get_value(raw_data_map, f"{operation_prefix}destination")
starting_balance = _get_amount_value(
raw_data_map, f"{operation_prefix}startingBalance"
)
return CreateAccount(
destination=destination, starting_balance=starting_balance, source=source
)
def _get_asset(raw_data_map: Dict[str, str], key: str) -> Asset:
return _decode_asset(_get_value(raw_data_map, key))
def _get_amount_value(raw_data_map: Dict[str, str], key: str) -> str:
value = _get_int_value(raw_data_map, key)
return Operation.from_xdr_amount(value)
def _get_int_value(raw_data_map: Dict[str, str], key: str) -> int:
value = _get_value(raw_data_map, key)
try:
return int(value)
except ValueError as e:
raise SdkValueError(f"Failed to convert `{value}` to int type.") from e
def _get_bool_value(raw_data_map: Dict[str, str], key: str) -> bool:
value = _get_value(raw_data_map, key)
if value == _true:
return True
elif value == _false:
return False
else:
raise SdkValueError(f"Failed to convert `{value}` to bool type.")
def _get_bytes_value(raw_data_map: Dict[str, str], key: str) -> bytes:
value = _get_value(raw_data_map, key)
if value[0] == '"':
# for text memo.
return _get_string_value(raw_data_map, key).encode()
try:
return bytes.fromhex(value)
except ValueError as e:
raise SdkValueError(f"Failed to convert `{value}` to bytes type.") from e
def _get_string_value(raw_data_map: Dict[str, str], key: str) -> str:
value = _get_value(raw_data_map, key)
if len(value) == 0:
return value
return value[1:-1]
def _get_value(raw_data_map: Dict[str, str], key: str) -> str:
try:
return raw_data_map[key]
except KeyError as e:
raise SdkValueError(f"`{key}` is missing from txrep.") from e
def _decode_asset(asset: str) -> Asset:
# native (or any string up to 12 characters not containing an unescaped colon) for the native asset
if ":" not in asset and len(asset) <= 12:
return Asset.native()
parts = asset.split(":")
if len(parts) != 2:
raise SdkValueError("Failed to decode asset string.")
return Asset(parts[0], parts[1])
def _add_line(key: str, value: Union[str, int], lines: List[str]) -> None:
lines.append(f"{key}: {value}")
def _add_time_bounds(
time_bounds: Optional[TimeBounds], prefix: str, lines: List[str]
) -> None:
if time_bounds is None:
_add_line(f"{prefix}timeBounds._present", _false, lines)
else:
_add_line(f"{prefix}timeBounds._present", _true, lines)
_add_line(f"{prefix}timeBounds.minTime", time_bounds.min_time, lines)
_add_line(f"{prefix}timeBounds.maxTime", time_bounds.max_time, lines)
def _add_memo(memo: Memo, prefix: str, lines: List[str]) -> None:
if isinstance(memo, NoneMemo):
_add_line(f"{prefix}memo.type", "MEMO_NONE", lines)
if isinstance(memo, TextMemo):
_add_line(f"{prefix}memo.type", "MEMO_TEXT", lines)
# I don't think we should decode it.
_add_line(f"{prefix}memo.text", _to_string(memo.memo_text), lines)
if isinstance(memo, IdMemo):
_add_line(f"{prefix}memo.type", "MEMO_ID", lines)
_add_line(f"{prefix}memo.id", memo.memo_id, lines)
if isinstance(memo, HashMemo):
_add_line(f"{prefix}memo.type", "MEMO_HASH", lines)
_add_line(f"{prefix}memo.hash", _to_opaque(memo.memo_hash), lines)
if isinstance(memo, ReturnHashMemo):
_add_line(f"{prefix}memo.type", "MEMO_RETURN", lines)
_add_line(f"{prefix}memo.retHash", _to_opaque(memo.memo_return), lines)
def _add_operations(operations: List[Operation], prefix: str, lines: List[str]) -> None:
_add_line(f"{prefix}operations.len", len(operations), lines)
for index, operation in enumerate(operations):
_add_operation(index, operation, prefix, lines)
def _add_operation(
index: int, operation: Operation, prefix: str, lines: List[str]
) -> None:
prefix = f"{prefix}operations[{index}]."
operation_type = operation.__class__.__name__
def add_operation_line(key: str, value: Union[str, int]) -> None:
_add_line(f"{prefix}{key}", value, lines)
if operation.source is not None:
add_operation_line("sourceAccount._present", _true)
add_operation_line("sourceAccount", operation.source)
else:
add_operation_line("sourceAccount._present", _false)
add_operation_line("body.type", _to_caps_with_under(operation_type))
def add_body_line(
key: str, value: Union[str, int, None], optional: bool = False
) -> None:
operation_type = operation.__class__.__name__
key = f"body.{_to_camel_case(operation_type)}Op.{key}"
if optional:
present = True if value is not None else False
add_operation_line(f"{key}._present", _true if present else _false)
if present:
assert value is not None
add_operation_line(key, value)
else:
assert value is not None
add_operation_line(key, value)
def add_signer(signer: Optional[Signer]) -> None:
add_body_line("signer._present", _false if signer is None else _true)
if signer is None:
return
if (
signer.signer_key.signer_key.type
== stellar_xdr.SignerKeyType.SIGNER_KEY_TYPE_ED25519
):
assert signer.signer_key.signer_key.ed25519 is not None
add_body_line(
"signer.key",
StrKey.encode_ed25519_public_key(
signer.signer_key.signer_key.ed25519.uint256
),
)
if (
signer.signer_key.signer_key.type
== stellar_xdr.SignerKeyType.SIGNER_KEY_TYPE_PRE_AUTH_TX
):
assert signer.signer_key.signer_key.pre_auth_tx is not None
add_body_line(
"signer.key",
StrKey.encode_pre_auth_tx(
signer.signer_key.signer_key.pre_auth_tx.uint256
),
)
if (
signer.signer_key.signer_key.type
== stellar_xdr.SignerKeyType.SIGNER_KEY_TYPE_HASH_X
):
assert signer.signer_key.signer_key.hash_x is not None
add_body_line(
"signer.key",
StrKey.encode_sha256_hash(signer.signer_key.signer_key.hash_x.uint256),
)
add_body_line("signer.weight", signer.weight)
def add_price(price: Union[Price, str, Decimal]) -> None:
price = _to_price(price)
add_body_line("price.n", price.n)
add_body_line("price.d", price.d)
def add_home_domain(home_domain: Optional[str]) -> None:
if home_domain is None:
add_body_line("homeDomain", None, True)
else:
add_body_line("homeDomain", _to_string(home_domain), True)
if isinstance(operation, CreateAccount):
add_body_line("destination", operation.destination)
add_body_line("startingBalance", _to_amount(operation.starting_balance))
elif isinstance(operation, Payment):
add_body_line("destination", operation.destination)
add_body_line("asset", _to_asset(operation.asset))
add_body_line("amount", _to_amount(operation.amount))
elif isinstance(operation, PathPaymentStrictReceive):
add_body_line("sendAsset", _to_asset(operation.send_asset))
add_body_line("sendMax", _to_amount(operation.send_max))
add_body_line("destination", operation.destination)
add_body_line("destAsset", _to_asset(operation.dest_asset))
add_body_line("destAmount", _to_amount(operation.dest_amount))
add_body_line("path.len", len(operation.path))
for index, asset in enumerate(operation.path):
add_body_line(f"path[{index}]", _to_asset(asset))
elif isinstance(operation, ManageSellOffer):
add_body_line("selling", _to_asset(operation.selling))
add_body_line("buying", _to_asset(operation.buying))
add_body_line("amount", _to_amount(operation.amount))
add_price(operation.price)
add_body_line("offerID", operation.offer_id)
elif isinstance(operation, CreatePassiveSellOffer):
add_body_line("selling", _to_asset(operation.selling))
add_body_line("buying", _to_asset(operation.buying))
add_body_line("amount", _to_amount(operation.amount))
add_price(operation.price)
elif isinstance(operation, SetOptions):
add_body_line("inflationDest", operation.inflation_dest, True)
add_body_line("clearFlags", operation.clear_flags, True)
add_body_line("setFlags", operation.set_flags, True)
add_body_line("masterWeight", operation.master_weight, True)
add_body_line("lowThreshold", operation.low_threshold, True)
add_body_line("medThreshold", operation.med_threshold, True)
add_body_line("highThreshold", operation.high_threshold, True)
add_home_domain(operation.home_domain)
add_signer(operation.signer)
elif isinstance(operation, ChangeTrust):
add_body_line("line", _to_asset(operation.asset))
add_body_line("limit", _to_amount(operation.limit))
elif isinstance(operation, AllowTrust):
add_body_line("trustor", operation.trustor)
add_body_line("asset", operation.asset_code)
add_body_line("authorize", _true if operation.authorize else _false)
elif isinstance(operation, AccountMerge):
# AccountMerge does not include 'accountMergeOp' prefix
# see https://github.com/StellarCN/py-stellar-base/blob/master/.xdr/Stellar-transaction.x#L282
add_operation_line("body.destination", operation.destination)
elif isinstance(operation, ManageData):
add_body_line("dataName", _to_string(operation.data_name))
if operation.data_value is None:
add_body_line("dataValue._present", _false)
else:
add_body_line("dataValue._present", _true)
add_body_line("dataValue", _to_opaque(operation.data_value))
elif isinstance(operation, BumpSequence):
add_body_line("bumpTo", operation.bump_to)
elif isinstance(operation, ManageBuyOffer):
add_body_line("selling", _to_asset(operation.selling))
add_body_line("buying", _to_asset(operation.buying))
add_body_line("buyAmount", _to_amount(operation.amount))
add_price(operation.price)
add_body_line("offerID", operation.offer_id)
elif isinstance(operation, PathPaymentStrictSend):
add_body_line("sendAsset", _to_asset(operation.send_asset))
add_body_line("sendAmount", _to_amount(operation.send_amount))
add_body_line("destination", operation.destination)
add_body_line("destAsset", _to_asset(operation.dest_asset))
add_body_line("destMin", _to_amount(operation.dest_min))
add_body_line("path.len", len(operation.path))
for index, asset in enumerate(operation.path):
add_body_line(f"path[{index}]", _to_asset(asset))
elif isinstance(operation, Inflation):
pass
else:
raise SdkValueError(
f"This operation has not been implemented yet, "
f"operation type: {operation}."
)
def _add_signatures(
signatures: List[stellar_xdr.DecoratedSignature], prefix: str, lines: List[str]
) -> None:
_add_line(f"{prefix}signatures.len", len(signatures), lines)
for index, signature in enumerate(signatures):
_add_signature(index, signature, prefix, lines)
def _add_signature(
index: int, signature: stellar_xdr.DecoratedSignature, prefix: str, lines: List[str]
) -> None:
prefix = f"{prefix}signatures[{index}]."
_add_line(f"{prefix}hint", _to_opaque(signature.hint.signature_hint), lines)
_add_line(f"{prefix}signature", _to_opaque(signature.signature.signature), lines)
def _to_asset(asset: Asset) -> str:
if asset.is_native():
return "native"
return f"{asset.code}:{asset.issuer}"
def _to_amount(amount: Union[Decimal, str]) -> int:
return Operation.to_xdr_amount(amount)
def _to_price(price: Union[Price, str, Decimal]) -> Price:
if isinstance(price, Price):
price_fraction = price
else:
price_fraction = Price.from_raw_price(price)
return price_fraction
def _to_camel_case(cap_words: str) -> str:
return cap_words[0].lower() + cap_words[1:]
def _to_caps_with_under(word):
return "".join(["_" + c if c.isupper() else c for c in word]).lstrip("_").upper()
def _to_string(value: Union[str, bytes]) -> str:
if isinstance(value, str):
return json.dumps(value)
# We are not following the standard here, it needs more discussion.
try:
value = value.decode("utf-8")
except UnicodeDecodeError:
assert isinstance(value, bytes)
return _to_opaque(value)
return json.dumps(value)
def _to_opaque(value: Union[bytes]) -> str:
return value.hex()