remote_cmd

Remote CMD - SSH 远程服务器管理工具

一个功能强大的 Python 库，用于管理远程服务器。提供简洁的 API 用于 SSH 连接、命令执行和文件传输。

主要功能： - SSH 连接管理（支持密码和密钥认证） - 远程命令执行（包括 sudo 命令） - 文件上传和下载 - 主机配置管理 - 标签分类系统 - 批量连接测试 - 凭据加密存储 - 结构化日志系统

快速开始：

from remote_cmd import SSHClient from remote_cmd.core.ssh_client import ConnectionConfig

创建连接配置

config = ConnectionConfig( ... hostname="192.168.1.100", ... username="admin", ... key_filename="~/.ssh/id_rsa" ... )

执行远程命令

with SSHClient(config) as client: ... result = client.execute("ls -la") ... print(result.stdout)

新架构（推荐）:

from remote_cmd.repository.json_host_repository import JsonHostRepository from remote_cmd.service.host_service import HostService

repo = JsonHostRepository("hosts.json") service = HostService(repo) service.add_host(host)

命令行使用： $ remote-cmd host add server1 192.168.1.100 admin -k ~/.ssh/id_rsa $ remote-cmd host list $ remote-cmd run server1 "uptime"

更多信息： - GitHub: https://github.com/Vae-Scrooge/remote-cmd - 文档: 参见 docs/ 目录

Author: Vae-Scrooge Version: 2.1.0（单一真相源见 remote_cmd._version） License: MIT

View Source

  1"""
  2Remote CMD - SSH 远程服务器管理工具
  3
  4一个功能强大的 Python 库，用于管理远程服务器。
  5提供简洁的 API 用于 SSH 连接、命令执行和文件传输。
  6
  7主要功能：
  8    - SSH 连接管理（支持密码和密钥认证）
  9    - 远程命令执行（包括 sudo 命令）
 10    - 文件上传和下载
 11    - 主机配置管理
 12    - 标签分类系统
 13    - 批量连接测试
 14    - 凭据加密存储
 15    - 结构化日志系统
 16
 17快速开始：
 18    >>> from remote_cmd import SSHClient
 19    >>> from remote_cmd.core.ssh_client import ConnectionConfig
 20    >>>
 21    >>> # 创建连接配置
 22    >>> config = ConnectionConfig(
 23    ...     hostname="192.168.1.100",
 24    ...     username="admin",
 25    ...     key_filename="~/.ssh/id_rsa"
 26    ... )
 27    >>>
 28    >>> # 执行远程命令
 29    >>> with SSHClient(config) as client:
 30    ...     result = client.execute("ls -la")
 31    ...     print(result.stdout)
 32
 33新架构（推荐）:
 34    >>> from remote_cmd.repository.json_host_repository import JsonHostRepository
 35    >>> from remote_cmd.service.host_service import HostService
 36    >>>
 37    >>> repo = JsonHostRepository("hosts.json")
 38    >>> service = HostService(repo)
 39    >>> service.add_host(host)
 40
 41命令行使用：
 42    $ remote-cmd host add server1 192.168.1.100 admin -k ~/.ssh/id_rsa
 43    $ remote-cmd host list
 44    $ remote-cmd run server1 "uptime"
 45
 46更多信息：
 47    - GitHub: https://github.com/Vae-Scrooge/remote-cmd
 48    - 文档: 参见 docs/ 目录
 49
 50Author: Vae-Scrooge
 51Version: 2.1.0（单一真相源见 remote_cmd._version）
 52License: MIT
 53"""
 54
 55# 版本号单一真相源（见 remote_cmd._version）
 56from remote_cmd._version import __version__
 57
 58__author__ = "Vae-Scrooge"
 59__email__ = "vae-scrooge@example.com"
 60__license__ = "MIT"
 61
 62import logging
 63
 64logging.getLogger(__name__).addHandler(logging.NullHandler())
 65
 66# 异步原生模块依赖 asyncssh（optional extra [async]），未安装时优雅降级，
 67# 避免 `import remote_cmd` 直接失败
 68try:
 69    from remote_cmd.core.async_connection_pool import AsyncConnectionPool
 70    from remote_cmd.core.async_ssh_client import AsyncSSHClient
 71    from remote_cmd.service.async_batch_executor import AsyncBatchExecutor
 72
 73    _HAS_ASYNC = True
 74except ImportError:  # pragma: no cover - 依赖 asyncssh，未安装时不导出异步符号
 75    _HAS_ASYNC = False
 76
 77from remote_cmd.core.host import Host
 78from remote_cmd.core.host_manager import HostManager
 79from remote_cmd.core.ssh_client import SSHClient
 80from remote_cmd.core.sync_connection_pool import SyncConnectionPool
 81
 82if _HAS_ASYNC:
 83    # 兼容别名（历史名称）
 84    NativeAsyncSSHClient = AsyncSSHClient
 85    ConnectionPool = AsyncConnectionPool
 86
 87# 新架构导出（推荐）
 88from remote_cmd.repository import HostRepository, JsonHostRepository
 89
 90# Phase 2 新组件
 91from remote_cmd.repository.sqlite_host_repository import SqliteHostRepository
 92from remote_cmd.service import (
 93    ChainCredentialProvider,
 94    CredentialProvider,
 95    EnvCredentialProvider,
 96    HostService,
 97    SSHService,
 98)
 99from remote_cmd.service.batch_executor import BatchExecutor, BatchHostResult, BatchResult
100from remote_cmd.service.credential_provider import KeyringCredentialProvider
101from remote_cmd.service.task_runner import Task, TaskRunner, TaskStatus
102from remote_cmd.utils.crypto import CredentialEncryption
103from remote_cmd.utils.logging_utils import (
104    SensitiveDataFilter,
105    get_logger,
106    setup_logging,
107)
108
109if _HAS_ASYNC:
110    __all__ = [
111        # 原有导出（向后兼容）
112        "SSHClient",
113        "AsyncSSHClient",
114        "ConnectionPool",
115        "NativeAsyncSSHClient",
116        "AsyncConnectionPool",
117        "AsyncBatchExecutor",
118        "Host",
119        "HostManager",
120        # 新架构导出
121        "HostRepository",
122        "JsonHostRepository",
123        "HostService",
124        "SSHService",
125        "CredentialProvider",
126        "EnvCredentialProvider",
127        "ChainCredentialProvider",
128        "CredentialEncryption",
129        "setup_logging",
130        "SensitiveDataFilter",
131        "get_logger",
132        # Phase 2 新组件
133        "SqliteHostRepository",
134        "BatchExecutor",
135        "BatchResult",
136        "BatchHostResult",
137        "SyncConnectionPool",
138        "TaskRunner",
139        "Task",
140        "TaskStatus",
141        "KeyringCredentialProvider",
142        # 元信息
143        "__version__",
144        "__author__",
145        "__license__",
146    ]
147else:
148    __all__ = [
149        # 原有导出（向后兼容，不含异步符号）
150        "SSHClient",
151        "Host",
152        "HostManager",
153        # 新架构导出
154        "HostRepository",
155        "JsonHostRepository",
156        "HostService",
157        "SSHService",
158        "CredentialProvider",
159        "EnvCredentialProvider",
160        "ChainCredentialProvider",
161        "CredentialEncryption",
162        "setup_logging",
163        "SensitiveDataFilter",
164        "get_logger",
165        # Phase 2 新组件
166        "SqliteHostRepository",
167        "BatchExecutor",
168        "BatchResult",
169        "BatchHostResult",
170        "SyncConnectionPool",
171        "TaskRunner",
172        "Task",
173        "TaskStatus",
174        "KeyringCredentialProvider",
175        # 元信息
176        "__version__",
177        "__author__",
178        "__license__",
179    ]

ConnectionPool = <class 'AsyncConnectionPool'>

NativeAsyncSSHClient = <class 'AsyncSSHClient'>

class AsyncConnectionPool: View Source

 34class AsyncConnectionPool:
 35    """原生异步 SSH 连接池。
 36
 37    Args:
 38        config: 用于建立 SSH 连接的配置
 39        max_connections: 同一最大连接数（同一配置可复用）
 40        max_lifetime: 连接最大生命周期（秒），超过自动关闭
 41        idle_timeout: 空闲超时（秒），超过自动关闭
 42        health_check_interval: 后台清理任务周期（秒）
 43    """
 44
 45    def __init__(
 46        self,
 47        config: ConnectionConfig,
 48        max_connections: int = 10,
 49        max_lifetime: int = 3600,
 50        idle_timeout: int = 300,
 51        health_check_interval: int = 60,
 52        client_factory: Optional[Any] = None,
 53    ) -> None:
 54        """
 55        Args:
 56            config: 用于建立 SSH 连接的配置
 57            max_connections: 同一最大连接数（同一配置可复用）
 58            max_lifetime: 连接最大生命周期（秒），超过自动关闭
 59            idle_timeout: 空闲超时（秒），超过自动关闭
 60            health_check_interval: 后台清理任务周期（秒）
 61            client_factory: 客户端工厂，默认为 AsyncSSHClient；测试可注入
 62                mock（与 SyncConnectionPool 对齐）
 63        """
 64        self.config = config
 65        self._max = max_connections
 66        self._max_lifetime = max_lifetime
 67        self._idle_timeout = idle_timeout
 68        self._health_check_interval = health_check_interval
 69        # 客户端工厂：默认为 AsyncSSHClient；测试可注入 mock
 70        self._client_factory = client_factory or AsyncSSHClient
 71
 72        # 容器
 73        self._connections: list[AsyncSSHClient] = []
 74        self._free: asyncio.Queue[AsyncSSHClient] = asyncio.Queue()
 75        self._semaphore = asyncio.Semaphore(max_connections)
 76        self._lock = asyncio.Lock()
 77
 78        # 生命周期状态：close_all() 后置 True，禁止再借用/归还
 79        self._closed = False
 80
 81        # 指标
 82        self._total_created = 0
 83        self._total_reconnects = 0
 84        self._total_failed = 0
 85        self._total_released = 0
 86
 87        # 后台清理任务
 88        self._monitor_task: Optional[asyncio.Task[None]] = None
 89
 90        # 连接元数据（副表，避免侵入 AsyncSSHClient 私有属性）
 91        self._meta: dict[int, ConnectionMeta] = {}
 92
 93    # ------------------------------------------------------------------
 94    # 指标
 95    # ------------------------------------------------------------------
 96    def get_metrics(self) -> dict[str, Any]:
 97        """获取连接池指标快照。"""
 98        return {
 99            # 当前在用的连接数 = 存活连接总数 - 空闲连接数。
100            # 不能用 total_created - total_released：复用连接时
101            # total_released 会超过 total_created，导致 active 为负。
102            "active": len(self._connections) - self._free.qsize(),
103            "idle": self._free.qsize(),
104            "total_connections": len(self._connections),
105            "total_created": self._total_created,
106            "reconnects": self._total_reconnects,
107            "failed": self._total_failed,
108            "max_connections": self._max,
109            "max_lifetime": self._max_lifetime,
110            "idle_timeout": self._idle_timeout,
111        }
112
113    # ------------------------------------------------------------------
114    # 获取 / 释放
115    # ------------------------------------------------------------------
116    async def acquire(self) -> AsyncSSHClient:
117        """从池中获取一个可用连接，必要时创建新连接。
118
119        Returns:
120            AsyncSSHClient: 可用的异步客户端
121
122        Raises:
123            SSHConnectionError: 创建连接失败
124            RuntimeError: 连接池已关闭（close_all 之后）
125        """
126        if self._closed:
127            raise RuntimeError("connection pool is closed")
128        await self._semaphore.acquire()
129        # 竞态守卫：等待信号量期间 close_all() 可能已完成——
130        # 取得槽位后必须复查，已关闭则归还槽位并抛出既有错误，
131        # 否则会向调用方发放来自已关闭池的连接
132        if self._closed:
133            self._semaphore.release()
134            raise RuntimeError("connection pool is closed")
135        try:
136            # 优先复用空闲连接
137            while not self._free.empty():
138                conn = self._free.get_nowait()
139                if await self._check_connection(conn):
140                    self._touch(conn)
141                    return conn
142                await self._close_connection(conn)
143
144            # 创建新连接（信号量已保证未超额）
145            return await self._create_connection()
146        except BaseException:
147            self._semaphore.release()
148            raise
149
150    async def release(self, conn: Optional[AsyncSSHClient]) -> None:
151        """归还连接到池中（如已断开/超时则关闭）。"""
152        if conn is None:
153            return
154        # 池已关闭：不把连接放回空闲队列（避免游离连接），直接关闭并释放槽位
155        if self._closed:
156            await self._close_connection(conn)
157            self._semaphore.release()
158            self._total_released += 1
159            return
160        meta = self._meta.get(id(conn))
161        if meta is not None:
162            meta.last_used = time.time()
163
164        if not conn.is_connected():
165            await self._close_connection(conn)
166            self._semaphore.release()
167            return
168
169        # 生命周期 / 空闲超时则关闭
170        if meta and should_close(meta, self._max_lifetime, self._idle_timeout, True):
171            await self._close_connection(conn)
172            self._semaphore.release()
173            return
174
175        try:
176            async with self._lock:
177                self._free.put_nowait(conn)
178            # 放回 free 后释放许可：free 中的连接不再占用并发槽位，
179            # 后续 acquire 会从 free 直接复用（无需再次获取许可）
180            self._semaphore.release()
181        except asyncio.QueueFull:
182            await self._close_connection(conn)
183            self._semaphore.release()
184        finally:
185            self._total_released += 1
186
187    # ------------------------------------------------------------------
188    # 内部
189    # ------------------------------------------------------------------
190    async def _create_connection(self) -> AsyncSSHClient:
191        client = self._client_factory(self.config)
192        try:
193            await client.connect()
194        except Exception:  # noqa: BLE001
195            # 信号量由 acquire() 的 except 统一释放，此处不再释放
196            self._total_failed += 1
197            raise
198        self._connections.append(client)
199        now = time.time()
200        self._meta[id(client)] = ConnectionMeta(
201            created_at=now,
202            last_used=now,
203            conn_id=uuid.uuid4().hex,
204        )
205        self._total_created += 1
206        return client
207
208    def _touch(self, conn: AsyncSSHClient) -> None:
209        meta = self._meta.get(id(conn))
210        if meta is not None:
211            meta.last_used = time.time()
212
213    async def _check_connection(self, conn: AsyncSSHClient) -> bool:
214        if not conn.is_connected():
215            return False
216        meta = self._meta.get(id(conn))
217        if meta is None:
218            return True
219        if lifetime_expired(meta.created_at, self._max_lifetime):
220            logger.debug("connection %s exceeded max lifetime", meta.conn_id[:8])
221            return False
222        # 连接刚使用过（空闲未超时）则信任其状态，避免频繁探活开销
223        # （与 SyncConnectionPool._check_connection 保持一致）
224        if not idle_expired(meta.last_used, self._idle_timeout):
225            return True
226        # 空闲较久才触发轻量探活：发出一个无害命令
227        try:
228            result = await conn.execute("true", timeout=5)
229            return result.success
230        except Exception as e:  # noqa: BLE001
231            self._total_reconnects += 1
232            logger.debug("connection liveness check failed: %s", e)
233            return False
234
235    async def _close_connection(self, conn: AsyncSSHClient) -> None:
236        with contextlib.suppress(Exception):
237            await conn.disconnect()
238        self._meta.pop(id(conn), None)
239        if conn in self._connections:
240            self._connections.remove(conn)
241
242    # ------------------------------------------------------------------
243    # 后台监控
244    # ------------------------------------------------------------------
245    def _start_monitor(self) -> None:
246        if self._monitor_task is None or self._monitor_task.done():
247            self._monitor_task = asyncio.create_task(self._monitor_loop())
248
249    def stop_monitor(self) -> None:
250        if self._monitor_task and not self._monitor_task.done():
251            self._monitor_task.cancel()
252
253    async def _monitor_loop(self) -> None:
254        while True:
255            try:
256                await asyncio.sleep(self._health_check_interval)
257                await self._cleanup_expired()
258            except asyncio.CancelledError:
259                break
260            except Exception:  # noqa: BLE001
261                logger.warning("connection pool monitor error")
262
263    async def _cleanup_expired(self) -> None:
264        now = time.time()
265        # 在锁保护下排空 _free 快照，绝不替换队列对象。
266        # 旧实现 self._free = kept 会替换队列对象，在 await 让出点 release()
267        # 可能 put 到旧队列导致连接泄漏（asyncio 协程交错使竞态比同步版更易触发）。
268        async with self._lock:
269            snapshot: list[AsyncSSHClient] = []
270            while not self._free.empty():
271                snapshot.append(self._free.get_nowait())
272        keep: list[AsyncSSHClient] = []
273        for conn in snapshot:
274            meta = self._meta.get(id(conn))
275            if should_close(meta, self._max_lifetime, self._idle_timeout, conn.is_connected(), now):
276                await self._close_connection(conn)
277                continue
278            keep.append(conn)
279        # 将存活连接放回同一队列对象
280        async with self._lock:
281            for conn in keep:
282                self._free.put_nowait(conn)
283
284    # ------------------------------------------------------------------
285    # 上下文管理
286    # ------------------------------------------------------------------
287    class _AcquireContext:
288        def __init__(self, pool: "AsyncConnectionPool") -> None:
289            self._pool = pool
290            self._conn: Optional[AsyncSSHClient] = None
291
292        async def __aenter__(self) -> AsyncSSHClient:
293            self._conn = await self._pool.acquire()
294            return self._conn
295
296        async def __aexit__(self, exc_type, exc, tb) -> None:
297            await self._pool.release(self._conn)
298            self._conn = None
299
300    def acquire_context(self) -> "_AcquireContext":
301        """获取连接的上下文管理器。"""
302        return AsyncConnectionPool._AcquireContext(self)
303
304    async def close_all(self) -> None:
305        """关闭池中所有连接并停止监控。"""
306        self._closed = True
307        self.stop_monitor()
308        for conn in list(self._connections):
309            await self._close_connection(conn)
310        # 释放所有信号量
311        while self._free.empty() is False:
312            self._free.get_nowait()
313
314    async def __aenter__(self) -> "AsyncConnectionPool":
315        self._start_monitor()
316        return self
317
318    async def __aexit__(self, exc_type, exc, tb) -> None:
319        await self.close_all()

