Fix mock sqlite - Add option to use plaintext credentials

Author: Maxteabag

sqlit/domains/connections/app/credentials.py

@@ -78,27 +78,49 @@ def is_keyring_usable() -> bool:
         return _is_keyring_usable()
 
 
+_keyring_probe_error: str | None = None
+
+
+def get_keyring_probe_error() -> str | None:
+    """Return the last keyring probe error, if any."""
+    return _keyring_probe_error
+
+
 def _is_keyring_usable() -> bool:
     """Internal keyring probe (wrapped for profiling)."""
+    global _keyring_probe_error
+    _keyring_probe_error = None
+
     try:
         import keyring
     except ImportError:
+        _keyring_probe_error = "keyring module not installed"
         return False
 
-    try:
-        backend = keyring.get_keyring()
-        module_name = getattr(backend, "__module__", "") or ""
-        priority = getattr(backend, "priority", None)
-        if "keyring.backends.fail" in module_name:
-            return False
-        if isinstance(priority, (int, float)) and priority <= 0:
-            return False
-
-        # Minimal probe: read-only call to surface obvious misconfiguration.
-        keyring.get_password(KEYRING_SERVICE_NAME, f"probe:{secrets.token_hex(8)}")
-        return True
-    except Exception:
-        return False
+    # Retry probe to handle transient D-Bus/keyring daemon issues
+    last_exc = None
+    for attempt in range(3):
+        try:
+            backend = keyring.get_keyring()
+            module_name = getattr(backend, "__module__", "") or ""
+            priority = getattr(backend, "priority", None)
+            if "keyring.backends.fail" in module_name:
+                _keyring_probe_error = "No usable keyring backend found"
+                return False
+            if isinstance(priority, (int, float)) and priority <= 0:
+                _keyring_probe_error = "Keyring backend has low priority"
+                return False
+
+            # Minimal probe: read-only call to surface obvious misconfiguration.
+            keyring.get_password(KEYRING_SERVICE_NAME, f"probe:{secrets.token_hex(8)}")
+            return True
+        except Exception as exc:
+            last_exc = exc
+            if attempt < 2:
+                time.sleep(0.1)  # Brief delay before retry
+
+    _keyring_probe_error = f"Keyring probe failed: {last_exc}"
+    return False
 
 class CredentialsService(ABC):
     """Abstract base class for credential storage services."""
@@ -403,18 +425,23 @@ def delete_ssh_password(self, connection_name: str) -> None:
 
 def build_credentials_service(settings_store: Any | None = None) -> CredentialsService:
     """Build a credentials service with an optional settings store."""
-    if is_keyring_usable():
-        return KeyringCredentialsService()
-
     if settings_store is None:
         from sqlit.domains.shell.store.settings import SettingsStore
 
         settings_store = SettingsStore.get_instance()
 
     settings = settings_store.load_all()
-    allow_plaintext = bool(settings.get(ALLOW_PLAINTEXT_CREDENTIALS_SETTING))
-    if allow_plaintext:
+    allow_plaintext = settings.get(ALLOW_PLAINTEXT_CREDENTIALS_SETTING)
+
+    # If user explicitly chose plaintext, use it regardless of keyring availability
+    if allow_plaintext is True:
         return PlaintextFileCredentialsService()
+
+    # Otherwise, try keyring first
+    if is_keyring_usable():
+        return KeyringCredentialsService()
+
+    # Keyring unavailable - fall back to in-memory (not persisted)
     return PlaintextCredentialsService()
 
 

sqlit/domains/connections/app/mock_profiles.py

@@ -4,6 +4,7 @@
 
 from collections.abc import Callable
 from dataclasses import dataclass, field
+from pathlib import Path
 
 from sqlit.domains.connections.domain.config import ConnectionConfig
 from sqlit.domains.connections.providers.adapters.base import ColumnInfo
@@ -64,12 +65,13 @@ def get_provider(self, db_type: str) -> DatabaseProvider:
 
 def _create_sqlite_demo_profile() -> MockProfile:
     """Create the sqlite-demo profile with pre-configured connection."""
+    demo_db_path = Path(__file__).resolve().parents[4] / "docs" / "demos" / "demo.db"
     connections = [
         ConnectionConfig.from_dict(
             {
                 "name": "Demo SQLite",
                 "db_type": "sqlite",
-                "endpoint": {"kind": "file", "path": "./docs/demos/demo.db"},
+                "endpoint": {"kind": "file", "path": str(demo_db_path)},
             }
         ),
     ]

sqlit/domains/explorer/ui/mixins/tree.py

@@ -353,7 +353,10 @@ async def work_async() -> None:
 
             columns: list[Any] = []
             try:
