Sign all requests sent to a validator

cspell.json

@@ -12,6 +12,7 @@
         "levelname",
         "NETUID",
         "OMRON",
+        "opentensor",
         "pydantic",
         "substrateinterface",
         "subtensor",

proof_of_weights/main.py

@@ -1,12 +1,13 @@
 import base64
+import datetime
 import hashlib
 import json
-import typing
 import logging
-from yarl import URL
+import typing
 
 import bittensor
 import requests
+from yarl import URL
 
 __version__: typing.Final[str] = "0.0.5"
 OMRON_NETUID_FINNEY: typing.Final[int] = 2
@@ -16,12 +17,24 @@
 
 
 def get_omron_validator_axon(
-    omron_validator_ss58: str, network: str = "finney"
+    omron_validator_ss58: str,
+    network: str = "finney",
+    chain_endpoint: str = "",
 ) -> bittensor.AxonInfo:
     """
     Get the axon of a validator on the omron subnet.
     """
-    btnetwork = bittensor.subtensor(network=network)
+    config = bittensor.config()
+    config.subtensor = bittensor.config()
+    if chain_endpoint:
+        config.subtensor.chain_endpoint = chain_endpoint
+        config.subtensor.network = network
+    else:
+        config.subtensor.network, config.subtensor.chain_endpoint = (
+            bittensor.subtensor.determine_chain_endpoint_and_network(network)
+        )
+
+    btnetwork = bittensor.subtensor(config)
     omron_validator_axon = btnetwork.get_axon_info(
         netuid=(OMRON_NETUID_FINNEY if network == "finney" else OMRON_NETUID_TESTNET),
         hotkey_ss58=omron_validator_ss58,
@@ -37,13 +50,14 @@ def __init__(
         omron_validator_ss58: str,
         netuid: int,
         network: str = "finney",
+        chain_endpoint: str = "",
     ):
         """
         Initialize the Proof of Weights class with your wallet and a validator's hotkey from the omron subnet.
         """
         self._wallet = bittensor.wallet(wallet_name, wallet_hotkey)
         self._omron_validator_axon = get_omron_validator_axon(
-            omron_validator_ss58, network
+            omron_validator_ss58, network, chain_endpoint
         )
         self._netuid = netuid
         self._last_input_hash = ""
@@ -53,27 +67,42 @@ def __init__(
             port=self._omron_validator_axon.port,
         )
 
+    def get_signature_headers(
+        self,
+        url: str,
+        data: str = "",
+    ) -> dict[str, str]:
+        """
+        Sign the request with the wallet's hotkey and return proper headers with the signature.
+        """
+        request_time = datetime.datetime.now(datetime.timezone.utc).strftime(
+            "%Y-%m-%dT%H:%M:%S.%f%z"
+        )
+        payload = f"{url}{request_time}{self._netuid}{data}".encode()
+        signature = self._wallet.hotkey.sign(payload)
+        return {
+            "X-Request-Datetime": request_time,
+            "X-Request-Signature": base64.b64encode(signature).decode("utf-8"),
+            "X-Netuid": str(self._netuid),
+            "X-ss58-Address": self._wallet.hotkey.ss58_address,
+        }
+
     def submit_inputs(self, reward_function_inputs: dict | list) -> str:
         """
         Submit reward function inputs from network with netuid to a validator on the omron subnet.
         """
-        # serialize the reward function inputs as json bytes
-        input_bytes = json.dumps(reward_function_inputs).encode()
-        # sign the inputs with your hotkey
-        signature = self._wallet.hotkey.sign(input_bytes)
-        # encode the inputs and signature as base64
-        input_str = base64.b64encode(input_bytes).decode("utf-8")
-        signature_str = base64.b64encode(signature).decode("utf-8")
         self._last_input_hash = _hash_inputs(reward_function_inputs)
+        # serialize the reward function inputs as json bytes
+        input_str = json.dumps(reward_function_inputs)
+        url = self._base_url.with_path("submit-inputs")
 
         # send the reward function inputs and signature to the omron subnet on port API_PORT
         response = requests.post(
-            self._base_url.with_path("submit-inputs"),
-            json={
-                "inputs": input_str,
-                "signature": signature_str,
-                "sender": self._wallet.hotkey.ss58_address,
-                "netuid": self._netuid,
+            url=url,
+            data=input_str.encode(),
+            headers={
+                "Content-Type": "application/json",
+                **self.get_signature_headers(url=url, data=input_str),
             },
         )
         if response.status_code != 200:
@@ -97,11 +126,14 @@ def get_proof(self) -> dict:
         """
         Get the proof of weights from the omron subnet validator.
         """
+        url = self._base_url.with_path(f"get-proof-of-weights").with_query(
+            {"input_hash": self._last_input_hash}
+        )
         response = requests.get(
-            self._base_url.with_path(f"get-proof-of-weights").with_query(
-                {"input_hash": self._last_input_hash}
-            )
+            url=url,
+            headers=self.get_signature_headers(url=str(url)),
         )
+        print(f"Status code: {response.status_code}")
         if response.status_code != 200:
             return {}
         return response.json()