原生异步 SSH 连接池。

Args: config: 用于建立 SSH 连接的配置 max_connections: 同一最大连接数（同一配置可复用） max_lifetime: 连接最大生命周期（秒），超过自动关闭 idle_timeout: 空闲超时（秒），超过自动关闭 health_check_interval: 后台清理任务周期（秒）

AsyncConnectionPool( config: remote_cmd.core.ssh_client.ConnectionConfig, max_connections: int = 10, max_lifetime: int = 3600, idle_timeout: int = 300, health_check_interval: int = 60, client_factory: Any | None = None) View Source

45    def __init__(
46        self,
47        config: ConnectionConfig,
48        max_connections: int = 10,
49        max_lifetime: int = 3600,
50        idle_timeout: int = 300,
51        health_check_interval: int = 60,
52        client_factory: Optional[Any] = None,
53    ) -> None:
54        """
55        Args:
56            config: 用于建立 SSH 连接的配置
57            max_connections: 同一最大连接数（同一配置可复用）
58            max_lifetime: 连接最大生命周期（秒），超过自动关闭
59            idle_timeout: 空闲超时（秒），超过自动关闭
60            health_check_interval: 后台清理任务周期（秒）
61            client_factory: 客户端工厂，默认为 AsyncSSHClient；测试可注入
62                mock（与 SyncConnectionPool 对齐）
63        """
64        self.config = config
65        self._max = max_connections
66        self._max_lifetime = max_lifetime
67        self._idle_timeout = idle_timeout
68        self._health_check_interval = health_check_interval
69        # 客户端工厂：默认为 AsyncSSHClient；测试可注入 mock
70        self._client_factory = client_factory or AsyncSSHClient
71
72        # 容器
73        self._connections: list[AsyncSSHClient] = []
74        self._free: asyncio.Queue[AsyncSSHClient] = asyncio.Queue()
75        self._semaphore = asyncio.Semaphore(max_connections)
76        self._lock = asyncio.Lock()
77
78        # 生命周期状态：close_all() 后置 True，禁止再借用/归还
79        self._closed = False
80
81        # 指标
82        self._total_created = 0
83        self._total_reconnects = 0
84        self._total_failed = 0
85        self._total_released = 0
86
87        # 后台清理任务
88        self._monitor_task: Optional[asyncio.Task[None]] = None
89
90        # 连接元数据（副表，避免侵入 AsyncSSHClient 私有属性）
91        self._meta: dict[int, ConnectionMeta] = {}

Args: config: 用于建立 SSH 连接的配置 max_connections: 同一最大连接数（同一配置可复用） max_lifetime: 连接最大生命周期（秒），超过自动关闭 idle_timeout: 空闲超时（秒），超过自动关闭 health_check_interval: 后台清理任务周期（秒） client_factory: 客户端工厂，默认为 AsyncSSHClient；测试可注入 mock（与 SyncConnectionPool 对齐）

config

def get_metrics(self) -> dict[str, typing.Any]: View Source

 96    def get_metrics(self) -> dict[str, Any]:
 97        """获取连接池指标快照。"""
 98        return {
 99            # 当前在用的连接数 = 存活连接总数 - 空闲连接数。
100            # 不能用 total_created - total_released：复用连接时
101            # total_released 会超过 total_created，导致 active 为负。
102            "active": len(self._connections) - self._free.qsize(),
103            "idle": self._free.qsize(),
104            "total_connections": len(self._connections),
105            "total_created": self._total_created,
106            "reconnects": self._total_reconnects,
107            "failed": self._total_failed,
108            "max_connections": self._max,
109            "max_lifetime": self._max_lifetime,
110            "idle_timeout": self._idle_timeout,
111        }

获取连接池指标快照。

async def acquire(self) -> AsyncSSHClient: View Source

116    async def acquire(self) -> AsyncSSHClient:
117        """从池中获取一个可用连接，必要时创建新连接。
118
119        Returns:
120            AsyncSSHClient: 可用的异步客户端
121
122        Raises:
123            SSHConnectionError: 创建连接失败
124            RuntimeError: 连接池已关闭（close_all 之后）
125        """
126        if self._closed:
127            raise RuntimeError("connection pool is closed")
128        await self._semaphore.acquire()
129        # 竞态守卫：等待信号量期间 close_all() 可能已完成——
130        # 取得槽位后必须复查，已关闭则归还槽位并抛出既有错误，
131        # 否则会向调用方发放来自已关闭池的连接
132        if self._closed:
133            self._semaphore.release()
134            raise RuntimeError("connection pool is closed")
135        try:
136            # 优先复用空闲连接
137            while not self._free.empty():
138                conn = self._free.get_nowait()
139                if await self._check_connection(conn):
140                    self._touch(conn)
141                    return conn
142                await self._close_connection(conn)
143
144            # 创建新连接（信号量已保证未超额）
145            return await self._create_connection()
146        except BaseException:
147            self._semaphore.release()
148            raise

从池中获取一个可用连接，必要时创建新连接。

Returns: AsyncSSHClient: 可用的异步客户端

Raises: SSHConnectionError: 创建连接失败 RuntimeError: 连接池已关闭（close_all 之后）

async def release( self, conn: AsyncSSHClient | None) -> None: View Source

150    async def release(self, conn: Optional[AsyncSSHClient]) -> None:
151        """归还连接到池中（如已断开/超时则关闭）。"""
152        if conn is None:
153            return
154        # 池已关闭：不把连接放回空闲队列（避免游离连接），直接关闭并释放槽位
155        if self._closed:
156            await self._close_connection(conn)
157            self._semaphore.release()
158            self._total_released += 1
159            return
160        meta = self._meta.get(id(conn))
161        if meta is not None:
162            meta.last_used = time.time()
163
164        if not conn.is_connected():
165            await self._close_connection(conn)
166            self._semaphore.release()
167            return
168
169        # 生命周期 / 空闲超时则关闭
170        if meta and should_close(meta, self._max_lifetime, self._idle_timeout, True):
171            await self._close_connection(conn)
172            self._semaphore.release()
173            return
174
175        try:
176            async with self._lock:
177                self._free.put_nowait(conn)
178            # 放回 free 后释放许可：free 中的连接不再占用并发槽位，
179            # 后续 acquire 会从 free 直接复用（无需再次获取许可）
180            self._semaphore.release()
181        except asyncio.QueueFull:
182            await self._close_connection(conn)
183            self._semaphore.release()
184        finally:
185            self._total_released += 1

归还连接到池中（如已断开/超时则关闭）。

def stop_monitor(self) -> None: View Source

249    def stop_monitor(self) -> None:
250        if self._monitor_task and not self._monitor_task.done():
251            self._monitor_task.cancel()

def acquire_context(self) -> '_AcquireContext': View Source

300    def acquire_context(self) -> "_AcquireContext":
301        """获取连接的上下文管理器。"""
302        return AsyncConnectionPool._AcquireContext(self)

获取连接的上下文管理器。

async def close_all(self) -> None: View Source

304    async def close_all(self) -> None:
305        """关闭池中所有连接并停止监控。"""
306        self._closed = True
307        self.stop_monitor()
308        for conn in list(self._connections):
309            await self._close_connection(conn)
310        # 释放所有信号量
311        while self._free.empty() is False:
312            self._free.get_nowait()

关闭池中所有连接并停止监控。

class AsyncBatchExecutor: View Source

 50class AsyncBatchExecutor:
 51    """异步批量命令执行器。
 52
 53    Args:
 54        host_service: HostService 实例（提供主机配置与凭据解析）
 55        max_concurrency: 最大并发主机数，默认 10
 56        command_timeout: 单条命令超时（秒），默认 30
 57        pool_factory: 外部连接池工厂（可选）。提供时执行器从工厂获取
 58            池并复用其连接，**绝不关闭**返回的池（所有权归调用方）；
 59            未提供时与同步 BatchExecutor 行为对齐——多主机或需重试时
 60            内部按主机创建 AsyncConnectionPool，执行结束后自动关闭。
 61
 62    连接池所有权约定（与同步 BatchExecutor 一致）：
 63
 64    - 外部注入（``pool_factory``）→ 调用方拥有，executor 只借用不关闭；
 65      适合长驻服务跨批次复用连接。
 66    - 内部创建 → executor 拥有，单次 ``execute`` 结束后 ``close_all``；
 67      适合一次性脚本。
 68    """
 69
 70    def __init__(
 71        self,
 72        host_service: HostService,
 73        max_concurrency: int = 10,
 74        command_timeout: int = 30,
 75        pool_factory: Optional[PoolFactory] = None,
 76    ) -> None:
 77        if max_concurrency < 1:
 78            raise ValueError(f"max_concurrency must be >= 1, got: {max_concurrency}")
 79        if command_timeout <= 0:
 80            raise ValueError(f"command_timeout must be > 0, got: {command_timeout}")
 81        self._host_service = host_service
 82        self._max_concurrency = max_concurrency
 83        self._command_timeout = command_timeout
 84        self._pool_factory = pool_factory
 85
 86    async def execute(
 87        self,
 88        host_names: list[str],
 89        command: str,
 90        retry_count: int = 0,
 91        retry_delay: float = 1.0,
 92        progress_callback: Optional[ProgressCallback] = None,
 93    ) -> BatchResult:
 94        """在多台主机上异步并发执行同一命令。
 95
 96        Args:
 97            host_names: 主机名称列表
 98            command: 要执行的命令
 99            retry_count: 失败重试次数（默认 0）。仅对瞬态错误
100                （超时/网络中断等）重试；认证、凭据、配置等永久性错误
101                立即失败（分类见 service/retry_policy.py）
102            retry_delay: 重试基础延迟（秒），默认 1.0。实际等待为
103                指数退避 + full jitter：第 n 次失败后等待
104                0 到 retry_delay * 2^n（含端点）内的随机值（上限 60s）
105            progress_callback: 进度回调，签名为 `(completed, total, host_name)`；
106                回调可为同步或 async 函数（async 时会被 await）。
107
108        Returns:
109            BatchResult: 批量结果（与同步 BatchExecutor 完全一致）
110
111        Raises:
112            ValueError: host_names 为空
113        """
114        if not host_names:
115            raise ValueError("host_names must not be empty")
116        if retry_count < 0:
117            raise ValueError(f"retry_count must be >= 0, got: {retry_count}")
118        if retry_delay < 0:
119            raise ValueError(f"retry_delay must be >= 0, got: {retry_delay}")
120
121        # 去重（保留首次出现顺序）：重复主机名只执行一次，避免 results 覆盖导致
122        # total/success/failed 统计错位
123        host_names = list(dict.fromkeys(host_names))
124
125        total = len(host_names)
126        semaphore = asyncio.Semaphore(self._max_concurrency)
127        start = time.time()
128
129        logger.info(
130            "异步批量执行开始: %s 台主机, 并发=%s",
131            total,
132            self._max_concurrency,
133        )
134
135        # 连接池准备（与同步 BatchExecutor 对齐）：
136        # - 外部 pool_factory 提供时始终启用（调用方持有所有权，绝不关闭）
137        # - 否则多主机或需重试时创建内部池，批结束后统一 close_all
138        pools: dict[str, AsyncConnectionPool] = {}
139        internal_pools: list[AsyncConnectionPool] = []
140        if self._pool_factory is not None or retry_count > 0 or total > 1:
141            for name in host_names:
142                self._prepare_pool(name, pools, internal_pools)
143
144        completed_counter = 0
145        completed_lock = asyncio.Lock()
146        results: dict[str, BatchHostResult] = {}
147
148        async def _per_host(name: str) -> None:
149            nonlocal completed_counter
150            async with semaphore:
151                result = await self._execute_on_host(
152                    name, command, retry_count, retry_delay, pool=pools.get(name)
153                )
154            async with completed_lock:
155                results[name] = result
156                completed = completed_counter + 1
157                completed_counter = completed
158
159                if progress_callback is not None:
160                    # 包裹回调：用户提供的 progress_callback 抛异常时不应中断整个批次，
161                    # 否则 gather 会向上抛出首异常、BatchResult 永不构建、
162                    # 已完成结果丢失且其余 task 沦为孤儿。
163                    try:
164                        rv = progress_callback(completed, total, name)
165                        if asyncio.iscoroutine(rv):
166                            await rv
167                    except Exception as e:  # noqa: BLE001
168                        logger.warning("progress_callback for %s raised: %s", name, e)
169
170                logger.debug(
171                    "[%s/%s] %s: %s (%.1fs)",
172                    completed,
173                    total,
174                    name,
175                    "✓" if result.success else "✗",
176                    result.duration,
177                )
178
179        tasks = [asyncio.create_task(_per_host(n)) for n in host_names]
180        try:
181            # return_exceptions=True 作为兜底：即使 _per_host 意外抛出异常，
182            # 也不会中断其他任务或使 BatchResult 构建被跳过。
183            # KeyboardInterrupt 属于 BaseException，仍会被下方 except 捕获。
184            await asyncio.gather(*tasks, return_exceptions=True)
185        except KeyboardInterrupt:
186            logger.warning("batch execution interrupted by user")
187            await self._cancel_and_mark_interrupted(tasks, host_names, results, command)
188        finally:
189            # 仅关闭内部创建的池；外部 pool_factory 提供的池所有权归调用方
190            await self._cleanup_pools(internal_pools)
191
192        duration = time.time() - start
193        success_count = sum(1 for r in results.values() if r.success)
194        failed_count = total - success_count
195        logger.info(
196            "async batch execution finished: %s/%s succeeded, took %.1fs",
197            success_count,
198            total,
199            duration,
200        )
201
202        return BatchResult(
203            total=total,
204            success=success_count,
205            failed=failed_count,
206            duration=duration,
207            results=results,
208        )
209
210    async def _cancel_and_mark_interrupted(
211        self,
212        tasks: list[asyncio.Task],
213        host_names: list[str],
214        results: dict[str, BatchHostResult],
215        command: str,
216    ) -> None:
217        """用户中断时取消所有任务，并为未完成主机创建失败记录。"""
218        for t in tasks:
219            t.cancel()
220        # 等待取消完成，避免 pending task 告警
221        await asyncio.gather(*tasks, return_exceptions=True)
222        # 为尚未有结果的主机创建失败记录
223        for name in host_names:
224            if name not in results:
225                results[name] = BatchHostResult(
226                    host=name,
227                    success=False,
228                    command=command,
229                    error="user interrupted",
230                )
231
232    def _prepare_pool(
233        self,
234        host_name: str,
235        pools: dict[str, AsyncConnectionPool],
236        internal_pools: list[AsyncConnectionPool],
237    ) -> None:
238        """为指定主机准备连接池（与同步 BatchExecutor._prepare_pool 对称）。
239
240        主机解析失败时跳过（不写入 pools），由 _execute_on_host 的
241        resolve_host_or_error 记录失败条目，保持 execute 的
242        "错误结果而非异常" 契约。
243
244        池所有权：
245        - 外部 ``pool_factory`` 提供的池：调用方持有，绝不登记进
246          internal_pools（executor 不负责关闭）
247        - 内部创建的池：登记进 internal_pools，批结束后统一 close_all
248        """
249        if host_name in pools:
250            return
251        try:
252            host = self._host_service.resolve_host(host_name)
253        except Exception as e:  # noqa: BLE001
254            logger.debug("pool preparation skipped for %s: %s", host_name, e)
255            return
256        config = build_connection_config(host, self._command_timeout)
257        if self._pool_factory is not None:
258            pools[host_name] = self._pool_factory(config)
259            return
260        pool = AsyncConnectionPool(
261            config,
262            max_connections=max(1, self._max_concurrency),
263            client_factory=AsyncSSHClient,
264        )
265        pools[host_name] = pool
266        internal_pools.append(pool)
267
268    async def _cleanup_pools(self, internal_pools: list[AsyncConnectionPool]) -> None:
269        """关闭本批次内部创建的所有连接池（外部提供的池绝不关闭）。"""
270        for pool in internal_pools:
271            await pool.close_all()
272
273    async def _execute_on_host(
274        self,
275        host_name: str,
276        command: str,
277        retry_count: int,
278        retry_delay: float,
279        pool: Optional[AsyncConnectionPool] = None,
280    ) -> BatchHostResult:
281        """在单台主机上异步执行命令（含重试逻辑）。
282
283        Args:
284            host_name: 主机名称
285            command: 要执行的命令
286            retry_count: 重试次数
287            retry_delay: 重试基础延迟（秒，指数退避基准）
288            pool: 可选连接池（外部注入或内部创建），提供时复用连接
289        """
290        # 主机解析（失败返回错误结果）
291        outcome = resolve_host_or_error(self._host_service, host_name, command)
292        if isinstance(outcome, BatchHostResult):
293            return outcome
294        host: Host = outcome
295
296        last_error: Optional[str] = None
297        last_duration = 0.0
298        for attempt in range(retry_count + 1):
299            start = time.time()
300            try:
301                if pool is not None:
302                    # 连接池模式：复用主机连接，避免每次操作握手
303                    # （外部池与内部池语义一致，仅生命周期归属不同）
304                    async with pool.acquire_context() as client:
305                        cmd_result = await client.execute(
306                            command,
307                            timeout=self._command_timeout,
308                        )
309                else:
310                    config = build_connection_config(host, self._command_timeout)
311                    async with AsyncSSHClient(config) as client:
312                        cmd_result = await client.execute(
313                            command,
314                            timeout=self._command_timeout,
315                        )
316                return to_host_result(host_name, command, cmd_result, time.time() - start)
317            except Exception as e:  # noqa: BLE001
318                last_error = str(e)
319                last_duration = time.time() - start
320                logger.debug(
321                    "%s 第 %s/%s 次尝试失败: %s",
322                    host_name,
323                    attempt + 1,
324                    retry_count + 1,
325                    e,
326                )
327                # 已是最后一次尝试，或异常为永久性（认证/凭据/配置错误等），
328                # 立即放弃重试——详见 service/retry_policy.py 的分类契约
329                if attempt >= retry_count:
330                    break
331                if not is_retryable(e):
332                    logger.debug("non-retryable error for %s, giving up: %s", host_name, e)
333                    break
334                # 指数退避 + full jitter（避免多主机同步重试的惊群）
335                delay = compute_backoff_delay(attempt, retry_delay)
336                await asyncio.sleep(delay)
337
338        return BatchHostResult(
339            host=host_name,
340            success=False,
341            command=command,
342            error=last_error,
343            duration=last_duration,
344        )

