add support for custom HTTP headers on proxy and faucet API requests

Author: BubuMVX

CLI.md

@@ -30,7 +30,7 @@
 import multiversx_sdk_cli.cli_wallet
 import multiversx_sdk_cli.version
 from multiversx_sdk_cli import config, errors, utils, ux
-from multiversx_sdk_cli.cli_shared import set_proxy_from_config_if_not_provided
+from multiversx_sdk_cli.cli_shared import parse_proxy_headers, set_proxy_from_config_if_not_provided
 from multiversx_sdk_cli.config_env import get_address_hrp
 from multiversx_sdk_cli.constants import LOG_LEVELS, SDK_PATH
 
@@ -81,6 +81,7 @@ def _do_main(cli_args: list[str]):
         parser.print_help()
     else:
         set_proxy_from_config_if_not_provided(args)
+        config.set_proxy_headers(parse_proxy_headers(getattr(args, "proxy_headers", None)))
         args.func(args)
 
 

multiversx_sdk_cli/cli.py

@@ -29,6 +29,12 @@ def setup_parser(args: list[str], subparsers: Any) -> Any:
     cli_shared.add_wallet_args(args, sub)
     sub.add_argument("--chain", choices=["D", "T"], help="the chain identifier")
     sub.add_argument("--api", type=str, help="custom api url for the native auth client")
+    sub.add_argument(
+        "--api-headers",
+        nargs="+",
+        metavar="KEY=VALUE",
+        help="extra HTTP headers for API requests, e.g. 'Api-Key=mytoken'"
+    )
     sub.add_argument("--wallet-url", type=str, help="custom wallet url to call the faucet from")
     sub.set_defaults(func=faucet)
 
@@ -40,7 +46,8 @@ def faucet(args: Any):
     account = cli_shared.prepare_account(args)
     wallet, api = get_wallet_and_api_urls(args)
 
-    config = NativeAuthClientConfig(origin=wallet, api_url=api)
+    extra_headers = cli_shared.parse_proxy_headers(getattr(args, "api_headers", None))
+    config = NativeAuthClientConfig(origin=wallet, api_url=api, extra_request_headers=extra_headers or None)
     client = NativeAuthClient(config)
 
     init_token = client.initialize()

multiversx_sdk_cli/cli_faucet.py

@@ -6,6 +6,7 @@
 from multiversx_sdk import ProxyNetworkProvider, Token, TokenComputer
 
 from multiversx_sdk_cli import cli_shared
