Add SSH tunnel support for database connections

- Add SSH tab in connection dialog with host, port, username, auth type options
- Support both password and key-based SSH authentication
- Implement SSH tunnel creation in adapters.py using sshtunnel library
- Add SSH CLI options for connection create command
- Add SSH integration tests with Docker-based SSH server
- Pin paramiko<4.0.0 to avoid DSSKey compatibility issue
- Update CI workflow with SSH tunnel test job

Author: Maxteabag

.github/workflows/ci.yml

@@ -365,3 +365,86 @@ jobs:
           COCKROACHDB_DATABASE: test_sqlit
         run: |
           pytest tests/test_cockroachdb.py -v --timeout=120
+
+  test-ssh:
+    runs-on: ubuntu-latest
+    needs: build
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python 3.12
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -e ".[test]"
+          pip install psycopg2-binary
+
+      - name: Create Docker network
+        run: docker network create ssh-test-net
+
+      - name: Start PostgreSQL
+        run: |
+          docker run -d --name postgres --network ssh-test-net \
+            -e POSTGRES_USER=testuser \
+            -e POSTGRES_PASSWORD=TestPassword123! \
+            -e POSTGRES_DB=test_sqlit \
+            -p 5433:5432 \
+            postgres:16-alpine
+          # Wait for PostgreSQL to be ready
+          for i in {1..30}; do
+            if docker exec postgres pg_isready -U testuser -d test_sqlit > /dev/null 2>&1; then
+              echo "PostgreSQL is ready"
+              break
+            fi
+            echo "Waiting for PostgreSQL... ($i/30)"
+            sleep 2
+          done
+
+      - name: Start SSH server
+        run: |
+          docker run -d --name sshserver --network ssh-test-net \
+            -e PUID=1000 \
+            -e PGID=1000 \
+            -e USER_NAME=testuser \
+            -e USER_PASSWORD=testpass \
+            -e PASSWORD_ACCESS=true \
+            -e DOCKER_MODS=linuxserver/mods:openssh-server-ssh-tunnel \
+            -p 2222:2222 \
+            lscr.io/linuxserver/openssh-server:latest
+          # Wait for SSH to be ready
+          for i in {1..30}; do
+            if nc -z localhost 2222 2>/dev/null; then
+              echo "SSH server is ready"
+              break
+            fi
+            echo "Waiting for SSH server... ($i/30)"
+            sleep 2
+          done
+
+      - name: Run SSH tunnel integration tests
+        env:
+          SSH_HOST: localhost
+          SSH_PORT: 2222
+          SSH_USER: testuser
+          SSH_PASSWORD: testpass
+          SSH_REMOTE_DB_HOST: postgres
+          SSH_REMOTE_DB_PORT: 5432
+          SSH_DIRECT_PG_HOST: localhost
+          SSH_DIRECT_PG_PORT: 5433
+          POSTGRES_USER: testuser
+          POSTGRES_PASSWORD: TestPassword123!
+          POSTGRES_DATABASE: test_sqlit
+        run: |
+          pytest tests/test_ssh.py -v --timeout=120
+
+      - name: Cleanup
+        if: always()
+        run: |
+          docker stop sshserver postgres || true
+          docker rm sshserver postgres || true
+          docker network rm ssh-test-net || true

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "sqlit-tui"
-version = "0.3.1"
+version = "0.3.5"
 description = "A terminal UI for SQL Server, PostgreSQL, MySQL, SQLite, and Oracle"
 readme = "README.md"
 license = "MIT"
@@ -30,6 +30,8 @@ dependencies = [
     "textual[syntax]>=0.50.0",
     "pyodbc>=5.0.0",
     "pyperclip>=1.8.2",
+    "sshtunnel>=0.4.0",
+    "paramiko>=2.0.0,<4.0.0",  # sshtunnel requires paramiko<4.0.0 (DSSKey removed in 4.0)
 ]
 
 [project.optional-dependencies]

sqlit/adapters.py

