feature: kavach share to address implemented and tested

src/lighthouseweb3/__init__.py

@@ -2,6 +2,10 @@
 
 import os
 import io
+from typing import Any, Dict, List
+
+from .functions.encryption import share_to_address as shareToAddress
+
 from .functions import (
     upload as d,
     deal_status, 
@@ -224,3 +228,20 @@ def getTagged(self, tag: str):
         except Exception as e:
             raise e
 
+
+class Kavach:
+    @staticmethod
+    def shareToAddress(address: str, cid: str, auth_token: Dict[str, Any], share_to: List[str]) -> Dict[str, Any]:
+        """
+        Share an encrypted file with a list of addresses.
+
+        :param address: str, The public address of the file owner.
+        :param cid: str, The CID of the file to share.
+        :param auth_token: Dict[str, Any], The authentication token.
+        :param share_to: List[str], A list of public addresses to share the file with.
+        :return: Dict[str, Any], A dictionary indicating the result of the share operation.
+        """
+        try:
+            return shareToAddress.share_to_address(address, cid, auth_token, share_to)
+        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/share_to_address.py

@@ -0,0 +1,54 @@
+import json
+import time
+
+from typing import List, Dict, Any
+from .utils import is_equal, is_cid_reg, api_node_handler
+
+def share_to_address(address: str, cid: str, auth_token: Dict[str, Any], share_to: List[str]) -> Dict[str, Any]:
+    if not is_cid_reg(cid):
+        return {
+            "isSuccess": False,
+            "error": "Invalid CID"
+        }
+
+    try:
+        node_id = [1, 2, 3, 4, 5]
+        node_url = [f"/api/setSharedKey/{elem}" for elem in node_id]
+
+        def request_data(url: str) -> Dict[str, Any]:
+            try:
+                response = api_node_handler(url, "PUT", auth_token, {
+                    "address": address,
+                    "cid": cid,
+                    "shareTo": share_to
+                })
+                return response
+            except Exception as error:
+                return {"error": str(error)}
+
+        data = []
+        for url in node_url:
+            response = request_data(url)
+            if "error" in response:
+                try:
+                    error_message = json.loads(response.get("error", {}))
+                except json.JSONDecodeError:
+                    error_message = response.get("error")
+                return {
+                    "isSuccess": False,
+                    "error": error_message
+                }
+            time.sleep(1)  
+            data.append(response)
+
+        temp = [{**elem, "data": None} for elem in data]
+        return {
+            "isSuccess": is_equal(*temp) and temp[0].get("message") == "success",
+            "error": None
+        }
+
+    except Exception as err:
+        return {
+            "isSuccess": False,
+            "error": str(err)
+        }

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_share_to_address.py

@@ -0,0 +1,70 @@
+import unittest
+import os
+from eth_account.messages import encode_defunct
+from web3 import Web3
+from src.lighthouseweb3.functions.config import Config
+from src.lighthouseweb3 import Kavach
+from src.lighthouseweb3.functions.encryption.utils import api_node_handler
+
+
+def get_auth_message(public_key: str) -> dict:
+    response = api_node_handler(f"/api/message/{public_key}", "GET")
+    return response[0]['message']
+
+class TestKavach(unittest.TestCase):
+    def test_share_to_address(self):
+      public_key = os.environ.get("PUBLIC_KEY")
+
+      verification_message = get_auth_message(public_key)
+      self.assertIn(
+        "Please sign this message to prove you are owner of this account", 
+        verification_message, 
+        "Owner response should come"
+      )
+
+      auth_token = Web3().eth.account.sign_message(
+        encode_defunct(text=verification_message), 
+        private_key=os.environ.get("PRIVATE_KEY")
+      ).signature.hex()
+
+      result = Kavach.shareToAddress(
+        address=public_key,
+        cid = "QmbFMke1KXqnYyBBWxB74N4c5SBnJMVAiMNRcGu6x1AwQH",
+        auth_token=f"0x{auth_token}",
+        share_to=["0xF0Bc72fA04aea04d04b1fA80B359Adb566E1c8B1"]
+      )
+
+      self.assertIsNone(result["error"])
+      self.assertTrue(result["isSuccess"])
+
+    def test_share_to_address_invalid_address(self):
+      public_key = os.environ.get("PUBLIC_KEY")
+
+      verification_message = get_auth_message(public_key)
+      self.assertIn(
+        "Please sign this message to prove you are owner of this account", 
+        verification_message, 
+        "Owner response should come"
+      )
+
+      auth_token = Web3().eth.account.sign_message(
+        encode_defunct(text=verification_message), 
+        private_key=os.environ.get("PRIVATE_KEY")
+      ).signature.hex()
+
+
+      result = Kavach.shareToAddress(
+        address='0x344b0b6C1C5b8f4519db43dFb388b65ecA667243',
+        cid = "QmbFMke1KXqnYyBBWxB74N4c5SBnJMVAiMNRcGu6x1AwQH",
+        auth_token=f"0x{auth_token}",
+        share_to=["0xF0Bc72fA04aea04d04b1fA80B359Adb566E1c8B1"]
+      )
+
+      self.assertFalse(result["isSuccess"])
+      self.assertIsNotNone(result["error"])
+
+
+
+
+
+