diff --git a/src/lighthouseweb3/__init__.py b/src/lighthouseweb3/__init__.py index b1d8d7c..e337899 100644 --- a/src/lighthouseweb3/__init__.py +++ b/src/lighthouseweb3/__init__.py @@ -2,6 +2,9 @@ import os import io + +from .functions.encryption import get_access_condition as getAccessCondition + from .functions import ( upload as d, deal_status, @@ -224,3 +227,19 @@ def getTagged(self, tag: str): except Exception as e: raise e + +class Kavach: + @staticmethod + def getAccessCondition(cid: str): + """ + Get Access Condition for cid from the node + + :param cid: str, Content Identifier for the data to be downloaded + :return: conditions dict of access conditions + """ + + try: + return getAccessCondition.get_access_condition(cid) + except Exception as e: + raise e + 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/get_access_condition.py b/src/lighthouseweb3/functions/encryption/get_access_condition.py new file mode 100644 index 0000000..bafd210 --- /dev/null +++ b/src/lighthouseweb3/functions/encryption/get_access_condition.py @@ -0,0 +1,9 @@ +from typing import Any +from .utils import api_node_handler + +def get_access_condition(cid: str) -> dict[str, Any]: + try: + conditions = api_node_handler(f"/api/fileAccessConditions/get/{cid}", "GET") + return {'data': conditions} + except Exception as error: + raise error \ 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..abd56f6 --- /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_get_access_condition.py b/tests/test_encryption/test_get_access_condition.py new file mode 100644 index 0000000..5fade95 --- /dev/null +++ b/tests/test_encryption/test_get_access_condition.py @@ -0,0 +1,18 @@ +import unittest +from src.lighthouseweb3 import Kavach + + +class TestGetAccessCondition(unittest.TestCase): + def test_access_condition(self): + condition = Kavach.getAccessCondition("QmbFMke1KXqnYyBBWxB74N4c5SBnJMVAiMNRcGu6x1AwQH") + self.assertIsInstance(condition, dict, "condition is a dict") + self.assertIsInstance(condition.get('data'), dict, "data is a dict") + self.assertIn("([2] and [1])", str(condition.get('data'))) + self.assertEqual("0xeaf4e24ffc1a2f53c07839a74966a6611b8cb8a1", condition.get('data').get('owner')) + + def test_access_condition_invalid_cid(self): + condition = Kavach.getAccessCondition("invalid_cid") + self.assertIsInstance(condition, dict, "condition is a dict") + self.assertIsInstance(condition.get('data'), dict, "data is a dict") + self.assertEqual(condition.get('data').get('conditions'), [], "conditions is a list") +