异步批量命令执行器。

Args: host_service: HostService 实例（提供主机配置与凭据解析） max_concurrency: 最大并发主机数，默认 10 command_timeout: 单条命令超时（秒），默认 30 pool_factory: 外部连接池工厂（可选）。提供时执行器从工厂获取池并复用其连接，绝不关闭返回的池（所有权归调用方）；未提供时与同步 BatchExecutor 行为对齐——多主机或需重试时内部按主机创建 AsyncConnectionPool，执行结束后自动关闭。

连接池所有权约定（与同步 BatchExecutor 一致）：

外部注入（pool_factory）→ 调用方拥有，executor 只借用不关闭；适合长驻服务跨批次复用连接。
内部创建 → executor 拥有，单次 execute 结束后 close_all；适合一次性脚本。

AsyncBatchExecutor( host_service: HostService, max_concurrency: int = 10, command_timeout: int = 30, pool_factory: Callable[[remote_cmd.core.ssh_client.ConnectionConfig], AsyncConnectionPool] | None = None) View Source

70    def __init__(
71        self,
72        host_service: HostService,
73        max_concurrency: int = 10,
74        command_timeout: int = 30,
75        pool_factory: Optional[PoolFactory] = None,
76    ) -> None:
77        if max_concurrency < 1:
78            raise ValueError(f"max_concurrency must be >= 1, got: {max_concurrency}")
79        if command_timeout <= 0:
80            raise ValueError(f"command_timeout must be > 0, got: {command_timeout}")
81        self._host_service = host_service
82        self._max_concurrency = max_concurrency
83        self._command_timeout = command_timeout
84        self._pool_factory = pool_factory

async def execute( self, host_names: list[str], command: str, retry_count: int = 0, retry_delay: float = 1.0, progress_callback: Callable[[int, int, str], Awaitable[None] | None] | None = None) -> BatchResult: View Source

 86    async def execute(
 87        self,
 88        host_names: list[str],
 89        command: str,
 90        retry_count: int = 0,
 91        retry_delay: float = 1.0,
 92        progress_callback: Optional[ProgressCallback] = None,
 93    ) -> BatchResult:
 94        """在多台主机上异步并发执行同一命令。
 95
 96        Args:
 97            host_names: 主机名称列表
 98            command: 要执行的命令
 99            retry_count: 失败重试次数（默认 0）。仅对瞬态错误
100                （超时/网络中断等）重试；认证、凭据、配置等永久性错误
101                立即失败（分类见 service/retry_policy.py）
102            retry_delay: 重试基础延迟（秒），默认 1.0。实际等待为
103                指数退避 + full jitter：第 n 次失败后等待
104                0 到 retry_delay * 2^n（含端点）内的随机值（上限 60s）
105            progress_callback: 进度回调，签名为 `(completed, total, host_name)`；
106                回调可为同步或 async 函数（async 时会被 await）。
107
108        Returns:
109            BatchResult: 批量结果（与同步 BatchExecutor 完全一致）
110
111        Raises:
112            ValueError: host_names 为空
113        """
114        if not host_names:
115            raise ValueError("host_names must not be empty")
116        if retry_count < 0:
117            raise ValueError(f"retry_count must be >= 0, got: {retry_count}")
118        if retry_delay < 0:
119            raise ValueError(f"retry_delay must be >= 0, got: {retry_delay}")
120
121        # 去重（保留首次出现顺序）：重复主机名只执行一次，避免 results 覆盖导致
122        # total/success/failed 统计错位
123        host_names = list(dict.fromkeys(host_names))
124
125        total = len(host_names)
126        semaphore = asyncio.Semaphore(self._max_concurrency)
127        start = time.time()
128
129        logger.info(
130            "异步批量执行开始: %s 台主机, 并发=%s",
131            total,
132            self._max_concurrency,
133        )
134
135        # 连接池准备（与同步 BatchExecutor 对齐）：
136        # - 外部 pool_factory 提供时始终启用（调用方持有所有权，绝不关闭）
137        # - 否则多主机或需重试时创建内部池，批结束后统一 close_all
138        pools: dict[str, AsyncConnectionPool] = {}
139        internal_pools: list[AsyncConnectionPool] = []
140        if self._pool_factory is not None or retry_count > 0 or total > 1:
141            for name in host_names:
142                self._prepare_pool(name, pools, internal_pools)
143
144        completed_counter = 0
145        completed_lock = asyncio.Lock()
146        results: dict[str, BatchHostResult] = {}
147
148        async def _per_host(name: str) -> None:
149            nonlocal completed_counter
150            async with semaphore:
151                result = await self._execute_on_host(
152                    name, command, retry_count, retry_delay, pool=pools.get(name)
153                )
154            async with completed_lock:
155                results[name] = result
156                completed = completed_counter + 1
157                completed_counter = completed
158
159                if progress_callback is not None:
160                    # 包裹回调：用户提供的 progress_callback 抛异常时不应中断整个批次，
161                    # 否则 gather 会向上抛出首异常、BatchResult 永不构建、
162                    # 已完成结果丢失且其余 task 沦为孤儿。
163                    try:
164                        rv = progress_callback(completed, total, name)
165                        if asyncio.iscoroutine(rv):
166                            await rv
167                    except Exception as e:  # noqa: BLE001
168                        logger.warning("progress_callback for %s raised: %s", name, e)
169
170                logger.debug(
171                    "[%s/%s] %s: %s (%.1fs)",
172                    completed,
173                    total,
174                    name,
175                    "✓" if result.success else "✗",
176                    result.duration,
177                )
178
179        tasks = [asyncio.create_task(_per_host(n)) for n in host_names]
180        try:
181            # return_exceptions=True 作为兜底：即使 _per_host 意外抛出异常，
182            # 也不会中断其他任务或使 BatchResult 构建被跳过。
183            # KeyboardInterrupt 属于 BaseException，仍会被下方 except 捕获。
184            await asyncio.gather(*tasks, return_exceptions=True)
185        except KeyboardInterrupt:
186            logger.warning("batch execution interrupted by user")
187            await self._cancel_and_mark_interrupted(tasks, host_names, results, command)
188        finally:
189            # 仅关闭内部创建的池；外部 pool_factory 提供的池所有权归调用方
190            await self._cleanup_pools(internal_pools)
191
192        duration = time.time() - start
193        success_count = sum(1 for r in results.values() if r.success)
194        failed_count = total - success_count
195        logger.info(
196            "async batch execution finished: %s/%s succeeded, took %.1fs",
197            success_count,
198            total,
199            duration,
200        )
201
202        return BatchResult(
203            total=total,
204            success=success_count,
205            failed=failed_count,
206            duration=duration,
207            results=results,
208        )

在多台主机上异步并发执行同一命令。

Args: host_names: 主机名称列表 command: 要执行的命令 retry_count: 失败重试次数（默认 0）。仅对瞬态错误（超时/网络中断等）重试；认证、凭据、配置等永久性错误立即失败（分类见 service/retry_policy.py） retry_delay: 重试基础延迟（秒），默认 1.0。实际等待为指数退避 + full jitter：第 n 次失败后等待 0 到 retry_delay * 2^n（含端点）内的随机值（上限 60s） progress_callback: 进度回调，签名为 (completed, total, host_name)；回调可为同步或 async 函数（async 时会被 await）。

Returns: BatchResult: 批量结果（与同步 BatchExecutor 完全一致）

Raises: ValueError: host_names 为空

@dataclass

class Host: View Source

 20@dataclass
 21class Host:
 22    """
 23    远程主机配置类
 24
 25    存储单个远程主机的所有连接信息和元数据。
 26    支持通过标签进行分类管理。
 27
 28    Attributes:
 29        name: 主机名称（唯一标识符）
 30        hostname: 主机地址（IP 或域名）
 31        username: SSH 登录用户名
 32        port: SSH 端口号，默认为 22
 33        password: 登录密码（可选，可能被加密）
 34        key_filename: SSH 私钥文件路径（可选）
 35        tags: 主机标签列表，用于分类和筛选
 36        description: 主机描述信息
 37
 38    Example:
 39        >>> host = Host(
 40        ...     name="web-server",
 41        ...     hostname="192.168.1.100",
 42        ...     username="admin",
 43        ...     key_filename="~/.ssh/id_rsa",
 44        ...     tags=["production", "web"]
 45        ... )
 46    """
 47
 48    name: str
 49    hostname: str
 50    username: str
 51    port: int = 22
 52    password: Optional[str] = None
 53    key_filename: Optional[str] = None
 54    tags: list[str] = field(default_factory=list)
 55    description: str = ""
 56
 57    def __post_init__(self):
 58        """初始化后处理：归一化外部传入的 None 标签（兼容旧数据）"""
 59        if self.tags is None:
 60            self.tags = []
 61
 62    def to_connection_config(self) -> ConnectionConfig:
 63        """
 64        将主机配置转换为 SSH 连接配置
 65
 66        Returns:
 67            ConnectionConfig: 可直接用于 SSHClient 的连接配置对象
 68        """
 69        return ConnectionConfig(
 70            hostname=self.hostname,
 71            username=self.username,
 72            port=self.port,
 73            password=self.password,
 74            key_filename=self.key_filename,
 75        )
 76
 77    def to_dict(self) -> dict:
 78        """
 79        将主机配置转换为字典
 80
 81        Returns:
 82            Dict: 包含所有主机属性的字典
 83        """
 84        return asdict(self)
 85
 86    @classmethod
 87    def from_dict(cls, data: dict) -> "Host":
 88        """
 89        从字典创建主机配置对象
 90
 91        Args:
 92            data: 包含主机属性的字典
 93
 94        Returns:
 95            Host: 主机配置对象
 96        """
 97        # 只提取 Host 已知的字段，忽略未知字段
 98        known_fields = {
 99            "name",
100            "hostname",
101            "username",
102            "port",
103            "password",
104            "key_filename",
105            "tags",
106            "description",
107        }
108        filtered = {k: v for k, v in data.items() if k in known_fields}
109        return cls(**filtered)
110
111    def sanitized_dict(self) -> dict[str, Any]:
112        """
113        返回脱敏后的主机字典，用于日志、显示、API 响应等场景。
114
115        敏感字段（password、key_filename）被替换为安全标识。
116
117        Returns:
118            Dict: 不包含敏感明文的主机信息字典
119        """
120        data = asdict(self)
121        # 脱敏密码：显示加密状态而非明文
122        pw = data.get("password")
123        if pw:
124            data["password"] = (
125                "***encrypted***" if isinstance(pw, str) and pw.startswith("$encrypted$") else "***"
126            )
127        # 脱敏密钥路径：仅显示文件名
128        if data.get("key_filename"):
129            data["key_filename"] = Path(data["key_filename"]).name
130        return data
131
132    def __repr__(self) -> str:
133        """安全的字符串表示：自动脱敏敏感字段"""
134        safe = self.sanitized_dict()
135        fields = ", ".join(f"{k}={v!r}" for k, v in safe.items())
136        return f"Host({fields})"

远程主机配置类

存储单个远程主机的所有连接信息和元数据。支持通过标签进行分类管理。

Attributes: name: 主机名称（唯一标识符） hostname: 主机地址（IP 或域名） username: SSH 登录用户名 port: SSH 端口号，默认为 22 password: 登录密码（可选，可能被加密） key_filename: SSH 私钥文件路径（可选） tags: 主机标签列表，用于分类和筛选 description: 主机描述信息

Example:

host = Host( ... name="web-server", ... hostname="192.168.1.100", ... username="admin", ... key_filename="~/.ssh/id_rsa", ... tags=["production", "web"] ... )

Host( name: str, hostname: str, username: str, port: int = 22, password: str | None = None, key_filename: str | None = None, tags: list[str] = <factory>, description: str = '')

name: str

hostname: str

username: str

port: int = 22

password: str | None = None

key_filename: str | None = None

tags: list[str]

description: str = ''

def to_connection_config(self) -> remote_cmd.core.ssh_client.ConnectionConfig: View Source

62    def to_connection_config(self) -> ConnectionConfig:
63        """
64        将主机配置转换为 SSH 连接配置
65
66        Returns:
67            ConnectionConfig: 可直接用于 SSHClient 的连接配置对象
68        """
69        return ConnectionConfig(
70            hostname=self.hostname,
71            username=self.username,
72            port=self.port,
73            password=self.password,
74            key_filename=self.key_filename,
75        )

