diff --git a/src/lighthouseweb3/__init__.py b/src/lighthouseweb3/__init__.py index b1d8d7c..53995e0 100644 --- a/src/lighthouseweb3/__init__.py +++ b/src/lighthouseweb3/__init__.py @@ -2,6 +2,12 @@ import os import io +from typing import Any + +from .functions.encryption import ( + transfer_ownership as transferOwnership +) + from .functions import ( upload as d, deal_status, @@ -224,3 +230,21 @@ def getTagged(self, tag: str): except Exception as e: raise e + +class Kavach: + @staticmethod + def transferOwnership(address: str, cid: str, new_owner: str, auth_token: str, reset_shared_to: bool = True) -> dict[str, Any]: + """ + Transfer ownership of a file from the current owner to a new owner. + + :param address: str, The address of the current owner. + :param cid: str, The Content Identifier (CID) of the file to transfer. + :param new_owner: str, The address of the new owner. + :param auth_token: str, The authentication token for the current owner. + :param reset_shared_to: bool, Whether to reset the list of users the file is shared with (default: True). + :return: dict, A dictionary indicating the success or failure of the operation. + """ + try: + return transferOwnership.transfer_ownership(address, cid, new_owner, auth_token, reset_shared_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..2f62d25 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/transfer_ownership.py b/src/lighthouseweb3/functions/encryption/transfer_ownership.py new file mode 100644 index 0000000..e065706 --- /dev/null +++ b/src/lighthouseweb3/functions/encryption/transfer_ownership.py @@ -0,0 +1,53 @@ +import json +import time +from typing import Any +from .utils import api_node_handler, is_cid_reg, is_equal + +def transfer_ownership(address: str, cid: str, new_owner: str, auth_token: str, reset_shared_to: bool = True) -> dict[str, Any]: + if not is_cid_reg(cid): + return { + "isSuccess": False, + "error": "Invalid CID" + } + + try: + node_index_selected = [1, 2, 3, 4, 5] + node_urls = [f"/api/transferOwnership/{elem}" for elem in node_index_selected] + + def request_data(url: str) -> dict: + try: + response = api_node_handler(url, "POST", auth_token, { + "address": address, + "cid": cid, + "newOwner": new_owner, + "resetSharedTo": reset_shared_to + }) + return response + except Exception as error: + return {"error": str(error)} + + data = [] + for url in node_urls: + 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) # Delay between requests + data.append(response) + + return { + "isSuccess": is_equal(*data) and data[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_transfer_ownership.py b/tests/test_encryption/test_transfer_ownership.py new file mode 100644 index 0000000..1071eeb --- /dev/null +++ b/tests/test_encryption/test_transfer_ownership.py @@ -0,0 +1,62 @@ +import unittest +import os +from eth_account.messages import encode_defunct +from web3 import Web3 +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 TestKavachTransferOwnership(unittest.TestCase): + def test_transfer_ownership(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.transferOwnership( + address=public_key, + cid = "QmbFMke1KXqnYyBBWxB74N4c5SBnJMVAiMNRcGu6x1AwQH", + auth_token=f"0x{auth_token}", + new_owner="0xF0Bc72fA04aea04d04b1fA80B359Adb566E1c8B1" + ) + + self.assertIsNone(result["error"]) + self.assertTrue(result["isSuccess"]) + + def test_transfer_ownership_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.transferOwnership( + address='0x344b0b6C1C5b8f4519db43dFb388b65ecA667243', + cid = "QmbFMke1KXqnYyBBWxB74N4c5SBnJMVAiMNRcGu6x1AwQH", + auth_token=f"0x{auth_token}", + new_owner="0xF0Bc72fA04aea04d04b1fA80B359Adb566E1c8B1" + ) + + self.assertFalse(result["isSuccess"]) + self.assertIsNotNone(result["error"]) \ No newline at end of file