Fix: Improve code structure in pair-programming with Lyam

Author: nesitor

.gitignore

@@ -50,7 +50,7 @@ MANIFEST
 **/device.key
 
 # environment variables
-.env
+.config.json
 .env.local
 
 .gitsigners

src/aleph/sdk/account.py

@@ -1,102 +1,30 @@
 import asyncio
-import json
 import logging
 from pathlib import Path
-from typing import Dict, List, Optional, Type, TypeVar, Union, overload
+from typing import Dict, Optional, Type, TypeVar
 
-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,
-    parse_solana_private_key,
-    solana_private_key_from_bytes,
-)
-from aleph.sdk.conf import settings
+from aleph.sdk.chains.solana import SOLAccount
+from aleph.sdk.conf import load_main_configuration, settings
 from aleph.sdk.types import AccountFromPrivateKey
-from aleph.sdk.utils import load_account_key_context
 
 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 load_chain_account_type(chain: Chain) -> Type[AccountFromPrivateKey]:
+    chain_account_map: Dict[Chain, Type[AccountFromPrivateKey]] = {
+        Chain.ETH: ETHAccount,
+        Chain.AVAX: ETHAccount,
+        Chain.SOL: SOLAccount,
+        Chain.BASE: ETHAccount,
+    }
+    return chain_account_map.get(chain) or ETHAccount
 
 
 def account_from_hex_string(private_key_str: str, account_type: Type[T]) -> T:
@@ -107,72 +35,42 @@ 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)
 
 
 def _load_account(
     private_key_str: Optional[str] = None,
     private_key_path: Optional[Path] = None,
-    account_type: Type[AccountFromPrivateKey] = ETHAccount,
+    account_type: Optional[Type[AccountFromPrivateKey]] = None,
 ) -> AccountFromPrivateKey:
     """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():
-        if private_key_path:
-            account_type = ETHAccount  # Default account type
-
-            try:
-                account_data = load_account_key_context(settings.CONFIG_FILE)
-
-                if account_data:
-                    chain = Chain(account_data.chain)
-                    account_type = (
-                        CHAIN_TO_ACCOUNT_MAP.get(chain, ETHAccount) or ETHAccount
-                    )
-                    logger.debug(
-                        f"Detected {chain} account for path {private_key_path}"
-                    )
-                else:
-                    logger.warning(
-                        f"No account data found in {private_key_path}, defaulting to {account_type.__name__}"
-                    )
-
-            except FileNotFoundError:
-                logger.warning(
-                    f"{private_key_path} not found, using default account type {account_type.__name__}"
+    if private_key_str or (private_key_path and private_key_path.is_file()):
+        if account_type:
+            if private_key_path and private_key_path.is_file():
+                return account_from_file(private_key_path, account_type)
+            elif private_key_str:
+                return account_from_hex_string(private_key_str, account_type)
+            else:
+                raise ValueError("Any private key specified")
+        else:
+            main_configuration = load_main_configuration(settings.CONFIG_FILE)
+            if main_configuration:
+                account_type = load_chain_account_type(main_configuration.chain)
+                logger.debug(
+                    f"Detected {main_configuration.chain} account for path {settings.CONFIG_FILE}"
                 )
-            except json.JSONDecodeError:
-                logger.error(
-                    f"Invalid format in {private_key_path}, unable to load account info."
-                )
-                raise ValueError(f"Invalid format in {private_key_path}.")
-            except KeyError as e:
-                logger.error(f"Missing key in account config: {e}")
-                raise ValueError(
-                    f"Invalid account data in {private_key_path}. Key {e} is missing."
-                )
-            except Exception as e:
-                logger.error(f"Error loading account from {private_key_path}: {e}")
-                raise ValueError(
-                    f"Could not load account data from {private_key_path}."
+            else:
+                account_type = ETHAccount  # Defaults to ETHAccount
+                logger.warning(
+                    f"No main configuration data found in {settings.CONFIG_FILE}, defaulting to {account_type.__name__}"
                 )
+            if private_key_path and private_key_path.is_file():
+                return account_from_file(private_key_path, account_type)
+            elif private_key_str:
+                return account_from_hex_string(private_key_str, account_type)
+            else:
+                raise ValueError("Any private key specified")
 
-        return account_from_file(private_key_path, account_type)
     elif settings.REMOTE_CRYPTO_HOST:
         logger.debug("Using remote account")
         loop = asyncio.get_event_loop()
@@ -183,6 +81,7 @@ def _load_account(
             )
         )
     else:
+        account_type = ETHAccount  # Defaults to ETHAccount
         new_private_key = get_fallback_private_key()
         account = account_type(private_key=new_private_key)
         logger.info(

src/aleph/sdk/chains/common.py

@@ -3,7 +3,6 @@
 from pathlib import Path
 from typing import Dict, Optional
 
-import nacl.signing
 from coincurve.keys import PrivateKey
 from typing_extensions import deprecated
 
@@ -150,20 +149,6 @@ 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/chains/solana.py

@@ -22,7 +22,7 @@ class SOLAccount(BaseAccount):
     _private_key: PrivateKey
 
     def __init__(self, private_key: bytes):
-        self.private_key = private_key
+        self.private_key = parse_private_key(private_key_from_bytes(private_key))
         self._signing_key = SigningKey(self.private_key)
         self._private_key = self._signing_key.to_curve25519_private_key()
 
@@ -93,7 +93,7 @@ def verify_signature(
         raise BadSignatureError from e
 
 
-def solana_private_key_from_bytes(
+def private_key_from_bytes(
     private_key_bytes: bytes, output_format: str = "base58"
 ) -> Union[str, List[int], bytes]:
     """
