Comprehensive codebase review and analysis

.gitignore

@@ -100,9 +100,6 @@ tmp/
 .cache/coverage/
 .cache/coverage_html/
 
-# Tests (exclude from git tracking)
-tests/
-
 # Test documentation files
 DOCKER_MCP_COMPREHENSIVE_TEST_PLAN.md
 DOCKER_MCP_TEST_RESULTS.md

Dockerfile

@@ -1,5 +1,5 @@
 # Multi-stage build for FastMCP Docker Context Manager
-FROM python:3.13-slim AS builder
+FROM python:3.11-slim AS builder
 
 # Install build dependencies
 RUN apt-get update && apt-get install -y \
@@ -27,7 +27,7 @@ COPY docker_mcp/ ./docker_mcp/
 RUN uv sync --frozen --no-dev
 
 # Production stage
-FROM python:3.13-slim
+FROM python:3.11-slim
 
 # Install all runtime dependencies in one layer
 RUN apt-get update && apt-get install -y \

docker_mcp/core/config_loader.py

@@ -28,13 +28,19 @@ class DockerHost(BaseModel):
     compose_path: str | None = None  # Path where compose files are stored on this host
     appdata_path: str | None = None  # Path where container data volumes are stored
     enabled: bool = True
+    strict_host_checking: bool = Field(
+        default=True,
+        description="Enable SSH host key verification (recommended for security)"
+    )
 
 
 
 
 class ServerConfig(BaseModel):
     """Server configuration."""
 
+    model_config = {"populate_by_name": True}
+
     host: str = Field(
         default="127.0.0.1", alias="FASTMCP_HOST"
     )  # Use 0.0.0.0 for container deployment
@@ -46,6 +52,8 @@ class ServerConfig(BaseModel):
 class TransferConfig(BaseModel):
     """Transfer configuration for migration operations."""
 
+    model_config = {"populate_by_name": True}
+
     method: Literal["ssh", "containerized"] = Field(
         default="ssh",
         alias="DOCKER_MCP_TRANSFER_METHOD",
@@ -320,6 +328,7 @@ def _build_host_data(host_config: DockerHost) -> dict[str, Any]:
         ("docker_context", host_config.docker_context, bool(host_config.docker_context)),
         ("appdata_path", host_config.appdata_path, bool(host_config.appdata_path)),
         ("enabled", host_config.enabled, not host_config.enabled),
+        ("strict_host_checking", host_config.strict_host_checking, not host_config.strict_host_checking),
     ]
 
     # Add fields that meet their conditions

docker_mcp/core/docker_context.py

@@ -16,6 +16,7 @@
 
 from .config_loader import DockerHost, DockerMCPConfig
 from .exceptions import DockerContextError
+from .retry_utils import retry_docker_operation
 
 logger = structlog.get_logger()
 
@@ -87,6 +88,7 @@ async def _run_docker_command(
             timeout=timeout,
         )
 
+    @retry_docker_operation
     async def ensure_context(self, host_id: str) -> str:
         """Ensure Docker context exists for host."""
         if host_id not in self.config.hosts:

docker_mcp/core/retry_utils.py

@@ -0,0 +1,48 @@
+"""Retry utilities for resilient operations."""
+
+import logging
+
+import structlog
+from tenacity import (
+    before_sleep_log,
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_exponential,
+)
+
+from docker_mcp.core.exceptions import DockerCommandError, DockerContextError
+
+logger = structlog.get_logger(__name__)
+
+# SSH retry configuration
+retry_ssh_operation = retry(
+    stop=stop_after_attempt(3),
+    wait=wait_exponential(multiplier=1, min=2, max=30),
+    retry=retry_if_exception_type((ConnectionError, TimeoutError, OSError)),
+    before_sleep=before_sleep_log(logger, logging.INFO),
+    reraise=True,
+)
+
+# Docker API retry configuration
+retry_docker_operation = retry(
+    stop=stop_after_attempt(3),
+    wait=wait_exponential(multiplier=1, min=1, max=10),
+    retry=retry_if_exception_type((
+        ConnectionError,
+        TimeoutError,
+        DockerCommandError,
+        DockerContextError,
+    )),
+    before_sleep=before_sleep_log(logger, logging.INFO),
+    reraise=True,
+)
+
+# Network operation retry configuration
+retry_network_operation = retry(
+    stop=stop_after_attempt(4),
+    wait=wait_exponential(multiplier=1, min=2, max=30),
+    retry=retry_if_exception_type((ConnectionError, TimeoutError)),
+    before_sleep=before_sleep_log(logger, logging.INFO),
+    reraise=True,
+)