将主机配置转换为 SSH 连接配置

Returns: ConnectionConfig: 可直接用于 SSHClient 的连接配置对象

def to_dict(self) -> dict: View Source

77    def to_dict(self) -> dict:
78        """
79        将主机配置转换为字典
80
81        Returns:
82            Dict: 包含所有主机属性的字典
83        """
84        return asdict(self)

将主机配置转换为字典

Returns: Dict: 包含所有主机属性的字典

@classmethod

def from_dict(cls, data: dict) -> 'Host': View Source

 86    @classmethod
 87    def from_dict(cls, data: dict) -> "Host":
 88        """
 89        从字典创建主机配置对象
 90
 91        Args:
 92            data: 包含主机属性的字典
 93
 94        Returns:
 95            Host: 主机配置对象
 96        """
 97        # 只提取 Host 已知的字段，忽略未知字段
 98        known_fields = {
 99            "name",
100            "hostname",
101            "username",
102            "port",
103            "password",
104            "key_filename",
105            "tags",
106            "description",
107        }
108        filtered = {k: v for k, v in data.items() if k in known_fields}
109        return cls(**filtered)

从字典创建主机配置对象

Args: data: 包含主机属性的字典

Returns: Host: 主机配置对象

def sanitized_dict(self) -> dict[str, typing.Any]: View Source

111    def sanitized_dict(self) -> dict[str, Any]:
112        """
113        返回脱敏后的主机字典，用于日志、显示、API 响应等场景。
114
115        敏感字段（password、key_filename）被替换为安全标识。
116
117        Returns:
118            Dict: 不包含敏感明文的主机信息字典
119        """
120        data = asdict(self)
121        # 脱敏密码：显示加密状态而非明文
122        pw = data.get("password")
123        if pw:
124            data["password"] = (
125                "***encrypted***" if isinstance(pw, str) and pw.startswith("$encrypted$") else "***"
126            )
127        # 脱敏密钥路径：仅显示文件名
128        if data.get("key_filename"):
129            data["key_filename"] = Path(data["key_filename"]).name
130        return data

返回脱敏后的主机字典，用于日志、显示、API 响应等场景。

敏感字段（password、key_filename）被替换为安全标识。

Returns: Dict: 不包含敏感明文的主机信息字典

class HostRepository(abc.ABC): View Source

17class HostRepository(ABC):
18    """主机配置仓库抽象基类"""
19
20    @abstractmethod
21    def save(self, host: Host) -> None:
22        """保存主机（新增或覆盖）"""
23        ...
24
25    @abstractmethod
26    def get(self, name: str) -> Host:
27        """按名称获取主机，不存在时抛出 KeyError"""
28        ...
29
30    @abstractmethod
31    def delete(self, name: str) -> None:
32        """按名称删除主机，不存在时抛出 KeyError"""
33        ...
34
35    @abstractmethod
36    def list(self, tag: Optional[str] = None) -> list[Host]:
37        """列出主机，可选按标签筛选"""
38        ...
39
40    @abstractmethod
41    def list_tags(self) -> builtins.list[str]:
42        """列出所有标签"""
43        ...
44
45    @abstractmethod
46    def contains(self, name: str) -> bool:
47        """检查主机是否存在"""
48        ...
49
50    @abstractmethod
51    def count(self) -> int:
52        """返回主机数量"""
53        ...
54
55    @abstractmethod
56    def flush(self) -> None:
57        """将所有内存中更改写入存储"""
58        ...

主机配置仓库抽象基类

@abstractmethod

def save(self, host: Host) -> None: View Source

20    @abstractmethod
21    def save(self, host: Host) -> None:
22        """保存主机（新增或覆盖）"""
23        ...

保存主机（新增或覆盖）

@abstractmethod

def get(self, name: str) -> Host: View Source

25    @abstractmethod
26    def get(self, name: str) -> Host:
27        """按名称获取主机，不存在时抛出 KeyError"""
28        ...

按名称获取主机，不存在时抛出 KeyError

@abstractmethod

def delete(self, name: str) -> None: View Source

30    @abstractmethod
31    def delete(self, name: str) -> None:
32        """按名称删除主机，不存在时抛出 KeyError"""
33        ...

按名称删除主机，不存在时抛出 KeyError

@abstractmethod

def list(unknown): View Source

35    @abstractmethod
36    def list(self, tag: Optional[str] = None) -> list[Host]:
37        """列出主机，可选按标签筛选"""
38        ...

列出主机，可选按标签筛选

@abstractmethod

def contains(self, name: str) -> bool: View Source

45    @abstractmethod
46    def contains(self, name: str) -> bool:
47        """检查主机是否存在"""
48        ...

检查主机是否存在

@abstractmethod

def count(self) -> int: View Source

50    @abstractmethod
51    def count(self) -> int:
52        """返回主机数量"""
53        ...

返回主机数量

@abstractmethod

def flush(self) -> None: View Source

55    @abstractmethod
56    def flush(self) -> None:
57        """将所有内存中更改写入存储"""
58        ...

将所有内存中更改写入存储

class SSHService: View Source

 21class SSHService:
 22    """
 23    SSH 连接服务
 24
 25    提供连接管理、命令执行和健康检查功能。
 26    支持重试和超时控制。
 27    """
 28
 29    def __init__(self, timeout: int = 30) -> None:
 30        self._timeout = timeout
 31
 32    def create_client(
 33        self,
 34        hostname: str,
 35        username: str,
 36        port: int = 22,
 37        password: Optional[str] = None,
 38        key_filename: Optional[str] = None,
 39        known_hosts_file: Optional[str] = None,
 40    ) -> SSHClient:
 41        """
 42        创建并建立 SSH 连接
 43
 44        Returns:
 45            已连接的 SSHClient 实例
 46
 47        Raises:
 48            SSHConnectionError: 连接失败
 49        """
 50        config = ConnectionConfig(
 51            hostname=hostname,
 52            username=username,
 53            port=port,
 54            password=password,
 55            key_filename=key_filename,
 56            timeout=self._timeout,
 57            known_hosts_file=known_hosts_file,
 58        )
 59        client = SSHClient(config)
 60        return client.connect()
 61
 62    def test_connection(
 63        self,
 64        hostname: str,
 65        username: str,
 66        port: int = 22,
 67        password: Optional[str] = None,
 68        key_filename: Optional[str] = None,
 69    ) -> bool:
 70        """
 71        测试主机连接是否正常
 72
 73        Returns:
 74            bool: True if connected
 75        """
 76        try:
 77            with self.create_client(
 78                hostname=hostname,
 79                username=username,
 80                port=port,
 81                password=password,
 82                key_filename=key_filename,
 83            ) as client:
 84                return client.is_connected()
 85        except Exception as e:  # noqa: BLE001
 86            logger.debug(f"connection test failed {hostname}:{port}: {e}")
 87            return False
 88
 89    def execute_command(
 90        self,
 91        hostname: str,
 92        username: str,
 93        command: str,
 94        port: int = 22,
 95        password: Optional[str] = None,
 96        key_filename: Optional[str] = None,
 97        timeout: Optional[int] = None,
 98    ) -> CommandResult:
 99        """
100        在远程主机上执行命令
101
102        Returns:
103            CommandResult: 命令执行结果
104        """
105        with self.create_client(
106            hostname=hostname,
107            username=username,
108            port=port,
109            password=password,
110            key_filename=key_filename,
111        ) as client:
112            return client.execute(command, timeout=timeout)

SSH 连接服务

提供连接管理、命令执行和健康检查功能。支持重试和超时控制。

SSHService(timeout: int = 30) View Source

29    def __init__(self, timeout: int = 30) -> None:
30        self._timeout = timeout

def create_client( self, hostname: str, username: str, port: int = 22, password: str | None = None, key_filename: str | None = None, known_hosts_file: str | None = None) -> SSHClient: View Source

32    def create_client(
33        self,
34        hostname: str,
35        username: str,
36        port: int = 22,
37        password: Optional[str] = None,
38        key_filename: Optional[str] = None,
39        known_hosts_file: Optional[str] = None,
40    ) -> SSHClient:
41        """
42        创建并建立 SSH 连接
43
44        Returns:
45            已连接的 SSHClient 实例
46
47        Raises:
48            SSHConnectionError: 连接失败
49        """
50        config = ConnectionConfig(
51            hostname=hostname,
52            username=username,
53            port=port,
54            password=password,
55            key_filename=key_filename,
56            timeout=self._timeout,
57            known_hosts_file=known_hosts_file,
58        )
59        client = SSHClient(config)
60        return client.connect()

创建并建立 SSH 连接

Returns: 已连接的 SSHClient 实例

Raises: SSHConnectionError: 连接失败

def test_connection( self, hostname: str, username: str, port: int = 22, password: str | None = None, key_filename: str | None = None) -> bool: View Source

62    def test_connection(
63        self,
64        hostname: str,
65        username: str,
66        port: int = 22,
67        password: Optional[str] = None,
68        key_filename: Optional[str] = None,
69    ) -> bool:
70        """
71        测试主机连接是否正常
72
73        Returns:
74            bool: True if connected
75        """
76        try:
77            with self.create_client(
78                hostname=hostname,
79                username=username,
80                port=port,
81                password=password,
82                key_filename=key_filename,
83            ) as client:
84                return client.is_connected()
85        except Exception as e:  # noqa: BLE001
86            logger.debug(f"connection test failed {hostname}:{port}: {e}")
87            return False

测试主机连接是否正常

Returns: bool: True if connected

def execute_command( self, hostname: str, username: str, command: str, port: int = 22, password: str | None = None, key_filename: str | None = None, timeout: int | None = None) -> remote_cmd.core.ssh_client.CommandResult: View Source

 89    def execute_command(
 90        self,
 91        hostname: str,
 92        username: str,
 93        command: str,
 94        port: int = 22,
 95        password: Optional[str] = None,
 96        key_filename: Optional[str] = None,
 97        timeout: Optional[int] = None,
 98    ) -> CommandResult:
 99        """
100        在远程主机上执行命令
101
102        Returns:
103            CommandResult: 命令执行结果
104        """
105        with self.create_client(
106            hostname=hostname,
107            username=username,
108            port=port,
109            password=password,
110            key_filename=key_filename,
111        ) as client:
112            return client.execute(command, timeout=timeout)

在远程主机上执行命令

Returns: CommandResult: 命令执行结果

class CredentialProvider(abc.ABC): View Source

36class CredentialProvider(ABC):
37    """凭据提供者抽象基类"""
38
39    @abstractmethod
40    def get_password(self, host: Host) -> Optional[str]:
41        """get a host by name, or None if not found"""
42        ...

凭据提供者抽象基类

@abstractmethod

def get_password(self, host: Host) -> str | None: View Source

39    @abstractmethod
40    def get_password(self, host: Host) -> Optional[str]:
41        """get a host by name, or None if not found"""
42        ...

get a host by name, or None if not found

class EnvCredentialProvider(remote_cmd.CredentialProvider): View Source

45class EnvCredentialProvider(CredentialProvider):
46    """
47    从环境变量获取密码
48
49    适用于 CI/CD 或容器环境。
50    优先级低于交互式输入但高于默认值。
51
52    支持两种查找方式（按优先级）：
53    1. 主机专属变量 ``<env_var>_<HOST>``（主机名大写，非字母数字替换为 ``_``）
54    2. 全局变量 ``<env_var>``
55
56    示例：
57        - 全局：``REMOTE_CMD_PASSWORD=secret`` 对所有主机生效
58        - 专属：``REMOTE_CMD_PASSWORD_WEB1=secret`` 仅对名为 ``web1`` 的主机生效
59
60    Args:
61        env_var: 环境变量名（默认 REMOTE_CMD_PASSWORD）
62    """
63
64    def __init__(self, env_var: str = "REMOTE_CMD_PASSWORD") -> None:
65        self._env_var = env_var
66
67    @staticmethod
68    def _host_env_suffix(host_name: str) -> str:
69        """
70        将主机名转换为环境变量后缀：web1 -> WEB1，my-host -> MY_HOST
71
72        注意: 非字母数字字符统一归一化为下划线，因此 ``web-1`` 与 ``web_1``
73        会映射到同一个变量 ``..._WEB_1``。若同舰队同时存在这两种命名，
74        需避免依赖同名变量（如需区分请先统一主机命名规范）。
75        """
76        normalized = "".join(c if c.isalnum() else "_" for c in host_name).upper()
77        return normalized
78
79    def get_password(self, host: Host) -> Optional[str]:
80        # 优先主机专属变量，避免全局变量被应用到所有主机
81        if host and host.name:
82            host_var = f"{self._env_var}_{self._host_env_suffix(host.name)}"
83            host_password = os.environ.get(host_var)
84            if host_password is not None:
85                return host_password
86        return os.environ.get(self._env_var)

从环境变量获取密码

适用于 CI/CD 或容器环境。优先级低于交互式输入但高于默认值。

支持两种查找方式（按优先级）：

主机专属变量 <env_var>_<HOST>（主机名大写，非字母数字替换为 _）
全局变量 <env_var>

示例： - 全局：REMOTE_CMD_PASSWORD=secret 对所有主机生效 - 专属：REMOTE_CMD_PASSWORD_WEB1=secret 仅对名为 web1 的主机生效

Args: env_var: 环境变量名（默认 REMOTE_CMD_PASSWORD）

EnvCredentialProvider(env_var: str = 'REMOTE_CMD_PASSWORD') View Source

64    def __init__(self, env_var: str = "REMOTE_CMD_PASSWORD") -> None:
65        self._env_var = env_var

def get_password(self, host: Host) -> str | None: View Source

79    def get_password(self, host: Host) -> Optional[str]:
80        # 优先主机专属变量，避免全局变量被应用到所有主机
81        if host and host.name:
82            host_var = f"{self._env_var}_{self._host_env_suffix(host.name)}"
83            host_password = os.environ.get(host_var)
84            if host_password is not None:
85                return host_password
86        return os.environ.get(self._env_var)

get a host by name, or None if not found

class ChainCredentialProvider(remote_cmd.CredentialProvider): View Source

116class ChainCredentialProvider(CredentialProvider):
117    """
118    链式凭据提供者
119
120    按顺序尝试每个提供者，返回第一个非空结果。
121    适用于 "环境变量 → 加密文件 → 交互式输入" 的优先级链。
122
123    Args:
124        providers: 凭据提供者列表，按优先级降序排列
125    """
126
127    def __init__(self, providers: list[CredentialProvider]) -> None:
128        self._providers = list(providers)
129
130    def get_password(self, host: Host) -> Optional[str]:
131        for provider in self._providers:
132            password = provider.get_password(host)
133            if password is not None:
134                return password
135        return None
136
137    def add_provider(self, provider: CredentialProvider) -> None:
138        """在链尾添加一个提供者"""
139        self._providers.append(provider)

链式凭据提供者

按顺序尝试每个提供者，返回第一个非空结果。适用于 "环境变量 → 加密文件 → 交互式输入" 的优先级链。

Args: providers: 凭据提供者列表，按优先级降序排列

ChainCredentialProvider( providers: list[CredentialProvider]) View Source

127    def __init__(self, providers: list[CredentialProvider]) -> None:
128        self._providers = list(providers)

def get_password(self, host: Host) -> str | None: View Source

130    def get_password(self, host: Host) -> Optional[str]:
131        for provider in self._providers:
132            password = provider.get_password(host)
133            if password is not None:
134                return password
135        return None

get a host by name, or None if not found

def add_provider( self, provider: CredentialProvider) -> None: View Source

137    def add_provider(self, provider: CredentialProvider) -> None:
138        """在链尾添加一个提供者"""
139        self._providers.append(provider)

在链尾添加一个提供者