@@ -132,7 +132,7 @@ def solana_private_key_from_bytes(
         raise ValueError("Invalid output format. Choose 'base58', 'list', or 'bytes'.")
 
 
-def parse_solana_private_key(private_key: Union[str, List[int], bytes]) -> bytes:
+def parse_private_key(private_key: Union[str, List[int], bytes]) -> bytes:
     """
     Parse the private key which could be either:
     - a base58-encoded string (which may contain both private and public key)

src/aleph/sdk/conf.py

@@ -1,21 +1,24 @@
 import json
+import logging
 import os
 from pathlib import Path
 from shutil import which
 from typing import Dict, Optional, Union
 
 from aleph_message.models import Chain
 from aleph_message.models.execution.environment import HypervisorType
-from pydantic import BaseSettings, Field
+from pydantic import BaseModel, BaseSettings, Field
 
 from aleph.sdk.types import ChainInfo
 
+logger = logging.getLogger(__name__)
+
 
 class Settings(BaseSettings):
     CONFIG_HOME: Optional[str] = None
 
     CONFIG_FILE: Path = Field(
-        default=Path("chains_config.json"),
+        default=Path("config.json"),
         description="Path to the JSON file containing chain account configurations",
     )
 
@@ -145,6 +148,18 @@ class Config:
         env_file = ".env"
 
 
+class MainConfiguration(BaseModel):
+    """
+    Intern Chain Management with Account.
+    """
+
+    path: Path
+    chain: Chain
+
+    class Config:
+        use_enum_values = True
+
+
 # Settings singleton
 settings = Settings()
 
@@ -168,8 +183,8 @@ class Config:
     settings.PRIVATE_MNEMONIC_FILE = Path(
         settings.CONFIG_HOME, "private-keys", "substrate.mnemonic"
     )
-if str(settings.CONFIG_FILE) == "chains_config.json":
-    settings.CONFIG_FILE = Path(settings.CONFIG_HOME, "chains_config.json")
+if str(settings.CONFIG_FILE) == "config.json":
+    settings.CONFIG_FILE = Path(settings.CONFIG_HOME, "config.json")
     # If Config file exist and well filled we update the PRIVATE_KEY_FILE default
     if settings.CONFIG_FILE.exists():
         try:
@@ -191,3 +206,35 @@ class Config:
         field = field.lower()
         settings.CHAINS[chain].__dict__[field] = value
     settings.__delattr__(f"CHAINS_{fields}")
+
+
+def save_main_configuration(file_path: Path, data: MainConfiguration):
+    """
+    Synchronously save a single ChainAccount object as JSON to a file.
+    """
+    with file_path.open("w") as file:
+        data_serializable = data.dict()
+        data_serializable["path"] = str(data_serializable["path"])
+        json.dump(data_serializable, file, indent=4)
+
+
+def load_main_configuration(file_path: Path) -> Optional[MainConfiguration]:
+    """
+    Synchronously load the private key and chain type from a file.
+    If the file does not exist or is empty, return None.
+    """
+    if not file_path.exists() or file_path.stat().st_size == 0:
+        logger.debug(f"File {file_path} does not exist or is empty. Returning None.")
+        return None
+
+    try:
+        with file_path.open("rb") as file:
+            content = file.read()
+            data = json.loads(content.decode("utf-8"))
+            return MainConfiguration(**data)
+    except UnicodeDecodeError as e:
+        logger.error(f"Unable to decode {file_path} as UTF-8: {e}")
+    except json.JSONDecodeError:
+        logger.error(f"Invalid JSON format in {file_path}.")
+
+    return None