+from multiversx_sdk_cli.cli_shared import add_proxy_arg
 from multiversx_sdk_cli.config import get_config_for_network_providers
 from multiversx_sdk_cli.config_env import MxpyEnv
 from multiversx_sdk_cli.errors import (
@@ -25,7 +26,7 @@ def setup_parser(subparsers: Any) -> Any:
     sub = cli_shared.add_command_subparser(subparsers, "get", "account", "Get info about an account.")
     _add_alias_arg(sub)
     _add_address_arg(sub)
-    _add_proxy_arg(sub)
+    add_proxy_arg(sub)
     sub.add_argument(
         "--balance",
         action="store_true",
@@ -40,22 +41,22 @@ def setup_parser(subparsers: Any) -> Any:
     )
     _add_alias_arg(sub)
     _add_address_arg(sub)
-    _add_proxy_arg(sub)
+    add_proxy_arg(sub)
     sub.set_defaults(func=get_storage)
 
     sub = cli_shared.add_command_subparser(
         subparsers, "get", "storage-entry", "Get a specific storage entry (key-value pair) of an account."
     )
     _add_alias_arg(sub)
     _add_address_arg(sub)
-    _add_proxy_arg(sub)
+    add_proxy_arg(sub)
     sub.add_argument("--key", type=str, required=True, help="the storage key to read from")
     sub.set_defaults(func=get_key)
 
     sub = cli_shared.add_command_subparser(subparsers, "get", "token", "Get a token of an account.")
     _add_alias_arg(sub)
     _add_address_arg(sub)
-    _add_proxy_arg(sub)
+    add_proxy_arg(sub)
     sub.add_argument(
         "--identifier",
         type=str,
@@ -65,16 +66,16 @@ def setup_parser(subparsers: Any) -> Any:
     sub.set_defaults(func=get_token)
 
     sub = cli_shared.add_command_subparser(subparsers, "get", "transaction", "Get a transaction from the network.")
-    _add_proxy_arg(sub)
+    add_proxy_arg(sub)
     sub.add_argument("--hash", type=str, required=True, help="the transaction hash")
     sub.set_defaults(func=get_transaction)
 
     sub = cli_shared.add_command_subparser(subparsers, "get", "network-config", "Get the network configuration.")
-    _add_proxy_arg(sub)
+    add_proxy_arg(sub)
     sub.set_defaults(func=get_network_config)
 
     sub = cli_shared.add_command_subparser(subparsers, "get", "network-status", "Get the network status.")
-    _add_proxy_arg(sub)
+    add_proxy_arg(sub)
     sub.add_argument(
         "--shard",
         type=int,
@@ -96,10 +97,6 @@ def _add_address_arg(sub: Any):
     sub.add_argument("--address", type=str, help="the bech32 address")
 
 
-def _add_proxy_arg(sub: Any):
-    sub.add_argument("--proxy", type=str, help="the proxy url")
-
-
 def get_account(args: Any):
     if args.alias and args.address:
         raise BadUsage("Provide either '--alias' or '--address'")

multiversx_sdk_cli/cli_get.py

@@ -58,12 +58,11 @@
 from multiversx_sdk_cli.signing_wrapper import SigningWrapper
 from multiversx_sdk_cli.simulation import Simulator
 from multiversx_sdk_cli.transactions import send_and_wait_for_result
-from multiversx_sdk_cli.utils import log_explorer_transaction
+from multiversx_sdk_cli.utils import log_explorer_transaction, parse_headers_list
 from multiversx_sdk_cli.ux import confirm_continuation
 
 logger = logging.getLogger("cli_shared")
 
-
 trusted_cosigner_service_url_by_chain_id = {
     "1": "https://tools.multiversx.com/guardian",
     "D": "https://devnet-tools.multiversx.com/guardian",
@@ -118,11 +117,11 @@ def add_command_subparser(subparsers: Any, group: str, command: str, description
 
 
 def add_tx_args(
-    args: list[str],
-    sub: Any,
-    with_nonce: bool = True,
-    with_receiver: bool = True,
-    with_data: bool = True,
+        args: list[str],
+        sub: Any,
+        with_nonce: bool = True,
+        with_receiver: bool = True,
+        with_data: bool = True,
 ):
     if with_nonce:
         sub.add_argument(
@@ -270,6 +269,21 @@ def add_relayed_v3_wallet_args(args: list[str], sub: Any):
 
 def add_proxy_arg(sub: Any):
     sub.add_argument("--proxy", type=str, help="🔗 the URL of the proxy")
+    sub.add_argument(
+        "--proxy-headers",
+        nargs="+",
+        metavar="KEY=VALUE",
+        help="custom HTTP headers for proxy requests, e.g. 'Api-Key=mytoken'",
+    )
+
+
+def parse_proxy_headers(proxy_headers: Optional[list[str]]) -> dict[str, str]:
+    if not proxy_headers:
+        return {}
+    for item in proxy_headers:
+        if "=" not in item:
+            raise ArgumentsNotProvidedError(f"Invalid proxy header (expected KEY=VALUE): {item!r}")
+    return parse_headers_list(proxy_headers)
 
 
 def add_outfile_arg(sub: Any, what: str = ""):
@@ -302,7 +316,7 @@ def add_token_transfers_args(sub: Any):
         "--token-transfers",
         nargs="+",
         help="token transfers for transfer & execute, as [token, amount] "
-        "E.g. --token-transfers NFT-123456-0a 1 ESDT-987654 100000000",
+             "E.g. --token-transfers NFT-123456-0a 1 ESDT-987654 100000000",
     )
 
 
@@ -843,9 +857,9 @@ def initialize_gas_limit_estimator(args: Any) -> Union[GasLimitEstimator, None]:
 
 
 def set_options_for_hash_signing_if_needed(
-    transaction: Transaction,
-    guardian: Union[IAccount, None],
-    relayer: Union[IAccount, None],
+        transaction: Transaction,
+        guardian: Union[IAccount, None],
+        relayer: Union[IAccount, None],
 ):
     transaction_computer = TransactionComputer()
 
@@ -858,10 +872,10 @@ def set_options_for_hash_signing_if_needed(
 
 
 def alter_transaction_and_sign_again_if_needed(
-    args: Any,
-    tx: Transaction,
-    sender: IAccount,
-    guardian_and_relayer_data: GuardianRelayerData,
+        args: Any,
+        tx: Transaction,
+        sender: IAccount,
+        guardian_and_relayer_data: GuardianRelayerData,
 ):
     initial_tx = deepcopy(tx)
 
@@ -886,9 +900,9 @@ def alter_transaction_and_sign_again_if_needed(
 
 
 def _alter_version_and_options_if_provided(
-    args: Any,
-    initial_tx: Transaction,
-    transaction: Transaction,
+        args: Any,
+        initial_tx: Transaction,
+        transaction: Transaction,
 ) -> bool:
     """Alters the transaction version and options if they are provided in args.
     Returns True if any alteration was made, False otherwise.
@@ -913,9 +927,9 @@ def _alter_version_and_options_if_provided(
 
 
 def _sign_transaction(
-    transaction: Transaction,
-    sender: Optional[IAccount] = None,
-    guardian_and_relayer_data: GuardianRelayerData = GuardianRelayerData(),
+        transaction: Transaction,
+        sender: Optional[IAccount] = None,
+        guardian_and_relayer_data: GuardianRelayerData = GuardianRelayerData(),
 ):
     signer = SigningWrapper()
     signer.sign_transaction(

multiversx_sdk_cli/cli_shared.py

@@ -7,7 +7,7 @@
     TokenType,
 )
 
-from multiversx_sdk_cli import cli_shared
+from multiversx_sdk_cli import cli_shared, config
 from multiversx_sdk_cli.args_validation import (
     validate_broadcast_args,
     validate_chain_id_args,
@@ -828,7 +828,7 @@ def _initialize_controller(args: Any) -> TokenManagementController:
     proxy_url = args.proxy if args.proxy else ""
     return TokenManagementController(
         chain_id=chain_id,
-        network_provider=ProxyNetworkProvider(proxy_url),
+        network_provider=ProxyNetworkProvider(proxy_url, config=config.get_config_for_network_providers()),
         gas_limit_estimator=gas_estimator,
     )
 

multiversx_sdk_cli/cli_tokens.py

@@ -1,7 +1,7 @@
 import os
 from functools import cache
 from pathlib import Path
-from typing import Any
+from typing import Any, Optional
 
 from multiversx_sdk import NetworkProviderConfig
 
@@ -192,5 +192,14 @@ def get_dependency_parent_directory(key: str) -> Path:
     return SDK_PATH / key
 
 
+_proxy_headers: dict[str, str] = {}
+
+
+def set_proxy_headers(headers: dict[str, str]) -> None:
+    global _proxy_headers
+    _proxy_headers = headers
+
+
 def get_config_for_network_providers() -> NetworkProviderConfig:
-    return NetworkProviderConfig(client_name="mxpy")
+    requests_options: Optional[dict[str, Any]] = {"headers": _proxy_headers} if _proxy_headers else None
+    return NetworkProviderConfig(client_name="mxpy", requests_options=requests_options)

multiversx_sdk_cli/config.py

@@ -11,7 +11,7 @@
     InvalidEnvironmentValue,
     UnknownEnvironmentError,
 )
-from multiversx_sdk_cli.utils import read_json_file, write_json_file
+from multiversx_sdk_cli.utils import parse_headers_list, read_json_file, write_json_file
 
 LOCAL_ENV_PATH = Path("env.mxpy.json").resolve()
 GLOBAL_ENV_PATH = SDK_PATH / "env.mxpy.json"
@@ -21,6 +21,7 @@
 class MxpyEnv:
     address_hrp: str
     proxy_url: str
+    proxy_headers: dict[str, str]
     explorer_url: str
     ask_confirmation: bool
 
@@ -30,6 +31,7 @@ def from_active_env(cls) -> "MxpyEnv":
         return cls(
             address_hrp=get_address_hrp(),
             proxy_url=get_proxy_url(),
+            proxy_headers=get_proxy_headers(),
             explorer_url=get_explorer_url(),
             ask_confirmation=get_confirmation_setting(),
         )
@@ -39,6 +41,7 @@ def get_defaults() -> dict[str, str]:
     return {
         "default_address_hrp": "erd",
         "proxy_url": "",
+        "proxy_headers": "",
         "explorer_url": "",
         "ask_confirmation": "false",
     }
@@ -72,6 +75,16 @@ def get_proxy_url() -> str:
     return _get_env_value("proxy_url")
 
 
+@cache
+def get_proxy_headers() -> dict[str, str]:
+    """
+    Returns the proxy headers for the active environment as a dict.
+    Headers are stored as space-separated KEY=VALUE pairs (e.g. 'X-Api-Key=abc Authorization=Bearer token').
+    """
+    raw = _get_env_value("proxy_headers")
+    return parse_headers_list(raw.split()) if raw else {}
+
+
 @cache
 def get_explorer_url() -> str:
     """

multiversx_sdk_cli/config_env.py

@@ -1,4 +1,4 @@
-from multiversx_sdk import Address, ProxyNetworkProvider
+from multiversx_sdk import Address, NetworkProviderConfig, ProxyNetworkProvider
 
 from multiversx_sdk_cli.cli import main
 from multiversx_sdk_cli.config import get_config_for_network_providers
@@ -25,3 +25,22 @@ def test_query_contract():
         ]
     )
     assert False if result else True
+
+
+def test_proxy_extra_headers():
+    from unittest.mock import MagicMock, patch
+
+    config = NetworkProviderConfig(requests_options={"headers": {"x-custom-header": "test"}})
+    proxy = ProxyNetworkProvider("", config=config)
+
+    def echo_headers(url, **kwargs):
+        received_headers = kwargs.get("headers", {})
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.json.return_value = {"data": {"headers": received_headers}, "error": "", "code": "successful"}
+        return mock_resp
+
+    with patch("requests.Session.get", side_effect=echo_headers):
+        response = proxy.do_get_generic("headers")
+        headers = response.get("headers", {})
+        assert headers["x-custom-header"] == "test"

multiversx_sdk_cli/tests/test_proxy.py

@@ -183,3 +183,12 @@ def get_explorer_by_chain_id(chain_id: str) -> str:
         return explorers_by_chain_id[chain_id]
     except KeyError:
         return ""
+
+
+def parse_headers_list(items: list[str]) -> dict[str, str]:
+    """Parses a list of KEY=VALUE strings into a dict."""
+    headers: dict[str, str] = {}
+    for item in items:
+        key, _, value = item.partition("=")
+        headers[key.strip()] = value.strip()
+    return headers