class CredentialEncryption: View Source

 48class CredentialEncryption:
 49    """
 50    Fernet 凭据加解密器
 51
 52    使用 cryptography.fernet 实现对称加密。
 53    密钥自动管理：首次加密时生成，后续自动加载。
 54
 55    Attributes:
 56        key_path: 密钥文件存储路径
 57    """
 58
 59    _PREFIX = "$encrypted$"
 60
 61    def __init__(self, key_path: Optional[Path] = None) -> None:
 62        """
 63        Args:
 64            key_path: 密钥文件路径，默认 ~/.remote_cmd/.key
 65        """
 66        self._key_path = key_path or (Path.home() / ".remote_cmd" / ".key")
 67        self._fernet: Optional[Fernet] = None
 68
 69    @property
 70    def _cipher(self) -> "Fernet":
 71        """延迟初始化 Fernet 实例"""
 72        if self._fernet is None:
 73            key = self._load_or_create_key()
 74            from cryptography.fernet import Fernet
 75
 76            self._fernet = Fernet(key)
 77        return self._fernet
 78
 79    def encrypt(self, plaintext: str) -> str:
 80        """
 81        加密明文密码
 82
 83        Args:
 84            plaintext: 明文密码
 85
 86        Returns:
 87            str: Base64 编码的密文字符串（格式: $encrypted$<token>）
 88
 89        Raises:
 90            CredentialEncryptionError: 加密失败
 91        """
 92        try:
 93            token = self._cipher.encrypt(plaintext.encode("utf-8"))
 94            return self._PREFIX + token.decode("utf-8")
 95        except Exception as e:
 96            raise CredentialEncryptionError(f"encryption failed: {e}") from e
 97
 98    def decrypt(self, ciphertext: str) -> str:
 99        """
100        解密密文
101
102        Args:
103            ciphertext: 加密后的密文字符串
104
105        Returns:
106            str: 明文密码
107
108        Raises:
109            CredentialEncryptionError: 解密失败或格式错误
110        """
111        if not ciphertext.startswith(self._PREFIX):
112            raise CredentialEncryptionError("invalid ciphertext format")
113
114        try:
115            token = ciphertext[len(self._PREFIX) :].encode("utf-8")
116            return self._cipher.decrypt(token).decode("utf-8")
117        except Exception as e:
118            raise CredentialEncryptionError(f"decryption failed: {e}") from e
119
120    def is_encrypted(self, value: str) -> bool:
121        """
122        检查字符串是否为加密格式
123
124        采用双重校验降低格式碰撞风险：
125        1. 前缀匹配 ``$encrypted$``
126        2. token 部分必须是 Fernet 格式（version byte 0x80，base64 编码后以 ``g`` 开头，
127           且解码后总长度至少 57 字节 = 1 + 8 + 16 + 32）
128
129        这避免了与真实密码恰好以 ``$encrypted$`` 开头时被误判的场景：
130        真实密码很少以 ``$encrypted$g`` 开头，且 base64 解码长度必须 >= 57 字节。
131
132        Attention: 本函数不验证 HMAC/签名，仅为廉价标识符检测；
133        完整合法性校验在 :meth:`decrypt` 中完成。
134        """
135        if not value.startswith(self._PREFIX):
136            return False
137        token = value[len(self._PREFIX) :]
138        # Fernet token base64 解码后第一字节恒为 0x80（编码后开头为 "g"）
139        if not token.startswith("g"):
140            return False
141        # 重新填充 base64 并解码长度校验，避免误判纯字符串
142        import base64
143
144        padding = "=" * (-len(token) % 4)
145        try:
146            decoded = base64.urlsafe_b64decode(token + padding)
147        except ValueError:
148            return False
149        # 最小密文长度 = version(1) + timestamp(8) + IV(16) + HMAC(32)
150        return len(decoded) >= 57
151
152    def _load_or_create_key(self) -> bytes:
153        """
154        加载现有密钥或生成新密钥
155
156        密钥文件权限设为 0600，防止其他用户读取。
157        校验密钥格式：必须是合法的 Fernet key（32 字节 base64 编码）。
158        """
159        from cryptography.fernet import Fernet
160
161        if self._key_path.exists():
162            # 确保现有密钥文件权限正确
163            try:
164                current_perms = stat_module.S_IMODE(self._key_path.stat().st_mode)
165                if current_perms != 0o600:
166                    self._key_path.chmod(0o600)
167            except OSError:
168                pass  # Windows 或权限不足时忽略
169
170            key = self._key_path.read_bytes()
171            # 校验密钥格式：Fernet key 必须是 32 字节 base64 编码（44 字符，含 padding）
172            try:
173                decoded = base64.urlsafe_b64decode(key)
174                if len(decoded) != 32:
175                    raise ValueError(f"Invalid key length: {len(decoded)} bytes, expected 32")
176            except (ValueError, binascii.Error) as e:
177                logger.error(f"Invalid encryption key format at {self._key_path}: {e}")
178                raise CredentialEncryptionError(
179                    f"Corrupted or invalid key file: {self._key_path}. "
180                    f"Delete it to regenerate, or check permissions."
181                ) from e
182
183            return key
184
185        # 生成新密钥
186        key = Fernet.generate_key()
187        self._key_path.parent.mkdir(parents=True, exist_ok=True)
188        # 安全：原子创建密钥文件并直接设置 0600 权限，避免 write_bytes（受 umask
189        # 影响，默认可能为 0644）与 chmod(0600) 之间的 TOCTOU 窗口——在此窗口内
190        # 同机其他用户可读取主加密密钥，进而解密所有凭据。
191        # O_EXCL 同时消除多进程并发首次创建时互相覆盖密钥的竞态。
192        try:
193            fd = os.open(
194                str(self._key_path),
195                os.O_WRONLY | os.O_CREAT | os.O_EXCL,
196                0o600,
197            )
198        except FileExistsError:
199            # 并发首次创建：另一进程已生成密钥，重新加载
200            return self._load_or_create_key()
201        try:
202            os.write(fd, key)
203        finally:
204            os.close(fd)
205
206        logger.info(f"generated encryption key: {self._key_path}")
207        return key

Fernet 凭据加解密器

使用 cryptography.fernet 实现对称加密。密钥自动管理：首次加密时生成，后续自动加载。

Attributes: key_path: 密钥文件存储路径

CredentialEncryption(key_path: pathlib.Path | None = None) View Source

61    def __init__(self, key_path: Optional[Path] = None) -> None:
62        """
63        Args:
64            key_path: 密钥文件路径，默认 ~/.remote_cmd/.key
65        """
66        self._key_path = key_path or (Path.home() / ".remote_cmd" / ".key")
67        self._fernet: Optional[Fernet] = None

Args: key_path: 密钥文件路径，默认 ~/.remote_cmd/.key

def encrypt(self, plaintext: str) -> str: View Source

79    def encrypt(self, plaintext: str) -> str:
80        """
81        加密明文密码
82
83        Args:
84            plaintext: 明文密码
85
86        Returns:
87            str: Base64 编码的密文字符串（格式: $encrypted$<token>）
88
89        Raises:
90            CredentialEncryptionError: 加密失败
91        """
92        try:
93            token = self._cipher.encrypt(plaintext.encode("utf-8"))
94            return self._PREFIX + token.decode("utf-8")
95        except Exception as e:
96            raise CredentialEncryptionError(f"encryption failed: {e}") from e

加密明文密码

Args: plaintext: 明文密码

Returns: str: Base64 编码的密文字符串（格式: $encrypted$）

Raises: CredentialEncryptionError: 加密失败

def decrypt(self, ciphertext: str) -> str: View Source

 98    def decrypt(self, ciphertext: str) -> str:
 99        """
100        解密密文
101
102        Args:
103            ciphertext: 加密后的密文字符串
104
105        Returns:
106            str: 明文密码
107
108        Raises:
109            CredentialEncryptionError: 解密失败或格式错误
110        """
111        if not ciphertext.startswith(self._PREFIX):
112            raise CredentialEncryptionError("invalid ciphertext format")
113
114        try:
115            token = ciphertext[len(self._PREFIX) :].encode("utf-8")
116            return self._cipher.decrypt(token).decode("utf-8")
117        except Exception as e:
118            raise CredentialEncryptionError(f"decryption failed: {e}") from e

解密密文

Args: ciphertext: 加密后的密文字符串

Returns: str: 明文密码

Raises: CredentialEncryptionError: 解密失败或格式错误

def is_encrypted(self, value: str) -> bool: View Source

120    def is_encrypted(self, value: str) -> bool:
121        """
122        检查字符串是否为加密格式
123
124        采用双重校验降低格式碰撞风险：
125        1. 前缀匹配 ``$encrypted$``
126        2. token 部分必须是 Fernet 格式（version byte 0x80，base64 编码后以 ``g`` 开头，
127           且解码后总长度至少 57 字节 = 1 + 8 + 16 + 32）
128
129        这避免了与真实密码恰好以 ``$encrypted$`` 开头时被误判的场景：
130        真实密码很少以 ``$encrypted$g`` 开头，且 base64 解码长度必须 >= 57 字节。
131
132        Attention: 本函数不验证 HMAC/签名，仅为廉价标识符检测；
133        完整合法性校验在 :meth:`decrypt` 中完成。
134        """
135        if not value.startswith(self._PREFIX):
136            return False
137        token = value[len(self._PREFIX) :]
138        # Fernet token base64 解码后第一字节恒为 0x80（编码后开头为 "g"）
139        if not token.startswith("g"):
140            return False
141        # 重新填充 base64 并解码长度校验，避免误判纯字符串
142        import base64
143
144        padding = "=" * (-len(token) % 4)
145        try:
146            decoded = base64.urlsafe_b64decode(token + padding)
147        except ValueError:
148            return False
149        # 最小密文长度 = version(1) + timestamp(8) + IV(16) + HMAC(32)
150        return len(decoded) >= 57

检查字符串是否为加密格式

采用双重校验降低格式碰撞风险：

前缀匹配 $encrypted$
token 部分必须是 Fernet 格式（version byte 0x80，base64 编码后以 g 开头，且解码后总长度至少 57 字节 = 1 + 8 + 16 + 32）

这避免了与真实密码恰好以 $encrypted$ 开头时被误判的场景：真实密码很少以 $encrypted$g 开头，且 base64 解码长度必须 >= 57 字节。

Attention: 本函数不验证 HMAC/签名，仅为廉价标识符检测；完整合法性校验在 decrypt() 中完成。

def setup_logging( level: str = 'INFO', log_file: str | None = None, max_bytes: int = 10485760, backup_count: int = 5, structured: bool = False, verbose: bool = False) -> None: View Source

101def setup_logging(
102    level: str = "INFO",
103    log_file: Optional[str] = None,
104    max_bytes: int = 10 * 1024 * 1024,  # 10MB
105    backup_count: int = 5,
106    structured: bool = False,
107    verbose: bool = False,
108) -> None:
109    """
110    配置统一日志系统
111
112    Args:
113        level: 日志级别 (DEBUG/INFO/WARNING/ERROR/CRITICAL)
114        log_file: 日志文件路径（可选，不指定时只输出到控制台）
115        max_bytes: 单个日志文件最大字节数
116        backup_count: 保留的轮转文件数
117        structured: 是否使用 JSON 结构化格式
118        verbose: 是否启用详细格式
119    """
120    root_logger = logging.getLogger()
121    root_logger.setLevel(getattr(logging, level.upper(), logging.INFO))
122
123    # 清除已有处理器
124    root_logger.handlers.clear()
125
126    # 创建格式化器
127    if structured:
128        formatter = logging.Formatter(STRUCTURED_FORMAT)
129    elif verbose:
130        formatter = logging.Formatter(VERBOSE_FORMAT)
131    else:
132        formatter = logging.Formatter(DEFAULT_FORMAT)
133
134    # 添加敏感数据过滤器
135    sensitive_filter = SensitiveDataFilter()
136
137    # 控制台输出
138    console_handler = logging.StreamHandler(sys.stderr)
139    console_handler.setFormatter(formatter)
140    console_handler.addFilter(sensitive_filter)
141    root_logger.addHandler(console_handler)
142
143    # 文件输出（可选）
144    if log_file:
145        log_path = Path(log_file)
146        log_path.parent.mkdir(parents=True, exist_ok=True)
147
148        file_handler = RotatingFileHandler(
149            str(log_path),
150            maxBytes=max_bytes,
151            backupCount=backup_count,
152            encoding="utf-8",
153        )
154        file_handler.setFormatter(formatter)
155        file_handler.addFilter(sensitive_filter)
156        root_logger.addHandler(file_handler)
157
158    # 配置 remote_cmd 包日志
159    logging.getLogger("remote_cmd").setLevel(getattr(logging, level.upper(), logging.INFO))
160
161    logging.getLogger("remote_cmd").debug("日志系统已初始化")

配置统一日志系统

Args: level: 日志级别 (DEBUG/INFO/WARNING/ERROR/CRITICAL) log_file: 日志文件路径（可选，不指定时只输出到控制台） max_bytes: 单个日志文件最大字节数 backup_count: 保留的轮转文件数 structured: 是否使用 JSON 结构化格式 verbose: 是否启用详细格式

class SensitiveDataFilter(logging.Filter): View Source

51class SensitiveDataFilter(logging.Filter):
52    """
53    日志过滤器：自动脱敏敏感数据
54
55    重写 filter 方法，在日志记录发出前替换敏感字段。
56    """
57
58    def filter(self, record: logging.LogRecord) -> bool:
59        if isinstance(record.msg, str):
60            record.msg = redact_sensitive_data(record.msg)
61        if record.args:
62            # 处理 %s 格式的参数
63            cleaned_args = tuple(
64                redact_sensitive_data(str(a)) if isinstance(a, str) else a for a in record.args
65            )
66            record.args = cleaned_args
67        return True
68
69    @staticmethod
70    def redact_dict(data: dict[str, Any]) -> dict[str, Any]:
71        """递归脱敏字典中的敏感字段"""
72        result: dict[str, Any] = {}
73        for key, value in data.items():
74            if key.lower() in SENSITIVE_FIELDS:
75                result[key] = "[REDACTED]"
76            elif isinstance(value, dict):
77                result[key] = SensitiveDataFilter.redact_dict(value)
78            else:
79                result[key] = value
80        return result

日志过滤器：自动脱敏敏感数据

重写 filter 方法，在日志记录发出前替换敏感字段。

def filter(self, record: logging.LogRecord) -> bool: View Source

58    def filter(self, record: logging.LogRecord) -> bool:
59        if isinstance(record.msg, str):
60            record.msg = redact_sensitive_data(record.msg)
61        if record.args:
62            # 处理 %s 格式的参数
63            cleaned_args = tuple(
64                redact_sensitive_data(str(a)) if isinstance(a, str) else a for a in record.args
65            )
66            record.args = cleaned_args
67        return True

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

@staticmethod

def redact_dict(data: dict[str, typing.Any]) -> dict[str, typing.Any]: View Source

69    @staticmethod
70    def redact_dict(data: dict[str, Any]) -> dict[str, Any]:
71        """递归脱敏字典中的敏感字段"""
72        result: dict[str, Any] = {}
73        for key, value in data.items():
74            if key.lower() in SENSITIVE_FIELDS:
75                result[key] = "[REDACTED]"
76            elif isinstance(value, dict):
77                result[key] = SensitiveDataFilter.redact_dict(value)
78            else:
79                result[key] = value
80        return result

递归脱敏字典中的敏感字段

def get_logger(name: str, **context) -> 'logging.Logger | logging.LoggerAdapter': View Source

187def get_logger(name: str, **context) -> "logging.Logger | logging.LoggerAdapter":
188    """
189    获取带可选上下文的日志器
190
191    用法:
192        >>> log = get_logger(__name__, host="web-server")
193        >>> log.info("测试连接")
194    """
195    logger = logging.getLogger(name)
196    if context:
197        return LoggerAdapter(logger, context)
198    return logger

获取带可选上下文的日志器

用法:

log = get_logger(__name__, host="web-server") log.info("测试连接")

