feat: add authorize entry helper.

docs/en/api.rst

@@ -798,6 +798,11 @@ scval
 .. autofunction:: stellar_sdk.scval.to_struct
 .. autofunction:: stellar_sdk.scval.from_struct
 
+Auth
+^^^^
+.. autofunction:: stellar_sdk.auth.authorize_entry
+.. autofunction:: stellar_sdk.auth.authorize_invocation
+
 Helpers
 ^^^^^^^
 .. autofunction:: stellar_sdk.helpers.parse_transaction_envelope_from_xdr

examples/soroban_auth_atomic_swap.py

@@ -0,0 +1,113 @@
+"""This example demonstrates how to invoke an atomic swap contract to swap two tokens.
+
+See https://soroban.stellar.org/docs/how-to-guides/atomic-swap
+https://soroban.stellar.org/docs/learn/authorization
+"""
+import time
+
+from stellar_sdk import (
+    InvokeHostFunction,
+    Keypair,
+    Network,
+    SorobanServer,
+    TransactionBuilder,
+    scval,
+)
+from stellar_sdk.auth import authorize_entry
+from stellar_sdk.exceptions import PrepareTransactionException
+from stellar_sdk.soroban_rpc import GetTransactionStatus, SendTransactionStatus
+
+rpc_server_url = "https://rpc-futurenet.stellar.org:443/"
+soroban_server = SorobanServer(rpc_server_url)
+network_passphrase = Network.FUTURENET_NETWORK_PASSPHRASE
+
+submitter_kp = Keypair.from_secret(
+    "SAAPYAPTTRZMCUZFPG3G66V4ZMHTK4TWA6NS7U4F7Z3IMUD52EK4DDEV"
+)  # GDAT5HWTGIU4TSSZ4752OUC4SABDLTLZFRPZUJ3D6LKBNEPA7V2CIG54
+alice_kp = Keypair.from_secret(
+    "SBPTTA3D3QYQ6E2GSACAZDUFH2UILBNG3EBJCK3NNP7BE4O757KGZUGA"
+)  # GAERW3OYAVYMZMPMVKHSCDS4ORFPLT5Z3YXA4VM3BVYEA2W7CG3V6YYB
+bob_kp = Keypair.from_secret(
+    "SAEZSI6DY7AXJFIYA4PM6SIBNEYYXIEM2MSOTHFGKHDW32MBQ7KVO6EN"
+)  # GBMLPRFCZDZJPKUPHUSHCKA737GOZL7ERZLGGMJ6YGHBFJZ6ZKMKCZTM
+atomic_swap_contract_id = "CCOYWFXSCED6GIUYAYEDVZ7EA4MFDNBXTKSB2RWDSDYMR3EQFT6MUSU5"
+native_token_contract_id = "CB64D3G7SM2RTH6JSGG34DDTFTQ5CFDKVDZJZSODMCX4NJ2HV2KN7OHT"
+cat_token_contract_id = "CDOZ4ZTY2OSEHMTSBL3AFMIDF2EGP3AZTFB7YTCKXNSYZMRV6SROUFAY"
+
+source = soroban_server.load_account(submitter_kp.public_key)
+
+args = [
+    scval.to_address(alice_kp.public_key),  # a
+    scval.to_address(bob_kp.public_key),  # b
+    scval.to_address(native_token_contract_id),  # token_a
+    scval.to_address(cat_token_contract_id),  # token_b
+    scval.to_int128(1000),  # amount_a
+    scval.to_int128(4500),  # min_b_for_a
+    scval.to_int128(5000),  # amount_b
+    scval.to_int128(950),  # min_a_for_b
+]
+
+tx = (
+    TransactionBuilder(source, network_passphrase, base_fee=50000)
+    .add_time_bounds(0, 0)
+    .append_invoke_contract_function_op(
+        contract_id=atomic_swap_contract_id,
+        function_name="swap",
+        parameters=args,
+    )
+    .build()
+)
+
+try:
+    simulate_resp = soroban_server.simulate_transaction(tx)
+    # You need to check the error in the response,
+    # if the error is not None, you need to handle it.
+    op = tx.transaction.operations[0]
+    assert isinstance(op, InvokeHostFunction)
+    assert simulate_resp.results is not None
+    assert simulate_resp.results[0].auth is not None
+    op.auth = [
+        authorize_entry(
+            simulate_resp.results[0].auth[0],
+            alice_kp,
+            simulate_resp.latest_ledger + 20,
+            network_passphrase,
+        ),  # alice sign the entry
+        authorize_entry(
+            simulate_resp.results[0].auth[1],
+            bob_kp,
+            simulate_resp.latest_ledger + 20,
+            network_passphrase,
+        ),  # bob sign the entry
+    ]
+    tx = soroban_server.prepare_transaction(tx, simulate_resp)
+except PrepareTransactionException as e:
+    print(f"Got exception: {e.simulate_transaction_response}")
+    raise e
+
+# tx.transaction.soroban_data.resources.instructions = stellar_xdr.Uint32(
+#     tx.transaction.soroban_data.resources.instructions.uint32 * 2
+# )
+
+tx.sign(submitter_kp)
+print(f"Signed XDR:\n{tx.to_xdr()}")
+
+
+send_transaction_data = soroban_server.send_transaction(tx)
+print(f"sent transaction: {send_transaction_data}")
+if send_transaction_data.status != SendTransactionStatus.PENDING:
+    raise Exception("send transaction failed")
+
+while True:
+    print("waiting for transaction to be confirmed...")
+    get_transaction_data = soroban_server.get_transaction(send_transaction_data.hash)
+    if get_transaction_data.status != GetTransactionStatus.NOT_FOUND:
+        break
+    time.sleep(3)
+
+print(f"transaction: {get_transaction_data}")
+
+if get_transaction_data.status == GetTransactionStatus.SUCCESS:
+    print(f"Transaction successful: {get_transaction_data.result_xdr}")
+else:
+    print(f"Transaction failed: {get_transaction_data.result_xdr}")

