ssh: Introduce LocalHost vs. Remote via Host interface

Motivation: local builds and deployments without ssh

Add a new interface `Host` which is implemented bei either `Remote` or `Localhost`

This simplifies all interactions with hosts. THe caller does ot need to know if the Host is remote or local in mot cases anymore
This commit is contained in:
DavHau
2025-07-29 16:38:55 +07:00
committed by Jörg Thalheim
parent 98d5b3651b
commit c33fd4e504
3 changed files with 232 additions and 0 deletions

View File

@@ -0,0 +1,85 @@
"""Base Host interface for both local and remote command execution."""
import logging
from abc import ABC, abstractmethod
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass
from clan_lib.cmd import CmdOut, RunOpts
cmdlog = logging.getLogger(__name__)
@dataclass(frozen=True)
class Host(ABC):
"""
Abstract base class for host command execution.
This provides a common interface for both local and remote hosts.
"""
command_prefix: str
@property
@abstractmethod
def target(self) -> str:
"""Return a descriptive target string for this host."""
@property
@abstractmethod
def user(self) -> str:
"""Return the user for this host."""
@abstractmethod
def run(
self,
cmd: list[str],
opts: RunOpts | None = None,
extra_env: dict[str, str] | None = None,
tty: bool = False,
verbose_ssh: bool = False,
quiet: bool = False,
control_master: bool = True,
) -> CmdOut:
"""
Run a command on the host.
Args:
cmd: Command to execute
opts: Run options
extra_env: Additional environment variables
tty: Whether to allocate a TTY (for remote hosts)
verbose_ssh: Enable verbose SSH output (for remote hosts)
quiet: Suppress command logging
control_master: Use SSH ControlMaster (for remote hosts)
Returns:
Command output
"""
@contextmanager
@abstractmethod
def become_root(self) -> Iterator["Host"]:
"""
Context manager to execute commands as root.
"""
@contextmanager
@abstractmethod
def host_connection(self) -> Iterator["Host"]:
"""
Context manager to manage host connections.
For remote hosts, this manages SSH ControlMaster connections.
For local hosts, this is a no-op that returns self.
"""
@abstractmethod
def nix_ssh_env(
self,
env: dict[str, str] | None = None,
control_master: bool = True,
) -> dict[str, str]:
"""
Get environment variables for Nix operations.
Remote hosts will add NIX_SSHOPTS, local hosts won't.
"""

View File

@@ -0,0 +1,111 @@
import logging
import os
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from clan_lib.cmd import CmdOut, RunOpts, run
from clan_lib.colors import AnsiColor
from clan_lib.ssh.host import Host
cmdlog = logging.getLogger(__name__)
@dataclass(frozen=True)
class LocalHost(Host):
"""
A Host implementation that executes commands locally without SSH.
"""
command_prefix: str = "localhost"
_user: str = field(default_factory=lambda: os.environ.get("USER", "root"))
_askpass_path: str | None = None
@property
def target(self) -> str:
"""Return a descriptive target string for localhost."""
return "localhost"
@property
def user(self) -> str:
"""Return the user for localhost."""
return self._user
def run(
self,
cmd: list[str],
opts: RunOpts | None = None,
extra_env: dict[str, str] | None = None,
tty: bool = False,
verbose_ssh: bool = False,
quiet: bool = False,
control_master: bool = True,
) -> CmdOut:
"""
Run a command locally instead of via SSH.
"""
if opts is None:
opts = RunOpts()
# Set up environment
env = opts.env or os.environ.copy()
if extra_env:
env.update(extra_env)
# Handle sudo if needed
if self._askpass_path is not None:
# Prepend sudo command
sudo_cmd = ["sudo", "-A", "--"]
cmd = sudo_cmd + cmd
env["SUDO_ASKPASS"] = self._askpass_path
# Set options
opts.env = env
opts.prefix = opts.prefix or self.command_prefix
# Log the command
displayed_cmd = " ".join(cmd)
if not quiet:
cmdlog.info(
f"$ {displayed_cmd}",
extra={
"command_prefix": self.command_prefix,
"color": AnsiColor.GREEN.value,
},
)
# Run locally
return run(cmd, opts)
@contextmanager
def become_root(self) -> Iterator["LocalHost"]:
"""
Context manager to execute commands as root.
"""
if self._user == "root":
yield self
return
# For local execution, we can use sudo with askpass if GUI is available
# This is a simplified version - could be enhanced with sudo askpass proxy
yield self
@contextmanager
def host_connection(self) -> Iterator["LocalHost"]:
"""
For LocalHost, this is a no-op that just returns self.
"""
yield self
def nix_ssh_env(
self,
env: dict[str, str] | None = None,
control_master: bool = True,
) -> dict[str, str]:
"""
LocalHost doesn't need SSH environment variables.
"""
if env is None:
env = {}
# Don't set NIX_SSHOPTS for localhost
return env

36
test_host_interface.py Normal file
View File

@@ -0,0 +1,36 @@
#!/usr/bin/env python3
"""Test script for Host interface with LocalHost implementation."""
from clan_lib.cmd import RunOpts
from clan_lib.ssh.host import Host
from clan_lib.ssh.localhost import LocalHost
def test_localhost() -> None:
# Create LocalHost instance
localhost = LocalHost(command_prefix="local-test")
# Verify it's a Host instance
assert isinstance(localhost, Host), "LocalHost should be an instance of Host"
# Test basic command execution
result = localhost.run(["echo", "Hello from LocalHost"])
assert result.returncode == 0, f"Command failed with code {result.returncode}"
assert result.stdout.strip() == "Hello from LocalHost", (
f"Unexpected output: {result.stdout}"
)
# Test with environment variable
result = localhost.run(
["printenv", "TEST_VAR"],
opts=RunOpts(check=False), # Don't check return code
extra_env={"TEST_VAR": "LocalHost works!"},
)
assert result.returncode == 0, f"Command failed with code {result.returncode}"
assert result.stdout.strip() == "LocalHost works!", (
f"Expected 'LocalHost works!', got '{result.stdout.strip()}'"
)
if __name__ == "__main__":
test_localhost()