101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
"""Managed known_hosts synthesis for SSH clients.
|
|
|
|
Instead of mounting a host-side ``known_hosts`` file, the backend can discover
|
|
and persist host keys for configured SSH machines inside its own cache volume.
|
|
This keeps strict host-key checking enabled without exposing a whole SSH
|
|
configuration directory into the container.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import socket
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import paramiko
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _host_alias(host: str, port: int) -> str:
|
|
return host if int(port or 22) == 22 else f"[{host}]:{int(port or 22)}"
|
|
|
|
|
|
def _fetch_server_key(host: str, port: int, timeout: int = 30) -> paramiko.PKey:
|
|
sock = socket.create_connection((host, int(port or 22)), timeout=timeout)
|
|
transport = paramiko.Transport(sock)
|
|
try:
|
|
transport.start_client(timeout=timeout)
|
|
key = transport.get_remote_server_key()
|
|
if key is None:
|
|
raise RuntimeError(f"Unable to read SSH host key for {host}:{port}")
|
|
return key
|
|
except Exception as exc:
|
|
raise RuntimeError(
|
|
f"Unable to read SSH protocol banner from {host}:{port}. "
|
|
"Confirm the host is running an SSH server on that port and is reachable from the backend."
|
|
) from exc
|
|
finally:
|
|
transport.close()
|
|
sock.close()
|
|
|
|
|
|
def has_known_host(host: str, port: int, known_hosts_path: Path) -> bool:
|
|
"""Return whether the given host/port is already present in known_hosts."""
|
|
if not host or not known_hosts_path.exists():
|
|
return False
|
|
host_alias = _host_alias(host, port)
|
|
host_keys = paramiko.HostKeys()
|
|
host_keys.load(str(known_hosts_path))
|
|
return host_keys.lookup(host_alias) is not None
|
|
|
|
|
|
def ensure_known_host(host: str, port: int, known_hosts_path: Path, *, strict: bool = True) -> bool:
|
|
"""Ensure a host key entry exists for the given host/port.
|
|
|
|
Returns ``True`` when the file was changed. If ``strict`` is enabled and the
|
|
existing key differs, a ``RuntimeError`` is raised instead of silently
|
|
overwriting the entry.
|
|
"""
|
|
if not host:
|
|
return False
|
|
known_hosts_path.parent.mkdir(parents=True, exist_ok=True)
|
|
host_key = _fetch_server_key(host, port)
|
|
host_alias = _host_alias(host, port)
|
|
host_keys = paramiko.HostKeys()
|
|
if known_hosts_path.exists():
|
|
host_keys.load(str(known_hosts_path))
|
|
|
|
existing = host_keys.lookup(host_alias)
|
|
key_type = host_key.get_name()
|
|
if existing and key_type in existing:
|
|
if existing[key_type].get_base64() == host_key.get_base64():
|
|
return False
|
|
if strict:
|
|
raise RuntimeError(f"SSH host key mismatch for {host_alias}")
|
|
|
|
host_keys.add(host_alias, key_type, host_key)
|
|
host_keys.save(str(known_hosts_path))
|
|
logger.info("Recorded SSH host key host=%s port=%s file=%s", host, port, known_hosts_path)
|
|
return True
|
|
|
|
|
|
def ensure_known_hosts_for_machines(
|
|
machines: list[dict[str, Any]],
|
|
known_hosts_path: Path,
|
|
*,
|
|
strict: bool = True,
|
|
) -> int:
|
|
changed = 0
|
|
for machine in machines:
|
|
if str(machine.get("mode") or "").lower() != "ssh":
|
|
continue
|
|
host = str(machine.get("host") or "").strip()
|
|
port = int(machine.get("port") or 22)
|
|
if not host:
|
|
continue
|
|
if ensure_known_host(host, port, known_hosts_path, strict=strict):
|
|
changed += 1
|
|
return changed
|