Implemented a native auth client and faucet functionality

.github/workflows/build-windows.yml

@@ -30,7 +30,7 @@ jobs:
         run: |
           python3 -m pip install --upgrade pip
           pip3 install -r requirements.txt
-          pip3 install pytest
+          pip3 install -r requirements-dev.txt
       - name: Set github_api_token
         shell: bash
         run: |

.github/workflows/build.yml

@@ -29,7 +29,7 @@ jobs:
         run: |
           python3 -m pip install --upgrade pip
           pip3 install -r requirements.txt
-          pip3 install pytest
+          pip3 install -r requirements-dev.txt
       - name: Set github_api_token
         run: |
           mkdir ~/multiversx-sdk

CLI.md.sh

@@ -114,6 +114,9 @@ generate() {
     command "Data.Dump" "data parse"
     command "Data.Store" "data store"
     command "Data.Load" "data load"
+
+    group "Faucet" "faucet"
+    command "Faucet.Request" "faucet request"
 }
 
 generate

multiversx_sdk_cli/cli.py

@@ -15,6 +15,7 @@
 import multiversx_sdk_cli.cli_delegation
 import multiversx_sdk_cli.cli_deps
 import multiversx_sdk_cli.cli_dns
+import multiversx_sdk_cli.cli_faucet
 import multiversx_sdk_cli.cli_ledger
 import multiversx_sdk_cli.cli_localnet
 import multiversx_sdk_cli.cli_transactions
@@ -100,6 +101,7 @@ def setup_parser(args: List[str]):
     commands.append(multiversx_sdk_cli.cli_data.setup_parser(subparsers))
     commands.append(multiversx_sdk_cli.cli_delegation.setup_parser(args, subparsers))
     commands.append(multiversx_sdk_cli.cli_dns.setup_parser(args, subparsers))