class BatchExecutor: View Source

 63class BatchExecutor:
 64    """
 65    批量命令执行器
 66
 67    支持多主机并发执行，带超时控制、失败重试和进度回调。
 68
 69    Args:
 70        host_service: HostService 实例，用于获取主机配置和凭据
 71        max_concurrency: 最大并发数，默认 10
 72        command_timeout: 单个命令超时时间（秒），默认 30
 73        use_async: 是否启用异步内核（基于 asyncssh 的原生异步实现）。默认 False
 74            保持原有 ThreadPoolExecutor 行为；设为 True 时，execute 会在内部使用
 75            asyncio.run 调用 AsyncBatchExecutor 完成并发调度，从而在大规模场景下
 76            降低线程/CPU 开销。注意：启用时调用线程不应已运行 asyncio 事件循环，
 77            否则抛出 RuntimeError（应改用 AsyncBatchExecutor.execute()）。
 78        pool_factory: 外部连接池工厂（可选，v2.1）。提供时执行器从工厂获取
 79            池并复用其连接，**绝不关闭**返回的池（所有权归调用方）；
 80            未提供时多主机或需重试时内部按主机创建 SyncConnectionPool
 81            （use_async=True 时为 AsyncConnectionPool），执行结束后自动关闭。
 82            工厂返回的池类型须与内核匹配（见 PoolFactory 注释）。
 83
 84    连接池所有权约定（与 AsyncBatchExecutor 一致）：
 85
 86    - 外部注入（``pool_factory``）→ 调用方拥有，executor 只借用不关闭；
 87      适合长驻服务跨批次复用连接。
 88    - 内部创建 → executor 拥有，单次 ``execute`` 结束后 ``close_all``；
 89      适合一次性脚本。
 90
 91    Note:
 92        无论 `use_async` 取值，对外 `execute` 始终为同步接口，返回类型一致，
 93        便于上层无差别切换。
 94    """
 95
 96    def __init__(
 97        self,
 98        host_service: HostService,
 99        max_concurrency: int = 10,
100        command_timeout: int = 30,
101        use_async: bool = False,
102        pool_factory: Optional[PoolFactory] = None,
103    ) -> None:
104        if max_concurrency < 1:
105            raise ValueError(f"max_concurrency must be >= 1, got: {max_concurrency}")
106        if command_timeout <= 0:
107            raise ValueError(f"command_timeout must be > 0, got: {command_timeout}")
108        self._host_service = host_service
109        self._max_concurrency = max_concurrency
110        self._command_timeout = command_timeout
111        self._use_async = use_async
112        self._pool_factory = pool_factory
113        # 延迟导入以避免在未安装 asyncssh 的环境下的导入失败
114        # 使用前向引用避免在模块加载期引入 asyncssh 硬依赖（开启 use_async 时才惰性导入）
115        self._async_executor: Optional["AsyncBatchExecutor"] = None  # noqa: UP037
116        if use_async:
117            from remote_cmd.service.async_batch_executor import AsyncBatchExecutor
118
119            self._async_executor = AsyncBatchExecutor(
120                host_service=host_service,
121                max_concurrency=max_concurrency,
122                command_timeout=command_timeout,
123                pool_factory=pool_factory,
124            )
125
126    def execute(
127        self,
128        host_names: list[str],
129        command: str,
130        retry_count: int = 0,
131        retry_delay: float = 1.0,
132        progress_callback: Optional[ProgressCallback] = None,
133    ) -> BatchResult:
134        """
135        在指定主机上批量执行命令
136
137        Args:
138            host_names: 要执行命令的主机名称列表
139            command: 要执行的命令
140            retry_count: 失败重试次数，默认 0（不重试）。仅对瞬态错误
141                （超时/网络中断等）重试；认证、凭据、配置等永久性错误
142                立即失败（分类见 service/retry_policy.py）
143            retry_delay: 重试基础延迟（秒），默认 1.0。实际等待为
144                指数退避 + full jitter：第 n 次失败后等待
145                0 到 retry_delay * 2^n（含端点）内的随机值（上限 60s）
146            progress_callback: 进度回调，参数 (completed, total, current_host_name)。
147                同步内核下回调应为同步函数；异步回调请使用 use_async=True。
148
149        Returns:
150            BatchResult: 批量执行结果
151
152        Raises:
153            ValueError: host_names 为空
154        """
155        if not host_names:
156            raise ValueError("host_names must not be empty")
157        if retry_count < 0:
158            raise ValueError(f"retry_count must be >= 0, got: {retry_count}")
159        if retry_delay < 0:
160            raise ValueError(f"retry_delay must be >= 0, got: {retry_delay}")
161
162        # 去重（保留首次出现顺序）：重复主机名只执行一次，避免 results 覆盖导致
163        # total/success/failed 统计错位
164        host_names = list(dict.fromkeys(host_names))
165
166        # 异步内核委托路径：同步接口 + asyncio.run(异步实现)
167        if self._async_executor is not None:
168            return self._delegate_to_async(
169                host_names, command, retry_count, retry_delay, progress_callback
170            )
171
172        return self._execute_sync(host_names, command, retry_count, retry_delay, progress_callback)
173
174    def _delegate_to_async(
175        self,
176        host_names: list[str],
177        command: str,
178        retry_count: int,
179        retry_delay: float,
180        progress_callback: Optional[ProgressCallback],
181    ) -> BatchResult:
182        """异步内核委托路径：同步接口 + asyncio.run(异步实现)"""
183        # 仅当 _async_executor 已初始化时才进入此路径（见 execute() 的窄化判断）
184        assert self._async_executor is not None
185
186        import asyncio
187
188        # 前置检测运行中的事件循环：asyncio.run() 在事件循环内调用只会抛出
189        # 通用 Python 错误，替换为可操作的项目级错误提示。
190        # 仅此场景抛出，其余 RuntimeError（来自 asyncio.run 本身的其他
191        # 失败原因）不经本分支改写。
192        try:
193            asyncio.get_running_loop()
194        except RuntimeError:
195            pass  # 无运行中的事件循环，asyncio.run 可安全使用
196        else:
197            raise RuntimeError(
198                "BatchExecutor(use_async=True) cannot be used inside a running "
199                "event loop; use AsyncBatchExecutor.execute() directly instead"
200            )
201
202        return asyncio.run(
203            self._async_executor.execute(
204                host_names=host_names,
205                command=command,
206                retry_count=retry_count,
207                retry_delay=retry_delay,
208                progress_callback=progress_callback,
209            )
210        )
211
212    def _execute_sync(
213        self,
214        host_names: list[str],
215        command: str,
216        retry_count: int,
217        retry_delay: float,
218        progress_callback: Optional[ProgressCallback],
219    ) -> BatchResult:
220        """同步路径：ThreadPoolExecutor + 连接池复用"""
221        total = len(host_names)
222        results: dict[str, BatchHostResult] = {}
223        start_time = time.time()
224
225        logger.info(f"batch execution started: {total} hosts, concurrency={self._max_concurrency}")
226
227        pools: dict[str, SyncConnectionPool] = {}
228        # 内部创建的池由本批次负责关闭；外部 pool_factory 提供的池
229        # 所有权归调用方，绝不登记进此列表
230        internal_pools: list[SyncConnectionPool] = []
231
232        try:
233            with ThreadPoolExecutor(max_workers=self._max_concurrency) as executor:
234                future_map = self._submit_tasks(
235                    executor,
236                    host_names,
237                    command,
238                    retry_count,
239                    retry_delay,
240                    pools,
241                    internal_pools,
242                    total,
243                )
244                self._collect_results(
245                    future_map, host_names, command, progress_callback, results, total
246                )
247        finally:
248            # 仅关闭内部创建的池；外部 pool_factory 提供的池所有权归调用方
249            self._cleanup_pools(internal_pools)
250
251        duration = time.time() - start_time
252        return self._build_result(total, results, duration)
253
254    def _submit_tasks(
255        self,
256        executor: ThreadPoolExecutor,
257        host_names: list[str],
258        command: str,
259        retry_count: int,
260        retry_delay: float,
261        pools: dict[str, SyncConnectionPool],
262        internal_pools: list[SyncConnectionPool],
263        total: int,
264    ) -> dict:
265        """提交任务到线程池，返回 future_map"""
266        future_map = {}
267        for host_name in host_names:
268            # 连接池：外部注入时始终启用；否则多主机或需重试时创建
269            pool: Optional[SyncConnectionPool] = None
270            if self._pool_factory is not None or retry_count > 0 or total > 1:
271                pool = self._prepare_pool(host_name, pools, internal_pools)
272            future = executor.submit(
273                self._execute_on_host, host_name, command, retry_count, retry_delay, pool
274            )
275            future_map[future] = host_name
276        return future_map
277
278    def _prepare_pool(
279        self,
280        host_name: str,
281        pools: dict[str, SyncConnectionPool],
282        internal_pools: list[SyncConnectionPool],
283    ) -> Optional[SyncConnectionPool]:
284        """为指定主机创建或获取连接池
285
286        主机解析失败时返回 None 而非上抛：保持 execute 的
287        "未知主机 → BatchHostResult 错误条目" 契约（由
288        _execute_on_host 的 resolve_host_or_error 记录失败详情），
289        避免整个批次因单个坏主机以异常收场。
290
291        池所有权：外部 ``pool_factory`` 提供的池绝不登记进
292        internal_pools（executor 不负责关闭）；内部创建的池登记后
293        由 _cleanup_pools 统一 close_all。
294        """
295        if host_name in pools:
296            return pools[host_name]
297
298        try:
299            host = self._host_service.resolve_host(host_name)
300        except Exception as e:  # noqa: BLE001
301            logger.debug(f"pool preparation skipped for {host_name}: {e}")
302            return None
303        config = build_connection_config(host, self._command_timeout)
304        if self._pool_factory is not None:
305            pool = self._pool_factory(config)
306            pools[host_name] = pool
307            return pool
308        pool = SyncConnectionPool(
309            config,
310            max_connections=max(1, self._max_concurrency),
311            client_factory=SSHClient,
312        )
313        pools[host_name] = pool
314        internal_pools.append(pool)
315        return pool
316
317    def _collect_results(
318        self,
319        future_map: dict,
320        host_names: list[str],
321        command: str,
322        progress_callback: Optional[ProgressCallback],
323        results: dict[str, BatchHostResult],
324        total: int,
325    ) -> int:
326        """收集结果并处理进度回调与中断"""
327        completed = 0
328        try:
329            for future in as_completed(future_map):
330                host_name = future_map[future]
331                result = self._process_future_result(future, host_name, command)
332                results[host_name] = result
333                completed += 1
334                self._invoke_progress_callback(
335                    progress_callback, completed, total, host_name, result
336                )
337        except KeyboardInterrupt:
338            logger.warning("batch execution interrupted by user")
339            self._handle_interrupt(future_map, host_names, command, results)
340            completed = len(results)
341
342        return completed
343
344    def _process_future_result(self, future, host_name: str, command: str) -> BatchHostResult:
345        """处理单个 future 结果，捕获调度异常"""
346        try:
347            return future.result()
348        except Exception as e:  # noqa: BLE001
349            return BatchHostResult(
350                host=host_name,
351                success=False,
352                command=command,
353                error=f"scheduling error: {e}",
354            )
355
356    def _invoke_progress_callback(
357        self,
358        progress_callback: Optional[ProgressCallback],
359        completed: int,
360        total: int,
361        host_name: str,
362        result: BatchHostResult,
363    ) -> None:
364        """调用进度回调并记录日志"""
365        if progress_callback:
366            rv = progress_callback(completed, total, host_name)
367            if isinstance(rv, Coroutine):
368                logger.warning("同步内核不支持异步进度回调，请使用 use_async=True")
369                # 显式关闭未 await 的协程，避免 RuntimeWarning 与资源泄漏
370                rv.close()
371
372        logger.debug(
373            f"[{completed}/{total}] {host_name}: "
374            f"{'✓' if result.success else '✗'} "
375            f"({result.duration:.1f}s)"
376        )
377
378    def _handle_interrupt(
379        self,
380        future_map: dict,
381        host_names: list[str],
382        command: str,
383        results: dict[str, BatchHostResult],
384    ) -> None:
385        """处理键盘中断：取消任务并为未完成主机创建失败记录"""
386        # 取消所有未完成的任务
387        for future in future_map:
388            future.cancel()
389
390        # 为尚未有结果的主机创建失败记录
391        for host_name in host_names:
392            if host_name not in results:
393                results[host_name] = BatchHostResult(
394                    host=host_name,
395                    success=False,
396                    command=command,
397                    error="user interrupted",
398                )
399
400    def _cleanup_pools(self, internal_pools: list[SyncConnectionPool]) -> None:
401        """关闭本批次内部创建的所有连接池（外部提供的池绝不关闭）"""
402        for pool in internal_pools:
403            pool.close_all()
404
405    def _build_result(
406        self, total: int, results: dict[str, BatchHostResult], duration: float
407    ) -> BatchResult:
408        """构建批量执行汇总结果"""
409        success_count = sum(1 for r in results.values() if r.success)
410        failed_count = total - success_count
411
412        logger.info(
413            f"batch execution finished: {success_count}/{total} succeeded, took {duration:.1f}s"
414        )
415
416        return BatchResult(
417            total=total,
418            success=success_count,
419            failed=failed_count,
420            duration=duration,
421            results=results,
422        )
423
424    def _execute_on_host(
425        self,
426        host_name: str,
427        command: str,
428        retry_count: int,
429        retry_delay: float,
430        pool: Optional[SyncConnectionPool] = None,
431    ) -> BatchHostResult:
432        """
433        在单台主机上执行命令（包含重试逻辑）
434
435        Args:
436            host_name: 主机名称
437            command: 要执行的命令
438            retry_count: 重试次数
439            retry_delay: 重试间隔
440
441        Returns:
442            BatchHostResult: 单台主机的执行结果
443        """
444        # 解析主机配置（包括凭据解密）；失败返回错误结果
445        outcome = resolve_host_or_error(self._host_service, host_name, command)
446        if isinstance(outcome, BatchHostResult):
447            return outcome
448        host: Host = outcome
449
450        last_error: Optional[str] = None
451        last_duration = 0.0
452
453        for attempt in range(retry_count + 1):
454            start = time.time()
455            try:
456                config = build_connection_config(host, self._command_timeout)
457
458                if pool is not None:
459                    # 连接池模式：复用主机连接，避免每次操作握手
460                    with pool.acquire_context() as client:
461                        cmd_result = client.execute(command, timeout=self._command_timeout)
462                    return to_host_result(host_name, command, cmd_result, time.time() - start)
463
464                # 非连接池路径：try/finally 确保即使 execute() 抛异常，
465                # disconnect() 也会执行，避免 SSH 连接泄漏
466                client = SSHClient(config)
467                try:
468                    client.connect()
469                    cmd_result = client.execute(command, timeout=self._command_timeout)
470                finally:
471                    client.disconnect()
472
473                return to_host_result(host_name, command, cmd_result, time.time() - start)
474
475            except Exception as e:  # noqa: BLE001
476                duration = time.time() - start
477                last_error = str(e)
478                last_duration = duration
479                logger.debug(f"attempt {attempt + 1}/{retry_count + 1} failed for {host_name}: {e}")
480
481                # 已是最后一次尝试，或异常为永久性（认证/凭据/配置错误等），
482                # 立即放弃重试——详见 service/retry_policy.py 的分类契约
483                if attempt >= retry_count:
484                    break
485                if not is_retryable(e):
486                    logger.debug(f"non-retryable error for {host_name}, giving up: {e}")
487                    break
488
489                # 指数退避 + full jitter（避免多主机同步重试的惊群）
490                delay = compute_backoff_delay(attempt, retry_delay)
491                time.sleep(delay)
492
493        # 所有重试都失败
494        return BatchHostResult(
495            host=host_name,
496            success=False,
497            command=command,
498            error=last_error,
499            duration=last_duration,
500        )

批量命令执行器

支持多主机并发执行，带超时控制、失败重试和进度回调。

Args: host_service: HostService 实例，用于获取主机配置和凭据 max_concurrency: 最大并发数，默认 10 command_timeout: 单个命令超时时间（秒），默认 30 use_async: 是否启用异步内核（基于 asyncssh 的原生异步实现）。默认 False 保持原有 ThreadPoolExecutor 行为；设为 True 时，execute 会在内部使用 asyncio.run 调用 AsyncBatchExecutor 完成并发调度，从而在大规模场景下降低线程/CPU 开销。注意：启用时调用线程不应已运行 asyncio 事件循环，否则抛出 RuntimeError（应改用 AsyncBatchExecutor.execute()）。 pool_factory: 外部连接池工厂（可选，v2.1）。提供时执行器从工厂获取池并复用其连接，**绝不关闭**返回的池（所有权归调用方）；未提供时多主机或需重试时内部按主机创建 SyncConnectionPool （use_async=True 时为 AsyncConnectionPool），执行结束后自动关闭。工厂返回的池类型须与内核匹配（见 PoolFactory 注释）。

连接池所有权约定（与 AsyncBatchExecutor 一致）：

外部注入（pool_factory）→ 调用方拥有，executor 只借用不关闭；适合长驻服务跨批次复用连接。
内部创建 → executor 拥有，单次 execute 结束后 close_all；适合一次性脚本。

Note: 无论 use_async 取值，对外 execute 始终为同步接口，返回类型一致，便于上层无差别切换。

