feat: Add support for Permissioned Keys

v4-client-py-v2/dydx_v4_client/config.py

@@ -1 +1 @@
-GAS_MULTIPLIER = 1.4
+GAS_MULTIPLIER = 1.7

v4-client-py-v2/dydx_v4_client/network.py

@@ -46,7 +46,6 @@ def make_config(
 make_secure = partial(make_config, secure_channel)
 make_insecure = partial(make_config, insecure_channel)
 
-
 mainnet_node = partial(
     NodeConfig,
     "dydx-mainnet-1",
@@ -55,7 +54,6 @@ def make_config(
 )
 make_mainnet = partial(make_secure, mainnet_node)
 
-
 testnet_node = partial(
     NodeConfig,
     "dydx-testnet-4",
@@ -73,7 +71,6 @@ def make_config(
 TESTNET_FAUCET = "https://faucet.v4testnet.dydx.exchange"
 TESTNET_NOBLE = "https://rpc.testnet.noble.strange.love"
 
-
 local_node = partial(
     NodeConfig,
     "localdydxprotocol",

v4-client-py-v2/dydx_v4_client/node/authenticators.py

@@ -0,0 +1,119 @@
+import base64
+import binascii
+import json
+from dataclasses import dataclass
+from typing import Literal, Union
+
+
+AuthenticatorType = Literal["AllOf", "AnyOf"]
+
+
+@dataclass
+class SubAuthenticator:
+    type: str
+    config: Union[str, bytes, int]
+
+
+@dataclass
+class Authenticator:
+    type: AuthenticatorType
+    config: list[SubAuthenticator]
+
+
+def b64encode(value: bytes) -> str:
+    """Encodes bytes into a base64 string."""
+    return base64.b64encode(value).decode()
+
+
+def b64decode(value: str) -> bytes:
+    """Decodes a base64 string into bytes."""
+    return base64.b64decode(value)
+
+
+def decode_authenticator(
+    config: str, auth_type: str
+) -> Union[SubAuthenticator, Authenticator]:
+    """Decodes a sub-authenticator configuration."""
+    if auth_type == "SignatureVerification":
+        return SubAuthenticator(type=auth_type, config=b64decode(config))
+    elif auth_type == "MessageFilter":
+        return SubAuthenticator(type=auth_type, config=b64decode(config).decode())
+    elif auth_type == "SubaccountFilter":
+        return SubAuthenticator(
+            type=auth_type, config=int.from_bytes(b64decode(config))
+        )
+    elif auth_type == "ClobPairIdFilter":
+        return SubAuthenticator(
+            type=auth_type, config=int.from_bytes(b64decode(config))
+        )
+    elif auth_type in ["AllOf", "AnyOf"]:
+        decoded_subs: list[SubAuthenticator] = []
+        try:
+            # Try decoding assuming the config is base64 encoded JSON.
+            subauth_config = json.loads(b64decode(config))
+        except (binascii.Error, UnicodeDecodeError):
+            subauth_config = json.loads(config)
+        for subauth in subauth_config:
+            # Each sub–authenticator is itself encoded as a dict.
+            decoded_sub = decode_authenticator(subauth["config"], subauth["type"])
+            # In our overall design, we want a list of SubAuthenticators.
+            # If a composite authenticator was decoded, we “prepare” it for inclusion.
+            if isinstance(decoded_sub, Authenticator):
+                decoded_sub = _prepare_authenticator(decoded_sub)
+            decoded_subs.append(decoded_sub)
+        return Authenticator(
+            type=auth_type, config=decoded_subs  # type: ignore[literal-required]
+        )
+    else:
+        raise ValueError(f"Unknown SubAuthenticator type: {auth_type}")
+
+
+def _prepare_authenticator(
+    auth: Union[Authenticator, SubAuthenticator]
+) -> SubAuthenticator:
+    """
+    Converts an Authenticator (with a list of sub–authenticators) into a SubAuthenticator.
+    This is needed for composition so that all sub–authenticators have the same (flat)
+    shape.
+    """
+    if isinstance(auth.config, list):
+        # Convert the list of sub–authenticators into a JSON string.
+        # (We convert each dataclass to a dict via __dict__ so that it is JSON serializable.)
+        subs_as_dicts = [sa.__dict__ for sa in auth.config]
+        encoded_config = b64encode(json.dumps(subs_as_dicts).encode())
+        return SubAuthenticator(type=auth.type, config=encoded_config)
+    return auth
+
+
+def composed_authenticator(
+    authenticator_type: AuthenticatorType,
+    sub_authenticators: list[Union[Authenticator, SubAuthenticator]],
+) -> Authenticator:
+    """Combines multiple sub-authenticators into a single one."""
+    dumped_subs = [_prepare_authenticator(sa) for sa in sub_authenticators]
+    return Authenticator(type=authenticator_type, config=dumped_subs)
+
+
+def signature_verification(pub_key: bytes) -> SubAuthenticator:
+    """Enables authentication via a specific key."""
+    return SubAuthenticator(type="SignatureVerification", config=b64encode(pub_key))
+
+
+def message_filter(msg_type: str) -> SubAuthenticator:
+    """Restricts authentication to certain message types."""
+    assert msg_type.startswith("/"), msg_type
+    return SubAuthenticator(type="MessageFilter", config=b64encode(msg_type.encode()))
+
+
+def subaccount_filter(subaccount_id: int) -> SubAuthenticator:
+    """Restricts authentication to certain subaccount constraints."""
+    return SubAuthenticator(
+        type="SubaccountFilter", config=b64encode(subaccount_id.to_bytes())
+    )
+
+
+def clob_pair_id_filter(clob_pair_id: int) -> SubAuthenticator:
+    """Restricts transactions to specific CLOB pair IDs."""
+    return SubAuthenticator(
+        type="ClobPairIdFilter", config=b64encode(clob_pair_id.to_bytes())
+    )

v4-client-py-v2/dydx_v4_client/node/builder.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import List
+from typing import List, Optional
 
 import google
 from google.protobuf.message import Message
@@ -17,6 +17,7 @@
 
 from dydx_v4_client.node.fee import calculate_fee, Denom
 from dydx_v4_client.wallet import Wallet
+from v4_proto.dydxprotocol.accountplus.tx_pb2 import TxExtension
 
 
 def as_any(message: Message):
@@ -50,6 +51,13 @@ def get_signature(key_pair, body, auth_info, account_number, chain_id):
 )
 
 
+@dataclass
+class TxOptions:
+    authenticator_id: int
+    sequence: int
+    account_number: int
+
+
 @dataclass
 class Builder:
     chain_id: str
@@ -69,17 +77,48 @@ def fee(self, gas_limit: int, *amount: List[Coin]) -> Fee:
             gas_limit=gas_limit,
         )
 
-    def build_transaction(self, wallet: Wallet, messages: List[Message], fee: Fee):
-        body = TxBody(messages=messages, memo=self.memo)
+    def build_transaction(
+        self,
+        wallet: Wallet,
+        messages: List[Message],
+        fee: Fee,
+        tx_options: Optional[TxOptions] = None,
+    ):
+        non_critical_extension_options = []
+        if tx_options is not None:
+            tx_extension = TxExtension(
+                selected_authenticators=[tx_options.authenticator_id],
+            )
+            non_critical_extension_options.append(as_any(tx_extension))
+        body = TxBody(
+            messages=messages,
+            memo=self.memo,
+            non_critical_extension_options=non_critical_extension_options,
+        )
         auth_info = AuthInfo(
-            signer_infos=[get_signer_info(wallet.public_key, wallet.sequence)],
+            signer_infos=[
+                get_signer_info(
+                    wallet.public_key,
+                    tx_options.sequence if tx_options else wallet.sequence,
+                )
+            ],
             fee=fee,
         )
         signature = get_signature(
-            wallet.key, body, auth_info, wallet.account_number, self.chain_id
+            wallet.key,
+            body,
+            auth_info,
+            tx_options.account_number if tx_options else wallet.account_number,
+            self.chain_id,
         )
 
         return Tx(body=body, auth_info=auth_info, signatures=[signature])
 
-    def build(self, wallet: Wallet, message: Message, fee: Fee = DEFAULT_FEE):
-        return self.build_transaction(wallet, [as_any(message)], fee)
+    def build(
+        self,
+        wallet: Wallet,
+        message: Message,
+        fee: Fee = DEFAULT_FEE,
+        tx_options: Optional[dict] = None,
+    ):
+        return self.build_transaction(wallet, [as_any(message)], fee, tx_options)