feature: kavach get auth message implemented and tested

src/lighthouseweb3/__init__.py

@@ -2,6 +2,10 @@
 
 import os
 import io
+from typing import Any
+
+from .functions.encryption import get_auth_message as getAuthMessage
+
 from .functions import (
     upload as d,
     deal_status, 
@@ -224,3 +228,19 @@ def getTagged(self, tag: str):
         except Exception as e:
             raise e
 
+
+
+class Kavach:
+    @staticmethod
+    def getAuthMessage(address: str) -> dict[str, Any]:
+        """
+        Get Authentication message from the server
+
+        :param address: str, The public key of the user
+        :return: dict, A dict with authentication message or error
+
+        """
+        try:
+            return getAuthMessage.get_auth_message(address)
+        except Exception as e:
+            raise e

src/lighthouseweb3/functions/config.py

@@ -9,3 +9,6 @@ class Config:
     lighthouse_node = "https://node.lighthouse.storage"
     lighthouse_bls_node = "https://encryption.lighthouse.storage"
     lighthouse_gateway = "https://gateway.lighthouse.storage/ipfs"
+
+    is_dev = False
+    lighthouse_bls_node_dev = "http://enctest.lighthouse.storage"

src/lighthouseweb3/functions/encryption/get_auth_message.py

@@ -0,0 +1,10 @@
+from typing import Any
+from .utils import api_node_handler
+
+
+def get_auth_message(address: str) -> dict[str, Any]:
+  try:
+    response = api_node_handler(f"/api/message/{address}", "GET")
+    return {'message': response[0]['message'], 'error': None}
+  except Exception as e:
+    return {'message': None, 'error':str(e)}

src/lighthouseweb3/functions/encryption/utils.py

@@ -0,0 +1,69 @@
+import re
+import json
+import time
+import requests
+from typing import Dict, Any
+from dataclasses import dataclass
+from src.lighthouseweb3.functions.config import Config
+
+
+def is_cid_reg(cid: str) -> bool:
+
+    pattern = r'Qm[1-9A-HJ-NP-Za-km-z]{44}|b[A-Za-z2-7]{58}|B[A-Z2-7]{58}|z[1-9A-HJ-NP-Za-km-z]{48}|F[0-9A-F]{50}'
+    return bool(re.match(pattern, cid))
+
+def is_equal(*objects: Any) -> bool:
+
+    if not objects:
+        return True
+    first = json.dumps(objects[0], sort_keys=True)
+    return all(json.dumps(obj, sort_keys=True) == first for obj in objects)
+
+def api_node_handler(
+    endpoint: str,
+    verb: str,
+    auth_token: str = "",
+    body: Any = None,
+    retry_count: int = 3
+) -> Dict[str, Any]:
+
+    url = f"{Config.is_dev and Config.lighthouse_bls_node_dev or Config.lighthouse_bls_node}{endpoint}"
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {auth_token}" if auth_token else ""
+    }
+
+    for attempt in range(retry_count):
+        try:
+            if verb in ["POST", "PUT", "DELETE"] and body is not None:
+                response = requests.request(
+                    method=verb,
+                    url=url,
+                    headers=headers,
+                    json=body
+                )
+            else:
+                response = requests.request(
+                    method=verb,
+                    url=url,
+                    headers=headers
+                )
+
+            if not response.ok:
+                if response.status_code == 404:
+                    raise Exception(json.dumps({
+                        "message": "fetch Error",
+                        "statusCode": response.status_code
+                    }))
+                error_body = response.json()
+                raise Exception(json.dumps({
+                    **error_body,
+                    "statusCode": response.status_code
+                }))
+            return response.json()
+        except Exception as error:
+            if "fetch" not in str(error):
+                raise
+            if attempt == retry_count - 1:
+                raise
+            time.sleep(1)  

tests/test_encryption/test_get_auth_message.py

@@ -0,0 +1,18 @@
+import unittest
+import os
+from src.lighthouseweb3 import Kavach
+
+class TestGetAuthMessage(unittest.TestCase):
+
+  def test_get_auth_message(self):
+    address = os.environ.get("PUBLIC_KEY")
+    auth_message = Kavach.getAuthMessage(address=address)
+    self.assertIn("Please sign this message to prove you are owner of this account", auth_message['message'], "Owner response should come")
+    self.assertEqual(None, auth_message['error'])
+
+  def test_get_auth_message_invalid_address(self):
+
+    auth_message = Kavach.getAuthMessage(address="0x9a40b8EE3B8Fe7eB621cd142a651560Fa7")
+    self.assertEqual(None, auth_message['message'])
+    self.assertNotEqual(None, auth_message['error'])
+    self.assertIn("invalid address", str(auth_message["error"]).lower())