-                use_worker = bool(getattr(self.services.runtime, "process_worker", False))
+                runtime = getattr(self.services, "runtime", None)
+                use_worker = bool(getattr(runtime, "process_worker", False)) and not bool(
+                    getattr(getattr(runtime, "mock", None), "enabled", False)
+                )
                 client = None
                 if use_worker and hasattr(self, "_get_process_worker_client_async"):
                     client = await self._get_process_worker_client_async()  # type: ignore[attr-defined]

sqlit/domains/explorer/ui/tree/loaders.py

@@ -62,7 +62,10 @@ async def work_async() -> None:
 
         try:
             columns = []
-            use_worker = bool(getattr(host.services.runtime, "process_worker", False))
+            runtime = getattr(host.services, "runtime", None)
+            use_worker = bool(getattr(runtime, "process_worker", False)) and not bool(
+                getattr(getattr(runtime, "mock", None), "enabled", False)
+            )
             if use_worker and hasattr(host, "_get_process_worker_client_async"):
                 client = await host._get_process_worker_client_async()  # type: ignore[attr-defined]
             else:
@@ -96,9 +99,10 @@ async def work_async() -> None:
                 lambda: on_columns_loaded(host, node, db_name, schema_name, obj_name, columns),
             )
         except Exception as error:
+            error_message = f"Error loading columns: {error}"
             host.set_timer(
                 MIN_TIMER_DELAY_S,
-                lambda: on_tree_load_error(host, node, f"Error loading columns: {error}"),
+                lambda: on_tree_load_error(host, node, error_message),
             )
 
     host.run_worker(work_async(), name=f"load-columns-{obj_name}", exclusive=False)
@@ -151,7 +155,10 @@ async def work_async() -> None:
 
         try:
             items: list[Any] = []
-            use_worker = bool(getattr(host.services.runtime, "process_worker", False))
+            runtime = getattr(host.services, "runtime", None)
+            use_worker = bool(getattr(runtime, "process_worker", False)) and not bool(
+                getattr(getattr(runtime, "mock", None), "enabled", False)
+            )
             client = None
             if use_worker and hasattr(host, "_get_process_worker_client_async"):
                 client = await host._get_process_worker_client_async()  # type: ignore[attr-defined]
@@ -183,9 +190,10 @@ async def work_async() -> None:
                 lambda: on_folder_loaded(host, node, db_name, folder_type, items),
             )
         except Exception as error:
+            error_message = f"Error loading: {error}"
             host.set_timer(
                 MIN_TIMER_DELAY_S,
-                lambda: on_tree_load_error(host, node, f"Error loading: {error}"),
+                lambda: on_tree_load_error(host, node, error_message),
             )
 
     host.run_worker(work_async(), name=f"load-folder-{folder_type}", exclusive=False)

sqlit/domains/process_worker/ui/mixins/process_worker_lifecycle.py

@@ -20,6 +20,8 @@ def _use_process_worker(self: QueryMixinHost, provider: Any) -> bool:
         runtime = getattr(self.services, "runtime", None)
         if not runtime or not getattr(runtime, "process_worker", False):
             return False
+        if bool(getattr(getattr(runtime, "mock", None), "enabled", False)):
+            return False
         return True
 
     def _get_process_worker_client(self: QueryMixinHost) -> Any | None:
@@ -146,6 +148,8 @@ def _schedule_process_worker_warm(self: QueryMixinHost) -> None:
         runtime = getattr(self.services, "runtime", None)
         if runtime is None or not getattr(runtime, "process_worker_warm_on_idle", False):
             return
+        if bool(getattr(getattr(runtime, "mock", None), "enabled", False)):
+            return
         from sqlit.domains.shell.app.idle_scheduler import Priority, get_idle_scheduler
 
         scheduler = get_idle_scheduler()
@@ -158,6 +162,8 @@ def _warm() -> None:
                 return
             if not getattr(self.services.runtime, "process_worker_warm_on_idle", False):
                 return
