Method save shards

src/lighthouseweb3/__init__.py

@@ -17,6 +17,14 @@
     create_wallet as createWallet
 )
 
+from .functions.kavach import(
+    get_auth_message as getAuthMessage,
+    save_shards as saveShards
+)
+
+from typing import List, Dict, Any, Union
+from .functions.kavach.types import AuthToken, KeyShard
+
 
 class Lighthouse:
     def __init__(self, token: str = ""):
@@ -224,3 +232,39 @@ def getTagged(self, tag: str):
         except Exception as e:
             raise e
 
+class Kavach:
+
+    @staticmethod
+    def getAuthMessage(address: str):
+        """
+        Retrieves an authentication message for a given address.
+
+        :param address: str, The address for which to retrieve the authentication message.
+        :return: dict, A dictionary containing the authentication message.
+        """
+        try:
+            return getAuthMessage.get_auth_message(address)
+        except Exception as e:
+            raise e
+
+    def saveShards(
+        address: str,
+        cid: str,
+        auth_token: AuthToken,
+        key_shards: List[KeyShard],
+        share_to: List[str] = []
+    ) -> Dict[str, Union[bool, str, None]]:
+        """
+        Save shards for a given address and CID.
+
+        :param address: str, The address for which to save the shards.
+        :param cid: str, The content identifier for the data.
+        :param auth_token: AuthToken, The authentication token.
+        :param key_shards: List[KeyShard], The list of key shards to save.
+        :param share_to: List[str], The list of addresses to share the shards with (optional).
+        :return: dict, A dictionary containing the result of the operation.
+        """
+        try:
+            return saveShards.save_shards(address, cid, auth_token, key_shards, share_to)
+        except Exception as e:
+            raise e

src/lighthouseweb3/functions/config.py

@@ -9,3 +9,7 @@ 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/kavach/get_auth_message.py

@@ -0,0 +1,10 @@
+from typing import Any
+from .util import api_node_handler
+
+
+async def get_auth_message(address: str) -> dict[str, Any]:
+  try:
+    response = await api_node_handler(f"/api/message/{address}", "GET")
+    return {'message': response[0]['message'], 'error': None}
+  except Exception as e:
+    return {'message': None, 'error':str(e)}

src/lighthouseweb3/functions/kavach/save_shards.py

@@ -0,0 +1,75 @@
+import asyncio
+from typing import List, Dict, Any, Union
+from .util import api_node_handler, is_cid_reg, is_equal
+from .types import AuthToken, KeyShard  
+
+async def save_shards(
+    address: str,
+    cid: str,
+    auth_token: AuthToken,
+    key_shards: List[KeyShard],
+    share_to: List[str] = []
+) -> Dict[str, Union[bool, str, None]]:
+
+    if not is_cid_reg(cid):
+        return {
+            "isSuccess": False,
+            "error": "Invalid CID"
+        }
+
+    if not isinstance(key_shards, list) or len(key_shards) != 5:
+        return {
+            "isSuccess": False,
+            "error": "keyShards must be an array of 5 objects"
+        }
+
+    try:
+        node_ids = [1, 2, 3, 4, 5]
+        node_urls = [f"/api/setSharedKey/{i}" for i in node_ids]
+
+        async def request_data(url: str, index: int) -> Dict[str, Any]:
+            try:
+                payload = {
+                    "address": address,
+                    "cid": cid,
+                    "payload": key_shards[index]
+                }
+                if share_to:
+                    payload["sharedTo"] = share_to
+
+                response = await api_node_handler(url, "POST", auth_token, payload)
+                return response
+
+            except Exception as error:
+                return {
+                    "error": error
+                }
+
+        data = []
+        for index, url in enumerate(node_urls):
+            response = await request_data(url, index)
+            if "error" in response:
+                try:
+                    return {
+                        "isSuccess": False,
+                        "error": str(response["error"])
+                    }
+                except Exception:
+                    return {
+                        "isSuccess": False,
+                        "error": "Unknown error"
+                    }
+            await asyncio.sleep(1)
+            data.append(response)
+
+        temp = [{**elem, "data": None} for elem in data]
+        return {
+            "isSuccess": is_equal(*temp) and data[0].get("message") == "success",
+            "error": None
+        }
+
+    except Exception as err:
+        return {
+            "isSuccess": False,
+            "error": str(err)
+        }

src/lighthouseweb3/functions/kavach/types.py