examples/soroban_auth_with_stellar_account.py

@@ -0,0 +1,99 @@
+"""This example demonstrates how to invoke an auth contract with [Stellar Account] authorization.
+
+See https://soroban.stellar.org/docs/how-to-guides/auth
+See https://soroban.stellar.org/docs/learn/authorization#stellar-account
+"""
+import time
+
+from stellar_sdk import (
+    InvokeHostFunction,
+    Keypair,
+    Network,
+    SorobanServer,
+    TransactionBuilder,
+    scval,
+)
+from stellar_sdk import xdr as stellar_xdr
+from stellar_sdk.auth import authorize_entry
+from stellar_sdk.exceptions import PrepareTransactionException
+from stellar_sdk.soroban_rpc import GetTransactionStatus, SendTransactionStatus
+
+rpc_server_url = "https://rpc-futurenet.stellar.org:443/"
+network_passphrase = Network.FUTURENET_NETWORK_PASSPHRASE
+soroban_server = SorobanServer(rpc_server_url)
+
+# https://github.com/stellar/soroban-examples/tree/v0.6.0/auth
+contract_id = "CDGPP5TBQIVN4ADNH6PL4METZNJ35OX4DIXKAQ3ENWYLBAJZMHHZE3EV"
+tx_submitter_kp = Keypair.from_secret(
+    "SAAPYAPTTRZMCUZFPG3G66V4ZMHTK4TWA6NS7U4F7Z3IMUD52EK4DDEV"
+)
+op_invoker_kp = Keypair.from_secret(
+    "SAEZSI6DY7AXJFIYA4PM6SIBNEYYXIEM2MSOTHFGKHDW32MBQ7KVO6EN"
+)
+
+func_name = "increment"
+args = [scval.to_address(op_invoker_kp.public_key), scval.to_uint32(10)]
+
+source = soroban_server.load_account(tx_submitter_kp.public_key)
+tx = (
+    TransactionBuilder(source, network_passphrase, base_fee=50000)
+    .add_time_bounds(0, 0)
+    .append_invoke_contract_function_op(
+        contract_id=contract_id,
+        function_name=func_name,
+        parameters=args,
+    )
+    .build()
+)
+
+try:
+    simulate_resp = soroban_server.simulate_transaction(tx)
+    # You need to check the error in the response,
+    # if the error is not None, you need to handle it.
+    op = tx.transaction.operations[0]
+    assert isinstance(op, InvokeHostFunction)
+    assert simulate_resp.results is not None
+    assert simulate_resp.results[0].auth is not None
+    op.auth = [
+        authorize_entry(
+            simulate_resp.results[0].auth[0],
+            op_invoker_kp,
+            simulate_resp.latest_ledger + 20,
+            network_passphrase,
+        )
+    ]
+    tx = soroban_server.prepare_transaction(tx, simulate_resp)
+except PrepareTransactionException as e:
+    print(f"Got exception: {e.simulate_transaction_response}")
+    raise e
+
+# tx.transaction.soroban_data.resources.instructions = stellar_xdr.Uint32(
+#     tx.transaction.soroban_data.resources.instructions.uint32 * 2
+# )
+
+tx.sign(tx_submitter_kp)
+print(f"Signed XDR:\n{tx.to_xdr()}")
+
+send_transaction_data = soroban_server.send_transaction(tx)
+print(f"sent transaction: {send_transaction_data}")
+if send_transaction_data.status != SendTransactionStatus.PENDING:
+    raise Exception("send transaction failed")
+
+while True:
+    print("waiting for transaction to be confirmed...")
+    get_transaction_data = soroban_server.get_transaction(send_transaction_data.hash)
+    if get_transaction_data.status != GetTransactionStatus.NOT_FOUND:
+        break
+    time.sleep(3)
+
+print(f"transaction: {get_transaction_data}")
+
+if get_transaction_data.status == GetTransactionStatus.SUCCESS:
+    assert get_transaction_data.result_meta_xdr is not None
+    transaction_meta = stellar_xdr.TransactionMeta.from_xdr(
+        get_transaction_data.result_meta_xdr
+    )
+    result = transaction_meta.v3.soroban_meta.return_value.u32  # type: ignore
+    print(f"Function result: {result}")
+else:
+    print(f"Transaction failed: {get_transaction_data.result_xdr}")

