diff --git a/starknet_py/net/paymaster/__init__.py b/starknet_py/net/paymaster/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/starknet_py/net/paymaster/client.py b/starknet_py/net/paymaster/client.py new file mode 100644 index 000000000..aefc12b21 --- /dev/null +++ b/starknet_py/net/paymaster/client.py @@ -0,0 +1,137 @@ +""" +Client for interacting with the Applicative Paymaster API. + +This module provides a client for interacting with the Applicative Paymaster API +as specified in the JSON-RPC specification. +""" + +from abc import ABC +from typing import List, Optional, cast + +import aiohttp + +from starknet_py.net.client_models import Hash +from starknet_py.net.client_utils import _to_rpc_felt +from starknet_py.net.http_client import RpcHttpClient +from starknet_py.net.paymaster.models import ( + BuildTransactionResponse, + ExecutableUserTransaction, + ExecuteTransactionResponse, + TokenData, + TrackingIdResponse, + UserParameters, + UserTransaction, +) +from starknet_py.net.schemas.paymaster import ( + BuildTransactionResponseSchema, + ExecutableUserTransactionSchema, + ExecuteTransactionResponseSchema, + TokenDataSchema, + TrackingIdResponseSchema, + UserParametersSchema, + UserTransactionSchema, +) +from starknet_py.utils.sync import add_sync_methods + + +@add_sync_methods +class PaymasterClient(ABC): + """ + Client for interacting with the Applicative Paymaster API. + """ + + def __init__( + self, + node_url: str, + session: Optional[aiohttp.ClientSession] = None, + ): + """ + Initialize PaymasterClient. + + :param node_url: URL of the paymaster service + :param session: Aiohttp session to be used for requests. If not provided, a client will create a session for + every request. When using a custom session, the user is responsible for closing it manually. + """ + self.url = node_url + self._client = RpcHttpClient( + url=node_url, session=session, method_prefix="paymaster" + ) + + async def is_available(self) -> bool: + """ + Returns the status of the paymaster service. + + :return: If the paymaster service is correctly functioning, return true. Else, return false. + """ + return await self._client.call(method_name="isAvailable") + + async def get_supported_tokens(self) -> List[TokenData]: + """ + Get a list of the tokens that the paymaster supports, together with their prices in STRK. + + :return: An array of token data + """ + res = await self._client.call(method_name="getSupportedTokens") + return cast(List[TokenData], TokenDataSchema().load(res, many=True)) + + async def tracking_id_to_latest_hash(self, tracking_id: Hash) -> TrackingIdResponse: + """ + Get the hash of the latest transaction broadcasted by the paymaster corresponding to the requested ID + and the status of the ID. + + :param tracking_id: A unique identifier used to track an execution request of a user + :return: The hash of the latest transaction broadcasted by the paymaster corresponding to the requested ID + and the status of the ID + """ + res = await self._client.call( + method_name="trackingIdToLatestHash", + params={"tracking_id": (_to_rpc_felt(tracking_id))}, + ) + return cast(TrackingIdResponse, TrackingIdResponseSchema().load(res)) + + async def build_transaction( + self, transaction: UserTransaction, parameters: UserParameters + ) -> BuildTransactionResponse: + """ + Receives the transaction the user wants to execute. Returns the typed data along with the estimated gas cost + and the maximum gas cost suggested to ensure execution. + + :param transaction: Transaction to be executed by the paymaster + :param parameters: Execution parameters to be used when executing the transaction + :return: The transaction data required for execution along with an estimation of the fee + """ + # Convert transaction to dict for JSON serialization + + res = await self._client.call( + method_name="buildTransaction", + params={ + "transaction": UserTransactionSchema().dump(obj=transaction), + "parameters": UserParametersSchema().dump(obj=parameters), + }, + ) + + return BuildTransactionResponseSchema().load(res) + + async def execute_transaction( + self, transaction: ExecutableUserTransaction, parameters: UserParameters + ) -> ExecuteTransactionResponse: + """ + Sends the signed typed data to the paymaster service for execution. + + :param transaction: Typed data build by calling paymaster_buildTransaction signed by the user + to be executed by the paymaster service + :param parameters: Execution parameters to be used when executing the transaction + :return: The hash of the transaction broadcasted by the paymaster and the tracking ID + corresponding to the user `execute` request + """ + res = await self._client.call( + method_name="executeTransaction", + params={ + "transaction": ExecutableUserTransactionSchema().dump(obj=transaction), + "parameters": UserParametersSchema().dump(obj=parameters), + }, + ) + + return cast( + ExecuteTransactionResponse, ExecuteTransactionResponseSchema().load(res) + ) diff --git a/starknet_py/net/paymaster/models.py b/starknet_py/net/paymaster/models.py new file mode 100644 index 000000000..dd75d04ec --- /dev/null +++ b/starknet_py/net/paymaster/models.py @@ -0,0 +1,316 @@ +""" +Models for the Paymaster API. + +This module contains data models used by the PaymasterClient for interacting with +the Applicative Paymaster API. +""" + +from dataclasses import dataclass +from enum import Enum +from typing import List, Optional + +from starknet_py.net.client_models import Call, OutsideExecutionTimeBounds +from starknet_py.utils.typed_data import TypedData + + +@dataclass +class TokenData: + """ + Object containing data about the token: contract address, number of decimals and current price in STRK. + """ + + token_address: str + """Token contract address""" + + decimals: int + """The number of decimals of the token""" + + price_in_strk: int + """Price in STRK (in FRI units)""" + + +class TrackingStatus(str, Enum): + """ + Status of a transaction tracking ID. + """ + + ACTIVE = "active" + ACCEPTED = "accepted" + DROPPED = "dropped" + + +@dataclass +class TrackingIdResponse: + """ + Response for tracking ID to latest int request. + """ + + transaction_hash: int + status: TrackingStatus + + +@dataclass +class FeeMode: + """ + Base class for fee modes. + """ + + mode: str + + +@dataclass +class SponsoredFeeMode(FeeMode): + """ + Specify that the transaction should be sponsored. + """ + + mode: str = "sponsored" + + +@dataclass +class DefaultFeeMode(FeeMode): + """ + Default fee mode where the transaction is paid by the user in the given gas token. + """ + + gas_token: int + """The token to use for paying gas fees""" + + mode: str = "default" + + +@dataclass +class PriorityFeeMode(FeeMode): + """ + Fee mode where the transaction is paid by the user in the given gas token and the user can specify a tip. + """ + + gas_token: int + """The token to use for paying gas fees""" + + tip_in_strk: int + """Additional tip in STRK""" + + mode: str = "priority" + + +@dataclass +class UserParameters: + """ + Execution parameters to be used when executing the transaction through the paymaster. + """ + + fee_mode: FeeMode + version: str = "0x1" + time_bounds: Optional[OutsideExecutionTimeBounds] = None + + +@dataclass +class AccountDeploymentData: + """ + Data required to deploy an account at an address. + """ + + address: int + class_hash: int + salt: int + calldata: List[int] + version: int = 1 + sigdata: Optional[List[int]] = None + + +@dataclass +class UserInvoke: + """ + Invoke data to a transaction on behalf of the user. + """ + + user_address: int + """The address of the user account""" + + calls: List[Call] + """The sequence of calls that the user wishes to perform""" + + +@dataclass +class UserTransaction: + """ + Base class for user transactions. + """ + + type: str + + +@dataclass +class DeployTransaction(UserTransaction): + """ + Deployment transaction. + """ + + deployment: AccountDeploymentData + """Deployment data necessary to deploy the account""" + + type: str = "deploy" + + +@dataclass +class InvokeTransaction(UserTransaction): + """ + Invoke transaction. + """ + + invoke: UserInvoke + """Invoke data to a transaction on behalf of the user""" + + type: str = "invoke" + + +@dataclass +class DeployAndInvokeTransaction(UserTransaction): + """ + Deploy and invoke transaction. + """ + + deployment: AccountDeploymentData + """Deployment data necessary to deploy the account""" + + invoke: UserInvoke + """Invoke data to a transaction on behalf of the user""" + + type: str = "deploy_and_invoke" + + +@dataclass +class FeeEstimate: + """ + Fee estimation for a transaction. + """ + + gas_token_price_in_strk: int + estimated_fee_in_strk: int + estimated_fee_in_gas_token: int + suggested_max_fee_in_strk: int + suggested_max_fee_in_gas_token: int + + +@dataclass +class ExecutableUserInvoke: + """ + Invoke data signed by the user to be executed by the paymaster service. + """ + + user_address: int + """The address of the user account""" + + typed_data: TypedData + """Typed data returned by the endpoint paymaster_buildTransaction""" + + signature: List[int] + """Signature of the associated Typed Data""" + + +@dataclass +class ExecutableUserTransaction: + """ + Base class for executable user transactions. + """ + + type: str + + +@dataclass +class ExecutableDeployTransaction(ExecutableUserTransaction): + """ + Executable deployment transaction. + """ + + deployment: AccountDeploymentData + """Deployment data necessary to deploy the account""" + + type: str = "deploy" + + +@dataclass +class ExecutableInvokeTransaction(ExecutableUserTransaction): + """ + Executable invoke transaction. + """ + + invoke: ExecutableUserInvoke + """Invoke data signed by the user to be executed by the paymaster service""" + + type: str = "invoke" + + +@dataclass +class ExecutableDeployAndInvokeTransaction(ExecutableUserTransaction): + """ + Executable deploy and invoke transaction. + """ + + deployment: AccountDeploymentData + """Deployment data necessary to deploy the account""" + + invoke: ExecutableUserInvoke + """Invoke data signed by the user to be executed by the paymaster service""" + + type: str = "deploy_and_invoke" + + +@dataclass +class ExecuteTransactionResponse: + """ + Response for execute transaction request. + """ + + tracking_id: int + transaction_hash: int + + +class BuildTransactionResponseType(str, Enum): + """ + Type of build transaction response. + """ + + DEPLOY = "deploy" + INVOKE = "invoke" + DEPLOY_AND_INVOKE = "deploy_and_invoke" + + +@dataclass +class BuildTransactionResponse: + """ + Base class for build transaction responses. + """ + + type: BuildTransactionResponseType + parameters: UserParameters + fee: FeeEstimate + + +@dataclass +class DeployBuildTransactionResponse(BuildTransactionResponse): + """ + Response for build transaction request for deployment. + """ + + deployment: AccountDeploymentData + + +@dataclass +class InvokeBuildTransactionResponse(BuildTransactionResponse): + """ + Response for build transaction request for invoke. + """ + + typed_data: TypedData + + +@dataclass +class DeployAndInvokeBuildTransactionResponse(BuildTransactionResponse): + """ + Response for build transaction request for deploy and invoke. + """ + + deployment: AccountDeploymentData + typed_data: TypedData diff --git a/starknet_py/net/schemas/paymaster.py b/starknet_py/net/schemas/paymaster.py new file mode 100644 index 000000000..0c38e3a74 --- /dev/null +++ b/starknet_py/net/schemas/paymaster.py @@ -0,0 +1,264 @@ +from marshmallow import fields, post_load +from marshmallow_oneofschema import OneOfSchema + +from starknet_py.net.paymaster.models import ( + AccountDeploymentData, + DefaultFeeMode, + DeployAndInvokeBuildTransactionResponse, + DeployBuildTransactionResponse, + ExecuteTransactionResponse, + FeeEstimate, + FeeMode, + InvokeBuildTransactionResponse, + PriorityFeeMode, + SponsoredFeeMode, + TokenData, + TrackingIdResponse, + TrackingStatus, + UserParameters, + UserTransaction, +) +from starknet_py.net.schemas.common import Felt +from starknet_py.utils.schema import Schema +from starknet_py.utils.typed_data import TypedDataSchema + + +class AccountDeploymentDataSchema(Schema): + address = Felt(data_key="address", required=True) + class_hash = Felt(data_key="class_hash", required=True) + salt = Felt(data_key="salt", required=True) + calldata = fields.List(Felt(), data_key="calldata", required=True) + sigdata = fields.List(Felt(), data_key="sigdata", required=False) + version = fields.Integer(data_key="version", required=True) + + @post_load + def make_dataclass(self, data, **kwargs) -> AccountDeploymentData: + return AccountDeploymentData(**data) + + +class CallSchema(Schema): + to = Felt(data_key="to", required=True) + selector = Felt(data_key="selector", required=True) + calldata = fields.List(Felt(), data_key="calldata", required=True) + + +class UserInvokeSchema(Schema): + user_address = Felt(data_key="user_address", required=True) + calls = fields.List(fields.Nested(CallSchema), data_key="calls", required=True) + + +class InvokeTransactionSchema(Schema): + invoke = fields.Nested(UserInvokeSchema, data_key="invoke", required=True) + + +class DeployTransactionSchema(Schema): + deployment = fields.Nested( + AccountDeploymentDataSchema, data_key="deployment", required=True + ) + + +class DeployAndInvokeTransactionSchema(Schema): + invoke = fields.Nested(UserInvokeSchema, data_key="invoke", required=True) + deployment = fields.Nested( + AccountDeploymentDataSchema, data_key="deployment", required=True + ) + + +class UserTransactionSchema(OneOfSchema): + type_field = "type" + type_schemas = { + "invoke": InvokeTransactionSchema(), + "deploy": DeployTransactionSchema(), + "deploy_and_invoke": DeployAndInvokeTransactionSchema(), + } + + def get_obj_type(self, obj: UserTransaction): + return obj.type + + +class TokenDataSchema(Schema): + token_address = Felt(data_key="token_address", required=True) + decimals = fields.Integer(data_key="decimals", required=True) + price_in_strk = Felt(data_key="price_in_strk", required=True) + + @post_load + def make_dataclass(self, data, **kwargs) -> TokenData: + return TokenData(**data) + + +class TrackingStatusField(fields.Field): + def _deserialize(self, value, attr, data, **kwargs) -> TrackingStatus: + return TrackingStatus(value) + + +class TrackingIdResponseSchema(Schema): + transaction_hash = Felt(data_key="transaction_hash", required=True) + status = TrackingStatusField(data_key="status", required=True) + + @post_load + def make_dataclass(self, data, **kwargs) -> TrackingIdResponse: + return TrackingIdResponse(**data) + + +class FeeEstimateSchema(Schema): + gas_token_price_in_strk = Felt(data_key="gas_token_price_in_strk", required=True) + estimated_fee_in_strk = Felt(data_key="estimated_fee_in_strk", required=True) + estimated_fee_in_gas_token = Felt( + data_key="estimated_fee_in_gas_token", required=True + ) + suggested_max_fee_in_strk = Felt( + data_key="suggested_max_fee_in_strk", required=True + ) + suggested_max_fee_in_gas_token = Felt( + data_key="suggested_max_fee_in_gas_token", required=True + ) + + @post_load + def make_dataclass(self, data, **kwargs) -> FeeEstimate: + return FeeEstimate(**data) + + +class TimeBoundsSchema(Schema): + execute_after = fields.DateTime( + data_key="execute_after", format="timestamp_ms", required=True + ) + execute_before = fields.Integer( + data_key="execute_before", format="timestamp_ms", required=True + ) + + +class SponsoredFeeModeSchema(Schema): + pass + + +class DefaultFeeModeSchema(Schema): + gas_token = Felt(data_key="gas_token", required=True) + + +class PriorityFeeModeSchema(Schema): + gas_token = Felt(data_key="gas_token", required=True) + tip_in_strk = Felt(data_key="tip_in_strk", required=True) + + +class FeeModeSchema(OneOfSchema): + type_field = "mode" + type_schemas = { + "sponsored": SponsoredFeeModeSchema(), + "default": DefaultFeeModeSchema(), + "priority": PriorityFeeModeSchema(), + } + + def get_obj_type(self, obj: FeeMode): + if isinstance(obj, SponsoredFeeMode): + return "sponsored" + if isinstance(obj, DefaultFeeMode): + return "default" + if isinstance(obj, PriorityFeeMode): + return "priority" + raise Exception(f"Unknown FeeMode type: {type(obj)}") + + +class UserParametersSchema(Schema): + version = fields.String(data_key="version", required=True) + fee_mode = fields.Nested(FeeModeSchema, data_key="fee_mode", required=False) + time_bounds = fields.Nested( + TimeBoundsSchema, data_key="time_bounds", required=False + ) + + @post_load + def make_dataclass(self, data, **kwargs) -> UserParameters: + return UserParameters(**data) + + +class ExecuteTransactionResponseSchema(Schema): + tracking_id = Felt(data_key="tracking_id", required=True) + transaction_hash = Felt(data_key="transaction_hash", required=True) + + @post_load + def make_dataclass(self, data, **kwargs) -> ExecuteTransactionResponse: + return ExecuteTransactionResponse(**data) + + +class DeployBuildTransactionResponseSchema(Schema): + deployment = fields.Nested( + AccountDeploymentDataSchema, data_key="deployment", required=True + ) + parameters = fields.Nested( + UserParametersSchema, data_key="parameters", required=True + ) + fee = fields.Nested(FeeEstimateSchema, data_key="fee", required=True) + + @post_load + def make_dataclass(self, data, **kwargs) -> DeployBuildTransactionResponse: + return DeployBuildTransactionResponse(**data) + + +class InvokeBuildTransactionResponseSchema(Schema): + typed_data = fields.Nested(TypedDataSchema, data_key="typed_data", required=True) + parameters = fields.Nested( + UserParametersSchema, data_key="parameters", required=True + ) + fee = fields.Nested(FeeEstimateSchema, data_key="fee", required=True) + + @post_load + def make_dataclass(self, data, **kwargs) -> InvokeBuildTransactionResponse: + return InvokeBuildTransactionResponse(**data) + + +class DeployAndInvokeBuildTransactionResponseSchema(Schema): + deployment = fields.Nested( + AccountDeploymentDataSchema, data_key="deployment", required=True + ) + typed_data = fields.Dict(data_key="typed_data", required=True) + parameters = fields.Nested( + UserParametersSchema, data_key="parameters", required=True + ) + fee = fields.Nested(FeeEstimateSchema, data_key="fee", required=True) + + @post_load + def make_dataclass(self, data, **kwargs) -> DeployAndInvokeBuildTransactionResponse: + return DeployAndInvokeBuildTransactionResponse(**data) + + +class BuildTransactionResponseSchema(OneOfSchema): + type_field = "type" + type_schemas = { + "deploy": DeployBuildTransactionResponseSchema(), + "invoke": InvokeBuildTransactionResponseSchema(), + "deploy_and_invoke": DeployAndInvokeBuildTransactionResponseSchema(), + } + + +class ExecutableUserInvokeSchema(Schema): + user_address = Felt(data_key="user_address", required=True) + typed_data = fields.Nested(TypedDataSchema, data_key="typed_data", required=True) + signature = fields.List(Felt(), data_key="signature", required=True) + + +class ExecutableInvokeTransactionSchema(Schema): + invoke = fields.Nested(ExecutableUserInvokeSchema, data_key="invoke", required=True) + + +class ExecutableDeployTransactionSchema(Schema): + deployment = fields.Nested( + AccountDeploymentDataSchema, data_key="deployment", required=True + ) + + +class ExecutableDeployAndInvokeTransactionSchema(Schema): + deployment = fields.Nested( + AccountDeploymentDataSchema, data_key="deployment", required=True + ) + invoke = fields.Nested(ExecutableUserInvokeSchema, data_key="invoke", required=True) + + +class ExecutableUserTransactionSchema(OneOfSchema): + type_field = "type" + type_schemas = { + "invoke": ExecutableInvokeTransactionSchema(), + "deploy": ExecutableDeployTransactionSchema(), + "deploy_and_invoke": ExecutableDeployAndInvokeTransactionSchema(), + } + + def get_obj_type(self, obj: UserTransaction): + return obj.type diff --git a/starknet_py/tests/e2e/paymaster/__init__.py b/starknet_py/tests/e2e/paymaster/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/starknet_py/utils/typed_data.py b/starknet_py/utils/typed_data.py index ebdd90020..fed98ec6f 100644 --- a/starknet_py/utils/typed_data.py +++ b/starknet_py/utils/typed_data.py @@ -150,14 +150,15 @@ def from_dict(data: TypedDataDict) -> "TypedData": """ return cast(TypedData, TypedDataSchema().load(data)) - def to_dict(self) -> dict: + # TODO verify + def to_dict(self) -> TypedDataDict: """ Create TypedData dictionary from dataclass. :return: TypedData dictionary. """ - return cast(Dict, TypedDataSchema().dump(obj=self)) + return cast(TypedDataDict, TypedDataSchema().dump(obj=self)) def _is_struct(self, type_name: str) -> bool: return type_name in self.types