lighthouse-python-sdk/src/lighthouseweb3/functions/kavach/access_control/main.py at ffa08e3412f96da142c239e08a5722789fadc69b · lighthouse-web3/lighthouse-python-sdk · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
from .validator import UpdateConditionSchema as update_condition_schema, AccessConditionSchema as access_condition_schema
from ..types import (AuthToken,
                    Condition,
                    DecryptionType,
                    KeyShard,
                    ChainType,
                    LightHouseSDKResponse)


from src.lighthouseweb3.functions.config import Config
from src.lighthouseweb3.functions.kavach.util import is_equal, is_cid_reg, api_node_handler
from typing import List, Optional
import json
import time


async def access_control(
   address: str,
   cid: str,
   auth_token: AuthToken,
   conditions: List[Condition],
   aggregator: Optional[str] = None,
   chain_type: ChainType = "evm",
   key_shards: List[KeyShard] = [],
   decryption_type: DecryptionType = "ADDRESS"
) -> LightHouseSDKResponse:
   try:
       if not isinstance(key_shards, list) or (
           len(key_shards) != 5 and len(key_shards) != 0
       ):
           raise ValueError("keyShards must be an array of 5 objects")


       if not is_cid_reg(cid):
           raise ValueError("Invalid CID")


       try:
           if len(key_shards) == 5:
               access_condition_schema.model_validate({
                   "address": address,
                   "cid": cid,
                   "conditions": conditions,
                   "aggregator": aggregator,
                   "decryptionType": decryption_type,
                   "chainType": chain_type,
                   "keyShards": key_shards
               })
           else:
               update_condition_schema.model_validate({
                   "address": address,
                   "cid": cid,
                   "conditions": conditions,
                   "aggregator": aggregator,
                   "chainType": chain_type
               })
       except ValueError as e:
           raise ValueError(f"Condition validation error: {str(e)}")


       node_ids = [1, 2, 3, 4, 5]
       node_urls = [
           f":900{id}/api/fileAccessConditions/{id}" if Config.is_dev else f"/api/fileAccessConditions/{id}"
           for id in node_ids
       ]


       data = []


       for index, url in enumerate(node_urls):
           try:
               if len(key_shards) == 5:
                   response = await api_node_handler(
                       url, "POST", auth_token, {
                           "address": address,
                           "cid": cid,
                           "conditions": conditions,
                           "aggregator": aggregator,
                           "decryptionType": decryption_type,
                           "chainType": chain_type,
                           "payload": key_shards[index]
                       }
                   )
               else:
                   response = await api_node_handler(
                       url, "PUT", auth_token, {
                           "address": address,
                           "cid": cid,
                           "conditions": conditions,
                           "aggregator": aggregator,
                           "chainType": chain_type
                       }
                   )
           except Exception as e:
               try:
                   error_data = json.loads(str(e))
               except Exception:
                   error_data = {"message": str(e)}
               response = {"isSuccess": False, "error": error_data}


           if response.get("error"):
               time.sleep(1)


           data.append(response)


       success = (
           is_equal(*(resp.get("message") for resp in data)) and
           data[0].get("message") == "success"
       )


       return {"isSuccess": success, "error": None}


   except Exception as e:
       if isinstance(e, ValueError):
           return {"isSuccess": False, "error": str(e)}
       try:
           err = json.loads(str(e))
           if isinstance(err, dict) and "message" in err:
               return {"isSuccess": False, "error": err["message"]}
           return {"isSuccess": False, "error": str(e)}
       except Exception:
           return {"isSuccess": False, "error": str(e)}