feat: Encrypt session keys with SSH key for improved security (jahwag#69

)

.gitignore

@@ -171,5 +171,6 @@ config.json
 claudesync.log
 claude_chats
 some_value
+.claudesync
 
 ROADMAP.md

pyproject.toml

@@ -1,13 +1,13 @@
 [project]
 name = "claudesync"
-version = "0.5.8"
+version = "0.6.0"
 authors = [
-    {name = "Jahziah Wagner", email = "jahziah.wagner+pypi@gmail.com"},
+    {name = "Jahziah Wagner", email = "[email protected].com"},
 ]
 description = "A tool to synchronize local files with Claude.ai projects"
 license = {file = "LICENSE"}
 readme = "README.md"
-requires-python = ">=3.7"
+requires-python = ">=3.10"
 classifiers = [
     "Programming Language :: Python :: 3",
     "License :: OSI Approved :: MIT License",
@@ -27,7 +27,8 @@ dependencies = [
     "crontab>=1.0.1",
     "python-crontab>=3.2.0",
     "Brotli>=1.1.0",
-    "anthropic>=0.34.2"
+    "anthropic>=0.34.2",
+    "cryptography>=3.4.7"
 ]
 keywords = [
     "sync",

requirements.txt

@@ -11,4 +11,5 @@ claudesync>=0.5.4
 crontab>=1.0.1
 python-crontab>=3.2.0
 Brotli>=1.1.0
-anthropic>=0.34.2
+anthropic>=0.34.2
+cryptography>=3.4.7

src/claudesync/configmanager/file_config_manager.py

@@ -1,9 +1,11 @@
 import json
+import logging
 import os
 from datetime import datetime
 from pathlib import Path
 
 from claudesync.configmanager.base_config_manager import BaseConfigManager
+from claudesync.session_key_manager import SessionKeyManager
 
 
 class FileConfigManager(BaseConfigManager):
@@ -29,6 +31,7 @@ def __init__(self):
         self.local_config = {}
         self.local_config_dir = None
         self._load_local_config()
+        self.session_key_manager = SessionKeyManager()
 
     def _load_global_config(self):
         """
@@ -169,42 +172,62 @@ def set_session_key(self, provider, session_key, expiry):
             session_key (str): The session key to set.
             expiry (datetime): The expiry datetime for the session key.
         """
-        self.global_config_dir.mkdir(parents=True, exist_ok=True)
-        provider_key_file = self.global_config_dir / f"{provider}.key"
-        with open(provider_key_file, "w") as f:
-            json.dump(
-                {"session_key": session_key, "session_key_expiry": expiry.isoformat()},
-                f,
+        try:
+            encrypted_session_key, encryption_method = (
+                self.session_key_manager.encrypt_session_key(provider, session_key)
             )
 
-    def get_session_key(self, providerName):
+            self.global_config_dir.mkdir(parents=True, exist_ok=True)
+            provider_key_file = self.global_config_dir / f"{provider}.key"
+            with open(provider_key_file, "w") as f:
+                json.dump(
+                    {
+                        "session_key": encrypted_session_key,
+                        "session_key_encryption_method": encryption_method,
+                        "session_key_expiry": expiry.isoformat(),
+                    },
+                    f,
+                )
+        except RuntimeError as e:
+            logging.error(f"Failed to encrypt session key: {str(e)}")
+            raise
+
+    def get_session_key(self, provider):
         """
         Retrieves the session key for the specified provider if it's still valid.
 
         Args:
-            providerName (str): The name of the provider.
+            provider (str): The name of the provider.
 
         Returns:
             tuple: A tuple containing the session key and expiry if valid, (None, None) otherwise.
         """
-        provider_key_file = self.global_config_dir / f"{providerName}.key"
+        provider_key_file = self.global_config_dir / f"{provider}.key"
         if not provider_key_file.exists():
             return None, None
 
         with open(provider_key_file, "r") as f:
             data = json.load(f)
 
-        session_key = data.get("session_key")
+        encrypted_key = data.get("session_key")
+        encryption_method = data.get("session_key_encryption_method")
         expiry_str = data.get("session_key_expiry")
 
-        if not session_key or not expiry_str:
+        if not encrypted_key or not expiry_str:
             return None, None
 
         expiry = datetime.fromisoformat(expiry_str)
         if datetime.now() > expiry:
             return None, None
 
-        return session_key, expiry
+        try:
+            session_key = self.session_key_manager.decrypt_session_key(
+                provider, encryption_method, encrypted_key
+            )
+            return session_key, expiry
+        except RuntimeError as e:
+            logging.error(f"Failed to decrypt session key: {str(e)}")
+            return None, None
 
     def add_file_category(self, category_name, description, patterns):
         """

src/claudesync/session_key_manager.py

@@ -1,4 +1,6 @@
+import os
 import subprocess
+import tempfile
 import base64
 import logging
 from pathlib import Path
@@ -14,17 +16,12 @@ def __init__(self):
 
     def _find_ssh_key(self):
         ssh_dir = Path.home() / ".ssh"
-        key_names = ["id_ed25519", "id_ecdsa"]
+        key_names = ["id_ed25519", "id_rsa", "id_ecdsa"]
         for key_name in key_names:
             key_path = ssh_dir / key_name
             if key_path.exists():
                 return str(key_path)
-
-        # If no supported key is found, prompt the user to generate an Ed25519 key
-        print("* No supported SSH key found. RSA keys are no longer supported.")
-        print("* Please generate an Ed25519 key using the following command:")
-        print('ssh-keygen -t ed25519 -C "[email protected]"')
-        return input("Enter the full path to your new Ed25519 private key: ")
+        return input("Enter the full path to your SSH private key: ")
 
     def _get_key_type(self):
         try:
@@ -35,7 +32,9 @@ def _get_key_type(self):
                 check=True,
             )
             output = result.stdout.lower()
-            if "ecdsa" in output:
+            if "rsa" in output:
+                return "rsa"
+            elif "ecdsa" in output:
                 return "ecdsa"
             elif "ed25519" in output:
                 return "ed25519"
@@ -61,8 +60,67 @@ def _derive_key_from_ssh_key(self):
         return key
 
     def encrypt_session_key(self, provider, session_key):
-        self._get_key_type()
-        return self._encrypt_symmetric(session_key)
+        key_type = self._get_key_type()
+
+        if key_type == "rsa":
+            return self._encrypt_rsa(session_key)
+        else:  # For ed25519 and ecdsa
+            return self._encrypt_symmetric(session_key)
+
+    def _encrypt_rsa(self, session_key):
+        temp_file_path = None
+        pub_key_file_path = None
+        try:
+            with tempfile.NamedTemporaryFile(mode="w+", delete=False) as temp_file:
+                temp_file.write(session_key)
+                temp_file_path = temp_file.name
+
+            result = subprocess.run(
+                ["ssh-keygen", "-f", self.ssh_key_path, "-e", "-m", "PKCS8"],
+                capture_output=True,
+                text=True,
+                check=True,
+            )
+            public_key = result.stdout
+
+            with tempfile.NamedTemporaryFile(mode="w+", delete=False) as pub_key_file:
+                pub_key_file.write(public_key)
+                pub_key_file_path = pub_key_file.name
+
+            encrypted_output = subprocess.run(
+                [
+                    "openssl",
+                    "pkeyutl",
+                    "-encrypt",
+                    "-pubin",
+                    "-inkey",
+                    pub_key_file_path,
+                    "-in",
+                    temp_file_path,
+                    "-pkeyopt",
+                    "rsa_padding_mode:oaep",
+                    "-pkeyopt",
+                    "rsa_oaep_md:sha256",
+                ],
+                capture_output=True,
+                check=True,
+            )
+
+            encrypted_session_key = base64.b64encode(encrypted_output.stdout).decode(
+                "utf-8"
+            )
+
+            return encrypted_session_key, "rsa"
+        except subprocess.CalledProcessError as e:
+            self.logger.error(f"Encryption failed: {e}")
+            raise RuntimeError(
+                "Failed to encrypt session key. Check if openssl and ssh-keygen are installed and the SSH key is valid."
+            )
+        finally:
+            if temp_file_path and os.path.exists(temp_file_path):
+                os.unlink(temp_file_path)
+            if pub_key_file_path and os.path.exists(pub_key_file_path):
+                os.unlink(pub_key_file_path)
 
     def _encrypt_symmetric(self, session_key):
         key = self._derive_key_from_ssh_key()
@@ -74,11 +132,50 @@ def decrypt_session_key(self, provider, encryption_method, encrypted_session_key
         if not encrypted_session_key or not encryption_method:
             return None
 
-        if encryption_method == "symmetric":
+        if encryption_method == "rsa":
+            return self._decrypt_rsa(encrypted_session_key)
+        elif encryption_method == "symmetric":
             return self._decrypt_symmetric(encrypted_session_key)
         else:
             raise ValueError(f"Unknown encryption method: {encryption_method}")
 
+    def _decrypt_rsa(self, encrypted_session_key):
+        temp_file_path = None
+        try:
+            with tempfile.NamedTemporaryFile(mode="wb+", delete=False) as temp_file:
+                temp_file.write(base64.b64decode(encrypted_session_key))
+                temp_file_path = temp_file.name
+
+            decrypted_output = subprocess.run(
+                [
+                    "openssl",
+                    "pkeyutl",
+                    "-decrypt",
+                    "-inkey",
+                    self.ssh_key_path,
+                    "-in",
+                    temp_file_path,
+                    "-pkeyopt",
+                    "rsa_padding_mode:oaep",
+                    "-pkeyopt",
+                    "rsa_oaep_md:sha256",
+                ],
+                capture_output=True,
+                text=True,
+                check=True,
+            )
+
+            return decrypted_output.stdout.strip()
+
+        except subprocess.CalledProcessError as e:
+            self.logger.error(f"Decryption failed: {e}")
+            raise RuntimeError(
+                "Failed to decrypt session key. Make sure the SSH key is valid and matches the one used for encryption."
+            )
+        finally:
+            if temp_file_path and os.path.exists(temp_file_path):
+                os.unlink(temp_file_path)
+
     def _decrypt_symmetric(self, encrypted_session_key):
         key = self._derive_key_from_ssh_key()
         f = Fernet(key)