Feature: Support V3 Keystore encryption for agent EOA private keys (Pearl V1 compatibility)

backend/entrypoint.sh

@@ -57,7 +57,12 @@ if [ -f "/app/scripts/checkpoint.py" ] || [ -f "/app/scripts/trigger_agent_run.p
         # Write next checkpoint timestamp to file for API
         echo "$NEXT_CHECKPOINT_TIMESTAMP" > /app/next_checkpoint_time.txt
 
-        echo "$FUTURE_MINUTE $FUTURE_HOUR * * * cd /app && timeout 300 uv run --quiet --script scripts/checkpoint.py >> /app/logs/checkpoint.log 2>&1" >> "$CRON_FILE"
+        # Add password to checkpoint script if KEY_PASSWORD is set
+        if [ -n "$KEY_PASSWORD" ]; then
+            echo "$FUTURE_MINUTE $FUTURE_HOUR * * * cd /app && timeout 300 uv run --quiet --script scripts/checkpoint.py --password \"\$KEY_PASSWORD\" >> /app/logs/checkpoint.log 2>&1" >> "$CRON_FILE"
+        else
+            echo "$FUTURE_MINUTE $FUTURE_HOUR * * * cd /app && timeout 300 uv run --quiet --script scripts/checkpoint.py >> /app/logs/checkpoint.log 2>&1" >> "$CRON_FILE"
+        fi
 
         FIRST_RUN=$(date -d "+24 hours" '+%Y-%m-%d %H:%M:%S %Z')
         echo "Checkpoint schedule: Daily at ${FUTURE_HOUR}:${FUTURE_MINUTE} UTC"
@@ -100,7 +105,12 @@ echo "Starting Quorum AI application..."
 echo "Environment: $(printenv | grep -E '^(DEBUG|HOST|HEALTH_CHECK_PORT)=' || echo 'No relevant env vars set')"
 
 # Start the main application in the background
-uv run --no-sync python -O main.py &
+# Build command with optional password
+if [ -n "$KEY_PASSWORD" ]; then
+    uv run --no-sync python -O main.py --password "$KEY_PASSWORD" &
+else
+    uv run --no-sync python -O main.py &
+fi
 MAIN_PID=$!
 
 echo "Application started with PID: $MAIN_PID"

backend/main.py

@@ -1,7 +1,9 @@
 """Main FastAPI application for Quorum AI backend."""
 
+import argparse
 import hashlib
 import os
+import sys
 import time
 from contextlib import asynccontextmanager
 from typing import List, Optional, Any
@@ -69,6 +71,33 @@
 health_status_service: Optional[HealthStatusService] = None
 staking_service: Optional[StakingService] = None
 
+# Module-level password storage for encrypted keystore files
+_key_password: Optional[str] = None
+
+
+def get_key_password() -> Optional[str]:
+    """Get the key password for encrypted keystore files.
+
+    Returns password from CLI args or environment variable.
+    Safe to call from any context including tests.
+    """
+    return _key_password
+
+
+def create_key_manager():
+    """Create a KeyManager instance with appropriate password.
+
+    This is a helper function that services can use to simplify
+    KeyManager initialization with proper error handling.
+    """
+    try:
+        from services.key_manager import KeyManager
+        return KeyManager(password=get_key_password())
+    except ImportError:
+        # In case of import issues during tests
+        from services.key_manager import KeyManager
+        return KeyManager(password=None)
+
 
 @asynccontextmanager
 async def lifespan(_app: FastAPI):
@@ -226,6 +255,23 @@ async def lifespan(_app: FastAPI):
     logger.info("Application shutdown completed")
 
 
+# Parse CLI arguments for password before FastAPI app creation
+parser = argparse.ArgumentParser(description="Quorum AI Backend")
+parser.add_argument(
+    "--password",
+    type=str,
+    help="Password for decrypting V3 Keystore encrypted private keys",
+)
+args, _ = parser.parse_known_args(sys.argv[1:])
+
+# Set module-level password from CLI arg or environment variable
+_key_password = args.password or os.environ.get("KEY_PASSWORD")
+if _key_password:
+    logger.info("Key password provided for encrypted keystore support")
+else:
+    logger.debug("No key password provided, plaintext keys only")
+
+
 # Create FastAPI app
 app = FastAPI(
     title="Quorum AI",

backend/scripts/checkpoint.py

@@ -14,6 +14,8 @@
 # ///
 """Call checkpoint on staking contract every 24 hours."""
 
+import argparse
+import os
 import sys
 from pathlib import Path
 from web3 import Web3
@@ -28,6 +30,14 @@
 
 logger = setup_pearl_logger(__name__)
 
+# Parse command line arguments
+argument_parser = argparse.ArgumentParser(description="Checkpoint script for staking contract")
+argument_parser.add_argument("--password", type=str, help="Password to decrypt V3 Keystore private key")
+args = argument_parser.parse_args()
+
+# Get password from CLI or environment variable
+key_password = args.password or os.environ.get("KEY_PASSWORD")
+
 # Default staking contract address (can be overridden via env)
 STAKING_CONTRACT_ADDRESS = "0xeF662b5266db0AeFe55554c50cA6Ad25c1DA16fb"
 
@@ -72,7 +82,7 @@ def main():
         logger.info(f"Connected to chain_id={w3.eth.chain_id}")
 
         # Load private key using KeyManager
-        key_manager = KeyManager()
+        key_manager = KeyManager(password=key_password)
         private_key = key_manager.get_private_key()
         account = Account.from_key(private_key)
 

backend/services/key_manager.py

@@ -5,12 +5,15 @@
 and caching with expiration.
 """
 
+import json
 import os
 import re
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Optional
 
+from eth_account import Account
+
 from logging_config import setup_pearl_logger
 from config import settings
 
@@ -39,29 +42,34 @@ class KeyManager:
     - Secure error handling without exposing sensitive data
     """
 
-    def __init__(self):
+    def __init__(self, password: Optional[str] = None):
         """Initialize the KeyManager.
 
+        Args:
+            password: Optional password for decrypting V3 Keystore encrypted keys.
+
         Raises:
-            KeyManagerError: If key file doesn't exist or has insecure permissions.
+            KeyManagerError: If key file doesn't exist.
         """
         # Get key directory from environment or use default
-        key_dir_env = os.environ.get("AGENT_KEY_DIR")
-        if key_dir_env:
-            self.working_directory = Path(key_dir_env)
-        else:
-            # Use /agent_key as default (Docker-friendly)
-            self.working_directory = Path("/agent_key")
-        self.key_file_path = self.working_directory / KEY_FILE_NAME
+        key_dir = os.environ.get("AGENT_KEY_DIR", "/agent_key")
+        self.key_file_path = Path(key_dir) / KEY_FILE_NAME
         self._cached_key: Optional[str] = None
         self._cache_timestamp: Optional[datetime] = None
+        self._password = password
 
-        # Validate key file setup during initialization
-        self._validate_key_file_setup()
+        # Ensure key file exists early
+        if not self.key_file_path.exists():
+            raise KeyManagerError(
+                "Key file not found. Ensure the key file exists in the working directory."
+            )
 
         logger.info(
             "KeyManager initialized",
-            extra={"working_directory": str(self.working_directory)},
+            extra={
+                "key_file": str(self.key_file_path),
+                "password_provided": password is not None,
+            },
         )
 
     def get_private_key(self) -> str:
@@ -95,17 +103,6 @@ def clear_cache(self) -> None:
         self._cache_timestamp = None
         logger.info("Private key cache cleared")
 
-    def _validate_key_file_setup(self) -> None:
-        """Validate that the key file exists and has secure permissions.
-
-        This method is called during initialization to provide early feedback
-        about key file configuration issues.
-
-        Raises:
-            KeyManagerError: If file doesn't exist or has insecure permissions.
-        """
-        self._ensure_file_exists()
-        logger.info("Key file setup validated successfully")
 
     def _is_cache_valid(self) -> bool:
         """Check if the cached key is still valid.
@@ -122,49 +119,82 @@ def _is_cache_valid(self) -> bool:
         return cache_age < max_age
 
     def _read_key_file(self) -> str:
-        """Read the private key from file with security checks.
+        """Read and process the private key from file.
 
         Returns:
-            The raw key content.
+            The processed private key (decrypted if V3 keystore).
 
         Raises:
-            KeyManagerError: If file doesn't exist or has insecure permissions.
+            KeyManagerError: If file cannot be read or is invalid.
         """
-        # Validate file exists
-        self._ensure_file_exists()
+        try:
+            content = self.key_file_path.read_text().strip()
+            logger.debug("Key file read successfully")
 
-        # Read the key
-        return self._read_file_content()
+            # Handle V3 keystore format
+            if self._is_v3_keystore(content):
+                logger.info("Detected V3 Keystore format")
+                if not self._password:
+                    raise KeyManagerError(
+                        "V3 Keystore encrypted key file detected but no password provided. "
+                        "Please provide password via --password argument or KEY_PASSWORD environment variable."
+                    )
+                return self._decrypt_v3_keystore(content)
+
+            # Return plaintext content
+            return content
+
+        except KeyManagerError:
+            raise
+        except Exception as e:
+            logger.error(f"Failed to read key file: {type(e).__name__}")
+            raise KeyManagerError("Failed to read key file. Check file accessibility.")
 
-    def _ensure_file_exists(self) -> None:
-        """Ensure the key file exists.
 
-        Raises:
-            KeyManagerError: If file doesn't exist.
+    def _is_v3_keystore(self, content: str) -> bool:
+        """Check if the content is a V3 Keystore JSON format.
+
+        Args:
+            content: The file content to check.
+
+        Returns:
+            True if content is a valid V3 keystore, False otherwise.
         """
-        if not self.key_file_path.exists():
-            logger.error("Key file not found")
-            raise KeyManagerError(
-                "Key file not found. Ensure the key file exists in the working directory."
-            )
+        try:
+            keystore = json.loads(content)
+            return isinstance(keystore, dict) and keystore.get("version") == 3
+        except json.JSONDecodeError:
+            return False
 
-    def _read_file_content(self) -> str:
-        """Read and return the file content.
+    def _decrypt_v3_keystore(self, keystore_json: str) -> str:
+        """Decrypt a V3 Keystore JSON and return the private key.
+
+        Args:
+            keystore_json: The V3 keystore JSON string.
 
         Returns:
-            The file content, stripped of whitespace.
+            The decrypted private key as a hex string with 0x prefix.
 
         Raises:
-            KeyManagerError: If file cannot be read.
+            KeyManagerError: If decryption fails or password is incorrect.
         """
         try:
-            key_content = self.key_file_path.read_text().strip()
-            logger.debug("Key file read successfully")
-            return key_content
+            keystore = json.loads(keystore_json)
+            private_key_bytes = Account.decrypt(keystore, self._password)
+            private_key = "0x" + private_key_bytes.hex()
+            logger.debug("V3 keystore decrypted successfully")
+            return private_key
+        except ValueError as e:
+            # eth_account raises ValueError for incorrect password
+            logger.error("V3 keystore decryption failed: incorrect password")
+            raise KeyManagerError(
+                "Failed to decrypt V3 keystore: incorrect password provided"
+            ) from e
         except Exception as e:
-            logger.error(f"Failed to read key file: {type(e).__name__}")
-            raise KeyManagerError("Failed to read key file. Check file accessibility.")
-
+            logger.error(f"V3 keystore decryption failed: {type(e).__name__}")
+            raise KeyManagerError(
+                f"Failed to decrypt V3 keystore: {type(e).__name__}"
+            ) from e
 
     def _validate_key_format(self, key: str) -> str:
         """Validate and normalize the private key format.

backend/services/safe_service.py

@@ -79,7 +79,12 @@ def __init__(self):
         }
 
         # Initialize account from private key using KeyManager
-        self.key_manager = KeyManager()
+        try:
+            from main import create_key_manager
+            self.key_manager = create_key_manager()
+        except Exception:
+            # Fallback for tests or when main module not fully initialized
+            self.key_manager = KeyManager(password=None)
         self.private_key = self.key_manager.get_private_key()
         self.account = Account.from_key(self.private_key)
         self._web3_connections = {}

backend/services/voting_service.py

@@ -38,9 +38,16 @@ def __init__(self, key_manager=None):
             key_manager: Optional KeyManager instance. If not provided, creates a new one.
         """
         # Initialize KeyManager
-        from services.key_manager import KeyManager
-
-        self.key_manager = key_manager or KeyManager()
+        if key_manager:
+            self.key_manager = key_manager
+        else:
+            try:
+                from main import create_key_manager
+                self.key_manager = create_key_manager()
+            except Exception:
+                # Fallback for tests or when main module not fully initialized
+                from services.key_manager import KeyManager
+                self.key_manager = KeyManager(password=None)
 
         # Initialize account lazily
         self._account = None

backend/tests/conftest.py

@@ -12,6 +12,7 @@
 
 # Set test environment variables before imports
 os.environ.setdefault("OPENROUTER_API_KEY", "test-key")
+os.environ.setdefault("BASE_RPC_URL", "http://localhost:8545")
 
 
 @pytest.fixture