@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+import os
 import sqlite3
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 from .fields import (
@@ -20,6 +22,54 @@
     from .config import ConnectionConfig
 
 
+def create_ssh_tunnel(config: "ConnectionConfig") -> tuple[Any, str, int]:
+    """Create an SSH tunnel for the connection if SSH is enabled.
+
+    Returns:
+        Tuple of (tunnel_object, local_host, local_port) if SSH enabled,
+        or (None, original_server, original_port) if SSH not enabled.
+    """
+    if not config.ssh_enabled:
+        port = int(config.port) if config.port else 0
+        return None, config.server, port
+
+    from sshtunnel import SSHTunnelForwarder
+
+    # Parse remote database host and port
+    remote_host = config.server
+    remote_port = int(config.port) if config.port else 0
+
+    # SSH connection settings
+    ssh_host = config.ssh_host
+    ssh_port = int(config.ssh_port) if config.ssh_port else 22
+    ssh_username = config.ssh_username
+
+    # Build SSH auth kwargs
+    ssh_kwargs: dict[str, Any] = {
+        "ssh_username": ssh_username,
+    }
+
+    if config.ssh_auth_type == "key":
+        # Expand ~ in path
+        key_path = os.path.expanduser(config.ssh_key_path)
+        if Path(key_path).exists():
+            ssh_kwargs["ssh_pkey"] = key_path
+        else:
+            raise ValueError(f"SSH key file not found: {key_path}")
+    else:
+        ssh_kwargs["ssh_password"] = config.ssh_password
+
+    # Create tunnel
+    tunnel = SSHTunnelForwarder(
+        (ssh_host, ssh_port),
+        remote_bind_address=(remote_host, remote_port),
+        **ssh_kwargs,
+    )
+    tunnel.start()
+
+    return tunnel, "127.0.0.1", tunnel.local_bind_port
+
+
 @dataclass
 class ColumnInfo:
     """Information about a database column."""
@@ -316,12 +366,16 @@ def get_connection_fields(self) -> list[FieldGroup]:
                         placeholder="",
                         visible_when=lambda v: v.get("auth_type") in auth_needs_username,
                         required=True,
+                        row_group="credentials",
+                        width="flex",
                     ),
                     FieldDefinition(
                         name="password",
                         label="Password",
                         field_type=FieldType.PASSWORD,
                         visible_when=lambda v: v.get("auth_type") in auth_needs_password,
+                        row_group="credentials",
+                        width="flex",
                     ),
                 ],
             ),

sqlit/app.py

@@ -16,7 +16,7 @@
 from textual.containers import Container, Horizontal, Vertical
 from textual.widgets import DataTable, Static, TextArea, Tree
 
