fix: preserve credentials during connection renames

sqlit/domains/connections/app/credentials.py

@@ -43,7 +43,7 @@ def __init__(self, *, connection_name: str, kind: str, action: str, reason: Exce
 
     def user_message(self) -> str:
         kind_label = "database" if self.kind == "db" else "SSH"
-        action_label = "save" if self.action == "store" else "delete"
+        action_label = {"store": "save", "delete": "delete", "read": "read"}.get(self.action, self.action)
         return (
             f"Keyring error while trying to {action_label} {kind_label} password for "
             f"'{self.connection_name}': {self.reason}"
@@ -137,6 +137,10 @@ def get_password(self, connection_name: str) -> str | None:
         """
         ...
 
+    def get_password_for_migration(self, connection_name: str) -> str | None:
+        """Read a password for a destructive move, surfacing backend failures."""
+        return self.get_password(connection_name)
+
     @abstractmethod
     def set_password(self, connection_name: str, password: str) -> None:
         """Store the database password for a connection.
@@ -168,6 +172,10 @@ def get_ssh_password(self, connection_name: str) -> str | None:
         """
         ...
 
+    def get_ssh_password_for_migration(self, connection_name: str) -> str | None:
+        """Read an SSH password for a destructive move, surfacing backend failures."""
+        return self.get_ssh_password(connection_name)
+
     @abstractmethod
     def set_ssh_password(self, connection_name: str, password: str) -> None:
         """Store the SSH password for a connection.
@@ -253,15 +261,30 @@ def _make_key(self, connection_name: str, key_type: str) -> str:
         """
         return f"{connection_name}:{key_type}"
 
-    def _get_with_retry(self, key: str, retries: int = 2, delay_seconds: float = 0.2) -> str | None:
+    def _get_with_retry(
+        self,
+        key: str,
+        retries: int = 2,
+        delay_seconds: float = 0.2,
+        *,
+        migration_context: tuple[str, str] | None = None,
+    ) -> str | None:
         # A short retry helps with transient keyring/DBus/Keychain hiccups.
         for attempt in range(retries + 1):
             try:
                 keyring = self._get_keyring()
                 value = keyring.get_password(KEYRING_SERVICE_NAME, key)
                 return value if isinstance(value, str) else None
-            except Exception:
+            except Exception as exc:
                 if attempt >= retries:
+                    if migration_context is not None:
+                        connection_name, kind = migration_context
+                        self._raise_keyring_error(
+                            connection_name=connection_name,
+                            kind=kind,
+                            action="read",
+                            reason=exc,
+                        )
                     return None
                 time.sleep(delay_seconds)
         return None
@@ -278,6 +301,10 @@ def get_password(self, connection_name: str) -> str | None:
         key = self._make_key(connection_name, "db")
         return self._get_with_retry(key)
 
+    def get_password_for_migration(self, connection_name: str) -> str | None:
+        key = self._make_key(connection_name, "db")
+        return self._get_with_retry(key, migration_context=(connection_name, "db"))
+
     def set_password(self, connection_name: str, password: str) -> None:
         if password is None:
             self.delete_password(connection_name)
@@ -306,6 +333,10 @@ def get_ssh_password(self, connection_name: str) -> str | None:
         key = self._make_key(connection_name, "ssh")
         return self._get_with_retry(key)
 
+    def get_ssh_password_for_migration(self, connection_name: str) -> str | None:
+        key = self._make_key(connection_name, "ssh")
+        return self._get_with_retry(key, migration_context=(connection_name, "ssh"))
+
     def set_ssh_password(self, connection_name: str, password: str) -> None:
         if password is None:
             self.delete_ssh_password(connection_name)

sqlit/domains/connections/store/connections.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import copy
 from typing import TYPE_CHECKING
 
 from sqlit.domains.connections.app.credentials import CredentialsPersistError, CredentialsStoreError
@@ -175,6 +176,47 @@ def _save_credentials(self, config: ConnectionConfig) -> list[CredentialsStoreEr
 
         return errors
 
+    def _restore_credentials(
+        self,
+        connection_name: str,
+        db_password: str | None,
+        ssh_password: str | None,
+    ) -> list[CredentialsStoreError]:
+        """Best-effort restoration used when a credential move fails."""
+        errors: list[CredentialsStoreError] = []
+        operations = (
+            (
+                "db",
+                self.credentials_service.set_password,
+                self.credentials_service.delete_password,
+                db_password,
+            ),
+            (
+                "ssh",
+                self.credentials_service.set_ssh_password,
+                self.credentials_service.delete_ssh_password,
+                ssh_password,
+            ),
+        )
+        for kind, setter, deleter, password in operations:
+            try:
+                if password is None:
+                    deleter(connection_name)
+                else:
+                    setter(connection_name, password)
+            except CredentialsStoreError as exc:
+                errors.append(exc)
+            except Exception as exc:
+                errors.append(
+                    CredentialsStoreError(
+                        connection_name=connection_name,
+                        kind=kind,
+                        action="store" if password is not None else "delete",
+                        reason=exc,
+                    )
+                )
+        return errors
+
     def _config_to_dict_without_passwords(self, config: ConnectionConfig) -> dict:
         """Convert config to dict without password fields.
 
@@ -245,10 +287,59 @@ def save_one(
             if c.name != connection.name and not (renamed and c.name == previous_name)
         ]
         filtered.append(connection)
-        self._write_index(filtered)
 
         errors: list[CredentialsStoreError] = []
         if renamed:
+            # Connections are normally loaded without credentials. Hydrate the
+            # renamed copy from the old keyring name before moving anything.
+            target = copy.deepcopy(connection)
+            try:
+                endpoint = target.tcp_endpoint
+                if endpoint and endpoint.password is None:
+                    endpoint.password = self.credentials_service.get_password_for_migration(previous_name)  # type: ignore[arg-type]
+                if target.tunnel and target.tunnel.password is None:
+                    target.tunnel.password = self.credentials_service.get_ssh_password_for_migration(previous_name)  # type: ignore[arg-type]
+                destination_db_password = self.credentials_service.get_password_for_migration(connection.name)
+                destination_ssh_password = self.credentials_service.get_ssh_password_for_migration(
+                    connection.name
+                )
+            except CredentialsStoreError as exc:
+                raise CredentialsPersistError([exc]) from exc
+
+            # Store the new credentials before changing the index or deleting
+            # the old entries. A failed write must leave the original usable.
+            try:
+                errors.extend(self._save_credentials(target))
+            except Exception as exc:
+                rollback_errors = self._restore_credentials(
+                    connection.name,
+                    destination_db_password,
+                    destination_ssh_password,
+                )
+                if rollback_errors:
+                    raise CredentialsPersistError(rollback_errors) from exc
+                raise
+            if errors:
+                errors.extend(
+                    self._restore_credentials(
+                        connection.name,
+                        destination_db_password,
+                        destination_ssh_password,
+                    )
+                )
+                raise CredentialsPersistError(errors)
+
+            try:
+                self._write_index(filtered)
+            except Exception as exc:
+                rollback_errors = self._restore_credentials(
+                    connection.name,
+                    destination_db_password,
+                    destination_ssh_password,
+                )
+                if rollback_errors:
+                    raise CredentialsPersistError(rollback_errors) from exc
+                raise
             for deleter in (
                 self.credentials_service.delete_password,
                 self.credentials_service.delete_ssh_password,
@@ -257,9 +348,10 @@ def save_one(
                     deleter(previous_name)  # type: ignore[arg-type]
                 except CredentialsStoreError as exc:
                     errors.append(exc)
-
-        target = build_persist_connections([connection], self.credentials_service)[0]
-        errors.extend(self._save_credentials(target))
+        else:
+            self._write_index(filtered)
+            target = build_persist_connections([connection], self.credentials_service)[0]
+            errors.extend(self._save_credentials(target))
         if errors:
             raise CredentialsPersistError(errors)
 

tests/test_connection_store_save_one.py

@@ -37,6 +37,9 @@ def __init__(self) -> None:
         self.delete_db: list[str] = []
         self.delete_ssh: list[str] = []
         self.fail_set_for: set[str] = set()
+        self.fail_set_ssh_for: set[str] = set()
+        self.fail_set_ssh_raw_for: set[str] = set()
+        self.fail_migration_read_for: set[str] = set()
 
     def set_password(self, connection_name: str, password: str) -> None:
         self.set_db.append(connection_name)
@@ -51,8 +54,28 @@ def set_password(self, connection_name: str, password: str) -> None:
 
     def set_ssh_password(self, connection_name: str, password: str) -> None:
         self.set_ssh.append(connection_name)
+        if connection_name in self.fail_set_ssh_raw_for:
+            self.fail_set_ssh_raw_for.remove(connection_name)
+            raise OSError("disk unavailable")
+        if connection_name in self.fail_set_ssh_for:
+            raise CredentialsStoreError(
+                connection_name=connection_name,
+                kind="ssh",
+                action="store",
+                reason=RuntimeError("boom"),
+            )
         super().set_ssh_password(connection_name, password)
 
+    def get_password_for_migration(self, connection_name: str) -> str | None:
+        if connection_name in self.fail_migration_read_for:
+            raise CredentialsStoreError(
+                connection_name=connection_name,
+                kind="db",
+                action="read",
+                reason=RuntimeError("boom"),
+            )
+        return super().get_password_for_migration(connection_name)
+
     def delete_password(self, connection_name: str) -> None:
         self.delete_db.append(connection_name)
         super().delete_password(connection_name)
@@ -209,6 +232,99 @@ def test_save_one_rename_moves_credentials(self) -> None:
         assert self.creds.get_ssh_password("new") == "ssh_secret"
         assert {c["name"] for c in self._json()} == {"new"}
 
+    def test_save_one_rename_preserves_omitted_credentials(self) -> None:
+        store = self._create_store()
+        store.save_one(self._make("old", password="secret", ssh=True))
+        self.creds.reset_calls()
+
+        renamed = self._make("new", password=None, ssh=True)
+        assert renamed.tunnel is not None
+        renamed.tunnel.password = None
+        store.save_one(renamed, previous_name="old")
+
+        assert self.creds.get_password("old") is None
+        assert self.creds.get_ssh_password("old") is None
+        assert self.creds.get_password("new") == "secret"
+        assert self.creds.get_ssh_password("new") == "ssh_secret"
+
+    def test_save_one_failed_rename_keeps_original_usable(self) -> None:
+        store = self._create_store()
+        store.save_one(self._make("old", password="secret", ssh=True))
+        self.creds.reset_calls()
+        self.creds.fail_set_for = {"new"}
+
+        renamed = self._make("new", password=None, ssh=True)
+        assert renamed.tunnel is not None
+        renamed.tunnel.password = None
+        with pytest.raises(CredentialsPersistError):
+            store.save_one(renamed, previous_name="old")
+
+        assert self.creds.get_password("old") == "secret"
+        assert self.creds.get_ssh_password("old") == "ssh_secret"
+        assert "old" not in self.creds.delete_db
+        assert "old" not in self.creds.delete_ssh
+        assert {c["name"] for c in self._json()} == {"old"}
+
+    def test_save_one_rename_aborts_when_source_credentials_cannot_be_read(self) -> None:
+        store = self._create_store()
+        store.save_one(self._make("old", password="secret", ssh=True))
+        self.creds.reset_calls()
+        self.creds.fail_migration_read_for = {"old"}
+
+        renamed = self._make("new", password=None, ssh=True)
+        with pytest.raises(CredentialsPersistError):
+            store.save_one(renamed, previous_name="old")
+
+        self.creds.fail_migration_read_for.clear()
+        assert self.creds.get_password("old") == "secret"
+        assert self.creds.get_password("new") is None
+        assert {c["name"] for c in self._json()} == {"old"}
+
+    def test_save_one_partial_rename_write_restores_destination_credentials(self) -> None:
+        store = self._create_store()
+        old = self._make("old", password="source-db", ssh=True)
+        destination = self._make("new", password="destination-db", ssh=True)
+        assert destination.tunnel is not None
+        destination.tunnel.password = "destination-ssh"
+        store.save_all([old, destination])
+        self.creds.reset_calls()
+        self.creds.fail_set_ssh_for = {"new"}
+
+        renamed = self._make("new", password=None, ssh=True)
+        assert renamed.tunnel is not None
+        renamed.tunnel.password = None
+        with pytest.raises(CredentialsPersistError):
+            store.save_one(renamed, previous_name="old")
+
+        self.creds.fail_set_ssh_for.clear()
+        assert self.creds.get_password("old") == "source-db"
+        assert self.creds.get_ssh_password("old") == "ssh_secret"
+        assert self.creds.get_password("new") == "destination-db"
+        assert self.creds.get_ssh_password("new") == "destination-ssh"
+        assert {c["name"] for c in self._json()} == {"old", "new"}
+
+    def test_save_one_raw_backend_failure_restores_destination_credentials(self) -> None:
+        store = self._create_store()
+        old = self._make("old", password="source-db", ssh=True)
+        destination = self._make("new", password="destination-db", ssh=True)
+        assert destination.tunnel is not None
+        destination.tunnel.password = "destination-ssh"
+        store.save_all([old, destination])
+        self.creds.reset_calls()
+        self.creds.fail_set_ssh_raw_for = {"new"}
+
+        renamed = self._make("new", password=None, ssh=True)
+        assert renamed.tunnel is not None
+        renamed.tunnel.password = None
+        with pytest.raises(OSError):
+            store.save_one(renamed, previous_name="old")
+
+        self.creds.fail_set_ssh_raw_for.clear()
+        assert self.creds.get_password("old") == "source-db"
+        assert self.creds.get_password("new") == "destination-db"
+        assert self.creds.get_ssh_password("new") == "destination-ssh"
+        assert {c["name"] for c in self._json()} == {"old", "new"}
+
     def test_save_one_rename_does_not_delete_when_name_unchanged(self) -> None:
         store = self._create_store()
         store.save_one(self._make("a", password="secret"))

tests/test_credentials_keyring.py

@@ -4,9 +4,11 @@
 
 from unittest.mock import MagicMock
 
+import pytest
+
 from sqlit.domains.connections.app.credentials import (
-    CredentialsStoreError,
     KEYRING_SERVICE_NAME,
+    CredentialsStoreError,
     KeyringCredentialsService,
 )
 
@@ -124,6 +126,16 @@ def test_keyring_error_returns_none(self) -> None:
         result = service.get_password("test_conn")
         assert result is None
 
+    def test_keyring_error_during_migration_is_surfaced(self) -> None:
+        service, mock_keyring = self._create_service_with_mock_keyring()
+        mock_keyring.get_password.side_effect = Exception("Keyring error")
+
+        with pytest.raises(CredentialsStoreError) as exc_info:
+            service.get_password_for_migration("test_conn")
+
+        assert exc_info.value.action == "read"
+        assert exc_info.value.connection_name == "test_conn"
+
     def test_keyring_error_on_set_raises(self) -> None:
         """Test that keyring errors on set raise a storage error."""
         service, mock_keyring = self._create_service_with_mock_keyring()