Feature: Internal account management + fix on _load_account to handle SolAccount

Author: 1yam

pyproject.toml

@@ -34,6 +34,7 @@ dependencies = [
     "aleph-superfluid>=0.2.1",
     "eth_typing==4.3.1",
     "web3==6.3.0",
+    "aiofiles>=24.1.0",
 ]
 
 [project.optional-dependencies]

src/aleph/sdk/account.py

@@ -1,18 +1,99 @@
 import asyncio
+import json
 import logging
 from pathlib import Path
-from typing import Optional, Type, TypeVar
+from typing import Dict, List, Optional, Type, TypeVar, Union, overload
+
+import base58
+from aleph_message.models import Chain
 
 from aleph.sdk.chains.common import get_fallback_private_key
 from aleph.sdk.chains.ethereum import ETHAccount
 from aleph.sdk.chains.remote import RemoteAccount
+from aleph.sdk.chains.solana import SOLAccount
 from aleph.sdk.conf import settings
 from aleph.sdk.types import AccountFromPrivateKey
+from aleph.sdk.utils import parse_solana_private_key, solana_private_key_from_bytes
 
 logger = logging.getLogger(__name__)
 
 T = TypeVar("T", bound=AccountFromPrivateKey)
 
+CHAIN_TO_ACCOUNT_MAP: Dict[Chain, Type[AccountFromPrivateKey]] = {
+    Chain.ETH: ETHAccount,
+    Chain.AVAX: ETHAccount,
+    Chain.SOL: SOLAccount,
+    Chain.BASE: ETHAccount,
+}
+
+
+def detect_chain_from_private_key(private_key: Union[str, List[int], bytes]) -> Chain:
+    """
+    Detect the blockchain chain based on the private key format.
+    - Chain.ETH for Ethereum (EVM) private keys
+    - Chain.SOL for Solana private keys (base58 or uint8 format).
+
+    Raises:
+        ValueError: If the private key format is invalid or not recognized.
+    """
+    if isinstance(private_key, (str, bytes)) and is_valid_private_key(
+        private_key, ETHAccount
+    ):
+        return Chain.ETH
+
+    elif is_valid_private_key(private_key, SOLAccount):
+        return Chain.SOL
+
+    else:
+        raise ValueError("Unsupported private key format. Unable to detect chain.")
+
+
+@overload
+def is_valid_private_key(
+    private_key: Union[str, bytes], account_type: Type[ETHAccount]
+) -> bool: ...
+
+
+@overload
+def is_valid_private_key(
+    private_key: Union[str, List[int], bytes], account_type: Type[SOLAccount]
+) -> bool: ...
+
+
+def is_valid_private_key(
+    private_key: Union[str, List[int], bytes], account_type: Type[T]
+) -> bool:
+    """
+    Check if the private key is valid for either Ethereum or Solana based on the account type.
+    """
+    try:
+        if account_type == ETHAccount:
+            # Handle Ethereum private key validation
+            if isinstance(private_key, str):
+                if private_key.startswith("0x"):
+                    private_key = private_key[2:]
+                private_key = bytes.fromhex(private_key)
+            elif isinstance(private_key, list):
+                raise ValueError("Ethereum keys cannot be a list of integers")
+
+            account_type(private_key)
+
+        elif account_type == SOLAccount:
+            # Handle Solana private key validation
+            if isinstance(private_key, bytes):
+                return len(private_key) == 64
+            elif isinstance(private_key, str):
+                decoded_key = base58.b58decode(private_key)
+                return len(decoded_key) == 64
+            elif isinstance(private_key, list):
+                return len(private_key) == 64 and all(
+                    isinstance(i, int) and 0 <= i <= 255 for i in private_key
+                )
+
+        return True
+    except Exception:
+        return False
+
 
 def account_from_hex_string(private_key_str: str, account_type: Type[T]) -> T:
     if private_key_str.startswith("0x"):
@@ -22,6 +103,11 @@ def account_from_hex_string(private_key_str: str, account_type: Type[T]) -> T:
 
 def account_from_file(private_key_path: Path, account_type: Type[T]) -> T:
     private_key = private_key_path.read_bytes()
+    if account_type == SOLAccount:
+        private_key = parse_solana_private_key(
+            solana_private_key_from_bytes(private_key)
+        )
+
     return account_type(private_key)
 
 
