recipe-maintainer/lib/ssh.py

"""SSH command runner for direct server operations.

Used for docker service ls, reading secrets from containers,
checking service logs, and other operations that require
server-side access bypassing abra.
"""

from lib.abra import run, AbraResult


def ssh_run(server: str, command: str, *,
            timeout: int = 30) -> AbraResult:
    """Run a command on a remote server via SSH."""
    return run(f'ssh {server} "{command}"', check=False, timeout=timeout)


def docker_service_ls(server: str, stack_prefix: str) -> list[dict]:
    """List Docker services for a stack, parsed into dicts.

    Returns list of {"name": str, "replicas": str, "image": str}.
    """
    result = ssh_run(
        server,
        f"docker service ls --filter 'name={stack_prefix}_'"
        f" --format '{{{{.Name}}}}|{{{{.Replicas}}}}|{{{{.Image}}}}'",
        timeout=30,
    )
    if not result.ok or not result.stdout.strip():
        return []

    services = []
    for line in result.stdout.strip().split("\n"):
        line = line.strip()
        if not line:
            continue
        parts = line.split("|")
        if len(parts) >= 3:
            services.append({
                "name": parts[0],
                "replicas": parts[1],
                "image": parts[2],
            })
    return services


def all_replicas_ready(server: str, stack_prefix: str) -> bool:
    """Check if all Docker services have their desired replica count.

    Returns True if every service shows N/N replicas (e.g. "1/1").
    Returns False if any service is not ready or no services found.
    """
    result = ssh_run(
        server,
        f"docker service ls --filter 'name={stack_prefix}_'"
        f" --format '{{{{.Replicas}}}}'",
        timeout=30,
    )
    if not result.ok:
        return False

    lines = [l.strip() for l in result.stdout.strip().split("\n") if l.strip()]
    if not lines:
        return False

    for replicas in lines:
        parts = replicas.split("/")
        if len(parts) != 2:
            return False
        if parts[0] != parts[1]:
            return False
    return True


def read_container_secrets(server: str, stack_prefix: str) -> dict[str, str]:
    """Read all Docker secrets from running containers of a stack.

    SSHes into the server, finds all containers for the stack,
    reads /run/secrets/ from each container, returns {name: value}.
    """
    cmd = (
        f"containers=\\$(docker ps -q --filter 'name={stack_prefix}_' 2>/dev/null); "
        f"if [ -z \\\"\\$containers\\\" ]; then exit 0; fi; "
        f"for cid in \\$containers; do "
        f"  files=\\$(docker exec \\$cid ls /run/secrets/ 2>/dev/null) || continue; "
        f"  for f in \\$files; do "
        f"    val=\\$(docker exec \\$cid cat /run/secrets/\\$f 2>/dev/null) || continue; "
        f"    echo \\\"\\$f=\\$val\\\"; "
        f"  done; "
        f"done"
    )
    result = ssh_run(server, cmd, timeout=60)
    if not result.ok or not result.stdout.strip():
        return {}

    secrets = {}
    for line in result.stdout.strip().split("\n"):
        line = line.strip()
        if "=" in line:
            name, value = line.split("=", 1)
            if name not in secrets:  # first occurrence wins
                secrets[name] = value
    return secrets


def docker_service_logs(server: str, stack_name: str, service: str, *,
                        since: str = "5m", tail: int = 100) -> str:
    """Get Docker service logs via SSH (avoids abra app logs which hangs)."""
    result = ssh_run(
        server,
        f"docker service logs {stack_name}_{service} --since {since} --tail {tail} 2>&1",
        timeout=30,
    )
    return result.stdout if result.ok else ""


def docker_secret_ls(server: str, stack_prefix: str) -> list[str]:
    """List Docker secret names matching a stack prefix."""
    result = ssh_run(
        server,
        f"docker secret ls --filter 'name={stack_prefix}_' --format '{{{{.Name}}}}'",
        timeout=15,
    )
    if not result.ok or not result.stdout.strip():
        return []
    return [l.strip() for l in result.stdout.strip().split("\n") if l.strip()]


def docker_volume_ls(server: str, stack_prefix: str) -> list[str]:
    """List Docker volume names for a stack.

    Returns a list of volume names. Raises RuntimeError if the SSH
    command fails (a failed command is a real error, not "no volumes").
    """
    result = ssh_run(
        server,
        f"docker volume ls --filter 'name={stack_prefix}_' --format '{{{{.Name}}}}'",
        timeout=15,
    )
    if not result.ok:
        raise RuntimeError(
            f"Failed to list volumes on {server}: {result.stderr.strip()}"
        )
    if not result.stdout.strip():
        return []
    return [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]


def docker_ps(server: str, *, filter_name: str | None = None) -> list[dict]:
    """List Docker containers on the server."""
    cmd = "docker ps --format '{{.Names}}|{{.Status}}|{{.Image}}'"
    if filter_name:
        cmd = f"docker ps --filter 'name={filter_name}' --format '{{{{.Names}}}}|{{{{.Status}}}}|{{{{.Image}}}}'"
    result = ssh_run(server, cmd, timeout=30)
    if not result.ok or not result.stdout.strip():
        return []

    containers = []
    for line in result.stdout.strip().split("\n"):
        parts = line.strip().split("|")
        if len(parts) >= 3:
            containers.append({
                "name": parts[0],
                "status": parts[1],
                "image": parts[2],
            })
    return containers