feature: kavach get access condition function implemented and tested

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
+

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_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

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_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")
+