Custom HTTP headers for proxy and faucet API requests - Duplicate

multiversx_sdk_cli/cli.py

@@ -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_faucet.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_get.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_shared.py

@@ -58,7 +58,7 @@
 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")
@@ -270,6 +270,24 @@ 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 request header (expected KEY=VALUE): {item!r}")
+        key, _, _ = item.partition("=")
+        if not key.strip():
+            raise ArgumentsNotProvidedError(f"Invalid request header (expected non-empty KEY=VALUE): {item!r}")
+    return parse_headers_list(proxy_headers)
 
 
 def add_outfile_arg(sub: Any, what: str = ""):
@@ -810,6 +828,11 @@ def prepare_token_transfers(transfers: list[str]) -> list[TokenTransfer]:
 
 def set_proxy_from_config_if_not_provided(args: Any) -> None:
     """This function modifies the `args` object by setting the proxy from the config if not already set. If proxy is not needed (chainID and nonce are provided), the proxy will not be set."""
+    if hasattr(args, "proxy_headers") and not args.proxy_headers:
+        env = MxpyEnv.from_active_env()
+        if env.proxy_headers:
+            args.proxy_headers = [f"{key}={value}" for key, value in env.proxy_headers.items()]
+
     if not hasattr(args, "proxy"):
         return
 

multiversx_sdk_cli/cli_tokens.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/config.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_env.py

@@ -1,3 +1,4 @@
+import json
 from dataclasses import dataclass
 from functools import cache
 from pathlib import Path
@@ -11,7 +12,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 +22,7 @@
 class MxpyEnv:
     address_hrp: str
     proxy_url: str
+    proxy_headers: dict[str, str]
     explorer_url: str
     ask_confirmation: bool
 
@@ -30,6 +32,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 +42,7 @@ def get_defaults() -> dict[str, str]:
     return {
         "default_address_hrp": "erd",
         "proxy_url": "",
+        "proxy_headers": "",
         "explorer_url": "",
         "ask_confirmation": "false",
     }
@@ -72,6 +76,35 @@ 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.
+    Supports two formats:
+    - JSON object, e.g. {"Authorization": "Bearer token"}
+    - legacy space-separated KEY=VALUE pairs, e.g. "X-Api-Key=abc Another=value"
+    """
+    raw = _get_raw_env_value("proxy_headers")
+
+    if not raw:
+        return {}
+
+    if isinstance(raw, dict):
+        return {str(key): str(value) for key, value in raw.items()}
+
+    if not isinstance(raw, str):
+        raise InvalidEnvironmentValue(f"Invalid proxy_headers value: [{raw!r}]")
+
+    stripped = raw.strip()
+    if stripped.startswith("{"):
+        parsed = json.loads(stripped)
+        if not isinstance(parsed, dict):
+            raise InvalidEnvironmentValue("proxy_headers must be a JSON object")
+        return {str(key): str(value) for key, value in parsed.items()}
+
+    return parse_headers_list(raw.split())
+
+
 @cache
 def get_explorer_url() -> str:
     """
@@ -112,16 +145,32 @@ def get_value(name: str, env_name: str) -> str:
     return value
 
 
+def _get_raw_env_value(key: str) -> Any:
+    """Returns the raw value of a key for the active environment."""
+    data = read_env_file()
+    active_env_name: str = data.get("active", "default")
+
+    if active_env_name == "default":
+        return get_defaults()[key]
+
+    envs = data.get("environments", {})
+    env = envs.get(active_env_name, None)
+    if env is None:
+        raise UnknownEnvironmentError(active_env_name)
+
+    return env.get(key, get_defaults()[key])
+
+
 def _guard_valid_name(name: str):
     if name not in get_defaults().keys():
         raise InvalidEnvironmentValue(f"Key is not present in environment config: [{name}]")
 
 
-def get_active_env() -> dict[str, str]:
+def get_active_env() -> dict[str, Any]:
     data = read_env_file()
     envs: dict[str, Any] = data.get("environments", {})
     active_env_name: str = data.get("active", "default")
-    result: dict[str, str] = envs.get(active_env_name, {})
+    result: dict[str, Any] = envs.get(active_env_name, {})
 
     return result
 
@@ -150,12 +199,35 @@ def set_value(name: str, value: str, env_name: str):
     if env is None:
         raise UnknownEnvironmentError(env_name)
 
-    env[name] = value
+    env[name] = _prepare_value_for_storage(name, value)
     envs[env_name] = env
     data["environments"] = envs
     write_file(data)
 
 
+def _prepare_value_for_storage(name: str, value: str) -> Any:
+    if name != "proxy_headers":
+        return value
+
+    stripped = value.strip()
+    if not stripped.startswith("{"):
+        return value
+
+    try:
+        parsed = json.loads(stripped)
+    except json.JSONDecodeError as error:
+        raise InvalidEnvironmentValue(f"Invalid JSON for proxy_headers: [{error}]")
+
+    if not isinstance(parsed, dict):
+        raise InvalidEnvironmentValue("proxy_headers must be a JSON object")
+
+    for key, item in parsed.items():
+        if not isinstance(key, str) or not isinstance(item, str):
+            raise InvalidEnvironmentValue("proxy_headers JSON object must have string keys and values")
+
+    return parsed
+
+
 def write_file(data: dict[str, Any]):
     env_path = resolve_env_path()
     write_json_file(str(env_path), data)

multiversx_sdk_cli/tests/test_cli_shared.py

@@ -0,0 +1,20 @@
+import pytest
+
+from multiversx_sdk_cli.cli_shared import parse_proxy_headers
+from multiversx_sdk_cli.errors import ArgumentsNotProvidedError
+
+
+def test_parse_proxy_headers_accepts_valid_headers():
+    headers = parse_proxy_headers(["Authorization=Bearer token", "X-Api-Key=abc"])
+
+    assert headers == {"Authorization": "Bearer token", "X-Api-Key": "abc"}
+
+
+def test_parse_proxy_headers_rejects_empty_key():
+    with pytest.raises(ArgumentsNotProvidedError):
+        parse_proxy_headers(["=value"])
+
+
+def test_parse_proxy_headers_rejects_whitespace_only_key():
+    with pytest.raises(ArgumentsNotProvidedError):
+        parse_proxy_headers(["   =value"])