diff --git a/src/claudesync/configmanager/file_config_manager.py b/src/claudesync/configmanager/file_config_manager.py index 2ed9c845f..196daa768 100644 --- a/src/claudesync/configmanager/file_config_manager.py +++ b/src/claudesync/configmanager/file_config_manager.py @@ -187,7 +187,7 @@ def set_session_key(self, provider, session_key, expiry): expiry (datetime): The expiry datetime for the session key. """ try: - session_key_manager = SessionKeyManager() + session_key_manager = SessionKeyManager(self.get("ssh_key_path")) encrypted_session_key, encryption_method = ( session_key_manager.encrypt_session_key(provider, session_key) ) @@ -236,7 +236,7 @@ def get_session_key(self, provider): return None, None try: - session_key_manager = SessionKeyManager() + session_key_manager = SessionKeyManager(self.get("ssh_key_path")) session_key = session_key_manager.decrypt_session_key( provider, encryption_method, encrypted_key ) diff --git a/src/claudesync/session_key_manager.py b/src/claudesync/session_key_manager.py index 48f2bf083..082b8349b 100644 --- a/src/claudesync/session_key_manager.py +++ b/src/claudesync/session_key_manager.py @@ -8,22 +8,66 @@ class SessionKeyManager: - def __init__(self): - self.ssh_key_path = self._find_ssh_key() + def __init__(self, ssh_key_path=None): self.logger = logging.getLogger(__name__) + # Allow config-provided ssh_key_path to guide key discovery + self.ssh_key_path = self._find_ssh_key(ssh_key_path) - def _find_ssh_key(self): - ssh_dir = Path.home() / ".ssh" + def _find_ssh_key(self, configured_path=None): + """ + Locate an SSH private key for session encryption. + Priority: + 1. If configured_path points to a specific file, check it first + 2. If configured_path is a directory, search it alongside ~/.ssh + 3. Fall back to ~/.ssh with default key names + 4. Prompt the user as a last resort + """ + default_ssh_dir = Path.home() / ".ssh" key_names = ["id_ed25519", "id_ecdsa"] - for key_name in key_names: - key_path = ssh_dir / key_name - if key_path.exists(): - return str(key_path) + search_dirs = [default_ssh_dir] + + if configured_path: + configured = Path(configured_path) + + if configured.is_file(): + # Config points directly to a key file — use it immediately + return str(configured) + + if configured.is_dir(): + # Config is a directory — search it in addition to ~/.ssh + if configured != default_ssh_dir: + search_dirs.insert(0, configured) + else: + # Path doesn't exist — warn and fall through to defaults + self.logger.warning( + "Configured ssh_key_path not found: %s", configured_path + ) + + # Search all candidate directories for supported key names + for search_dir in search_dirs: + for key_name in key_names: + key_path = search_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 "your_email@example.com"') + self.logger.warning( + "* No supported SSH key found. RSA keys are no longer supported." + ) + self.logger.warning( + "* Please generate an Ed25519 key using the following command:" + ) + self.logger.warning(' ssh-keygen -t ed25519 -C "your_email@example.com"') + self.logger.warning( + "* If you have NOT specified a custom ssh_key_path in config," + ) + self.logger.warning("* have created a key, and are still seeing this message,") + self.logger.warning( + " be sure to name your key 'id_ed25519' or 'id_ecdsa' so it's found automatically." + ) + self.logger.warning( + "* Or set ssh_key_path with the full key name in your .claudesync/config.local.json" + ) return input("Enter the full path to your new Ed25519 private key: ") def _get_key_type(self): diff --git a/tests/test_session_key_manager.py b/tests/test_session_key_manager.py new file mode 100644 index 000000000..bb47bda0b --- /dev/null +++ b/tests/test_session_key_manager.py @@ -0,0 +1,66 @@ +from pathlib import Path +from unittest.mock import patch +from claudesync.session_key_manager import SessionKeyManager + + +class TestFindSshKey: + """Tests for SessionKeyManager._find_ssh_key key discovery logic.""" + + @patch("claudesync.session_key_manager.input", return_value="/fake/key") + def test_configured_file_path(self, mock_input, tmp_path): + """When ssh_key_path points to an existing file, use it directly.""" + key_file = tmp_path / "my_custom_key" + key_file.write_text("fake-key-content") + + mgr = SessionKeyManager(ssh_key_path=str(key_file)) + assert mgr.ssh_key_path == str(key_file) + mock_input.assert_not_called() + + @patch("claudesync.session_key_manager.input", return_value="/fake/key") + def test_configured_directory(self, mock_input, tmp_path): + """When ssh_key_path is a directory, search it for standard key names.""" + key_file = tmp_path / "id_ed25519" + key_file.write_text("fake-key-content") + + mgr = SessionKeyManager(ssh_key_path=str(tmp_path)) + assert mgr.ssh_key_path == str(key_file) + mock_input.assert_not_called() + + @patch("claudesync.session_key_manager.input", return_value="/fake/key") + def test_default_fallback(self, mock_input, tmp_path): + """When no config provided, search ~/.ssh for standard key names.""" + fake_ssh = tmp_path / ".ssh" + fake_ssh.mkdir() + key_file = fake_ssh / "id_ed25519" + key_file.write_text("fake-key-content") + + with patch.object(Path, "home", return_value=tmp_path): + mgr = SessionKeyManager() + assert mgr.ssh_key_path == str(key_file) + mock_input.assert_not_called() + + @patch("claudesync.session_key_manager.input", return_value="/user/entered/key") + def test_no_key_found_prompts_user(self, mock_input, tmp_path): + """When no key exists anywhere, prompt the user.""" + fake_ssh = tmp_path / ".ssh" + fake_ssh.mkdir() + # No key files created + + with patch.object(Path, "home", return_value=tmp_path): + mgr = SessionKeyManager() + assert mgr.ssh_key_path == "/user/entered/key" + mock_input.assert_called_once() + + @patch("claudesync.session_key_manager.input", return_value="/fake/key") + def test_nonexistent_configured_path_warns(self, mock_input, tmp_path): + """When ssh_key_path doesn't exist, log warning and fall through.""" + fake_ssh = tmp_path / ".ssh" + fake_ssh.mkdir() + key_file = fake_ssh / "id_ed25519" + key_file.write_text("fake-key-content") + + with patch.object(Path, "home", return_value=tmp_path): + mgr = SessionKeyManager(ssh_key_path="/nonexistent/path") + # Should fall through to default and find the key + assert mgr.ssh_key_path == str(key_file) + mock_input.assert_not_called()