docker_mcp/core/ssh_utils.py

@@ -0,0 +1,124 @@
+"""SSH utilities for secure connection management."""
+import asyncio
+import shlex
+from pathlib import Path
+
+from docker_mcp.core.config_loader import DockerHost
+
+
+def get_ssh_config_dir() -> Path:
+    """Get SSH configuration directory for docker-mcp."""
+    config_dir = Path.home() / ".config" / "docker-mcp" / "ssh"
+    config_dir.mkdir(parents=True, exist_ok=True)
+    return config_dir
+
+
+def get_known_hosts_file() -> Path:
+    """Get path to docker-mcp known_hosts file."""
+    return get_ssh_config_dir() / "known_hosts"
+
+
+def build_ssh_command(
+    host: DockerHost,
+    remote_command: list[str] | None = None,
+    strict_host_checking: bool = True
+) -> list[str]:
+    """Build secure SSH command with proper options.
+
+    Args:
+        host: Docker host configuration
+        remote_command: Optional remote command to execute
+        strict_host_checking: Enable strict host key checking (default: True)
+
+    Returns:
+        Complete SSH command as list
+
+    Example:
+        >>> host = DockerHost(hostname="server.com", user="docker", port=22)
+        >>> build_ssh_command(host)
+        ['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10',
+         '-o', 'StrictHostKeyChecking=accept-new', '[email protected]']
+    """
+    # Get known_hosts file
+    known_hosts = get_known_hosts_file()
+
+    # Build SSH command
+    ssh_cmd = [
+        "ssh",
+        "-o", "BatchMode=yes",
+        "-o", "ConnectTimeout=10",
+        "-o", "ServerAliveInterval=30",
+        "-o", "ServerAliveCountMax=3",
+        "-o", "LogLevel=ERROR",
+    ]
+
+    # Host key checking configuration
+    if strict_host_checking:
+        # Accept new host keys but verify existing ones
+        ssh_cmd.extend([
+            "-o", "StrictHostKeyChecking=accept-new",
+            "-o", f"UserKnownHostsFile={known_hosts}",
+            "-o", "HashKnownHosts=yes",
+        ])
+    else:
+        # Legacy mode - log warning
+        ssh_cmd.extend([
+            "-o", "StrictHostKeyChecking=no",
+            "-o", "UserKnownHostsFile=/dev/null",
+        ])
+
+    # Add identity file if specified
+    if host.identity_file:
+        ssh_cmd.extend(["-i", host.identity_file])
+
+    # Add port if non-standard
+    if host.port and host.port != 22:
+        ssh_cmd.extend(["-p", str(host.port)])
+
+    # Handle hostname with proper quoting and IPv6 support
+    hostname = host.hostname
+    if ":" in hostname and not (hostname.startswith("[") and hostname.endswith("]")):
+        # IPv6 address needs brackets
+        hostname = f"[{hostname}]"
+
+    # Use proper quoting for hostname
+    user_host = f"{host.user}@{shlex.quote(hostname)}"
+    ssh_cmd.append(user_host)
+
+    # Add remote command if provided
+    if remote_command:
+        ssh_cmd.extend(remote_command)
+
+    return ssh_cmd
+
+
+async def test_ssh_connection(host: DockerHost) -> tuple[bool, str]:
+    """Test SSH connection to host.
+
+    Args:
+        host: Docker host to test
+
+    Returns:
+        Tuple of (success: bool, message: str)
+    """
+    # Use strict host checking from host config or default to True
+    strict = getattr(host, 'strict_host_checking', True)
+    ssh_cmd = build_ssh_command(host, ["echo", "OK"], strict_host_checking=strict)
+
+    try:
+        process = await asyncio.create_subprocess_exec(
+            *ssh_cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE
+        )
+        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=15.0)
+
+        if process.returncode == 0 and b"OK" in stdout:
+            return True, "SSH connection successful"
+        else:
+            return False, f"SSH connection failed: {stderr.decode()}"
+
+    except asyncio.TimeoutError:
+        return False, "SSH connection timeout"
+    except Exception as e:
+        return False, f"SSH connection error: {str(e)}"

