diff --git a/src/lighthouseweb3/__init__.py b/src/lighthouseweb3/__init__.py index b1d8d7c..159dc3b 100644 --- a/src/lighthouseweb3/__init__.py +++ b/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 \ No newline at end of file diff --git a/src/lighthouseweb3/functions/config.py b/src/lighthouseweb3/functions/config.py index 000c5ef..c2e70ec 100644 --- a/src/lighthouseweb3/functions/config.py +++ b/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" \ No newline at end of file diff --git a/src/lighthouseweb3/functions/encryption/share_to_address.py b/src/lighthouseweb3/functions/encryption/share_to_address.py new file mode 100644 index 0000000..212f0dd --- /dev/null +++ b/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) + } \ No newline at end of file diff --git a/src/lighthouseweb3/functions/encryption/utils.py b/src/lighthouseweb3/functions/encryption/utils.py new file mode 100644 index 0000000..d5b859a --- /dev/null +++ b/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) \ No newline at end of file diff --git a/tests/test_encryption/test_share_to_address.py b/tests/test_encryption/test_share_to_address.py new file mode 100644 index 0000000..c9ad203 --- /dev/null +++ b/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"]) + + + + + + \ No newline at end of file