@@ -33,10 +119,59 @@ def _load_account(
     """Load private key from a string or a file. takes the string argument in priority"""
 
     if private_key_str:
+        # Check Account type based on private-key string format (base58 / uint for solana)
+        private_key_chain = detect_chain_from_private_key(private_key=private_key_str)
+        if private_key_chain == Chain.SOL:
+            account_type = SOLAccount
+            logger.debug("Solana private key is detected")
+            parsed_key = parse_solana_private_key(private_key_str)
+            return account_type(parsed_key)
         logger.debug("Using account from string")
         return account_from_hex_string(private_key_str, account_type)
     elif private_key_path and private_key_path.is_file():
-        logger.debug("Using account from file")
+        if private_key_path:
+            try:
+                # Look for the account by private_key_path in CHAINS_CONFIG_FILE
+                with open(settings.CHAINS_CONFIG_FILE, "r") as file:
+                    accounts = json.load(file)
+
+                matching_account = next(
+                    (
+                        account
+                        for account in accounts
+                        if account["path"] == str(private_key_path)
+                    ),
+                    None,
+                )
+
+                if matching_account:
+                    chain = Chain(matching_account["chain"])
+                    account_type = CHAIN_TO_ACCOUNT_MAP.get(chain, ETHAccount)
+                    if account_type is None:
+                        account_type = ETHAccount
+                    logger.debug(
+                        f"Detected {chain} account for path {private_key_path}"
+                    )
+                else:
+                    logger.warning(
+                        f"No matching account found for path {private_key_path}, defaulting to {account_type.__name__}"
+                    )
+
+            except FileNotFoundError:
+                logger.warning(
+                    f"CHAINS_CONFIG_FILE not found, using default account type {account_type.__name__}"
+                )
+            except json.JSONDecodeError:
+                logger.error(
+                    f"Invalid format in CHAINS_CONFIG_FILE, unable to load account info."
+                )
+                raise ValueError(f"Invalid format in {settings.CHAINS_CONFIG_FILE}.")
+            except Exception as e:
+                logger.error(f"Error loading accounts from config: {e}")
+                raise ValueError(
+                    f"Could not find matching account for path {private_key_path}."
+                )
+
         return account_from_file(private_key_path, account_type)
     elif settings.REMOTE_CRYPTO_HOST:
         logger.debug("Using remote account")

src/aleph/sdk/chains/common.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 from typing import Dict, Optional
 
+import nacl.signing
 from coincurve.keys import PrivateKey
 from typing_extensions import deprecated
 
@@ -149,6 +150,20 @@ def generate_key() -> bytes:
     return privkey.secret
 
 
+def generate_key_solana() -> bytes:
+    """
+    Generate a new Solana private key (32 bytes) using Ed25519.
+
+    Returns:
+        A bytes object representing the 32-byte Solana private key.
+    """
+    # Generate a new signing key (this is the private key part)
+    private_key = nacl.signing.SigningKey.generate()
+
+    # Return only the private key (32 bytes)
+    return private_key.encode()
+
+
 def get_fallback_private_key(path: Optional[Path] = None) -> bytes:
     path = path or settings.PRIVATE_KEY_FILE
     private_key: bytes

src/aleph/sdk/conf.py

@@ -25,8 +25,13 @@ class Settings(BaseSettings):
         description="Path to the mnemonic used to create Substrate keypairs",
     )
 
+    CHAINS_CONFIG_FILE: Path = Field(
+        default=Path("chains_config.json"),
+        description="Path to the JSON file containing chain account configurations",
+    )
+
     PRIVATE_KEY_STRING: Optional[str] = None
-    API_HOST: str = "https://api2.aleph.im"
+    API_HOST: str = "https://api2.aleph.im/"
     MAX_INLINE_SIZE: int = 50000
     API_UNIX_SOCKET: Optional[str] = None
     REMOTE_CRYPTO_HOST: Optional[str] = None
@@ -162,13 +167,3 @@ class Config:
     settings.PRIVATE_MNEMONIC_FILE = Path(
         settings.CONFIG_HOME, "private-keys", "substrate.mnemonic"
     )
-
-# Update CHAINS settings and remove placeholders
-CHAINS_ENV = [(key[7:], value) for key, value in settings if key.startswith("CHAINS_")]
-for fields, value in CHAINS_ENV:
-    if value:
-        chain, field = fields.split("_", 1)
-        chain = chain if chain not in Chain.__members__ else Chain[chain]
-        field = field.lower()
-        settings.CHAINS[chain].__dict__[field] = value
-    settings.__delattr__(f"CHAINS_{fields}")

src/aleph/sdk/types.py

@@ -1,12 +1,13 @@
 from abc import abstractmethod
 from enum import Enum
+from pathlib import Path
 from typing import Dict, Optional, Protocol, TypeVar
 
 from pydantic import BaseModel
 
 __all__ = ("StorageEnum", "Account", "AccountFromPrivateKey", "GenericMessage")
 
-from aleph_message.models import AlephMessage
+from aleph_message.models import AlephMessage, Chain
 
 
 class StorageEnum(str, Enum):
@@ -76,3 +77,16 @@ class ChainInfo(BaseModel):
     token: str
     super_token: Optional[str] = None
     active: bool = True
+
+
+class ChainAccount(BaseModel):
+    """
+    Intern Chain Management with Account.
+    """
+
+    name: str
+    path: Path
+    chain: Chain
+
+    class Config:
+        use_enum_values = True