michael/compose-farm
1
0
Fork 0
mirror of https://github.com/basnijholt/compose-farm.git synced 2026-02-03 06:03:25 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(cli): Add ssh subcommand for SSH key management (#22)

Browse Source
This commit is contained in:
Bas Nijholt
2025-12-18 11:58:33 -08:00
committed by GitHub
parent 2c5308aea3
commit 282de12336
10 changed files with 812 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
README.md
View File

@@ -23,6 +23,9 @@ A minimal CLI tool to run Docker Compose commands across multiple hosts via SSH.
- [Best practices](#best-practices)
- [What Compose Farm doesn't do](#what-compose-farm-doesnt-do)
- [Installation](#installation)
- [SSH Authentication](#ssh-authentication)
- [SSH Agent (default)](#ssh-agent-default)
- [Dedicated SSH Key (recommended for Docker/Web UI)](#dedicated-ssh-key-recommended-for-dockerweb-ui)
- [Configuration](#configuration)
- [Multi-Host Services](#multi-host-services)
- [Config Command](#config-command)
@@ -159,6 +162,56 @@ docker run --rm \
</details>
## SSH Authentication
Compose Farm uses SSH to run commands on remote hosts. There are two authentication methods:
### SSH Agent (default)
Works out of the box if you have an SSH agent running with your keys loaded:
```bash
# Verify your agent has keys
ssh-add -l
# Run compose-farm commands
cf up --all
```
### Dedicated SSH Key (recommended for Docker/Web UI)
When running compose-farm in Docker, the SSH agent connection can be lost (e.g., after container restart). The `cf ssh` command sets up a dedicated key that persists:
```bash
# Generate key and copy to all configured hosts
cf ssh setup
# Check status
cf ssh status
```
This creates `~/.ssh/compose-farm` (ED25519, no passphrase) and copies the public key to each host's `authorized_keys`. Compose Farm tries the SSH agent first, then falls back to this key.
<details><summary>🐳 Docker volume options for SSH keys</summary>
When running in Docker, mount a volume to persist the SSH keys:
**Option 1: Named volume (default)**
```yaml
volumes:
- cf-ssh:/root/.ssh
```
**Option 2: Host path (easier to backup/inspect)**
```yaml
volumes:
- ~/.ssh/compose-farm:/root/.ssh
```
Run `cf ssh setup` once after starting the container (while the SSH agent still works), and the keys will persist across restarts.
</details>
## Configuration
Create `~/.config/compose-farm/compose-farm.yaml` (or `./compose-farm.yaml` in your working directory):
@@ -344,6 +397,7 @@ Full `--help` output for each command. See the [Usage](#usage) table above for a
│ check Validate configuration, traefik labels, mounts, and networks. │
│ init-network Create Docker network on hosts with consistent settings. │
│ config Manage compose-farm configuration files. │
│ ssh Manage SSH keys for passwordless authentication. │
╰──────────────────────────────────────────────────────────────────────────────╯
╭─ Monitoring ─────────────────────────────────────────────────────────────────╮
│ logs Show service logs. │
@@ -773,6 +827,21 @@ Full `--help` output for each command. See the [Usage](#usage) table above for a
</details>
<details>
<summary>See the output of <code>cf ssh --help</code></summary>
<!-- CODE:BASH:START -->
<!-- echo '```yaml' -->
<!-- export NO_COLOR=1 -->
<!-- export TERM=dumb -->
<!-- export TERMINAL_WIDTH=90 -->
<!-- cf ssh --help -->
<!-- echo '```' -->
<!-- CODE:END -->
</details>
**Monitoring**
<details>

12
docker-compose.yml
View File

@@ -5,6 +5,11 @@ services:
- ${SSH_AUTH_SOCK}:/ssh-agent:ro
# Compose directory (contains compose files AND compose-farm.yaml config)
- ${CF_COMPOSE_DIR:-/opt/stacks}:${CF_COMPOSE_DIR:-/opt/stacks}
# SSH keys for passwordless auth (generated by `cf ssh setup`)
# Option 1: Named volume (default) - managed by Docker
- cf-ssh:/root/.ssh
# Option 2: Host path - easier to backup/inspect, uncomment to use:
# - ${CF_SSH_DIR:-~/.ssh/compose-farm}:/root/.ssh
environment:
- SSH_AUTH_SOCK=/ssh-agent
# Config file path (state stored alongside it)
@@ -17,6 +22,9 @@ services:
volumes:
- ${SSH_AUTH_SOCK}:/ssh-agent:ro
- ${CF_COMPOSE_DIR:-/opt/stacks}:${CF_COMPOSE_DIR:-/opt/stacks}
# SSH keys - use same option as cf service above
- cf-ssh:/root/.ssh
# - ${CF_SSH_DIR:-~/.ssh/compose-farm}:/root/.ssh
environment:
- SSH_AUTH_SOCK=/ssh-agent
- CF_CONFIG=${CF_COMPOSE_DIR:-/opt/stacks}/compose-farm.yaml
@@ -33,3 +41,7 @@ services:
networks:
mynetwork:
external: true
volumes:
cf-ssh:
# Persists SSH keys across container restarts

1
src/compose_farm/cli/__init__.py
View File

@@ -8,6 +8,7 @@ from compose_farm.cli import (
lifecycle, # noqa: F401
management, # noqa: F401
monitoring, # noqa: F401
ssh, # noqa: F401
web, # noqa: F401
)

284
src/compose_farm/cli/ssh.py Normal file
View File

@@ -0,0 +1,284 @@
"""SSH key management commands for compose-farm."""
from __future__ import annotations
import subprocess
from typing import Annotated
import typer
from compose_farm.cli.app import app
from compose_farm.cli.common import ConfigOption, load_config_or_exit
from compose_farm.console import console, err_console
from compose_farm.ssh_keys import (
SSH_KEY_PATH,
SSH_PUBKEY_PATH,
get_pubkey_content,
get_ssh_env,
key_exists,
)
_DEFAULT_SSH_PORT = 22
_PUBKEY_DISPLAY_THRESHOLD = 60
ssh_app = typer.Typer(
name="ssh",
help="Manage SSH keys for passwordless authentication.",
no_args_is_help=True,
)
_ForceOption = Annotated[
bool,
typer.Option("--force", "-f", help="Regenerate key even if it exists."),
]
def _generate_key(*, force: bool = False) -> bool:
"""Generate an ED25519 SSH key with no passphrase.
Returns True if key was generated, False if skipped.
"""
if key_exists() and not force:
console.print(f"[yellow]![/] SSH key already exists: {SSH_KEY_PATH}")
console.print("[dim]Use --force to regenerate[/]")
return False
# Create .ssh directory if it doesn't exist
SSH_KEY_PATH.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
# Remove existing key if forcing regeneration
if force:
SSH_KEY_PATH.unlink(missing_ok=True)
SSH_PUBKEY_PATH.unlink(missing_ok=True)
console.print(f"[dim]Generating SSH key at {SSH_KEY_PATH}...[/]")
try:
subprocess.run(
[ # noqa: S607
"ssh-keygen",
"-t",
"ed25519",
"-N",
"", # No passphrase
"-f",
str(SSH_KEY_PATH),
"-C",
"compose-farm",
],
check=True,
capture_output=True,
)
except subprocess.CalledProcessError as e:
err_console.print(f"[red]Failed to generate SSH key:[/] {e.stderr.decode()}")
return False
except FileNotFoundError:
err_console.print("[red]ssh-keygen not found. Is OpenSSH installed?[/]")
return False
# Set correct permissions
SSH_KEY_PATH.chmod(0o600)
SSH_PUBKEY_PATH.chmod(0o644)
console.print(f"[green]Generated SSH key:[/] {SSH_KEY_PATH}")
return True
def _copy_key_to_host(host_name: str, address: str, user: str, port: int) -> bool:
"""Copy public key to a host's authorized_keys.
Uses ssh-copy-id which handles agent vs password fallback automatically.
Returns True on success, False on failure.
"""
target = f"{user}@{address}"
console.print(f"[dim]Copying key to {host_name} ({target})...[/]")
cmd = ["ssh-copy-id"]
# Disable strict host key checking (consistent with executor.py)
cmd.extend(["-o", "StrictHostKeyChecking=no"])
cmd.extend(["-o", "UserKnownHostsFile=/dev/null"])
if port != _DEFAULT_SSH_PORT:
cmd.extend(["-p", str(port)])
cmd.extend(["-i", str(SSH_PUBKEY_PATH), target])
try:
# Don't capture output so user can see password prompt
result = subprocess.run(cmd, check=False, env=get_ssh_env())
if result.returncode == 0:
console.print(f"[green]Key copied to {host_name}[/]")
return True
err_console.print(f"[red]Failed to copy key to {host_name}[/]")
return False
except FileNotFoundError:
err_console.print("[red]ssh-copy-id not found. Is OpenSSH installed?[/]")
return False
@ssh_app.command("keygen")
def ssh_keygen(
force: _ForceOption = False,
) -> None:
"""Generate SSH key (does not distribute to hosts).
Creates an ED25519 key at ~/.ssh/compose-farm with no passphrase.
Use 'cf ssh setup' to also distribute the key to all configured hosts.
"""
success = _generate_key(force=force)
if not success and not key_exists():
raise typer.Exit(1)
@ssh_app.command("setup")
def ssh_setup(
config: ConfigOption = None,
force: _ForceOption = False,
) -> None:
"""Generate SSH key and distribute to all configured hosts.
Creates an ED25519 key at ~/.ssh/compose-farm (no passphrase) and
copies the public key to authorized_keys on each host.
For each host, tries SSH agent first. If agent is unavailable,
prompts for password.
"""
cfg = load_config_or_exit(config)
# Skip localhost hosts
remote_hosts = {
name: host
for name, host in cfg.hosts.items()
if host.address.lower() not in ("localhost", "127.0.0.1")
}
if not remote_hosts:
console.print("[yellow]No remote hosts configured.[/]")
raise typer.Exit(0)
# Generate key if needed
if not key_exists() or force:
if not _generate_key(force=force):
raise typer.Exit(1)
else:
console.print(f"[dim]Using existing key: {SSH_KEY_PATH}[/]")
console.print()
console.print(f"[bold]Distributing key to {len(remote_hosts)} host(s)...[/]")
console.print()
# Copy key to each host
succeeded = 0
failed = 0
for host_name, host in remote_hosts.items():
if _copy_key_to_host(host_name, host.address, host.user, host.port):
succeeded += 1
else:
failed += 1
console.print()
if failed == 0:
console.print(
f"[green]Setup complete.[/] {succeeded}/{len(remote_hosts)} hosts configured."
)
else:
console.print(
f"[yellow]Setup partially complete.[/] {succeeded}/{len(remote_hosts)} hosts configured, "
f"[red]{failed} failed[/]."
)
raise typer.Exit(1)
@ssh_app.command("status")
def ssh_status( # noqa: PLR0912 - branches are clear and readable
config: ConfigOption = None,
) -> None:
"""Show SSH key status and host connectivity."""
from rich.table import Table # noqa: PLC0415
cfg = load_config_or_exit(config)
# Key status
console.print("[bold]SSH Key Status[/]")
console.print()
if key_exists():
console.print(f" [green]Key exists:[/] {SSH_KEY_PATH}")
pubkey = get_pubkey_content()
if pubkey:
# Show truncated public key
if len(pubkey) > _PUBKEY_DISPLAY_THRESHOLD:
console.print(f" [dim]Public key:[/] {pubkey[:30]}...{pubkey[-20:]}")
else:
console.print(f" [dim]Public key:[/] {pubkey}")
else:
console.print(f" [yellow]No key found:[/] {SSH_KEY_PATH}")
console.print(" [dim]Run 'cf ssh setup' to generate and distribute a key[/]")
console.print()
console.print("[bold]Host Connectivity[/]")
console.print()
# Skip localhost hosts
remote_hosts = {
name: host
for name, host in cfg.hosts.items()
if host.address.lower() not in ("localhost", "127.0.0.1")
}
if not remote_hosts:
console.print(" [dim]No remote hosts configured[/]")
return
table = Table(show_header=True, header_style="bold")
table.add_column("Host")
table.add_column("Address")
table.add_column("Status")
for host_name, host in remote_hosts.items():
target = f"{host.user}@{host.address}"
if host.port != _DEFAULT_SSH_PORT:
target += f":{host.port}"
# Test connectivity with a simple command
cmd = [
"ssh",
"-o",
"StrictHostKeyChecking=no",
"-o",
"UserKnownHostsFile=/dev/null",
"-o",
"BatchMode=yes", # Fail immediately if password required
"-o",
"ConnectTimeout=5",
]
# Add key file if it exists
if key_exists():
cmd.extend(["-i", str(SSH_KEY_PATH)])
if host.port != _DEFAULT_SSH_PORT:
cmd.extend(["-p", str(host.port)])
cmd.extend([f"{host.user}@{host.address}", "echo ok"])
try:
result = subprocess.run(
cmd, check=False, capture_output=True, timeout=10, env=get_ssh_env()
)
if result.returncode == 0:
table.add_row(host_name, target, "[green]OK[/]")
else:
table.add_row(host_name, target, "[red]Auth failed[/]")
except subprocess.TimeoutExpired:
table.add_row(host_name, target, "[red]Timeout[/]")
except Exception as e:
table.add_row(host_name, target, f"[red]Error: {e}[/]")
console.print(table)
# Register ssh subcommand on the shared app
app.add_typer(ssh_app, name="ssh", rich_help_panel="Configuration")

19
src/compose_farm/executor.py
View File

@@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any
from rich.markup import escape
from .console import console, err_console
from .ssh_keys import get_key_path, get_ssh_auth_sock, get_ssh_env
if TYPE_CHECKING:
from collections.abc import Callable
@@ -73,12 +74,21 @@ def is_local(host: Host) -> bool:
def ssh_connect_kwargs(host: Host) -> dict[str, Any]:
"""Get kwargs for asyncssh.connect() from a Host config."""
return {
kwargs: dict[str, Any] = {
"host": host.address,
"port": host.port,
"username": host.user,
"known_hosts": None,
}
# Add SSH agent path (auto-detect forwarded agent if needed)
agent_path = get_ssh_auth_sock()
if agent_path:
kwargs["agent_path"] = agent_path
# Add key file fallback for when SSH agent is unavailable
key_path = get_key_path()
if key_path:
kwargs["client_keys"] = [str(key_path)]
return kwargs
async def _run_local_command(
@@ -172,11 +182,16 @@ async def _run_ssh_command(
"-o",
"LogLevel=ERROR", # Suppress warnings about known_hosts
]
# Add key file if it exists (fallback for when agent is unavailable)
key_path = get_key_path()
if key_path:
ssh_args.extend(["-i", str(key_path)])
if host.port != _DEFAULT_SSH_PORT:
ssh_args.extend(["-p", str(host.port)])
ssh_args.extend([f"{host.user}@{host.address}", command])
# Run in thread to avoid blocking the event loop
result = await asyncio.to_thread(subprocess.run, ssh_args, check=False)
# Use get_ssh_env() to auto-detect SSH agent socket
result = await asyncio.to_thread(subprocess.run, ssh_args, check=False, env=get_ssh_env())
return CommandResult(
service=service,
exit_code=result.returncode,

65
src/compose_farm/ssh_keys.py Normal file
View File

@@ -0,0 +1,65 @@
"""SSH key utilities for compose-farm."""
from __future__ import annotations
import os
from pathlib import Path
# Default key paths for compose-farm SSH key
SSH_KEY_PATH = Path.home() / ".ssh" / "compose-farm"
SSH_PUBKEY_PATH = SSH_KEY_PATH.with_suffix(".pub")
def get_ssh_auth_sock() -> str | None:
"""Get SSH_AUTH_SOCK, auto-detecting forwarded agent if needed.
Checks in order:
1. SSH_AUTH_SOCK environment variable (if socket exists)
2. Forwarded agent sockets in ~/.ssh/agent/ (most recent first)
Returns the socket path or None if no valid socket found.
"""
sock = os.environ.get("SSH_AUTH_SOCK")
if sock and Path(sock).is_socket():
return sock
# Try to find a forwarded SSH agent socket
agent_dir = Path.home() / ".ssh" / "agent"
if agent_dir.is_dir():
sockets = sorted(
agent_dir.glob("s.*.sshd.*"), key=lambda p: p.stat().st_mtime, reverse=True
)
for s in sockets:
if s.is_socket():
return str(s)
return None
def get_ssh_env() -> dict[str, str]:
"""Get environment dict for SSH subprocess with auto-detected agent.
Returns a copy of the current environment with SSH_AUTH_SOCK set
to the auto-detected agent socket (if found).
"""
env = os.environ.copy()
sock = get_ssh_auth_sock()
if sock:
env["SSH_AUTH_SOCK"] = sock
return env
def key_exists() -> bool:
"""Check if the compose-farm SSH key pair exists."""
return SSH_KEY_PATH.exists() and SSH_PUBKEY_PATH.exists()
def get_key_path() -> Path | None:
"""Get the SSH key path if it exists, None otherwise."""
return SSH_KEY_PATH if key_exists() else None
def get_pubkey_content() -> str | None:
"""Get the public key content if it exists, None otherwise."""
if not SSH_PUBKEY_PATH.exists():
return None
return SSH_PUBKEY_PATH.read_text().strip()

24
src/compose_farm/web/streaming.py
View File

@@ -4,9 +4,10 @@ from __future__ import annotations
import asyncio
import os
from pathlib import Path
from typing import TYPE_CHECKING, Any
from compose_farm.ssh_keys import get_ssh_auth_sock
if TYPE_CHECKING:
from compose_farm.config import Config
@@ -17,25 +18,6 @@ DIM = "\x1b[2m"
RESET = "\x1b[0m"
CRLF = "\r\n"
def _get_ssh_auth_sock() -> str | None:
"""Get SSH_AUTH_SOCK, auto-detecting forwarded agent if needed."""
sock = os.environ.get("SSH_AUTH_SOCK")
if sock and Path(sock).is_socket():
return sock
# Try to find a forwarded SSH agent socket
agent_dir = Path.home() / ".ssh" / "agent"
if agent_dir.is_dir():
sockets = sorted(
agent_dir.glob("s.*.sshd.*"), key=lambda p: p.stat().st_mtime, reverse=True
)
for s in sockets:
if s.is_socket():
return str(s)
return None
# In-memory task registry
tasks: dict[str, dict[str, Any]] = {}
@@ -69,7 +51,7 @@ async def run_cli_streaming(
env = {"FORCE_COLOR": "1", "TERM": "xterm-256color", "COLUMNS": "120"}
# Ensure SSH agent is available (auto-detect if needed)
ssh_sock = _get_ssh_auth_sock()
ssh_sock = get_ssh_auth_sock()
if ssh_sock:
env["SSH_AUTH_SOCK"] = ssh_sock

7
src/compose_farm/web/ws.py
View File

@@ -18,7 +18,7 @@ from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from compose_farm.executor import is_local, ssh_connect_kwargs
from compose_farm.web.deps import get_config
from compose_farm.web.streaming import CRLF, DIM, GREEN, RED, RESET, _get_ssh_auth_sock, tasks
from compose_farm.web.streaming import CRLF, DIM, GREEN, RED, RESET, tasks
if TYPE_CHECKING:
from compose_farm.config import Host
@@ -155,13 +155,10 @@ async def _run_remote_exec(
websocket: WebSocket, host: Host, exec_cmd: str, *, agent_forwarding: bool = False
) -> None:
"""Run docker exec on remote host via SSH with PTY."""
# Get SSH agent socket for authentication
agent_path = _get_ssh_auth_sock()
# ssh_connect_kwargs includes agent_path and client_keys fallback
async with asyncssh.connect(
**ssh_connect_kwargs(host),
agent_forwarding=agent_forwarding,
agent_path=agent_path,
) as conn:
proc: asyncssh.SSHClientProcess[Any] = await conn.create_process(
exec_cmd,

114
tests/test_cli_ssh.py Normal file
View File

@@ -0,0 +1,114 @@
"""Tests for CLI ssh commands."""
from pathlib import Path
from unittest.mock import patch
from typer.testing import CliRunner
from compose_farm.cli.app import app
runner = CliRunner()
class TestSshKeygen:
"""Tests for cf ssh keygen command."""
def test_keygen_generates_key(self, tmp_path: Path) -> None:
"""Generate SSH key when none exists."""
key_path = tmp_path / "compose-farm"
pubkey_path = tmp_path / "compose-farm.pub"
with (
patch("compose_farm.cli.ssh.SSH_KEY_PATH", key_path),
patch("compose_farm.cli.ssh.SSH_PUBKEY_PATH", pubkey_path),
patch("compose_farm.cli.ssh.key_exists", return_value=False),
):
result = runner.invoke(app, ["ssh", "keygen"])
# Command runs (may fail if ssh-keygen not available in test env)
assert result.exit_code in (0, 1)
def test_keygen_skips_if_exists(self, tmp_path: Path) -> None:
"""Skip key generation if key already exists."""
key_path = tmp_path / "compose-farm"
pubkey_path = tmp_path / "compose-farm.pub"
with (
patch("compose_farm.cli.ssh.SSH_KEY_PATH", key_path),
patch("compose_farm.cli.ssh.SSH_PUBKEY_PATH", pubkey_path),
patch("compose_farm.cli.ssh.key_exists", return_value=True),
):
result = runner.invoke(app, ["ssh", "keygen"])
assert "already exists" in result.output
class TestSshStatus:
"""Tests for cf ssh status command."""
def test_status_shows_no_key(self, tmp_path: Path) -> None:
"""Show message when no key exists."""
config_file = tmp_path / "compose-farm.yaml"
config_file.write_text("""
hosts:
local:
address: localhost
services:
test: local
""")
with patch("compose_farm.cli.ssh.key_exists", return_value=False):
result = runner.invoke(app, ["ssh", "status", f"--config={config_file}"])
assert "No key found" in result.output
def test_status_shows_key_exists(self, tmp_path: Path) -> None:
"""Show key info when key exists."""
config_file = tmp_path / "compose-farm.yaml"
config_file.write_text("""
hosts:
local:
address: localhost
services:
test: local
""")
with (
patch("compose_farm.cli.ssh.key_exists", return_value=True),
patch("compose_farm.cli.ssh.get_pubkey_content", return_value="ssh-ed25519 AAAA..."),
):
result = runner.invoke(app, ["ssh", "status", f"--config={config_file}"])
assert "Key exists" in result.output
class TestSshSetup:
"""Tests for cf ssh setup command."""
def test_setup_no_remote_hosts(self, tmp_path: Path) -> None:
"""Show message when no remote hosts configured."""
config_file = tmp_path / "compose-farm.yaml"
config_file.write_text("""
hosts:
local:
address: localhost
services:
test: local
""")
result = runner.invoke(app, ["ssh", "setup", f"--config={config_file}"])
assert "No remote hosts" in result.output
class TestSshHelp:
"""Tests for cf ssh help."""
def test_ssh_help(self) -> None:
"""Show help for ssh command."""
result = runner.invoke(app, ["ssh", "--help"])
assert result.exit_code == 0
assert "setup" in result.output
assert "status" in result.output
assert "keygen" in result.output

245
tests/test_ssh_keys.py Normal file
View File

@@ -0,0 +1,245 @@
"""Tests for ssh_keys module."""
import os
from pathlib import Path
from unittest.mock import MagicMock, patch
from compose_farm.config import Host
from compose_farm.executor import ssh_connect_kwargs
from compose_farm.ssh_keys import (
SSH_KEY_PATH,
get_key_path,
get_pubkey_content,
get_ssh_auth_sock,
get_ssh_env,
key_exists,
)
class TestGetSshAuthSock:
"""Tests for get_ssh_auth_sock function."""
def test_returns_env_var_when_socket_exists(self) -> None:
"""Return SSH_AUTH_SOCK env var if the socket exists."""
mock_path = MagicMock()
mock_path.is_socket.return_value = True
with (
patch.dict(os.environ, {"SSH_AUTH_SOCK": "/tmp/agent.sock"}),
patch("compose_farm.ssh_keys.Path", return_value=mock_path),
):
result = get_ssh_auth_sock()
assert result == "/tmp/agent.sock"
def test_returns_none_when_env_var_not_socket(self, tmp_path: Path) -> None:
"""Return None if SSH_AUTH_SOCK points to non-socket."""
regular_file = tmp_path / "not_a_socket"
regular_file.touch()
with (
patch.dict(os.environ, {"SSH_AUTH_SOCK": str(regular_file)}),
patch("compose_farm.ssh_keys.Path.home", return_value=tmp_path),
):
# Should fall through to agent dir check, which won't exist
result = get_ssh_auth_sock()
assert result is None
def test_finds_agent_in_ssh_agent_dir(self, tmp_path: Path) -> None:
"""Find agent socket in ~/.ssh/agent/ directory."""
# Create agent directory structure with a regular file
agent_dir = tmp_path / ".ssh" / "agent"
agent_dir.mkdir(parents=True)
sock_path = agent_dir / "s.12345.sshd.67890"
sock_path.touch() # Create as regular file
with (
patch.dict(os.environ, {}, clear=False),
patch("compose_farm.ssh_keys.Path.home", return_value=tmp_path),
patch.object(Path, "is_socket", return_value=True),
):
os.environ.pop("SSH_AUTH_SOCK", None)
result = get_ssh_auth_sock()
assert result == str(sock_path)
def test_returns_none_when_no_agent_found(self, tmp_path: Path) -> None:
"""Return None when no SSH agent socket is found."""
with (
patch.dict(os.environ, {}, clear=False),
patch("compose_farm.ssh_keys.Path.home", return_value=tmp_path),
):
os.environ.pop("SSH_AUTH_SOCK", None)
result = get_ssh_auth_sock()
assert result is None
class TestGetSshEnv:
"""Tests for get_ssh_env function."""
def test_returns_env_with_ssh_auth_sock(self) -> None:
"""Return env dict with SSH_AUTH_SOCK set."""
with patch("compose_farm.ssh_keys.get_ssh_auth_sock", return_value="/tmp/agent.sock"):
result = get_ssh_env()
assert result["SSH_AUTH_SOCK"] == "/tmp/agent.sock"
# Should include other env vars too
assert "PATH" in result or len(result) > 1
def test_returns_env_without_ssh_auth_sock_when_none(self, tmp_path: Path) -> None:
"""Return env without SSH_AUTH_SOCK when no agent found."""
with (
patch.dict(os.environ, {}, clear=False),
patch("compose_farm.ssh_keys.Path.home", return_value=tmp_path),
):
os.environ.pop("SSH_AUTH_SOCK", None)
result = get_ssh_env()
# SSH_AUTH_SOCK should not be set if no agent found
assert result.get("SSH_AUTH_SOCK") is None
class TestKeyExists:
"""Tests for key_exists function."""
def test_returns_true_when_both_keys_exist(self, tmp_path: Path) -> None:
"""Return True when both private and public keys exist."""
key_path = tmp_path / "compose-farm"
pubkey_path = tmp_path / "compose-farm.pub"
key_path.touch()
pubkey_path.touch()
with (
patch("compose_farm.ssh_keys.SSH_KEY_PATH", key_path),
patch("compose_farm.ssh_keys.SSH_PUBKEY_PATH", pubkey_path),
):
assert key_exists() is True
def test_returns_false_when_private_key_missing(self, tmp_path: Path) -> None:
"""Return False when private key doesn't exist."""
key_path = tmp_path / "compose-farm"
pubkey_path = tmp_path / "compose-farm.pub"
pubkey_path.touch() # Only public key exists
with (
patch("compose_farm.ssh_keys.SSH_KEY_PATH", key_path),
patch("compose_farm.ssh_keys.SSH_PUBKEY_PATH", pubkey_path),
):
assert key_exists() is False
def test_returns_false_when_public_key_missing(self, tmp_path: Path) -> None:
"""Return False when public key doesn't exist."""
key_path = tmp_path / "compose-farm"
pubkey_path = tmp_path / "compose-farm.pub"
key_path.touch() # Only private key exists
with (
patch("compose_farm.ssh_keys.SSH_KEY_PATH", key_path),
patch("compose_farm.ssh_keys.SSH_PUBKEY_PATH", pubkey_path),
):
assert key_exists() is False
class TestGetKeyPath:
"""Tests for get_key_path function."""
def test_returns_path_when_key_exists(self) -> None:
"""Return key path when key exists."""
with patch("compose_farm.ssh_keys.key_exists", return_value=True):
result = get_key_path()
assert result == SSH_KEY_PATH
def test_returns_none_when_key_missing(self) -> None:
"""Return None when key doesn't exist."""
with patch("compose_farm.ssh_keys.key_exists", return_value=False):
result = get_key_path()
assert result is None
class TestGetPubkeyContent:
"""Tests for get_pubkey_content function."""
def test_returns_content_when_exists(self, tmp_path: Path) -> None:
"""Return public key content when file exists."""
pubkey_content = "ssh-ed25519 AAAA... compose-farm"
pubkey_path = tmp_path / "compose-farm.pub"
pubkey_path.write_text(pubkey_content + "\n")
with patch("compose_farm.ssh_keys.SSH_PUBKEY_PATH", pubkey_path):
result = get_pubkey_content()
assert result == pubkey_content
def test_returns_none_when_missing(self, tmp_path: Path) -> None:
"""Return None when public key doesn't exist."""
pubkey_path = tmp_path / "compose-farm.pub" # Doesn't exist
with patch("compose_farm.ssh_keys.SSH_PUBKEY_PATH", pubkey_path):
result = get_pubkey_content()
assert result is None
class TestSshConnectKwargs:
"""Tests for ssh_connect_kwargs function."""
def test_basic_kwargs(self) -> None:
"""Return basic connection kwargs."""
host = Host(address="example.com", port=22, user="testuser")
with (
patch("compose_farm.executor.get_ssh_auth_sock", return_value=None),
patch("compose_farm.executor.get_key_path", return_value=None),
):
result = ssh_connect_kwargs(host)
assert result["host"] == "example.com"
assert result["port"] == 22
assert result["username"] == "testuser"
assert result["known_hosts"] is None
assert "agent_path" not in result
assert "client_keys" not in result
def test_includes_agent_path_when_available(self) -> None:
"""Include agent_path when SSH agent is available."""
host = Host(address="example.com")
with (
patch("compose_farm.executor.get_ssh_auth_sock", return_value="/tmp/agent.sock"),
patch("compose_farm.executor.get_key_path", return_value=None),
):
result = ssh_connect_kwargs(host)
assert result["agent_path"] == "/tmp/agent.sock"
def test_includes_client_keys_when_key_exists(self, tmp_path: Path) -> None:
"""Include client_keys when compose-farm key exists."""
host = Host(address="example.com")
key_path = tmp_path / "compose-farm"
with (
patch("compose_farm.executor.get_ssh_auth_sock", return_value=None),
patch("compose_farm.executor.get_key_path", return_value=key_path),
):
result = ssh_connect_kwargs(host)
assert result["client_keys"] == [str(key_path)]
def test_includes_both_agent_and_key(self, tmp_path: Path) -> None:
"""Include both agent_path and client_keys when both available."""
host = Host(address="example.com")
key_path = tmp_path / "compose-farm"
with (
patch("compose_farm.executor.get_ssh_auth_sock", return_value="/tmp/agent.sock"),
patch("compose_farm.executor.get_key_path", return_value=key_path),
):
result = ssh_connect_kwargs(host)
assert result["agent_path"] == "/tmp/agent.sock"
assert result["client_keys"] == [str(key_path)]
def test_custom_port(self) -> None:
"""Handle custom SSH port."""
host = Host(address="example.com", port=2222)
with (
patch("compose_farm.executor.get_ssh_auth_sock", return_value=None),
patch("compose_farm.executor.get_key_path", return_value=None),
):
result = ssh_connect_kwargs(host)
assert result["port"] == 2222