-from .adapters import DatabaseAdapter, get_adapter
+from .adapters import DatabaseAdapter, create_ssh_tunnel, get_adapter
 from .config import (
     ConnectionConfig,
     load_connections,
@@ -197,6 +197,7 @@ def __init__(self):
         self.current_connection: Any | None = None
         self.current_config: ConnectionConfig | None = None
         self.current_adapter: DatabaseAdapter | None = None
+        self.current_ssh_tunnel: Any | None = None
         self.vim_mode: VimMode = VimMode.NORMAL
         self._expanded_paths: set[str] = set()
         self._schema_cache: dict = {
@@ -1228,27 +1229,55 @@ def handle_connection_result(self, result: tuple | None) -> None:
 
     def connect_to_server(self, config: ConnectionConfig) -> None:
         """Connect to a database."""
+        from dataclasses import replace
+
         # Check for pyodbc only if it's a SQL Server connection
         if config.db_type == "mssql" and not PYODBC_AVAILABLE:
             self.notify("pyodbc not installed. Run: pip install pyodbc", severity="error")
             return
 
         try:
+            # Close any existing SSH tunnel
+            if self.current_ssh_tunnel:
+                try:
+                    self.current_ssh_tunnel.stop()
+                except Exception:
+                    pass
+                self.current_ssh_tunnel = None
+
+            # Create SSH tunnel if enabled
+            tunnel, host, port = create_ssh_tunnel(config)
+            self.current_ssh_tunnel = tunnel
+
+            # If SSH tunnel was created, use the tunnel's local address
+            if tunnel:
+                connect_config = replace(config, server=host, port=str(port))
+            else:
+                connect_config = config
+
             adapter = get_adapter(config.db_type)
-            self.current_connection = adapter.connect(config)
-            self.current_config = config
+            self.current_connection = adapter.connect(connect_config)
+            self.current_config = config  # Store original config (not tunneled)
             self.current_adapter = adapter
             self._set_connection_health(config.name, True)
 
             status = self.query_one("#status-bar", Static)
             display_info = config.get_display_info()
-            status.update(f"[#90EE90]Connected to {config.name}[/] ({display_info})")
+            ssh_indicator = " [SSH]" if tunnel else ""
+            status.update(f"[#90EE90]Connected to {config.name}[/] ({display_info}){ssh_indicator}")
 
             self.refresh_tree()
             self._load_schema_cache()
             self.notify(f"Connected to {config.name}")
 
         except Exception as e:
+            # Clean up SSH tunnel on failure
+            if self.current_ssh_tunnel:
+                try:
+                    self.current_ssh_tunnel.stop()
+                except Exception:
+                    pass
+                self.current_ssh_tunnel = None
             self._set_connection_health(config.name, False)
             self.refresh_tree()
             self.notify(f"Connection failed: {e}", severity="error")
@@ -1336,6 +1365,14 @@ def _disconnect_silent(self) -> None:
             self.current_config = None
             self.current_adapter = None
 
+        # Close SSH tunnel if active
+        if self.current_ssh_tunnel:
+            try:
+                self.current_ssh_tunnel.stop()
+            except Exception:
+                pass
+            self.current_ssh_tunnel = None
+
     def action_disconnect(self) -> None:
         """Disconnect from current database."""
         if self.current_connection:

sqlit/cli.py

@@ -54,6 +54,14 @@ def main() -> int:
     )
     # SQLite options
     create_parser.add_argument("--file-path", help="Database file path (SQLite only)")
+    # SSH tunnel options
+    create_parser.add_argument("--ssh-enabled", action="store_true", help="Enable SSH tunnel")
+    create_parser.add_argument("--ssh-host", help="SSH server hostname")
+    create_parser.add_argument("--ssh-port", default="22", help="SSH server port (default: 22)")
+    create_parser.add_argument("--ssh-username", help="SSH username")
+    create_parser.add_argument("--ssh-auth-type", default="key", choices=["key", "password"], help="SSH auth type")
+    create_parser.add_argument("--ssh-key-path", help="SSH private key path")
+    create_parser.add_argument("--ssh-password", help="SSH password")
 
     # connection edit
     edit_parser = conn_subparsers.add_parser("edit", help="Edit an existing connection")

sqlit/commands.py

@@ -4,7 +4,7 @@
 
 import json
 
-from .adapters import get_adapter
+from .adapters import create_ssh_tunnel, get_adapter
 from .config import (
     AUTH_TYPE_LABELS,
     AuthType,
@@ -97,6 +97,13 @@ def cmd_connection_create(args) -> int:
             database=args.database or "",
             username=args.username or "",
             password=args.password or "",
+            ssh_enabled=getattr(args, "ssh_enabled", False) or False,
+            ssh_host=getattr(args, "ssh_host", "") or "",
+            ssh_port=getattr(args, "ssh_port", "22") or "22",
+            ssh_username=getattr(args, "ssh_username", "") or "",
+            ssh_auth_type=getattr(args, "ssh_auth_type", "key") or "key",
+            ssh_key_path=getattr(args, "ssh_key_path", "") or "",
+            ssh_password=getattr(args, "ssh_password", "") or "",
         )
     else:
         # SQL Server connection (mssql)
@@ -122,6 +129,13 @@ def cmd_connection_create(args) -> int:
             password=args.password or "",
             auth_type=auth_type.value,
             trusted_connection=(auth_type == AuthType.WINDOWS),
+            ssh_enabled=getattr(args, "ssh_enabled", False) or False,
+            ssh_host=getattr(args, "ssh_host", "") or "",
+            ssh_port=getattr(args, "ssh_port", "22") or "22",
+            ssh_username=getattr(args, "ssh_username", "") or "",
+            ssh_auth_type=getattr(args, "ssh_auth_type", "key") or "key",
+            ssh_key_path=getattr(args, "ssh_key_path", "") or "",
+            ssh_password=getattr(args, "ssh_password", "") or "",
         )
 
     connections.append(config)
@@ -237,9 +251,19 @@ def cmd_query(args) -> int:
         print("Error: Either --query or --file must be provided.")
         return 1
 
+    tunnel = None
     try:
+        from dataclasses import replace
+
+        # Create SSH tunnel if enabled
+        tunnel, host, port = create_ssh_tunnel(config)
+        if tunnel:
+            connect_config = replace(config, server=host, port=str(port))
+        else:
+            connect_config = config
+
         adapter = get_adapter(config.db_type)
-        db_conn = adapter.connect(config)
+        db_conn = adapter.connect(connect_config)
 
         # Detect query type to avoid executing non-SELECT statements twice
         query_type = query.strip().upper().split()[0] if query.strip() else ""
@@ -294,11 +318,17 @@ def cmd_query(args) -> int:
             print(f"Query executed successfully. Rows affected: {affected}")
 
         db_conn.close()
+        if tunnel:
+            tunnel.stop()
         return 0
 
     except ImportError as e:
         print(f"Error: Required module not installed: {e}")
+        if tunnel:
+            tunnel.stop()
         return 1
     except Exception as e:
         print(f"Error: {e}")
+        if tunnel:
+            tunnel.stop()
         return 1

sqlit/config.py

@@ -70,6 +70,14 @@ class ConnectionConfig:
     trusted_connection: bool = False  # Legacy field for backwards compatibility
     # SQLite specific fields
     file_path: str = ""
+    # SSH tunnel fields
+    ssh_enabled: bool = False
+    ssh_host: str = ""
+    ssh_port: str = "22"
+    ssh_username: str = ""
+    ssh_auth_type: str = "key"  # "key" or "password"
+    ssh_password: str = ""
+    ssh_key_path: str = ""
 
     def __post_init__(self):
         """Handle backwards compatibility with old configs."""