docker_mcp/core/transfer/rsync.py

@@ -10,6 +10,7 @@
 
 from ..config_loader import DockerHost
 from ..exceptions import DockerMCPError
+from ..retry_utils import retry_network_operation
 from ..settings import RSYNC_TIMEOUT
 from .base import BaseTransfer
 
@@ -66,6 +67,7 @@ async def validate_requirements(self, host: DockerHost) -> tuple[bool, str]:
         except Exception as e:
             return False, f"Failed to check rsync availability: {str(e)}"
 
+    @retry_network_operation
     async def transfer(
         self,
         source_host: DockerHost,

docker_mcp/services/host.py

@@ -19,6 +19,7 @@
 
 from ..constants import APPDATA_PATH, COMPOSE_PATH, DOCKER_COMPOSE_WORKING_DIR, HOST_ID
 from ..core.config_loader import DockerHost, DockerMCPConfig, load_config, save_config
+from ..core.retry_utils import retry_ssh_operation
 from ..utils import build_ssh_command
 
 
@@ -401,6 +402,7 @@ async def remove_docker_host(self, host_id: str) -> dict[str, Any]:
                 ),
             }
 
+    @retry_ssh_operation
     async def test_connection(self, host_id: str) -> dict[str, Any]:
         """Test SSH connection to a Docker host.
 

docker_mcp/utils.py

@@ -12,8 +12,10 @@
 Total impact: ~120 lines of duplicate code eliminated.
 """
 
-from .constants import SSH_NO_HOST_CHECK
+import warnings
+
 from .core.config_loader import DockerHost, DockerMCPConfig
+from .core.ssh_utils import build_ssh_command as _build_ssh_command_secure
 
 
 def build_ssh_command(host: DockerHost) -> list[str]:
@@ -36,37 +38,27 @@ def build_ssh_command(host: DockerHost) -> list[str]:
     Example:
         >>> host = DockerHost(hostname="server.com", user="docker", port=22)
         >>> build_ssh_command(host)
-        ['ssh', '-o', 'StrictHostKeyChecking=no', '-o', 'ConnectTimeout=10', '[email protected]']
+        ['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10',
+         '-o', 'StrictHostKeyChecking=accept-new', '[email protected]']
+
+    Note:
+        This function now uses secure SSH host key verification by default.
+        To disable (not recommended), set host.strict_host_checking = False.
     """
-    import shlex
-
-    ssh_cmd = [
-        "ssh",
-        "-o", SSH_NO_HOST_CHECK,
-        "-o", "UserKnownHostsFile=/dev/null",  # Prevent host key issues
-        "-o", "LogLevel=ERROR",  # Reduce noise
-        "-o", "ConnectTimeout=10",  # Connection timeout for automation
-        "-o", "ServerAliveInterval=30",  # Keep connection alive
-        "-o", "BatchMode=yes",  # Fully automated connections (no prompts)
-    ]
-
-    if host.identity_file:
-        ssh_cmd.extend(["-i", host.identity_file])
-
-    if host.port != 22:
-        ssh_cmd.extend(["-p", str(host.port)])
-
-    # Handle hostname with proper quoting and IPv6 support
-    hostname = host.hostname
-    if ":" in hostname and not (hostname.startswith("[") and hostname.endswith("]")):
-        # IPv6 address needs brackets
-        hostname = f"[{hostname}]"
-
-    # Use proper quoting for hostname
-    user_host = f"{host.user}@{shlex.quote(hostname)}"
-    ssh_cmd.append(user_host)
-
-    return ssh_cmd
+    # Get strict_host_checking from host config (defaults to True for security)
+    strict = getattr(host, 'strict_host_checking', True)
+
+    # Issue deprecation warning if using insecure mode
+    if not strict:
+        warnings.warn(
+            "SSH host key verification is disabled. This is insecure and "
+            "vulnerable to MITM attacks. Enable strict_host_checking for security.",
+            DeprecationWarning,
+            stacklevel=2
+        )
+
+    # Use the new secure SSH command builder
+    return _build_ssh_command_secure(host, strict_host_checking=strict)
 
 
 def validate_host(config: DockerMCPConfig, host_id: str) -> tuple[bool, str]: