Merge branch 'dev'

examples/txrep_demo.py

@@ -0,0 +1,31 @@
+from stellar_sdk.sep.txrep import from_txrep, to_txrep
+from stellar_sdk import Keypair, TransactionBuilder, Network, Account
+
+source_secret_key = "SBFZCHU5645DOKRWYBXVOXY2ELGJKFRX6VGGPRYUWHQ7PMXXJNDZFMKD"
+
+source_keypair = Keypair.from_secret(source_secret_key)
+source_public_key = source_keypair.public_key
+
+receiver_public_key = "GA7YNBW5CBTJZ3ZZOWX3ZNBKD6OE7A7IHUQVWMY62W2ZBG2SGZVOOPVH"
+source_account = Account(source_public_key, 12345)
+
+
+transaction = (
+    TransactionBuilder(
+        source_account=source_account,
+        network_passphrase=Network.TESTNET_NETWORK_PASSPHRASE,
+        base_fee=100,
+    )
+    .add_text_memo("Hello, Stellar!")
+    .append_payment_op(receiver_public_key, "350.1234567", "XLM")
+    .set_timeout(30)
+    .build()
+)
+
+transaction.sign(source_keypair)
+
+# convert transaction to txrep
+txrep = to_txrep(transaction)
+print(f"txrep: \n{txrep}")
+# convert txrep to transaction
+tx = from_txrep(txrep, Network.TESTNET_NETWORK_PASSPHRASE)

stellar_sdk/helpers.py

@@ -1,9 +1,8 @@
 from typing import Union
 
-from .exceptions import ValueError
 from .fee_bump_transaction_envelope import FeeBumpTransactionEnvelope
 from .transaction_envelope import TransactionEnvelope
-from .xdr import Xdr
+from .utils import is_fee_bump_transaction
 
 
 __all__ = ["parse_transaction_envelope_from_xdr"]