@@ -0,0 +1,156 @@
+from typing import List, Dict, Union, Optional, Any, Literal
+from dataclasses import dataclass
+from enum import Enum
+
+ErrorValue = Union[str, List[str], int, bool, None, Dict[str, Any], Any]
+SignedMessage = str
+JWT = str 
+AuthToken = Union[SignedMessage, JWT]
+
+class ChainType(str, Enum):
+    EVM = "EVM"
+    EVM_LOWER = "evm"
+    SOLANA = "SOLANA"
+    SOLANA_LOWER = "solana"
+
+class DecryptionType(str, Enum):
+    ADDRESS = "ADDRESS"
+    ACCESS_CONDITIONS = "ACCESS_CONDITIONS"
+
+class StandardContractType(str, Enum):
+    ERC20 = "ERC20"
+    ERC721 = "ERC721"
+    ERC1155 = "ERC1155"
+    CUSTOM = "Custom"
+    EMPTY = ""
+
+class SolanaContractType(str, Enum):
+    SPL_TOKEN = "spl-token"
+    EMPTY = ""
+
+class Comparator(str, Enum):
+    EQUAL = "=="
+    GREATER_EQUAL = ">="
+    LESS_EQUAL = "<="
+    NOT_EQUAL = "!="
+    GREATER = ">"
+    LESS = "<"
+
+# Data Classes
+@dataclass
+class KeyShard:
+    key: str
+    index: str
+
+@dataclass
+class GeneratedKey:
+    master_key: Optional[str]
+    key_shards: List[KeyShard]
+
+@dataclass
+class GenerateInput:
+    threshold: Optional[int] = None
+    key_count: Optional[int] = None
+
+@dataclass
+class AuthMessage:
+    message: Optional[str]
+    error: Optional[ErrorValue]
+
+@dataclass
+class RecoveredKey:
+    master_key: Optional[str]
+    error: Optional[ErrorValue]
+
+@dataclass
+class RecoverShards:
+    shards: List[KeyShard]
+    error: ErrorValue
+
+@dataclass
+class LightHouseSDKResponse:
+    is_success: bool
+    error: ErrorValue
+
+@dataclass
+class ReturnValueTest:
+    comparator: Comparator
+    value: Union[int, str, List[Any]]
+
+@dataclass
+class PDAInterface:
+    offset: Optional[int] = None
+    selector: Optional[str] = None
+
+@dataclass
+class EVMCondition:
+    id: int
+    standard_contract_type: StandardContractType
+    chain: str
+    method: str
+    return_value_test: ReturnValueTest
+    contract_address: Optional[str] = None
+    parameters: Optional[List[Any]] = None
+    input_array_type: Optional[List[str]] = None
+    output_type: Optional[str] = None
+
+@dataclass
+class SolanaCondition:
+    id: int
+    chain: str
+    method: str
+    standard_contract_type: SolanaContractType
+    pda_interface: PDAInterface
+    return_value_test: ReturnValueTest
+    contract_address: Optional[str] = None
+    parameters: Optional[List[Any]] = None
+
+# Union Type for Conditions
+Condition = Union[EVMCondition, SolanaCondition]
+
+@dataclass
+class UpdateConditionSchema:
+    chain_type: Literal["EVM", "SOLANA"]
+    conditions: List[Condition]
+    decryption_type: Literal["ADDRESS", "ACCESS_CONDITIONS"]
+    address: str
+    cid: str
+    aggregator: Optional[str] = None
+
+@dataclass
+class AccessConditionSchema:
+    chain_type: Literal["EVM", "SOLANA"]
+    conditions: List[Condition]
+    decryption_type: Literal["ADDRESS", "ACCESS_CONDITIONS"]
+    address: str
+    cid: str
+    key_shards: List[Any]
+    aggregator: Optional[str] = None
+
+@dataclass
+class IGetAccessCondition:
+    aggregator: str
+    owner: str
+    cid: str
+    conditions: Optional[List[Condition]] = None
+    conditions_solana: Optional[List[Any]] = None
+    shared_to: Optional[List[Any]] = None
+
+def is_jwt(token: str) -> bool:
+    """Check if token is a JWT (starts with 'jwt:')"""
+    return token.startswith('jwt:')
+
+def create_jwt(token: str) -> JWT:
+    """Create a JWT token with proper prefix"""
+    if not token.startswith('jwt:'):
+        return f'jwt:{token}'
+    return token
+
+# Type Guards
+def is_evm_condition(condition: Condition) -> bool:
+    """Check if condition is an EVM condition"""
+    return isinstance(condition, EVMCondition)
+
+def is_solana_condition(condition: Condition) -> bool:
+    """Check if condition is a Solana condition"""
+    return isinstance(condition, SolanaCondition)