BatchExecutor( host_service: HostService, max_concurrency: int = 10, command_timeout: int = 30, use_async: bool = False, pool_factory: Callable[[remote_cmd.core.ssh_client.ConnectionConfig], Any] | None = None) View Source

 96    def __init__(
 97        self,
 98        host_service: HostService,
 99        max_concurrency: int = 10,
100        command_timeout: int = 30,
101        use_async: bool = False,
102        pool_factory: Optional[PoolFactory] = None,
103    ) -> None:
104        if max_concurrency < 1:
105            raise ValueError(f"max_concurrency must be >= 1, got: {max_concurrency}")
106        if command_timeout <= 0:
107            raise ValueError(f"command_timeout must be > 0, got: {command_timeout}")
108        self._host_service = host_service
109        self._max_concurrency = max_concurrency
110        self._command_timeout = command_timeout
111        self._use_async = use_async
112        self._pool_factory = pool_factory
113        # 延迟导入以避免在未安装 asyncssh 的环境下的导入失败
114        # 使用前向引用避免在模块加载期引入 asyncssh 硬依赖（开启 use_async 时才惰性导入）
115        self._async_executor: Optional["AsyncBatchExecutor"] = None  # noqa: UP037
116        if use_async:
117            from remote_cmd.service.async_batch_executor import AsyncBatchExecutor
118
119            self._async_executor = AsyncBatchExecutor(
120                host_service=host_service,
121                max_concurrency=max_concurrency,
122                command_timeout=command_timeout,
123                pool_factory=pool_factory,
124            )

def execute( self, host_names: list[str], command: str, retry_count: int = 0, retry_delay: float = 1.0, progress_callback: Callable[[int, int, str], Awaitable[None] | None] | None = None) -> BatchResult: View Source

126    def execute(
127        self,
128        host_names: list[str],
129        command: str,
130        retry_count: int = 0,
131        retry_delay: float = 1.0,
132        progress_callback: Optional[ProgressCallback] = None,
133    ) -> BatchResult:
134        """
135        在指定主机上批量执行命令
136
137        Args:
138            host_names: 要执行命令的主机名称列表
139            command: 要执行的命令
140            retry_count: 失败重试次数，默认 0（不重试）。仅对瞬态错误
141                （超时/网络中断等）重试；认证、凭据、配置等永久性错误
142                立即失败（分类见 service/retry_policy.py）
143            retry_delay: 重试基础延迟（秒），默认 1.0。实际等待为
144                指数退避 + full jitter：第 n 次失败后等待
145                0 到 retry_delay * 2^n（含端点）内的随机值（上限 60s）
146            progress_callback: 进度回调，参数 (completed, total, current_host_name)。
147                同步内核下回调应为同步函数；异步回调请使用 use_async=True。
148
149        Returns:
150            BatchResult: 批量执行结果
151
152        Raises:
153            ValueError: host_names 为空
154        """
155        if not host_names:
156            raise ValueError("host_names must not be empty")
157        if retry_count < 0:
158            raise ValueError(f"retry_count must be >= 0, got: {retry_count}")
159        if retry_delay < 0:
160            raise ValueError(f"retry_delay must be >= 0, got: {retry_delay}")
161
162        # 去重（保留首次出现顺序）：重复主机名只执行一次，避免 results 覆盖导致
163        # total/success/failed 统计错位
164        host_names = list(dict.fromkeys(host_names))
165
166        # 异步内核委托路径：同步接口 + asyncio.run(异步实现)
167        if self._async_executor is not None:
168            return self._delegate_to_async(
169                host_names, command, retry_count, retry_delay, progress_callback
170            )
171
172        return self._execute_sync(host_names, command, retry_count, retry_delay, progress_callback)

在指定主机上批量执行命令

Args: host_names: 要执行命令的主机名称列表 command: 要执行的命令 retry_count: 失败重试次数，默认 0（不重试）。仅对瞬态错误（超时/网络中断等）重试；认证、凭据、配置等永久性错误立即失败（分类见 service/retry_policy.py） retry_delay: 重试基础延迟（秒），默认 1.0。实际等待为指数退避 + full jitter：第 n 次失败后等待 0 到 retry_delay * 2^n（含端点）内的随机值（上限 60s） progress_callback: 进度回调，参数 (completed, total, current_host_name)。同步内核下回调应为同步函数；异步回调请使用 use_async=True。

Returns: BatchResult: 批量执行结果

Raises: ValueError: host_names 为空

@dataclass

class BatchResult: View Source

47@dataclass
48class BatchResult:
49    """
50    批量执行汇总结果
51
52    Attributes:
53        total: 总主机数
54        success: number of succeeded hosts
55        failed: 失败主机数
56        duration: total took (seconds)
57        results: 按主机名索引的详细结果
58    """
59
60    total: int
61    success: int
62    failed: int
63    duration: float
64    results: dict[str, BatchHostResult] = field(default_factory=dict)
65
66    @property
67    def success_rate(self) -> float:
68        """success rate (0.0 ~ 1.0)"""
69        if self.total == 0:
70            return 1.0
71        return self.success / self.total
72
73    @property
74    def failed_hosts(self) -> list[str]:
75        """失败主机列表"""
76        return [h for h, r in self.results.items() if not r.success]
77
78    @property
79    def success_hosts(self) -> list[str]:
80        """list of succeeded hosts"""
81        return [h for h, r in self.results.items() if r.success]
82
83    def summary(self) -> str:
84        """生成可读的汇总字符串"""
85        return (
86            f"Total: {self.total}, "
87            f"Succeeded: {self.success}, "
88            f"Failed: {self.failed}, "
89            f"Duration: {self.duration:.1f}s, "
90            f"Success rate: {self.success_rate:.1%}"
91        )

批量执行汇总结果

Attributes: total: 总主机数 success: number of succeeded hosts failed: 失败主机数 duration: total took (seconds) results: 按主机名索引的详细结果

BatchResult( total: int, success: int, failed: int, duration: float, results: dict[str, BatchHostResult] = <factory>)

total: int

success: int

failed: int

duration: float

results: dict[str, BatchHostResult]

success_rate: float View Source

66    @property
67    def success_rate(self) -> float:
68        """success rate (0.0 ~ 1.0)"""
69        if self.total == 0:
70            return 1.0
71        return self.success / self.total

success rate (0.0 ~ 1.0)

failed_hosts: list[str] View Source

73    @property
74    def failed_hosts(self) -> list[str]:
75        """失败主机列表"""
76        return [h for h, r in self.results.items() if not r.success]

失败主机列表

success_hosts: list[str] View Source

78    @property
79    def success_hosts(self) -> list[str]:
80        """list of succeeded hosts"""
81        return [h for h, r in self.results.items() if r.success]

list of succeeded hosts

def summary(self) -> str: View Source

83    def summary(self) -> str:
84        """生成可读的汇总字符串"""
85        return (
86            f"Total: {self.total}, "
87            f"Succeeded: {self.success}, "
88            f"Failed: {self.failed}, "
89            f"Duration: {self.duration:.1f}s, "
90            f"Success rate: {self.success_rate:.1%}"
91        )

生成可读的汇总字符串

@dataclass

class BatchHostResult: View Source

21@dataclass
22class BatchHostResult:
23    """
24    单个主机的批量执行结果
25
26    Attributes:
27        host: 主机名称
28        success: whether the command succeeded
29        command: 执行的命令
30        stdout: 标准输出
31        stderr: 标准错误
32        exit_code: 退出码
33        duration: execution took (seconds)
34        error: 错误信息（如果有）
35    """
36
37    host: str
38    success: bool
39    command: str
40    stdout: str = ""
41    stderr: str = ""
42    exit_code: int = -1
43    duration: float = 0.0
44    error: Optional[str] = None

单个主机的批量执行结果

Attributes: host: 主机名称 success: whether the command succeeded command: 执行的命令 stdout: 标准输出 stderr: 标准错误 exit_code: 退出码 duration: execution took (seconds) error: 错误信息（如果有）

BatchHostResult( host: str, success: bool, command: str, stdout: str = '', stderr: str = '', exit_code: int = -1, duration: float = 0.0, error: str | None = None)

host: str

success: bool

command: str

stdout: str = ''

stderr: str = ''

exit_code: int = -1

duration: float = 0.0

error: str | None = None

class SyncConnectionPool: View Source

 34class SyncConnectionPool:
 35    """同步 SSH 连接池。
 36
 37    Args:
 38        config: 用于建立 SSH 连接的配置
 39        max_connections: 最大连接数（同一配置可复用）
 40        max_lifetime: 连接最大生命周期（秒），超过自动关闭
 41        idle_timeout: 空闲超时（秒），超过自动关闭
 42        health_check_interval: 后台清理线程周期（秒）
 43    """
 44
 45    def __init__(
 46        self,
 47        config: ConnectionConfig,
 48        max_connections: int = 10,
 49        max_lifetime: int = 3600,
 50        idle_timeout: int = 300,
 51        health_check_interval: int = 60,
 52        client_factory: Optional[Any] = None,
 53    ) -> None:
 54        self.config = config
 55        self._max = max_connections
 56        self._max_lifetime = max_lifetime
 57        self._idle_timeout = idle_timeout
 58        self._health_check_interval = health_check_interval
 59        # 客户端工厂：默认为 SSHClient；测试可注入 mock
 60        self._client_factory = client_factory or SSHClient
 61
 62        # 容器
 63        self._connections: list[SSHClient] = []
 64        self._free: queue.Queue[SSHClient] = queue.Queue()
 65        self._semaphore = threading.Semaphore(max_connections)
 66        self._lock = threading.Lock()
 67
 68        # 生命周期状态：close_all() 后置 True，禁止再借用/归还
 69        self._closed = False
 70
 71        # 指标
 72        self._total_created = 0
 73        self._total_reconnects = 0
 74        self._total_failed = 0
 75        self._total_released = 0
 76
 77        # 后台清理线程
 78        self._monitor_thread: Optional[threading.Thread] = None
 79        self._stop_event = threading.Event()
 80
 81        # 连接元数据（副表，避免侵入 SSHClient 私有属性）
 82        self._meta: dict[int, ConnectionMeta] = {}
 83
 84    # ------------------------------------------------------------------
 85    # 指标
 86    # ------------------------------------------------------------------
 87    def get_metrics(self) -> dict[str, Any]:
 88        """获取连接池指标快照。"""
 89        return {
 90            # 当前在用的连接数 = 存活连接总数 - 空闲连接数。
 91            # 不能用 total_created - total_released：复用连接时
 92            # total_released 会超过 total_created，导致 active 为负。
 93            "active": len(self._connections) - self._free.qsize(),
 94            "idle": self._free.qsize(),
 95            "total_connections": len(self._connections),
 96            "total_created": self._total_created,
 97            "reconnects": self._total_reconnects,
 98            "failed": self._total_failed,
 99            "max_connections": self._max,
100            "max_lifetime": self._max_lifetime,
101            "idle_timeout": self._idle_timeout,
102        }
103
104    # ------------------------------------------------------------------
105    # 获取 / 释放
106    # ------------------------------------------------------------------
107    def acquire(self) -> SSHClient:
108        """从池中获取一个可用连接，必要时创建新连接。
109
110        Returns:
111            SSHClient: 可用的同步客户端
112
113        Raises:
114            SSHConnectionError: 创建连接失败
115            RuntimeError: 连接池已关闭（close_all 之后）
116        """
117        if self._closed:
118            raise RuntimeError("connection pool is closed")
119        self._semaphore.acquire()
120        # 竞态守卫：等待信号量期间 close_all() 可能已完成——
121        # 取得槽位后必须复查，已关闭则归还槽位并抛出既有错误，
122        # 否则会向调用方发放来自已关闭池的连接
123        if self._closed:
124            self._semaphore.release()
125            raise RuntimeError("connection pool is closed")
126        try:
127            # 优先复用空闲连接
128            while not self._free.empty():
129                conn = self._free.get_nowait()
130                if self._check_connection(conn):
131                    self._touch(conn)
132                    return conn
133                self._close_connection(conn)
134
135            # 创建新连接（信号量已保证未超额）
136            return self._create_connection()
137        except BaseException:
138            self._semaphore.release()
139            raise
140
141    def release(self, conn: Optional[SSHClient]) -> None:
142        """归还连接到池中（如已断开/超时则关闭）。"""
143        if conn is None:
144            return
145        # 池已关闭：不把连接放回空闲队列（避免游离连接），直接关闭并释放槽位
146        if self._closed:
147            self._close_connection(conn)
148            self._semaphore.release()
149            self._total_released += 1
150            return
151        meta = self._meta.get(id(conn))
152        if meta is not None:
153            meta.last_used = time.time()
154
155        if not conn.is_connected():
156            self._close_connection(conn)
157            self._semaphore.release()
158            return
159
160        # 生命周期 / 空闲超时则关闭
161        if meta and should_close(meta, self._max_lifetime, self._idle_timeout, True):
162            self._close_connection(conn)
163            self._semaphore.release()
164            return
165
166        try:
167            with self._lock:
168                self._free.put_nowait(conn)
169            # 放回 free 后释放许可：free 中的连接不再占用并发槽位，
170            # 后续 acquire 会从 free 直接复用（无需再次获取许可）
171            self._semaphore.release()
172        except queue.Full:
173            self._close_connection(conn)
174            self._semaphore.release()
175        finally:
176            self._total_released += 1
177
178    # ------------------------------------------------------------------
179    # 内部
180    # ------------------------------------------------------------------
181    def _create_connection(self) -> SSHClient:
182        client = self._client_factory(self.config)
183        try:
184            client.connect()
185        except Exception:  # noqa: BLE001
186            # 信号量由 acquire() 的 except 统一释放，此处不再释放
187            self._total_failed += 1
188            raise
189        with self._lock:
190            self._connections.append(client)
191        now = time.time()
192        self._meta[id(client)] = ConnectionMeta(
193            created_at=now,
194            last_used=now,
195            conn_id=uuid.uuid4().hex,
196        )
197        self._total_created += 1
198        return client
199
200    def _touch(self, conn: SSHClient) -> None:
201        meta = self._meta.get(id(conn))
202        if meta is not None:
203            meta.last_used = time.time()
204
205    def _check_connection(self, conn: SSHClient) -> bool:
206        if not conn.is_connected():
207            return False
208        meta = self._meta.get(id(conn))
209        if meta is None:
210            return True
211        if lifetime_expired(meta.created_at, self._max_lifetime):
212            logger.debug("connection %s exceeded max lifetime", meta.conn_id[:8])
213            return False
214        # 连接刚使用过（空闲未超时）则信任其状态，避免频繁探活开销
215        if not idle_expired(meta.last_used, self._idle_timeout):
216            return True
217        # 空闲较久才触发轻量探活：发出一个无害命令
218        try:
219            result = conn.execute("true", timeout=5)
220            return result.success
221        except Exception as e:  # noqa: BLE001
222            self._total_reconnects += 1
223            logger.debug("connection liveness check failed: %s", e)
224            return False
225
226    def _close_connection(self, conn: SSHClient) -> None:
227        with contextlib.suppress(Exception):
228            conn.disconnect()
229        self._meta.pop(id(conn), None)
230        with self._lock:
231            if conn in self._connections:
232                self._connections.remove(conn)
233
234    # ------------------------------------------------------------------
235    # 后台监控
236    # ------------------------------------------------------------------
237    def start_monitor(self) -> None:
238        """启动后台清理线程（幂等）。"""
239        if self._monitor_thread is not None and self._monitor_thread.is_alive():
240            return
241        self._stop_event.clear()
242        self._monitor_thread = threading.Thread(
243            target=self._monitor_loop,
244            name="sync-connection-pool-monitor",
245            daemon=True,
246        )
247        self._monitor_thread.start()
248
249    def stop_monitor(self) -> None:
250        """停止后台清理线程。"""
251        self._stop_event.set()
252        if self._monitor_thread and self._monitor_thread.is_alive():
253            self._monitor_thread.join(timeout=2.0)
254
255    def _monitor_loop(self) -> None:
256        while not self._stop_event.is_set():
257            try:
258                self._stop_event.wait(self._health_check_interval)
259                if self._stop_event.is_set():
260                    break
261                self._cleanup_expired()
262            except Exception:  # noqa: BLE001
263                logger.warning("connection pool monitor error")
264
265    def _cleanup_expired(self) -> None:
266        now = time.time()
267        # 在锁保护下排空 _free 快照，绝不替换队列对象。
268        # 旧实现 self._free = kept 会替换队列对象，在替换窗口内 release()
269        # 可能 put 到旧队列导致连接泄漏（未关闭、信号量已释放、池中不可见）。
270        with self._lock:
271            snapshot: list[SSHClient] = []
272            while not self._free.empty():
273                snapshot.append(self._free.get_nowait())
274        keep: list[SSHClient] = []
275        for conn in snapshot:
276            meta = self._meta.get(id(conn))
277            if should_close(meta, self._max_lifetime, self._idle_timeout, conn.is_connected(), now):
278                self._close_connection(conn)
279                continue
280            keep.append(conn)
281        # 将存活连接放回同一队列对象
282        with self._lock:
283            for conn in keep:
284                self._free.put_nowait(conn)
285
286    # ------------------------------------------------------------------
287    # 上下文管理
288    # ------------------------------------------------------------------
289    def acquire_context(self) -> "_AcquireContext":
290        """获取连接的上下文管理器。"""
291        return SyncConnectionPool._AcquireContext(self)
292
293    class _AcquireContext:
294        def __init__(self, pool: "SyncConnectionPool") -> None:
295            self._pool = pool
296            self._conn: Optional[SSHClient] = None
297
298        def __enter__(self) -> SSHClient:
299            self._conn = self._pool.acquire()
300            return self._conn
301
302        def __exit__(self, exc_type, exc, tb) -> None:
303            self._pool.release(self._conn)
304            self._conn = None
305
306    def close_all(self) -> None:
307        """关闭池中所有连接并停止监控。"""
308        self._closed = True
309        self.stop_monitor()
310        with self._lock:
311            conns = list(self._connections)
312        for conn in conns:
313            self._close_connection(conn)
314        # 释放所有信号量
315        while not self._free.empty():
316            self._free.get_nowait()
317
318    def __enter__(self) -> "SyncConnectionPool":
319        self.start_monitor()
320        return self
321
322    def __exit__(self, exc_type, exc, tb) -> None:
323        self.close_all()

