diff --git a/pkgs/clan-cli/clan_lib/ssh/host.py b/pkgs/clan-cli/clan_lib/ssh/host.py new file mode 100644 index 000000000..4c804de8f --- /dev/null +++ b/pkgs/clan-cli/clan_lib/ssh/host.py @@ -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. + """ diff --git a/pkgs/clan-cli/clan_lib/ssh/localhost.py b/pkgs/clan-cli/clan_lib/ssh/localhost.py new file mode 100644 index 000000000..ca68b2d73 --- /dev/null +++ b/pkgs/clan-cli/clan_lib/ssh/localhost.py @@ -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 diff --git a/test_host_interface.py b/test_host_interface.py new file mode 100644 index 000000000..aa62e943a --- /dev/null +++ b/test_host_interface.py @@ -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()