+    commands.append(multiversx_sdk_cli.cli_faucet.setup_parser(args, subparsers))
 
     parser.epilog = """
 ----------------------

multiversx_sdk_cli/cli_delegation.py

@@ -12,7 +12,7 @@ def setup_parser(args: List[str], subparsers: Any) -> Any:
 
     # create new delegation contract
     sub = cli_shared.add_command_subparser(subparsers, "staking-provider", "create-new-delegation-contract",
-                                           "Create a new delegation system smart contract, transferred value must be"
+                                           "Create a new delegation system smart contract, transferred value must be "
                                            "greater than baseIssuingCost + min deposit value")
     _add_common_arguments(args, sub)
     sub.add_argument("--total-delegation-cap", required=True, help="the total delegation contract capacity")

multiversx_sdk_cli/cli_faucet.py

@@ -0,0 +1,72 @@
+import logging
+import webbrowser
+from enum import Enum
+from typing import Any, List, Tuple
+
+from multiversx_sdk_cli import cli_shared
+from multiversx_sdk_cli.errors import BadUserInput
+from multiversx_sdk_cli.native_auth_client import (NativeAuthClient,
+                                                   NativeAuthClientConfig)
+
+logger = logging.getLogger("cli.faucet")
+
+
+class WebWalletUrls(Enum):
+    DEVNET = "https://devnet-wallet.multiversx.com"
+    TESTNET = "https://testnet-wallet.multiversx.com"
+
+
+class ApiUrls(Enum):
+    DEVNET = "https://devnet-api.multiversx.com"
+    TESTNET = "https://testnet-api.multiversx.com"
+
+
+def setup_parser(args: List[str], subparsers: Any) -> Any:
+    parser = cli_shared.add_group_subparser(subparsers, "faucet", "Get xEGLD on Devnet or Testnet")
+    subparsers = parser.add_subparsers()
+
+    sub = cli_shared.add_command_subparser(subparsers, "faucet", "request", "Request xEGLD.")
+    cli_shared.add_wallet_args(args, sub)
+    sub.add_argument("--chain", required=True, help="the chain identifier")
+    sub.set_defaults(func=faucet)
+
+    parser.epilog = cli_shared.build_group_epilog(subparsers)
+    return subparsers
+
+
+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)
+    client = NativeAuthClient(config)
+
+    init_token = client.initialize()
+    token_for_siginig = f"{account.address.to_bech32()}{init_token}"
+    signature = account.sign_message(token_for_siginig.encode())
+
+    access_token = client.get_token(
+        address=account.address.to_bech32(),
+        token=init_token,
+        signature=signature
+    )
+
+    logger.info(f"Requesting funds for address: {account.address.to_bech32()}")
+    call_web_wallet_faucet(wallet_url=wallet, access_token=access_token)
+
+
+def call_web_wallet_faucet(wallet_url: str, access_token: str):
+    faucet_url = f"{wallet_url}/faucet?accessToken={access_token}"
+    webbrowser.open_new_tab(faucet_url)
+
+
+def get_wallet_and_api_urls(args: Any) -> Tuple[str, str]:
+    chain: str = args.chain
+
+    if chain.upper() == "D":
+        return WebWalletUrls.DEVNET.value, ApiUrls.DEVNET.value
+
+    if chain.upper() == "T":
+        return WebWalletUrls.TESTNET.value, ApiUrls.TESTNET.value
+
+    raise BadUserInput("Invalid chain id. Choose between 'D' for devnet and 'T' for testnet.")

multiversx_sdk_cli/errors.py

@@ -177,3 +177,8 @@ def __init__(self, message: str):
 class QueryContractError(KnownError):
     def __init__(self, message: str, inner: Any = None):
         super().__init__(message, str(inner))
+
+
+class NativeAuthClientError(KnownError):
+    def __init__(self, message: str):
+        super().__init__(message)

multiversx_sdk_cli/native_auth_client.py

@@ -0,0 +1,104 @@
+import base64
+import json
+from typing import Any, Dict, Optional
+
+import requests
+
+from multiversx_sdk_cli.errors import NativeAuthClientError
+
+DEFAULT_EXPIRY_TIME_IN_SECONDS = 60 * 60 * 2
+DEFAULT_API_URL = "https://api.multiversx.com"
+
+
+class NativeAuthClientConfig:
+    def __init__(
+            self,
+            origin: str = '',
+            api_url: str = DEFAULT_API_URL,
+            expiry_seconds: int = DEFAULT_EXPIRY_TIME_IN_SECONDS,
+            block_hash_shard: Optional[int] = None,
+            gateway_url: Optional[str] = None,
+            extra_request_headers: Optional[Dict[str, str]] = None
+    ) -> None:
+        self.origin = origin
+        self.api_url = api_url
+        self.expiry_seconds = expiry_seconds
+        self.block_hash_shard = block_hash_shard
+        self.gateway_url = gateway_url
+        self.extra_request_headers = extra_request_headers
+
+
+class NativeAuthClient:
+    def __init__(self, config: Optional[NativeAuthClientConfig] = None) -> None:
+        self.config = config or NativeAuthClientConfig()
+
+    def initialize(self, extra_info: Dict[Any, Any] = {}) -> str:
+        block_hash = self.get_current_block_hash()
+        encoded_extra_info = self._encode_value(json.dumps(extra_info))
+        encoded_origin = self._encode_value(self.config.origin)
+
+        return f"{encoded_origin}.{block_hash}.{self.config.expiry_seconds}.{encoded_extra_info}"
+
+    def get_token(self, address: str, token: str, signature: str) -> str:
+        encoded_address = self._encode_value(address)
+        encoded_token = self._encode_value(token)
+
+        return f"{encoded_address}.{encoded_token}.{signature}"
+
+    def get_current_block_hash(self) -> str:
+        if self.config.gateway_url:
+            return self._get_current_block_hash_using_gateway()
+        return self._get_current_block_hash_using_api()
+
+    def _get_current_block_hash_using_gateway(self) -> str:
+        round = self._get_current_round()
+        url = f"{self.config.gateway_url}/blocks/by-round/{round}"
+        response = self._execute_request(url)
+        blocks = response["data"]["blocks"]
+        block = [b for b in blocks if b["shard"] == self.config.block_hash_shard][0]
+        return block["hash"]
+
+    def _get_current_round(self) -> int:
+        if self.config.gateway_url is None:
+            raise NativeAuthClientError("Gateway URL not set")
+
+        if self.config.block_hash_shard is None:
+            raise NativeAuthClientError("Blockhash shard not set")
+
+        url = f"{self.config.gateway_url}/network/status/{self.config.block_hash_shard}"
+        response = self._execute_request(url)
+        status = response["data"]["status"]
+
+        return status["erd_current_round"]
+
+    def _get_current_block_hash_using_api(self) -> str:
+        try:
+            url = f"{self.config.api_url}/blocks/latest?ttl={self.config.expiry_seconds}&fields=hash"
+            response = self._execute_request(url)
+            if response["hash"]:
+                return response["hash"]
+        except Exception:
+            pass
+
+        return self._get_current_block_hash_using_api_fallback()
+
+    def _get_current_block_hash_using_api_fallback(self) -> str:
+        url = f"{self.config.api_url}/blocks?size=1&fields=hash"
+
+        if self.config.block_hash_shard:
+            url += f"&shard={self.config.block_hash_shard}"
+
+        response = self._execute_request(url)
+        return response[0]["hash"]
+
+    def _encode_value(self, string: str) -> str:
+        encoded = base64.b64encode(string.encode('utf-8')).decode('utf-8')
+        return self._escape(encoded)
+
+    def _escape(self, string: str) -> str:
+        return string.replace("+", "-").replace("/", "_").replace("=", "")
+
+    def _execute_request(self, url: str) -> Any:
+        response = requests.get(url=url, headers=self.config.extra_request_headers)
+        response.raise_for_status()
+        return response.json()

multiversx_sdk_cli/tests/test_native_auth_client.py

@@ -0,0 +1,128 @@
+from typing import Any, List
+
+import pytest
+
+from multiversx_sdk_cli.native_auth_client import (NativeAuthClient,
+                                                   NativeAuthClientConfig)
+
+
+def mock(mocker: Any, code: int, response: Any):
+    mock_response = mocker.Mock()
+    mock_response.status_code = code
+    mock_response.json.return_value = response
+    mocker.patch("requests.get", return_value=mock_response)
+
+
+def mock_side_effect(mocker: Any, responses: List[Any]):
+    def side_effect(*args: Any, **kwargs: Any):
+        response = responses.pop(0)
+        mock_response = mocker.Mock()
+        mock_response.status_code = response["code"]
+        mock_response.json.return_value = response["response"]
+        return mock_response
+
+    mocker.patch("requests.get", side_effect=side_effect)
+
+
+class TestNativeAuth:
+    ADDRESS = 'erd1qnk2vmuqywfqtdnkmauvpm8ls0xh00k8xeupuaf6cm6cd4rx89qqz0ppgl'
+    SIGNATURE = '906e79d54e69e688680abee54ec0c49ce2561eb5abfd01865b31cb3ed738272c7cfc4fc8cc1c3590dd5757e622639b01a510945d7f7c9d1ceda20a50a817080d'
+    BLOCK_HASH = 'ab459013b27fdc6fe98eed567bd0c1754e0628a4cc16883bf0170a29da37ad46'
+    TTL = 86400
+    ORIGIN = 'https://api.multiversx.com'
+    TOKEN = f"aHR0cHM6Ly9hcGkubXVsdGl2ZXJzeC5jb20.{BLOCK_HASH}.{TTL}.e30"
+    ACCESS_TOKEN = 'ZXJkMXFuazJ2bXVxeXdmcXRkbmttYXV2cG04bHMweGgwMGs4eGV1cHVhZjZjbTZjZDRyeDg5cXF6MHBwZ2w.YUhSMGNITTZMeTloY0drdWJYVnNkR2wyWlhKemVDNWpiMjAuYWI0NTkwMTNiMjdmZGM2ZmU5OGVlZDU2N2JkMGMxNzU0ZTA2MjhhNGNjMTY4ODNiZjAxNzBhMjlkYTM3YWQ0Ni44NjQwMC5lMzA.906e79d54e69e688680abee54ec0c49ce2561eb5abfd01865b31cb3ed738272c7cfc4fc8cc1c3590dd5757e622639b01a510945d7f7c9d1ceda20a50a817080d'
+    INVALID_HASH_ERROR = 'Validation failed for block hash \'hash\'. Length should be 64.'
+
+    def test_latest_block_should_return_signable_token(self, mocker: Any):
+        mock(mocker, 200, [{"hash": self.BLOCK_HASH}])
+        config = NativeAuthClientConfig(origin=self.ORIGIN, expiry_seconds=self.TTL)
+        client = NativeAuthClient(config)
+        token = client.initialize()
+        assert token == self.TOKEN
+
+    def test_throws_internal_server_error(self, mocker: Any):
+        mock(mocker, 500, {})
+        client = NativeAuthClient()
+        with pytest.raises(Exception):
+            client.initialize()
+
+    # if `/blocks/latest` raises error should fallback to `/blocks?size=1`
+    def test_fallback_mechanism(self, mocker: Any):
+        mock(mocker, 400, [{"statusCode": 400,
+                            "message": self.INVALID_HASH_ERROR,
+                            "error": "Bad request"}])
+        mock(mocker, 200, {"hash": self.BLOCK_HASH})
+
+        config = NativeAuthClientConfig(origin=self.ORIGIN, expiry_seconds=self.TTL)
+        client = NativeAuthClient(config)
+
+        token = client.initialize()
+        assert token == self.TOKEN
+
+    def test_generate_access_token(self):
+        client = NativeAuthClient()
+        access_token = client.get_token(self.ADDRESS, self.TOKEN, self.SIGNATURE)
+        assert access_token == self.ACCESS_TOKEN
+
+
+class TestNativeAuthWithGateway:
+    ADDRESS = 'erd1qnk2vmuqywfqtdnkmauvpm8ls0xh00k8xeupuaf6cm6cd4rx89qqz0ppgl'
+    SIGNATURE = '906e79d54e69e688680abee54ec0c49ce2561eb5abfd01865b31cb3ed738272c7cfc4fc8cc1c3590dd5757e622639b01a510945d7f7c9d1ceda20a50a817080d'
+    BLOCK_HASH = 'ab459013b27fdc6fe98eed567bd0c1754e0628a4cc16883bf0170a29da37ad46'
+    TTL = 86400
+    ORIGIN = 'https://api.multiversx.com'
+    TOKEN = f"aHR0cHM6Ly9hcGkubXVsdGl2ZXJzeC5jb20.{BLOCK_HASH}.{TTL}.e30"
+    ACCESS_TOKEN = 'ZXJkMXFuazJ2bXVxeXdmcXRkbmttYXV2cG04bHMweGgwMGs4eGV1cHVhZjZjbTZjZDRyeDg5cXF6MHBwZ2w.YUhSMGNITTZMeTloY0drdWJYVnNkR2wyWlhKemVDNWpiMjAuYWI0NTkwMTNiMjdmZGM2ZmU5OGVlZDU2N2JkMGMxNzU0ZTA2MjhhNGNjMTY4ODNiZjAxNzBhMjlkYTM3YWQ0Ni44NjQwMC5lMzA.906e79d54e69e688680abee54ec0c49ce2561eb5abfd01865b31cb3ed738272c7cfc4fc8cc1c3590dd5757e622639b01a510945d7f7c9d1ceda20a50a817080d'
+    LATEST_ROUND = 115656
+    METASHARD = 4294967295
+    GATEWAY = 'https://gateway.multiversx.com'
+
+    def test_latest_block_should_return_signable_token(self, mocker: Any):
+        responses = [
+            {"code": 200, "response": {"data": {"status": {"erd_current_round": self.LATEST_ROUND}}}},
+            {"code": 200, "response": {"data": {"blocks": [{"shard": self.METASHARD, "hash": self.BLOCK_HASH}]}}}
+        ]
+        mock_side_effect(mocker, responses)
+
+        config = NativeAuthClientConfig(origin=self.ORIGIN, gateway_url=self.GATEWAY, block_hash_shard=self.METASHARD, expiry_seconds=self.TTL)
+        client = NativeAuthClient(config)
+        token = client.initialize()
+        assert token == self.TOKEN
+
+    def test_should_raise_internal_server_error(self, mocker: Any):
+        responses = [
+            {"code": 500, "response": {"data": {"status": {"erd_current_round": self.LATEST_ROUND}}}}
+        ]
+        mock_side_effect(mocker, responses)
+
+        config = NativeAuthClientConfig(gateway_url=self.GATEWAY, block_hash_shard=self.METASHARD)
+        client = NativeAuthClient(config)
+
+        with pytest.raises(Exception):
+            client.initialize()
+
+    def test_raises_internal_server_error_on_second_request(self, mocker: Any):
+        responses = [
+            {"code": 200, "response": {"data": {"status": {"erd_current_round": self.LATEST_ROUND}}}},
+            {"code": 500, "response": {""}}
+        ]
+        mock_side_effect(mocker, responses)
+
+        config = NativeAuthClientConfig(gateway_url=self.GATEWAY, block_hash_shard=self.METASHARD)
+        client = NativeAuthClient(config)
+
+        with pytest.raises(Exception):
+            client.initialize()
+
+    def test_generate_access_token(self):
+        config = NativeAuthClientConfig(gateway_url=self.GATEWAY, block_hash_shard=self.METASHARD)
+        client = NativeAuthClient(config)
+
+        access_token = client.get_token(
+            address=self.ADDRESS,
+            token=self.TOKEN,
+            signature=self.SIGNATURE
+        )
+
+        assert access_token == self.ACCESS_TOKEN

requirements-dev.txt

@@ -1,3 +1,4 @@
 pytest
 flake8
 autopep8
+pytest-mock