同步 SSH 连接池。

Args: config: 用于建立 SSH 连接的配置 max_connections: 最大连接数（同一配置可复用） max_lifetime: 连接最大生命周期（秒），超过自动关闭 idle_timeout: 空闲超时（秒），超过自动关闭 health_check_interval: 后台清理线程周期（秒）

SyncConnectionPool( config: remote_cmd.core.ssh_client.ConnectionConfig, max_connections: int = 10, max_lifetime: int = 3600, idle_timeout: int = 300, health_check_interval: int = 60, client_factory: Any | None = None) View Source

45    def __init__(
46        self,
47        config: ConnectionConfig,
48        max_connections: int = 10,
49        max_lifetime: int = 3600,
50        idle_timeout: int = 300,
51        health_check_interval: int = 60,
52        client_factory: Optional[Any] = None,
53    ) -> None:
54        self.config = config
55        self._max = max_connections
56        self._max_lifetime = max_lifetime
57        self._idle_timeout = idle_timeout
58        self._health_check_interval = health_check_interval
59        # 客户端工厂：默认为 SSHClient；测试可注入 mock
60        self._client_factory = client_factory or SSHClient
61
62        # 容器
63        self._connections: list[SSHClient] = []
64        self._free: queue.Queue[SSHClient] = queue.Queue()
65        self._semaphore = threading.Semaphore(max_connections)
66        self._lock = threading.Lock()
67
68        # 生命周期状态：close_all() 后置 True，禁止再借用/归还
69        self._closed = False
70
71        # 指标
72        self._total_created = 0
73        self._total_reconnects = 0
74        self._total_failed = 0
75        self._total_released = 0
76
77        # 后台清理线程
78        self._monitor_thread: Optional[threading.Thread] = None
79        self._stop_event = threading.Event()
80
81        # 连接元数据（副表，避免侵入 SSHClient 私有属性）
82        self._meta: dict[int, ConnectionMeta] = {}

config

def get_metrics(self) -> dict[str, typing.Any]: View Source

 87    def get_metrics(self) -> dict[str, Any]:
 88        """获取连接池指标快照。"""
 89        return {
 90            # 当前在用的连接数 = 存活连接总数 - 空闲连接数。
 91            # 不能用 total_created - total_released：复用连接时
 92            # total_released 会超过 total_created，导致 active 为负。
 93            "active": len(self._connections) - self._free.qsize(),
 94            "idle": self._free.qsize(),
 95            "total_connections": len(self._connections),
 96            "total_created": self._total_created,
 97            "reconnects": self._total_reconnects,
 98            "failed": self._total_failed,
 99            "max_connections": self._max,
100            "max_lifetime": self._max_lifetime,
101            "idle_timeout": self._idle_timeout,
102        }

获取连接池指标快照。

def acquire(self) -> SSHClient: View Source

107    def acquire(self) -> SSHClient:
108        """从池中获取一个可用连接，必要时创建新连接。
109
110        Returns:
111            SSHClient: 可用的同步客户端
112
113        Raises:
114            SSHConnectionError: 创建连接失败
115            RuntimeError: 连接池已关闭（close_all 之后）
116        """
117        if self._closed:
118            raise RuntimeError("connection pool is closed")
119        self._semaphore.acquire()
120        # 竞态守卫：等待信号量期间 close_all() 可能已完成——
121        # 取得槽位后必须复查，已关闭则归还槽位并抛出既有错误，
122        # 否则会向调用方发放来自已关闭池的连接
123        if self._closed:
124            self._semaphore.release()
125            raise RuntimeError("connection pool is closed")
126        try:
127            # 优先复用空闲连接
128            while not self._free.empty():
129                conn = self._free.get_nowait()
130                if self._check_connection(conn):
131                    self._touch(conn)
132                    return conn
133                self._close_connection(conn)
134
135            # 创建新连接（信号量已保证未超额）
136            return self._create_connection()
137        except BaseException:
138            self._semaphore.release()
139            raise

从池中获取一个可用连接，必要时创建新连接。

Returns: SSHClient: 可用的同步客户端

Raises: SSHConnectionError: 创建连接失败 RuntimeError: 连接池已关闭（close_all 之后）

def release(self, conn: SSHClient | None) -> None: View Source

141    def release(self, conn: Optional[SSHClient]) -> None:
142        """归还连接到池中（如已断开/超时则关闭）。"""
143        if conn is None:
144            return
145        # 池已关闭：不把连接放回空闲队列（避免游离连接），直接关闭并释放槽位
146        if self._closed:
147            self._close_connection(conn)
148            self._semaphore.release()
149            self._total_released += 1
150            return
151        meta = self._meta.get(id(conn))
152        if meta is not None:
153            meta.last_used = time.time()
154
155        if not conn.is_connected():
156            self._close_connection(conn)
157            self._semaphore.release()
158            return
159
160        # 生命周期 / 空闲超时则关闭
161        if meta and should_close(meta, self._max_lifetime, self._idle_timeout, True):
162            self._close_connection(conn)
163            self._semaphore.release()
164            return
165
166        try:
167            with self._lock:
168                self._free.put_nowait(conn)
169            # 放回 free 后释放许可：free 中的连接不再占用并发槽位，
170            # 后续 acquire 会从 free 直接复用（无需再次获取许可）
171            self._semaphore.release()
172        except queue.Full:
173            self._close_connection(conn)
174            self._semaphore.release()
175        finally:
176            self._total_released += 1

归还连接到池中（如已断开/超时则关闭）。

def start_monitor(self) -> None: View Source

237    def start_monitor(self) -> None:
238        """启动后台清理线程（幂等）。"""
239        if self._monitor_thread is not None and self._monitor_thread.is_alive():
240            return
241        self._stop_event.clear()
242        self._monitor_thread = threading.Thread(
243            target=self._monitor_loop,
244            name="sync-connection-pool-monitor",
245            daemon=True,
246        )
247        self._monitor_thread.start()

启动后台清理线程（幂等）。

def stop_monitor(self) -> None: View Source

249    def stop_monitor(self) -> None:
250        """停止后台清理线程。"""
251        self._stop_event.set()
252        if self._monitor_thread and self._monitor_thread.is_alive():
253            self._monitor_thread.join(timeout=2.0)

停止后台清理线程。

def acquire_context(self) -> '_AcquireContext': View Source

289    def acquire_context(self) -> "_AcquireContext":
290        """获取连接的上下文管理器。"""
291        return SyncConnectionPool._AcquireContext(self)

获取连接的上下文管理器。

def close_all(self) -> None: View Source

306    def close_all(self) -> None:
307        """关闭池中所有连接并停止监控。"""
308        self._closed = True
309        self.stop_monitor()
310        with self._lock:
311            conns = list(self._connections)
312        for conn in conns:
313            self._close_connection(conn)
314        # 释放所有信号量
315        while not self._free.empty():
316            self._free.get_nowait()

关闭池中所有连接并停止监控。

@dataclass

class Task: View Source

47@dataclass
48class Task:
49    """
50    任务数据类
51
52    Attributes:
53        id: 任务 ID（UUID）
54        name: 任务名称
55        status: 任务状态
56        created_at: 创建时间
57        started_at: 开始时间
58        completed_at: 完成时间
59        result: 任务结果
60        error: 错误信息
61        metadata: 附加元数据
62    """
63
64    id: str
65    name: str
66    status: TaskStatus
67    created_at: datetime
68    started_at: Optional[datetime] = None
69    completed_at: Optional[datetime] = None
70    result: Any = None
71    error: Optional[str] = None
72    metadata: dict[str, Any] = field(default_factory=dict)

任务数据类

Attributes: id: 任务 ID（UUID） name: 任务名称 status: 任务状态 created_at: 创建时间 started_at: 开始时间 completed_at: 完成时间 result: 任务结果 error: 错误信息 metadata: 附加元数据

Task( id: str, name: str, status: TaskStatus, created_at: datetime.datetime, started_at: datetime.datetime | None = None, completed_at: datetime.datetime | None = None, result: Any = None, error: str | None = None, metadata: dict[str, typing.Any] = <factory>)

id: str

name: str

status: TaskStatus

created_at: datetime.datetime

started_at: datetime.datetime | None = None

completed_at: datetime.datetime | None = None

result: Any = None

error: str | None = None

metadata: dict[str, typing.Any]

class TaskStatus(builtins.str, enum.Enum): View Source

34class TaskStatus(str, Enum):
35    """任务状态枚举"""
36
37    PENDING = "PENDING"
38    RUNNING = "RUNNING"
39    SUCCESS = "SUCCESS"
40    FAILED = "FAILED"
41    CANCELLED = "CANCELLED"
42
43    def __str__(self) -> str:
44        return self.value

任务状态枚举

PENDING = <TaskStatus.PENDING: 'PENDING'>

RUNNING = <TaskStatus.RUNNING: 'RUNNING'>

SUCCESS = <TaskStatus.SUCCESS: 'SUCCESS'>

FAILED = <TaskStatus.FAILED: 'FAILED'>

CANCELLED = <TaskStatus.CANCELLED: 'CANCELLED'>

class KeyringCredentialProvider(remote_cmd.CredentialProvider): View Source

142class KeyringCredentialProvider(CredentialProvider):
143    """
144    Keyring 凭据提供者
145
146    使用系统 Keyring 服务（Windows Credential Manager / macOS Keychain / Linux Secret Service）
147    获取密码。需要安装 keyring 库（pip install keyring）。
148
149    在凭据链中的位置：EnvCredentialProvider 之后，EncryptedFileCredentialProvider 之前。
150
151    Args:
152        service_name: Keyring 服务名称，默认 "remote-cmd"
153    """
154
155    def __init__(self, service_name: str = "remote-cmd") -> None:
156        self._service_name = service_name
157
158    def get_password(self, host: Host) -> Optional[str]:
159        """
160        从系统 Keyring 获取密码
161
162        使用 keyring.get_password(service_name, host.name) 获取。
163        keyring 库为可选依赖，未安装时静默返回 None。
164
165        Args:
166            host: 主机配置对象
167
168        Returns:
169            Optional[str]: 密码，未找到或不可用时返回 None
170        """
171        try:
172            import keyring
173
174            password = keyring.get_password(self._service_name, host.name)
175            if password:
176                logger.debug(f"retrieved password for {host.name} ")
177            return password
178        except ImportError:
179            logger.debug("keyring not installed, skipping KeyringCredentialProvider")
180            return None
181        except Exception as e:  # noqa: BLE001
182            logger.debug(f"keyring access failed: {e}")
183            return None
184
185    def set_password(self, host: Host, password: str) -> bool:
186        """
187        向 Keyring 存储密码
188
189        Args:
190            host: 主机配置对象
191            password: to store
192
193        Returns:
194            bool: True if the store succeeded
195        """
196        try:
197            import keyring
198
199            keyring.set_password(self._service_name, host.name, password)
200            return True
201        except Exception as e:  # noqa: BLE001
202            logger.debug(f"keyring store failed: {e}")
203            return False
204
205    def delete_password(self, host: Host) -> bool:
206        """
207        从 Keyring 删除密码
208
209        Args:
210            host: 主机配置对象
211
212        Returns:
213            bool: True if the delete succeeded
214        """
215        try:
216            import keyring
217
218            keyring.delete_password(self._service_name, host.name)
219            return True
220        except Exception as e:  # noqa: BLE001
221            logger.debug(f"keyring delete failed: {e}")
222            return False

Keyring 凭据提供者

使用系统 Keyring 服务（Windows Credential Manager / macOS Keychain / Linux Secret Service）获取密码。需要安装 keyring 库（pip install keyring）。

在凭据链中的位置：EnvCredentialProvider 之后，EncryptedFileCredentialProvider 之前。

Args: service_name: Keyring 服务名称，默认 "remote-cmd"

KeyringCredentialProvider(service_name: str = 'remote-cmd') View Source

155    def __init__(self, service_name: str = "remote-cmd") -> None:
156        self._service_name = service_name

def get_password(self, host: Host) -> str | None: View Source

158    def get_password(self, host: Host) -> Optional[str]:
159        """
160        从系统 Keyring 获取密码
161
162        使用 keyring.get_password(service_name, host.name) 获取。
163        keyring 库为可选依赖，未安装时静默返回 None。
164
165        Args:
166            host: 主机配置对象
167
168        Returns:
169            Optional[str]: 密码，未找到或不可用时返回 None
170        """
171        try:
172            import keyring
173
174            password = keyring.get_password(self._service_name, host.name)
175            if password:
176                logger.debug(f"retrieved password for {host.name} ")
177            return password
178        except ImportError:
179            logger.debug("keyring not installed, skipping KeyringCredentialProvider")
180            return None
181        except Exception as e:  # noqa: BLE001
182            logger.debug(f"keyring access failed: {e}")
183            return None

从系统 Keyring 获取密码

使用 keyring.get_password(service_name, host.name) 获取。 keyring 库为可选依赖，未安装时静默返回 None。

Args: host: 主机配置对象

Returns: Optional[str]: 密码，未找到或不可用时返回 None

def set_password(self, host: Host, password: str) -> bool: View Source

185    def set_password(self, host: Host, password: str) -> bool:
186        """
187        向 Keyring 存储密码
188
189        Args:
190            host: 主机配置对象
191            password: to store
192
193        Returns:
194            bool: True if the store succeeded
195        """
196        try:
197            import keyring
198
199            keyring.set_password(self._service_name, host.name, password)
200            return True
201        except Exception as e:  # noqa: BLE001
202            logger.debug(f"keyring store failed: {e}")
203            return False

向 Keyring 存储密码

Args: host: 主机配置对象 password: to store

Returns: bool: True if the store succeeded

def delete_password(self, host: Host) -> bool: View Source

205    def delete_password(self, host: Host) -> bool:
206        """
207        从 Keyring 删除密码
208
209        Args:
210            host: 主机配置对象
211
212        Returns:
213            bool: True if the delete succeeded
214        """
215        try:
216            import keyring
217
218            keyring.delete_password(self._service_name, host.name)
219            return True
220        except Exception as e:  # noqa: BLE001
221            logger.debug(f"keyring delete failed: {e}")
222            return False

从 Keyring 删除密码

Args: host: 主机配置对象

Returns: bool: True if the delete succeeded

__version__ = '2.1.0'

__author__ = 'Vae-Scrooge'

__license__ = 'MIT'

remote_cmd

创建连接配置

执行远程命令

或链式调用