@@ -23,18 +22,6 @@ def parse_transaction_envelope_from_xdr(
     :raises: :exc:`ValueError <stellar_sdk.exceptions.ValueError>` - XDR is neither :py:class:`TransactionEnvelope <stellar_sdk.transaction_envelope.TransactionEnvelope>`
         nor :py:class:`FeeBumpTransactionEnvelope <stellar_sdk.fee_bump_transaction_envelope.FeeBumpTransactionEnvelope>`
     """
-    xdr_object = Xdr.types.TransactionEnvelope.from_xdr(xdr)
-    te_type = xdr_object.type
-    if te_type == Xdr.const.ENVELOPE_TYPE_TX_FEE_BUMP:
-        return FeeBumpTransactionEnvelope.from_xdr_object(
-            xdr_object, network_passphrase
-        )
-    elif (
-        te_type == Xdr.const.ENVELOPE_TYPE_TX
-        or te_type == Xdr.const.ENVELOPE_TYPE_TX_V0
-    ):
-        return TransactionEnvelope.from_xdr_object(xdr_object, network_passphrase)
-    else:
-        raise ValueError(
-            f"This transaction envelope type is not supported, type = {te_type}."
-        )
+    if is_fee_bump_transaction(xdr):
+        return FeeBumpTransactionEnvelope.from_xdr(xdr, network_passphrase)
+    return TransactionEnvelope.from_xdr(xdr, network_passphrase)

stellar_sdk/sep/stellar_uri.py

@@ -10,14 +10,16 @@
 
 import abc
 import base64
-from typing import Optional, List, Union
-from urllib.parse import urlencode, quote
+from typing import Optional, List, Union, Dict, Tuple
+from urllib import parse
 
 from ..asset import Asset
 from ..exceptions import ValueError
+from ..fee_bump_transaction_envelope import FeeBumpTransactionEnvelope
 from ..keypair import Keypair
 from ..memo import Memo, NoneMemo, IdMemo, TextMemo, HashMemo, ReturnHashMemo
 from ..transaction_envelope import TransactionEnvelope
+from ..utils import is_fee_bump_transaction
 
 __all__ = ["PayStellarUri", "TransactionStellarUri", "Replacement"]
 
@@ -52,6 +54,18 @@ def sign(self, signer: Union[Keypair, str]) -> None:
         signature = signer.sign(sign_data)
         self.signature = base64.b64encode(signature).decode()
 
+    @staticmethod
+    def _parse_uri_query(uri_query) -> Dict[str, str]:
+        return dict(parse.parse_qsl(uri_query))
+
+    @staticmethod
+    def _parse_callback(callback: str) -> Optional[str]:
+        if callback is None:
+            return None
+        if not callback.startswith("url:"):
+            raise ValueError("`callback` should start with `url:`.")
+        return callback[4:]
+
 
 class PayStellarUri(StellarUri):
     """A request for a payment to be signed.
@@ -95,32 +109,59 @@ def __init__(
         self.memo = None
         self.memo_type = None
         self._memo = memo
+        self.memo_type, self.memo = self._encode_memo(memo)
+        self.destination = destination
+        self.amount = amount
+        self.callback = callback
+        self.msg = message
+        self.network_passphrase = network_passphrase
+        self.origin_domain = origin_domain
+
+    @staticmethod
+    def _encode_memo(memo) -> Union[Tuple[str, str], Tuple[None, None]]:
         if memo and not isinstance(memo, NoneMemo):
             if isinstance(memo, TextMemo):
-                self.memo = memo.memo_text
-                self.memo_type = "MEMO_TEXT"
+                memo_value = memo.memo_text
+                memo_type = "MEMO_TEXT"
             elif isinstance(memo, IdMemo):
-                self.memo = memo.memo_id
-                self.memo_type = "MEMO_ID"
+                memo_value = memo.memo_id
+                memo_type = "MEMO_ID"
             elif isinstance(memo, HashMemo):
-                self.memo = base64.b64encode(memo.memo_hash).decode()
-                self.memo_type = "MEMO_HASH"
+                memo_value = base64.b64encode(memo.memo_hash).decode()
+                memo_type = "MEMO_HASH"
             elif isinstance(memo, ReturnHashMemo):
-                self.memo = base64.b64encode(memo.memo_return).decode()
-                self.memo_type = "MEMO_RETURN"
+                memo_value = base64.b64encode(memo.memo_return).decode()
+                memo_type = "MEMO_RETURN"
             else:
                 raise ValueError("Invalid memo.")
-        self.destination = destination
-        self.amount = amount
-        self.callback = callback
-        self.msg = message
-        self.network_passphrase = network_passphrase
-        self.origin_domain = origin_domain
+            return memo_type, memo_value
+        return None, None
+
+    @staticmethod
+    def _decode_memo(memo_type: str, memo_value: str) -> Optional[Memo]:
+        if memo_type is None:
+            return None
+        if memo_value is None:
+            raise ValueError("`memo` is missing from uri.")
+        if memo_type == "MEMO_TEXT":
+            return TextMemo(memo_value)
+        elif memo_type == "MEMO_ID":
+            return IdMemo(int(memo_value))
+        elif memo_type == "MEMO_HASH":
+            value = base64.b64decode(memo_value.encode())
+            return HashMemo(value)
+        elif memo_type == "MEMO_RETURN":
+            value = base64.b64decode(memo_value.encode())
+            return ReturnHashMemo(value)
+        else:
+            raise ValueError("Invalid `memo_type`.")
 
     def to_uri(self) -> str:
         """Generate the request URI.
+
+        :return: Stellar Pay URI.
         """
-        query_params = dict()
+        query_params = {}
         query_params["destination"] = self.destination
         if self.amount is not None:
             query_params["amount"] = self.amount
@@ -142,7 +183,55 @@ def to_uri(self) -> str:
             query_params["origin_domain"] = self.origin_domain
         if self.signature is not None:
             query_params["signature"] = self.signature
-        return f"{STELLAR_SCHEME}:pay?{urlencode(query_params, quote_via=quote)}"
+        return f"{STELLAR_SCHEME}:pay?{parse.urlencode(query_params, quote_via=parse.quote)}"
+
+    @classmethod
+    def from_uri(cls, uri: str) -> "PayStellarUri":
+        """Parse Stellar Pay URI and generate :class:`PayStellarUri` object.
+
+        :param uri: Stellar Pay URI.
+        :return: :class:`PayStellarUri` object from uri.
+        """
+        parsed_uri = parse.urlparse(uri)
+        if parsed_uri.scheme != STELLAR_SCHEME:
+            raise ValueError(
+                f"Stellar URI scheme should be `{STELLAR_SCHEME}`, but got `{parsed_uri.scheme}`."
+            )
+        if parsed_uri.path != "pay":
+            raise ValueError(
+                f"Stellar URI path should be `pay`, but got `{parsed_uri.path}`."
+            )
+        query = cls._parse_uri_query(parsed_uri.query)
+        destination = query.get("destination")
+        amount = query.get("amount")
+        asset_code = query.get("asset_code")
+        asset_issuer = query.get("asset_issuer")
+        memo_value = query.get("memo")
+        memo_type = query.get("memo_type")
+        callback = cls._parse_callback(query.get("callback"))
+        msg = query.get("msg")
+        network_passphrase = query.get("network_passphrase")
+        origin_domain = query.get("origin_domain")
+        signature = query.get("signature")
+        asset = None
+        if asset_code is not None:
+            asset = Asset(asset_code, asset_issuer)
+        memo = cls._decode_memo(memo_type=memo_type, memo_value=memo_value)
+
+        if destination is None:
+            raise ValueError("`destination` is missing from uri.")
+
+        return cls(
+            destination=destination,
+            amount=amount,
+            asset=asset,
+            memo=memo,
+            callback=callback,
+            message=msg,
+            network_passphrase=network_passphrase,
+            origin_domain=origin_domain,
+            signature=signature,
+        )
 
     def __str__(self):
         return (
@@ -250,7 +339,7 @@ def __init__(
 
     @property
     def _replace(self) -> Optional[str]:
-        if self.replace is None:
+        if not self.replace:
             return None
         replaces = []
         hits = dict()
@@ -263,13 +352,16 @@ def _replace(self) -> Optional[str]:
 
     def to_uri(self) -> str:
         """Generate the request URI.
+
+        :return: Stellar Transaction URI.
         """
         query_params = dict()
         query_params["xdr"] = self.transaction_envelope.to_xdr()
         if self.callback is not None:
             query_params["callback"] = "url:" + self.callback
-        if self.replace is not None:
-            query_params["replace"] = self._replace
+        replace = self._replace
+        if replace is not None:
+            query_params["replace"] = replace
         if self.pubkey is not None:
             query_params["pubkey"] = self.pubkey
         if self.msg is not None:
@@ -280,7 +372,81 @@ def to_uri(self) -> str:
             query_params["origin_domain"] = self.origin_domain
         if self.signature is not None:
             query_params["signature"] = self.signature
-        return f"{STELLAR_SCHEME}:tx?{urlencode(query_params, quote_via=quote)}"
+        return f"{STELLAR_SCHEME}:tx?{parse.urlencode(query_params, quote_via=parse.quote)}"
+
+    @classmethod
+    def from_uri(
+        cls, uri: str, network_passphrase: Optional[str]
+    ) -> "TransactionStellarUri":
+        """Parse Stellar Transaction URI and generate :class:`TransactionStellarUri` object.
+
+        :param uri: Stellar Transaction URI.
+        :param network_passphrase: The network to connect to for verifying and retrieving xdr,
+            If it is set to `None`, the `network_passphrase` in the uri will not be verified.
+        :return: :class:`TransactionStellarUri` object from uri.
+        """
+        parsed_uri = parse.urlparse(uri)
+        if parsed_uri.scheme != STELLAR_SCHEME:
+            raise ValueError(
+                f"Stellar URI scheme should be `{STELLAR_SCHEME}`, but got `{parsed_uri.scheme}`."
+            )
+        if parsed_uri.path != "tx":
+            raise ValueError(
+                f"Stellar URI path should be `tx`, but got `{parsed_uri.path}`."
+            )
+        query = cls._parse_uri_query(parsed_uri.query)
+        uri_network_passphrase = query.get("network_passphrase")
+        if network_passphrase is None and uri_network_passphrase is None:
+            raise ValueError("`network_passphrase` is required.")
+
+        if (
+            uri_network_passphrase is not None
+            and network_passphrase is not None
+            and network_passphrase != uri_network_passphrase
+        ):
+            raise ValueError(
+                "The `network_passphrase` in the function parameter does not "
+                "match the `network_passphrase` in the uri."
+            )
+        network_passphrase = network_passphrase or uri_network_passphrase
+
+        xdr = query.get("xdr")
+        callback = cls._parse_callback(query.get("callback"))
+        pubkey = query.get("pubkey")
+        msg = query.get("msg")
+        origin_domain = query.get("origin_domain")
+        signature = query.get("signature")
+        if xdr is None:
+            raise ValueError("`xdr` is missing from uri.")
+        if is_fee_bump_transaction(xdr):
+            tx = FeeBumpTransactionEnvelope.from_xdr(xdr, network_passphrase)
+        else:
+            tx = TransactionEnvelope.from_xdr(xdr, network_passphrase)
+        raw_replacements = query.get("replace")
+        replacements = []
+        if raw_replacements is not None:
+            descriptions_map = {}
+            identifiers, descriptions = raw_replacements.split(";")
+            for description in descriptions.split(","):
+                k, v = description.split(":")
+                descriptions_map[k] = v
+            for identifier in identifiers.split(","):
+                k, v = identifier.split(":")
+                hint = descriptions_map.get(v)
+                if hint is None:
+                    raise ValueError("Invalid `replace`.")
+                replacement = Replacement(k, v, hint)
+                replacements.append(replacement)
+        return cls(
+            transaction_envelope=tx,
+            replace=replacements,
+            callback=callback,
+            pubkey=pubkey,
+            message=msg,
+            network_passphrase=network_passphrase,
+            origin_domain=origin_domain,
+            signature=signature,
+        )
 
     def __str__(self):
         return (

stellar_sdk/utils.py

@@ -93,3 +93,19 @@ def parse_ed25519_account_id_from_muxed_account_xdr_object(
     if data.ed25519 is not None:
         return StrKey.encode_ed25519_public_key(data.ed25519)
     return StrKey.encode_ed25519_public_key(data.med25519.ed25519)
+
+
+def is_fee_bump_transaction(xdr: str) -> bool:
+    xdr_object = Xdr.types.TransactionEnvelope.from_xdr(xdr)
+    te_type = xdr_object.type
+    if te_type == Xdr.const.ENVELOPE_TYPE_TX_FEE_BUMP:
+        return True
+    elif (
+        te_type == Xdr.const.ENVELOPE_TYPE_TX
+        or te_type == Xdr.const.ENVELOPE_TYPE_TX_V0
+    ):
+        return False
+    else:
+        raise ValueError(
+            f"This transaction envelope type is not supported, type = {te_type}."
+        )