+            if bool(getattr(getattr(self.services.runtime, "mock", None), "enabled", False)):
+                return
             self.run_worker(
                 self._get_process_worker_client_async(),
                 name="process-worker-warm",

sqlit/domains/shell/app/commands/__init__.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from .router import dispatch_command, register_command_handler
+from . import credentials as _credentials
 from . import debug as _debug
 from . import watchdog as _watchdog
 from . import worker as _worker

sqlit/domains/shell/app/commands/credentials.py

@@ -0,0 +1,156 @@
+"""Credentials storage command handler."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from .router import register_command_handler
+
+
+def _migrate_credentials(
+    app: Any,
+    old_service: Any,
+    new_service: Any,
+) -> tuple[int, int]:
+    """Migrate credentials from old service to new service.
+
+    Returns:
+        Tuple of (migrated_count, error_count).
+    """
+    migrated = 0
+    errors = 0
+
+    for conn in getattr(app, "connections", []):
+        name = getattr(conn, "name", None)
+        if not name:
+            continue
+
+        # Migrate database password
+        try:
+            db_pw = old_service.get_password(name)
+            if db_pw:
+                new_service.set_password(name, db_pw)
+                migrated += 1
+        except Exception:
+            errors += 1
+
+        # Migrate SSH password
+        try:
+            ssh_pw = old_service.get_ssh_password(name)
+            if ssh_pw:
+                new_service.set_ssh_password(name, ssh_pw)
+                migrated += 1
+        except Exception:
+            errors += 1
+
+    return migrated, errors
+
+
+def _handle_credentials_command(app: Any, cmd: str, args: list[str]) -> bool:
+    if cmd != "credentials":
+        return False
+
+    from sqlit.domains.connections.app.credentials import (
+        ALLOW_PLAINTEXT_CREDENTIALS_SETTING,
+        KeyringCredentialsService,
+        PlaintextFileCredentialsService,
+        build_credentials_service,
+        is_keyring_usable,
+        reset_credentials_service,
+    )
+
+    value = args[0].lower() if args else ""
+
+    if value == "plaintext":
+        settings = app.services.settings_store.load_all()
+        was_plaintext = settings.get(ALLOW_PLAINTEXT_CREDENTIALS_SETTING) is True
+
+        # Try to migrate from keyring if switching
+        migrated = 0
+        if not was_plaintext and is_keyring_usable():
+            try:
+                old_service = KeyringCredentialsService()
+                new_service = PlaintextFileCredentialsService()
+                migrated, _ = _migrate_credentials(app, old_service, new_service)
+            except Exception:
+                pass  # Migration is best-effort
+
+        # Enable plaintext storage
+        settings[ALLOW_PLAINTEXT_CREDENTIALS_SETTING] = True
+        app.services.settings_store.save_all(settings)
+
+        # Rebuild credentials service
+        reset_credentials_service()
+        app.services.credentials_service = build_credentials_service(app.services.settings_store)
+        if hasattr(app.services.connection_store, "set_credentials_service"):
+            app.services.connection_store.set_credentials_service(app.services.credentials_service)
+
+        msg = "Credentials will be stored as plaintext in ~/.sqlit/ (protected folder)"
+        if migrated > 0:
+            msg += f" ({migrated} password(s) migrated from keyring)"
+        app.notify(msg)
+        return True
+
+    if value == "keyring":
+        from sqlit.shared.core.store import CONFIG_DIR
+
+        settings = app.services.settings_store.load_all()
+        was_plaintext = settings.get(ALLOW_PLAINTEXT_CREDENTIALS_SETTING) is True
+
+        if not is_keyring_usable():
+            app.notify("Keyring unavailable. Cannot switch to keyring storage.", severity="warning")
+            return True
+
+        # Try to migrate from plaintext if switching
+        migrated = 0
+        if was_plaintext:
+            try:
+                old_service = PlaintextFileCredentialsService()
+                new_service = KeyringCredentialsService()
+                migrated, _ = _migrate_credentials(app, old_service, new_service)
+            except Exception:
+                pass  # Migration is best-effort
+
+            # Clear plaintext credentials file after migration
+            try:
+                creds_file = CONFIG_DIR / "credentials.json"
+                if creds_file.exists():
+                    creds_file.unlink()
+            except Exception:
+                pass  # Best-effort cleanup
+
+        # Switch to keyring
+        settings[ALLOW_PLAINTEXT_CREDENTIALS_SETTING] = False
+        app.services.settings_store.save_all(settings)
+
+        # Rebuild credentials service
+        reset_credentials_service()
+        app.services.credentials_service = build_credentials_service(app.services.settings_store)
+        if hasattr(app.services.connection_store, "set_credentials_service"):
+            app.services.connection_store.set_credentials_service(app.services.credentials_service)
+
+        msg = "Credentials will be stored in system keyring"
+        if migrated > 0:
+            msg += f" ({migrated} password(s) migrated from plaintext)"
+        app.notify(msg)
+        return True
+
+    if not value:
+        # Show current status
+        settings = app.services.settings_store.load_all()
+        allow_plaintext = settings.get(ALLOW_PLAINTEXT_CREDENTIALS_SETTING)
+        keyring_ok = is_keyring_usable()
+
+        if allow_plaintext:
+            app.notify("Credentials: plaintext (~/.sqlit/credentials.json)")
+        elif keyring_ok:
+            app.notify("Credentials: system keyring")
+        else:
+            app.notify("Credentials: keyring unavailable, passwords not persisted", severity="warning")
+        return True
+
+    app.notify("Usage: :credentials [plaintext|keyring]", severity="warning")
+    return True
+
+
+register_command_handler(_handle_credentials_command)