stellar_sdk/auth.py

@@ -0,0 +1,147 @@
+import copy
+import random
+from typing import Callable, Optional, Union
+
+from . import scval
+from . import xdr as stellar_xdr
+from .address import Address
+from .exceptions import BadSignatureError
+from .keypair import Keypair
+from .network import Network
+from .utils import sha256
+
+__all__ = ["authorize_entry", "authorize_invocation"]
+
+
+def authorize_entry(
+    entry: Union[stellar_xdr.SorobanAuthorizationEntry, str],
+    signer: Union[Keypair, Callable[[stellar_xdr.HashIDPreimage], bytes]],
+    valid_until_ledger_sequence: int,
+    network_passphrase: str,
+) -> stellar_xdr.SorobanAuthorizationEntry:
+    """Actually authorizes an existing authorization entry using the given the
+    credentials and expiration details, returning a signed copy.
+
+    This "fills out" the authorization entry with a signature, indicating to the
+    :class:`stellar_sdk.InvokeHostFunction` it's attached to that:
+        - a particular identity (i.e. signing :class:`stellar_sdk.Keypair` or other signer)
+        - approving the execution of an invocation tree (i.e. a
+            simulation-acquired :class:`stellar_xdr.SorobanAuthorizedInvocation` or otherwise built)
+        - on a particular network (uniquely identified by its passphrase, see :class:`stellar_sdk.Network`)
+        - until a particular ledger sequence is reached.
+
+
+    :param entry: an unsigned Soroban authorization entry.
+    :param signer: either a :class:`Keypair` or a function which takes a payload
+        (a :class:`stellar_xdr.HashIDPreimage` instance) input and returns a bytes signature,
+        the signing key should correspond to the address in the `entry`.
+    :param valid_until_ledger_sequence: the (exclusive) future ledger sequence number until which
+        this authorization entry should be valid (if `currentLedgerSeq==validUntil`, this is expired)
+    :param network_passphrase: the network passphrase is incorporated into the signature (see :class:`stellar_sdk.Network` for options)
+    :return: a signed Soroban authorization entry.
+    """
+
+    if isinstance(entry, str):
+        entry = stellar_xdr.SorobanAuthorizationEntry.from_xdr(entry)
+    else:
+        entry = copy.deepcopy(entry)
+
+    if (
+        entry.credentials.type
+        != stellar_xdr.SorobanCredentialsType.SOROBAN_CREDENTIALS_ADDRESS
+    ):
+        return entry
+
+    assert entry.credentials.address is not None
+    addr_auth = entry.credentials.address
+    addr_auth.signature_expiration_ledger = stellar_xdr.Uint32(
+        valid_until_ledger_sequence
+    )
+
+    network_id = Network(network_passphrase).network_id()
+    preimage = stellar_xdr.HashIDPreimage(
+        type=stellar_xdr.EnvelopeType.ENVELOPE_TYPE_SOROBAN_AUTHORIZATION,
+        soroban_authorization=stellar_xdr.HashIDPreimageSorobanAuthorization(
+            network_id=stellar_xdr.Hash(network_id),
+            nonce=addr_auth.nonce,
+            signature_expiration_ledger=addr_auth.signature_expiration_ledger,
+            invocation=entry.root_invocation,
+        ),
+    )
+    payload = sha256(preimage.to_xdr_bytes())
+    if isinstance(signer, Keypair):
+        signature = signer.sign(payload)
+    else:
+        signature = signer(preimage)
+
+    public_key = Address.from_xdr_sc_address(addr_auth.address).key
+    try:
+        Keypair.from_raw_ed25519_public_key(public_key).verify(payload, signature)
+    except BadSignatureError as e:
+        raise ValueError("signature doesn't match payload.") from e
+
+    # This structure is defined here:
+    # https://soroban.stellar.org/docs/fundamentals-and-concepts/invoking-contracts-with-transactions#stellar-account-signatures
+    addr_auth.signature = scval.to_vec(
+        [
+            scval.to_map(
+                {
+                    scval.to_symbol("public_key"): scval.to_bytes(public_key),
+                    scval.to_symbol("signature"): scval.to_bytes(signature),
+                }
+            )
+        ]
+    )
+    return entry
+
+
+def authorize_invocation(
+    signer: Union[Keypair, Callable[[stellar_xdr.HashIDPreimage], bytes]],
+    public_key: Optional[str],
+    valid_until_ledger_sequence: int,
+    invocation: stellar_xdr.SorobanAuthorizedInvocation,
+    network_passphrase: str,
+):
+    """This builds an entry from scratch, allowing you to express authorization as a function of:
+        - a particular identity (i.e. signing :class:`stellar_sdk.Keypair` or other signer)
+        - approving the execution of an invocation tree (i.e. a
+            simulation-acquired :class:`stellar_xdr.SorobanAuthorizedInvocation` or otherwise built)
+        - on a particular network (uniquely identified by its passphrase, see :class:`stellar_sdk.Network`)
+        - until a particular ledger sequence is reached.
+
+    This is in contrast to :func:`authorize_entry`, which signs an existing entry "in place".
+
+    :param signer: either a :class:`Keypair` or a function which takes a payload
+        (a :class:`stellar_xdr.HashIDPreimage` instance) input and returns a bytes signature,
+        the signing key should correspond to the address in the `entry`.
+    :param public_key: the public identity of the signer (when providing a :class:`Keypair` to `signer`,
+        this can be omitted, as it just uses the public key of the keypair)
+    :param valid_until_ledger_sequence: the (exclusive) future ledger sequence number until which
+        this authorization entry should be valid (if `currentLedgerSeq==validUntil`, this is expired)
+    :param invocation: invocation the invocation tree that we're authorizing (likely, this comes from transaction simulation)
+    :param network_passphrase: the network passphrase is incorporated into the signature (see :class:`stellar_sdk.Network` for options)
+    :return: a signed Soroban authorization entry.
+    """
+    nonce = random.randint(-(2**63), 2**63 - 1)
+    pk = public_key
+    if not pk and isinstance(signer, Keypair):
+        pk = signer.public_key
+
+    if not pk:
+        raise ValueError("`public_key` parameter is required.")
+
+    entry = stellar_xdr.SorobanAuthorizationEntry(
+        root_invocation=invocation,
+        credentials=stellar_xdr.SorobanCredentials(
+            type=stellar_xdr.SorobanCredentialsType.SOROBAN_CREDENTIALS_ADDRESS,
+            address=stellar_xdr.SorobanAddressCredentials(
+                address=Address(pk).to_xdr_sc_address(),
+                nonce=stellar_xdr.Int64(nonce),
+                signature_expiration_ledger=stellar_xdr.Uint32(0),
+                signature=stellar_xdr.SCVal(type=stellar_xdr.SCValType.SCV_VOID),
+            ),
+        ),
+    )
+    return authorize_entry(
+        entry, signer, valid_until_ledger_sequence, network_passphrase
+    )