Ssh Skill

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
/lib/
/lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg

# Virtual Environment
venv/
ENV/
env/

# IDE
.vscode/
.idea/
*.swp
*.swo
*~

# OS
.DS_Store
Thumbs.db

# Logs
*.log

# Temporary files
*.tmp
*.bak
*.orig

# SSH keys (安全起见)
*.pem
*.key
id_rsa*
id_ed25519*

# Local configuration
config.local.json
.env

# Docs (本地文档，不提交)
docs/

"""
SSH Skill 基本用法（可运行）

说明：
- 这是一个“回归用例”脚本：用本目录的示例配置调用 ssh-skill 的 CLI 入口。
- 运行前请先把示例配置中的占位符替换为你的真实信息（不要提交到仓库）。

用法：
  python basic_usage.py [config_json] [command]

示例：
  python basic_usage.py ./config_single_key.json "whoami && hostname"
"""

from __future__ import annotations

import os
import subprocess
import sys
from pathlib import Path


def main() -> int:
    examples_dir = Path(__file__).resolve().parent
    skill_root = examples_dir.parent
    script = skill_root / "scripts" / "ssh_execute.py"

    config_path = Path(sys.argv[1]).resolve() if len(sys.argv) >= 2 else (examples_dir / "config_single_key.json")
    command = sys.argv[2] if len(sys.argv) >= 3 else "whoami && hostname"

    if not script.exists():
        raise FileNotFoundError(f"未找到脚本入口：{script}")
    if not config_path.exists():
        raise FileNotFoundError(f"未找到配置文件：{config_path}")

    env = os.environ.copy()
    env.setdefault("MSYS_NO_PATHCONV", "1")

    proc = subprocess.run(
        [sys.executable, str(script), str(config_path), command],
        text=True,
        env=env,
    )
    return proc.returncode


if __name__ == "__main__":
    raise SystemExit(main())

"""
并发/批量场景示例（文档型）

建议把“批量执行/并发上传下载”收敛到固定入口，避免在不同项目里临时拼装命令。
"""


def main() -> None:
    print(
        """
多服务器（servers 字典）选择某个环境执行：

  python ../scripts/ssh_execute.py ./config_multi_servers.json "uptime" --server dev

如需对多台机器并发执行，请使用 ssh_cluster.py（更适合运维批量操作）：

  python ../scripts/ssh_cluster.py ./config_multi_servers.json "df -h"
"""
    )


if __name__ == "__main__":
    main()

{
  "name": "example-env-development",
  "summary": "示例：开发环境标注（environment=development）",
  "host": "192.0.2.50",
  "port": 22,
  "user": "developer",
  "key_file": "./keys/example_dev_id_ed25519",
  "timeout": 15,
  "strict_host_key_checking": false,
  "auto_add_hostkey": true,
  "use_controlmaster": true,
  "control_persist": "5m",
  "description": "开发环境配置示例 - 快速迭代使用",
  "created_at": "2025-11-01T10:00:00+08:00",
  "updated_at": "2025-11-01T10:00:00+08:00",
  "environment": "development",
  "tags": ["example", "development", "fast"],
  "review_interval_days": 365,
  "last_reviewed_at": "2025-11-01T10:00:00+08:00",
  "owner": "（填：负责人）",
  "contact": "user@example.invalid",
  "notes": "开发环境通常允许更宽松的主机密钥检查，但生产环境建议启用严格检查。",
  "config_version": "1.0"
}

{
  "name": "example-env-production",
  "summary": "示例：生产环境标注（environment=production，建议 strict_host_key_checking=true）",
  "host": "198.51.100.50",
  "port": 22,
  "user": "deploy",
  "key_file": "./keys/example_prod_id_ed25519",
  "timeout": 30,
  "strict_host_key_checking": true,
  "auto_add_hostkey": false,
  "use_controlmaster": true,
  "control_persist": "30m",
  "description": "生产环境配置示例 - 高安全性要求",
  "created_at": "2025-01-15T09:00:00+08:00",
  "updated_at": "2025-12-17T10:30:00+08:00",
  "environment": "production",
  "tags": ["example", "production", "secure"],
  "review_interval_days": 90,
  "last_reviewed_at": "2025-10-01T14:00:00+08:00",
  "expires_at": null,
  "owner": "（填：负责人）",
  "contact": "user@example.invalid",
  "team": "（填：团队）",
  "notes": "生产环境建议：\n1. 启用严格主机密钥检查\n2. 禁用自动添加 hostkey，手动管理 known_hosts\n3. 定期审查配置\n4. 使用短期有效的访问凭证",
  "documentation_url": "",
  "config_version": "1.0"
}

{
  "name": "example-jump-double-key",
  "summary": "示例：双跳板机-密钥认证（替换 host/user/key_file 与 jump_hosts）",
  "description": "双跳板机配置示例（连接深层内网服务器的典型场景）。",
  "host": "192.0.2.30",
  "port": 22,
  "user": "dbadmin",
  "key_file": "./keys/example_db_id_ed25519",
  "timeout": 60,
  "strict_host_key_checking": true,
  "auto_add_hostkey": true,
  "use_controlmaster": true,
  "control_persist": "1h",
  "jump_hosts": [
    {
      "name": "bastion-1",
      "host": "198.51.100.30",
      "port": 22,
      "user": "jumpuser1",
      "key_file": "./keys/example_jump1_id_ed25519",
      "description": "第一级跳板机"
    },
    {
      "name": "jump-2",
      "host": "198.51.100.31",
      "port": 22,
      "user": "jumpuser2",
      "key_file": "./keys/example_jump2_id_ed25519",
      "description": "第二级跳板机"
    }
  ],
  "environment": "production",
  "tags": ["example", "jump", "double", "key"],
  "owner": "（填：负责人）",
  "team": "（填：团队）",
  "notes": "每层跳板机可配置不同 user/key_file；也可按实际情况简化为字符串格式。",
  "config_version": "1.0"
}

{
  "name": "example-jump-password",
  "summary": "示例：跳板机 + 密码认证（替换 host/user/password 与 jump_hosts）",
  "description": "密码认证 + 跳板机示例（全链路密码）。",
  "host": "192.0.2.40",
  "port": 22,
  "user": "appuser",
  "password": "__REPLACE_ME__",
  "timeout": 30,
  "jump_hosts": [
    {
      "host": "198.51.100.40",
      "port": 22,
      "user": "jumpuser",
      "password": "__REPLACE_ME__"
    }
  ],
  "environment": "development",
  "tags": ["example", "jump", "password"],
  "owner": "（填：负责人）",
  "team": "（填：团队）",
  "notes": "注意：不要在仓库中提交真实密码；请使用安全方式注入。",
  "config_version": "1.0"
}

{
  "name": "example-jump-simple",
  "summary": "示例：简化跳板机写法（jump_hosts 仅写 host 字符串）",
  "description": "jump_hosts 仅写主机名，用户名/密钥默认沿用目标服务器配置。",
  "host": "203.0.113.20",
  "port": 22,
  "user": "deploy",
  "key_file": "./keys/example_deploy_id_ed25519",
  "timeout": 30,
  "strict_host_key_checking": true,
  "auto_add_hostkey": true,
  "use_controlmaster": true,
  "control_persist": "30m",
  "jump_hosts": ["bastion.example.invalid", "jump.internal.invalid"],
  "environment": "production",
  "tags": ["example", "jump", "simple"],
  "config_version": "1.0"
}

{
  "name": "example-jump-single-key",
  "summary": "示例：单跳板机-密钥认证（替换 host/user/key_file 与 jump_hosts）",
  "description": "单跳板机场景示例（密钥认证 + ControlMaster）。",
  "host": "192.0.2.20",
  "port": 22,
  "user": "appuser",
  "key_file": "./keys/example_app_id_ed25519",
  "timeout": 30,
  "strict_host_key_checking": true,
  "auto_add_hostkey": true,
  "use_controlmaster": true,
  "control_persist": "30m",
  "jump_hosts": [
    {
      "name": "bastion-1",
      "host": "198.51.100.20",
      "port": 22,
      "user": "jumpuser",
      "key_file": "./keys/example_jump_id_ed25519",
      "description": "示例跳板机（外网入口）"
    }
  ],
  "environment": "production",
  "tags": ["example", "jump", "single", "key"],
  "owner": "（填：负责人）",
  "team": "（填：团队）",
  "notes": "请分别配置目标机与跳板机的密钥路径（key_file）。",
  "config_version": "1.0"
}

{
  "name": "example-key-with-passphrase",
  "summary": "示例：单机-密钥认证-私钥带密码（替换 host/user/key_file/key_passphrase）",
  "description": "私钥带密码的密钥认证示例（建议配合 ssh-agent 或安全的密钥管理方案）。",
  "host": "198.51.100.10",
  "port": 22,
  "user": "deploy",
  "key_file": "./keys/example_id_ed25519",
  "key_passphrase": "__REPLACE_ME__",
  "timeout": 30,
  "strict_host_key_checking": true,
  "auto_add_hostkey": true,
  "use_controlmaster": true,
  "control_persist": "10m",
  "environment": "production",
  "tags": ["example", "key", "passphrase"],
  "owner": "（填：负责人）",
  "team": "（填：团队）",
  "notes": "注意：不要在仓库中提交真实 key_passphrase；请使用安全方式注入。",
  "config_version": "1.0"
}

{
  "summary": "示例：多服务器配置（servers 字典），用于批量执行/并发/集群操作。",
  "description": "多环境服务器配置示例（请替换各 server 的 host/user/key_file）。",
  "servers": {
    "dev": {
      "name": "dev-server",
      "summary": "示例：dev-开发环境-密钥认证",
      "description": "开发环境服务器 - 用于日常开发和测试",
      "host": "192.0.2.61",
      "port": 22,
      "user": "dev",
      "key_file": "../keys/example_dev_id_ed25519",
      "timeout": 30,
      "use_controlmaster": true,
      "control_persist": "10m",
      "environment": "development",
      "tags": ["example", "dev"]
    },
    "staging": {
      "name": "staging-server",
      "summary": "示例：staging-预发布-密钥认证",
      "description": "预发布环境服务器 - 上线前验证",
      "host": "198.51.100.61",
      "port": 22,
      "user": "deploy",
      "key_file": "../keys/example_staging_id_ed25519",
      "timeout": 30,
      "use_controlmaster": true,
      "control_persist": "30m",
      "environment": "staging",
      "tags": ["example", "staging"]
    },
    "prod": {
      "name": "prod-server",
      "summary": "示例：prod-生产-密钥认证",
      "description": "生产环境主服务器（示例域名）",
      "host": "prod.example.invalid",
      "port": 22,
      "user": "deploy",
      "key_file": "../keys/example_prod_id_ed25519",
      "timeout": 60,
      "use_controlmaster": true,
      "control_persist": "1h",
      "strict_host_key_checking": true,
      "environment": "production",
      "tags": ["example", "prod"]
    }
  },
  "config_version": "1.0"
}

{
  "name": "example-single-key",
  "summary": "示例：开发环境-单机-密钥认证（替换 host/user/key_file）",
  "description": "单机密钥认证示例，默认启用 ControlMaster 连接复用。",
  "host": "192.0.2.10",
  "port": 22,
  "user": "deploy",
  "key_file": "./keys/example_id_ed25519",
  "timeout": 30,
  "strict_host_key_checking": true,
  "auto_add_hostkey": true,
  "use_controlmaster": true,
  "control_persist": "10m",
  "jump_hosts": [],
  "environment": "development",
  "tags": ["example", "single", "key"],
  "owner": "（填：负责人）",
  "team": "（填：团队）",
  "notes": "请将 key_file 指向实际私钥路径，并确保权限正确（建议 chmod 600）。",
  "config_version": "1.0"
}

{
  "name": "example-single-password",
  "summary": "示例：单机-密码认证（替换 host/user/password）",
  "description": "密码认证示例（走 Paramiko 客户端连接池），适合无法使用密钥认证的场景。",
  "host": "203.0.113.10",
  "port": 22,
  "user": "admin",
  "password": "__REPLACE_ME__",
  "timeout": 30,
  "strict_host_key_checking": true,
  "auto_add_hostkey": true,
  "environment": "development",
  "tags": ["example", "single", "password"],
  "owner": "（填：负责人）",
  "team": "（填：团队）",
  "notes": "建议：优先使用密钥认证；如必须使用密码，请用安全方案注入，不要明文提交。",
  "config_version": "1.0"
}

{
  "name": "TEMPLATE",
  "summary": "（示例）中文摘要：用途/环境/责任人/系统，例如：生产-支付网关-上海机房-张三",
  "host": "YOUR_SERVER_IP_OR_HOSTNAME",
  "port": 22,
  "user": "YOUR_USERNAME",
  "key_file": "./YOUR_KEY_FILE_PATH",
  "key_passphrase": null,
  "timeout": 30,
  "strict_host_key_checking": false,
  "use_controlmaster": true,
  "control_persist": "10m",
  "jump_hosts": [],

  "description": "配置文件模板 - 请修改后使用",
  "created_at": "2025-12-17T12:00:00+08:00",
  "updated_at": "2025-12-17T12:00:00+08:00",
  "environment": "development",
  "tags": ["template", "example"],

  "review_interval_days": 90,
  "last_reviewed_at": "2025-12-17T12:00:00+08:00",
  "expires_at": null,

  "owner": "YOUR_NAME",
  "contact": "YOUR_EMAIL_OR_CONTACT",
  "team": "YOUR_TEAM",
  "notes": "这是配置模板，请根据实际情况修改所有字段。\n\n元数据字段说明：\n- created_at/updated_at: 自动管理的时间戳\n- environment: production/staging/development/testing\n- tags: 便于分类和筛选\n- review_interval_days: 审查间隔（生产环境建议90天）\n- expires_at: 临时访问配置的过期时间\n- owner/contact/team: 团队协作信息",
  "documentation_url": "",

  "config_version": "1.0"
}

{
  "name": "example-temporary-access",
  "summary": "示例：临时访问配置（expires_at 到期后应拒绝执行）",
  "host": "203.0.113.60",
  "port": 22,
  "user": "contractor",
  "key_file": "./keys/example_temp_id_ed25519",
  "timeout": 30,
  "use_controlmaster": true,
  "control_persist": "10m",
  "description": "临时访问配置示例 - 自动过期",
  "created_at": "2025-12-01T09:00:00+08:00",
  "updated_at": "2025-12-01T09:00:00+08:00",
  "environment": "staging",
  "tags": ["example", "temporary", "expires"],
  "review_interval_days": 30,
  "last_reviewed_at": "2025-12-01T09:00:00+08:00",
  "expires_at": "2025-12-31T23:59:59+08:00",
  "owner": "（填：负责人）",
  "contact": "user@example.invalid",
  "team": "（填：团队）",
  "notes": "此配置用于临时访问，过期后应自动拒绝执行。项目结束后应立即删除。",
  "config_version": "1.0"
}

"""
配置加载示例（文档型）

本文件用于给人/AI 提供“正确范式”，避免在不同项目中手写 `ssh/scp`。
更完整的索引请看：README.md
"""


def main() -> None:
    print(
        """
最常见用法（推荐：走 CLI 入口，稳定且不依赖 Python 导入路径）：

  # 单机（密钥认证）
  python ../scripts/ssh_execute.py ./config_single_key.json "whoami && hostname"

  # 单跳板机
  python ../scripts/ssh_execute.py ./config_jump_single_key.json "uptime"

  # 多服务器（选择某个 server）
  python ../scripts/ssh_execute.py ./config_multi_servers.json "hostname" --server dev

配置文件建议字段：
- summary（中文摘要，推荐必填）：一眼看懂用途/环境/责任人
- description（可选）：补充说明
- environment/tags/owner/team（可选）：便于团队协作与筛选
"""
    )


if __name__ == "__main__":
    main()

"""
交互式会话提示（文档型）

运维中常见的交互问题：
- `sudo` 需要输入密码
- `apt/yum` 需要确认
- 远端脚本提示 `Enter password:` / `Press any key`

建议：
- 优先把命令改成“非交互式”（例如 `DEBIAN_FRONTEND=noninteractive`、`-y`、`sudo -n` 等）
- 若必须交互，建议把交互步骤拆成“明确的多步命令”并逐步验证输出

本 skill 的核心目标是：把连接与复用封装起来，避免回退到手写 `ssh`。
"""


def main() -> None:
    print(
        """
示例：把 apt 安装变为非交互式（需要你按真实系统调整）：

  python ../scripts/ssh_execute.py ./config_single_key.json "sudo -n true && echo ok || echo 'sudo 需要免密或交互'"
  python ../scripts/ssh_execute.py ./config_single_key.json "sudo -n DEBIAN_FRONTEND=noninteractive apt-get update -y"
  python ../scripts/ssh_execute.py ./config_single_key.json "sudo -n DEBIAN_FRONTEND=noninteractive apt-get install -y nginx"
"""
    )


if __name__ == "__main__":
    main()

"""
跳板机场景示例（文档型）

目标：把“直连/单跳/双跳/简化跳板”四类典型拓扑固定成模板，减少跑偏与试错。
"""


def main() -> None:
    print(
        """
单跳板机（推荐：使用 config_jump_single_key.json）：

  python ../scripts/ssh_execute.py ./config_jump_single_key.json "whoami && hostname"

双跳板机（推荐：使用 config_jump_double.json）：

  python ../scripts/ssh_execute.py ./config_jump_double.json "whoami && hostname"

简化写法（jump_hosts 只写字符串主机名）：

  python ../scripts/ssh_execute.py ./config_jump_simple.json "uptime"

注意：
- 示例配置全部是占位符/保留网段 IP，请按你的真实环境替换后使用。
- 遇到需要一次性初始化的步骤（例如写入 authorized_keys）可以临时用系统 ssh，但后续日常操作必须回到本 skill。
"""
    )


if __name__ == "__main__":
    main()

param(
  [switch] $StrictMetadata = $true
)

Set-StrictMode -Version Latest
$ErrorActionPreference = 'Stop'

$examplesDir = Split-Path -Parent $MyInvocation.MyCommand.Path
$skillRoot = Split-Path -Parent $examplesDir
$manager = Join-Path $skillRoot 'scripts\\ssh_config_manager.py'

if (-not (Test-Path -LiteralPath $manager)) {
  throw "未找到 ssh_config_manager.py：$manager"
}

$python = Get-Command python -ErrorAction SilentlyContinue
if (-not $python) {
  $python = Get-Command py -ErrorAction SilentlyContinue
}
if (-not $python) {
  throw "未找到 Python（python/py）。请先安装 Python，并确保可在 PATH 中调用。"
}

$jsonFiles = Get-ChildItem -LiteralPath $examplesDir -File -Filter *.json | Sort-Object Name
if (-not $jsonFiles) {
  Write-Host "未找到示例 JSON（跳过）：$examplesDir"
  exit 0
}

$failed = New-Object System.Collections.Generic.List[string]

foreach ($file in $jsonFiles) {
  Write-Host "校验：$($file.Name)"

  $args = @($manager, 'validate', $file.FullName)
  if ($StrictMetadata) {
    $args += '--strict-metadata'
  }

  & $python @args | Write-Host
  if ($LASTEXITCODE -ne 0) {
    $failed.Add($file.Name) | Out-Null
  }
}

if ($failed.Count -gt 0) {
  Write-Host "示例冒烟测试失败：$($failed.Count) 个文件"
  $failed | ForEach-Object { Write-Host "  - $_" }
  exit 1
}

Write-Host "示例冒烟测试通过：$($jsonFiles.Count) 个文件"
exit 0

"""
ControlMaster 复用性能对比（可选回归）

说明：
- 该脚本用于直观展示“每次新连接” vs “ControlMaster 复用”差异。
- 需要你把 HOST/USER/KEY_FILE 替换成可用目标。
"""

from __future__ import annotations

import os
import subprocess
import time

# 测试配置（请替换）
HOST = "192.0.2.10"
USER = "deploy"
KEY_FILE = "./keys/example_id_ed25519"
ITERATIONS = 10


def _user_known_hosts_file_arg() -> str:
    return "UserKnownHostsFile=NUL" if os.name == "nt" else "UserKnownHostsFile=/dev/null"


def test_without_controlmaster() -> float:
    print("\n" + "=" * 60)
    print("测试1: 传统方式（每次建立新连接）")
    print("=" * 60)

    times_ms = []
    for i in range(ITERATIONS):
        cmd = [
            "ssh",
            "-i",
            KEY_FILE,
            "-o",
            "StrictHostKeyChecking=no",
            "-o",
            _user_known_hosts_file_arg(),
            f"{USER}@{HOST}",
            "echo test",
        ]

        start = time.time()
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
        elapsed = (time.time() - start) * 1000
        times_ms.append(elapsed)
        print(f"  第{i + 1:2d}次: {elapsed:6.1f}ms - {result.stdout.strip()}")

    avg = sum(times_ms) / len(times_ms)
    print(f"\n  平均耗时: {avg:.1f}ms")
    return avg


def test_with_controlmaster() -> float:
    print("\n" + "=" * 60)
    print("测试2: ControlMaster方式（连接复用）")
    print("=" * 60)

    control_path = os.path.join(os.environ.get("TEMP", "/tmp"), f"ssh-test-{USER}@{HOST}")
    if os.path.exists(control_path):
        try:
            os.remove(control_path)
        except OSError:
            pass

    times_ms = []
    for i in range(ITERATIONS):
        cmd = [
            "ssh",
            "-i",
            KEY_FILE,
            "-o",
            "StrictHostKeyChecking=no",
            "-o",
            _user_known_hosts_file_arg(),
            "-o",
            "ControlMaster=auto",
            "-o",
            f"ControlPath={control_path}",
            "-o",
            "ControlPersist=10s",
            f"{USER}@{HOST}",
            "echo test",
        ]

        start = time.time()
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
        elapsed = (time.time() - start) * 1000
        times_ms.append(elapsed)
        print(f"  第{i + 1:2d}次: {elapsed:6.1f}ms - {result.stdout.strip()}")

    avg = sum(times_ms) / len(times_ms)
    print(f"\n  平均耗时: {avg:.1f}ms")
    return avg


def main() -> None:
    avg1 = test_without_controlmaster()
    avg2 = test_with_controlmaster()
    if avg2 > 0:
        print(f"\n加速比（大致）：{avg1 / avg2:.1f}x")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
为现有 SSH config 添加标准注释

读取现有的 SSH config 文件，为每个 Host 添加标准注释元数据
"""

import os
import re
from datetime import datetime


def parse_existing_config(config_path):
    """
    解析现有配置，提取每个 Host 块

    Returns:
        List of (comments, host_line, config_lines) tuples
    """
    if not os.path.exists(config_path):
        return []

    with open(config_path, 'r', encoding='utf-8') as f:
        lines = f.readlines()

    hosts = []
    current_comments = []
    current_host_line = None
    current_config = []
    in_host_block = False

    i = 0
    while i < len(lines):
        line = lines[i]
        stripped = line.strip()

        # 检查是否是 Host 行
        if stripped.startswith('Host ') and not stripped.startswith('Host *'):
            # 保存上一个 Host 块
            if current_host_line:
                hosts.append((current_comments, current_host_line, current_config))

            # 开始新的 Host 块
            current_host_line = line
            current_config = []
            in_host_block = True
            # 保留之前收集的注释

        elif in_host_block:
            # 在 Host 块中
            if stripped and not stripped.startswith('#'):
                # 配置行（缩进的）
                if line.startswith((' ', '\t')):
                    current_config.append(line)
                else:
                    # 遇到非缩进的非注释行，Host 块结束
                    in_host_block = False
                    current_comments = []
                    if stripped.startswith('#'):
                        current_comments.append(line)
            elif stripped.startswith('#'):
                # Host 块中的注释（通常不应该有）
                current_config.append(line)
            elif not stripped:
                # 空行，Host 块可能结束
                current_config.append(line)
                in_host_block = False
                current_comments = []
        else:
            # 不在 Host 块中
            if stripped.startswith('#') or not stripped:
                current_comments.append(line)
            else:
                # 非注释非空行，清空注释缓存
                current_comments = []

        i += 1

    # 保存最后一个 Host 块
    if current_host_line:
        hosts.append((current_comments, current_host_line, current_config))

    return hosts


def extract_alias_from_host_line(host_line):
    """从 Host 行提取别名"""
    match = re.match(r'Host\s+(.+)', host_line.strip())
    if match:
        return match.group(1).strip()
    return None


def has_standard_comments(comments):
    """检查是否已有标准注释"""
    comment_text = ''.join(comments)
    return '# description:' in comment_text or '# environment:' in comment_text


def generate_standard_comments(alias):
    """生成标准注释"""
    now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

    comments = [
        f"\n# ===== {alias} =====\n",
        f"# description: \n",
        f"# environment: unknown\n",
        f"# tags: \n",
        f"# location: \n",
        f"# created_at: {now}\n",
        f"# updated_at: {now}\n"
    ]

    return comments


def add_comments_to_config(config_path, output_path=None):
    """
    为配置文件添加标准注释

    Args:
        config_path: 输入配置文件路径
        output_path: 输出配置文件路径（如果为 None，则覆盖原文件）
    """
    if output_path is None:
        output_path = config_path

    # 解析现有配置
    hosts = parse_existing_config(config_path)

    print(f"找到 {len(hosts)} 个 Host 配置")

    # 生成新配置
    new_lines = []
    processed_count = 0
    skipped_count = 0

    for comments, host_line, config_lines in hosts:
        alias = extract_alias_from_host_line(host_line)

        if not alias:
            # 无法提取别名，保持原样
            new_lines.extend(comments)
            new_lines.append(host_line)
            new_lines.extend(config_lines)
            skipped_count += 1
            continue

        # 检查是否已有标准注释
        if has_standard_comments(comments):
            # 已有标准注释，保持原样
            new_lines.extend(comments)
            new_lines.append(host_line)
            new_lines.extend(config_lines)
            skipped_count += 1
            print(f"  跳过 {alias}（已有标准注释）")
        else:
            # 添加标准注释
            standard_comments = generate_standard_comments(alias)
            new_lines.extend(standard_comments)
            new_lines.append(host_line)
            new_lines.extend(config_lines)
            processed_count += 1
            print(f"  添加注释: {alias}")

    # 写入新配置
    with open(output_path, 'w', encoding='utf-8') as f:
        f.writelines(new_lines)

    print(f"\n完成:")
    print(f"  处理: {processed_count} 个")
    print(f"  跳过: {skipped_count} 个")
    print(f"  输出: {output_path}")


if __name__ == '__main__':
    config_path = os.path.expanduser('~/.ssh/config')
    add_comments_to_config(config_path)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
部署公钥到远程服务器

将指定的公钥部署到远程服务器，实现从密码认证迁移到密钥认证。
"""

import sys
import os
import argparse

# 修复 Windows 终端 UTF-8 输出
if sys.platform == 'win32':
    sys.stdout.reconfigure(encoding='utf-8', errors='replace')
    sys.stderr.reconfigure(encoding='utf-8', errors='replace')

# 添加 lib 到路径
_script_dir = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(_script_dir, 'lib'))


def deploy_pubkey(alias, pubkey_content, key_name):
    """
    部署公钥到远程服务器

    Args:
        alias: 服务器别名
        pubkey_content: 公钥内容
        key_name: 密钥名称（用于标识）

    Returns:
        bool: 是否成功
    """
    from config_v3 import SSHConfigLoaderV3

    try:
        # 加载配置
        loader = SSHConfigLoaderV3()
        params = loader.get_connection_params(alias)

        # 检查是否有密码
        if not params.get('password'):
            print(f"错误: {alias} 没有配置密码，无法使用密码认证部署公钥")
            return False

        # 使用密码认证创建客户端
        from paramiko_client import ParamikoClient
        client = ParamikoClient(
            host=params['hostname'],
            user=params['user'],
            port=params['port'],
            password=params['password'],
            timeout=30
        )

        print(f"正在连接到 {alias}...")

        # 测试连接
        result = client.execute("echo 'Connection OK'")
        if not result.success:
            print(f"错误: 无法连接到 {alias}")
            return False

        print("连接成功，开始部署公钥...")

        # 创建 .ssh 目录（如果不存在）
        result = client.execute("mkdir -p ~/.ssh && chmod 700 ~/.ssh")
        if not result.success:
            print(f"错误: 无法创建 .ssh 目录")
            return False

        # 检查公钥是否已存在
        result = client.execute(f"grep -F '{pubkey_content.strip()}' ~/.ssh/authorized_keys 2>/dev/null")
        if result.success and result.stdout.strip():
            print(f"公钥已存在于 {alias}，无需重复添加")
            return True

        # 追加公钥到 authorized_keys
        escaped_pubkey = pubkey_content.strip().replace("'", "'\\''")
        result = client.execute(
            f"echo '{escaped_pubkey}' >> ~/.ssh/authorized_keys && "
            f"chmod 600 ~/.ssh/authorized_keys"
        )

        if not result.success:
            print(f"错误: 无法写入公钥到 authorized_keys")
            print(f"错误信息: {result.stderr}")
            return False

        print(f"✓ 公钥已成功部署到 {alias}")

        # 验证密钥认证是否工作
        print("正在验证密钥认证...")
        # 这里需要使用新的密钥文件测试连接
        # 暂时跳过验证，由用户手动测试

        return True

    except Exception as e:
        print(f"错误: {str(e)}")
        return False


def main():
    parser = argparse.ArgumentParser(description='部署公钥到远程服务器')
    parser.add_argument('alias', help='服务器别名')
    parser.add_argument('--pubkey-file', required=True, help='公钥文件路径')
    parser.add_argument('--key-name', required=True, help='密钥名称（如 id_rsa_sa_legacy）')

    args = parser.parse_args()

    # 读取公钥内容
    pubkey_file = os.path.expanduser(args.pubkey_file)
    if not os.path.exists(pubkey_file):
        print(f"错误: 公钥文件不存在: {pubkey_file}")
        sys.exit(1)

    with open(pubkey_file, 'r', encoding='utf-8') as f:
        pubkey_content = f.read().strip()

    if not pubkey_content:
        print(f"错误: 公钥文件为空")
        sys.exit(1)

    # 部署公钥
    success = deploy_pubkey(args.alias, pubkey_content, args.key_name)

    if success:
        print(f"\n成功！现在可以使用密钥 {args.key_name} 连接到 {args.alias}")
        print(f"建议: 使用 migrate_to_key_auth.py 更新 SSH config")
        sys.exit(0)
    else:
        print(f"\n失败: 无法部署公钥到 {args.alias}")
        sys.exit(1)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
修复 SSH Config 文件

1. 从原始 JSON 配置中提取元数据填充注释
2. 统一证书文件路径格式为 ~/.ssh/keyfile
3. 在注释中添加密码字段（如果有）
"""

import os
import json
import re
from datetime import datetime
from pathlib import Path


def load_json_config(json_path):
    """加载 JSON 配置文件"""
    try:
        with open(json_path, 'r', encoding='utf-8') as f:
            return json.load(f)
    except Exception as e:
        print(f"  警告: 无法加载 {json_path}: {e}")
        return None


def find_json_config_by_alias(alias, json_dir):
    """根据别名查找对应的 JSON 配置文件"""
    # 尝试直接匹配
    json_path = os.path.join(json_dir, f"{alias}.json")
    if os.path.exists(json_path):
        return load_json_config(json_path)

    # 尝试大小写不敏感匹配
    for filename in os.listdir(json_dir):
        if not filename.endswith('.json'):
            continue

        name_without_ext = os.path.splitext(filename)[0]
        if name_without_ext.upper() == alias.upper():
            json_path = os.path.join(json_dir, filename)
            return load_json_config(json_path)

    # 尝试从 JSON 中的 name 字段匹配
    for filename in os.listdir(json_dir):
        if not filename.endswith('.json'):
            continue

        json_path = os.path.join(json_dir, filename)
        config = load_json_config(json_path)
        if config and config.get('name') == alias:
            return config

    return None


def normalize_key_path(key_path):
    """
    统一证书文件路径格式为 ~/.ssh/keyfile

    Args:
        key_path: 原始路径

    Returns:
        标准化后的路径
    """
    if not key_path:
        return key_path

    # 已经是 ~/.ssh/ 格式，直接返回
    if key_path.startswith('~/.ssh/'):
        return key_path

    # Windows 绝对路径转换为 ~/.ssh/ 格式
    # C:\Users\zhangyang\.ssh\keyfile -> ~/.ssh/keyfile
    # C:\Users\zhangyang/.ssh\keyfile -> ~/.ssh/keyfile
    # C:/Users/zhangyang/.ssh/keyfile -> ~/.ssh/keyfile

    # 统一路径分隔符
    normalized = key_path.replace('\\', '/')

    # 提取 .ssh 之后的部分
    if '/.ssh/' in normalized:
        parts = normalized.split('/.ssh/')
        if len(parts) == 2:
            return f"~/.ssh/{parts[1]}"

    # 如果无法转换，返回原路径
    return key_path


def extract_metadata_from_json(config):
    """从 JSON 配置中提取元数据"""
    metadata = {
        'description': '',
        'environment': 'unknown',
        'tags': [],
        'location': '',
        'password': ''
    }

    # 描述
    if 'description' in config:
        metadata['description'] = config['description']
    elif 'notes' in config:
        metadata['description'] = config['notes']

    # 从 metadata 字段提取
    if 'metadata' in config:
        meta = config['metadata']
        metadata['environment'] = meta.get('environment', 'unknown')
        metadata['tags'] = meta.get('tags', [])
        metadata['location'] = meta.get('location', '')

    # 密码
    if 'password' in config and config['password']:
        metadata['password'] = config['password']

    return metadata


def parse_ssh_config(config_path):
    """
    解析 SSH config 文件

    Returns:
        List of blocks, each block is a dict with:
        - comments: list of comment lines
        - host_line: the Host line
        - config_lines: list of config lines
        - alias: extracted alias
    """
    if not os.path.exists(config_path):
        return []

    with open(config_path, 'r', encoding='utf-8') as f:
        lines = f.readlines()

    blocks = []
    current_comments = []
    current_host_line = None
    current_config = []
    in_host_block = False

    i = 0
    while i < len(lines):
        line = lines[i]
        stripped = line.strip()

        # 检查是否是 Host 行
        if stripped.startswith('Host ') and not stripped.startswith('Host *'):
            # 保存上一个 Host 块
            if current_host_line:
                alias = extract_alias_from_host_line(current_host_line)
                blocks.append({
                    'comments': current_comments,
                    'host_line': current_host_line,
                    'config_lines': current_config,
                    'alias': alias
                })

            # 开始新的 Host 块
            current_host_line = line
            current_config = []
            in_host_block = True

        elif in_host_block:
            # 在 Host 块中
            if stripped and not stripped.startswith('#'):
                # 配置行（缩进的）
                if line.startswith((' ', '\t')):
                    current_config.append(line)
                else:
                    # 遇到非缩进的非注释行，Host 块结束
                    in_host_block = False
                    current_comments = []
                    if stripped.startswith('#'):
                        current_comments.append(line)
            elif stripped.startswith('#'):
                current_config.append(line)
            elif not stripped:
                current_config.append(line)
                in_host_block = False
                current_comments = []
        else:
            # 不在 Host 块中
            if stripped.startswith('#') or not stripped:
                current_comments.append(line)
            else:
                current_comments = []

        i += 1

    # 保存最后一个 Host 块
    if current_host_line:
        alias = extract_alias_from_host_line(current_host_line)
        blocks.append({
            'comments': current_comments,
            'host_line': current_host_line,
            'config_lines': current_config,
            'alias': alias
        })

    return blocks


def extract_alias_from_host_line(host_line):
    """从 Host 行提取别名"""
    match = re.match(r'Host\s+(.+)', host_line.strip())
    if match:
        return match.group(1).strip()
    return None


def generate_updated_comments(alias, metadata):
    """生成更新后的注释"""
    now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

    comments = [
        f"\n# ===== {alias} =====\n",
        f"# description: {metadata.get('description', '')}\n",
        f"# environment: {metadata.get('environment', 'unknown')}\n",
    ]

    # 标签
    tags = metadata.get('tags', [])
    if tags:
        comments.append(f"# tags: {','.join(tags)}\n")
    else:
        comments.append(f"# tags: \n")

    # 位置
    location = metadata.get('location', '')
    comments.append(f"# location: {location}\n")

    # 密码（如果有）
    password = metadata.get('password', '')
    if password:
        comments.append(f"# password: {password}\n")

    # 时间
    comments.append(f"# created_at: {now}\n")
    comments.append(f"# updated_at: {now}\n")

    return comments


def normalize_config_lines(config_lines):
    """标准化配置行中的证书路径"""
    normalized = []

    for line in config_lines:
        # 检查是否是 IdentityFile 行
        if 'IdentityFile' in line:
            match = re.match(r'(\s*)IdentityFile\s+(.+)', line)
            if match:
                indent = match.group(1)
                key_path = match.group(2).strip()
                normalized_path = normalize_key_path(key_path)
                normalized.append(f"{indent}IdentityFile {normalized_path}\n")
                continue

        normalized.append(line)

    return normalized


def fix_ssh_config(config_path, json_dir, output_path=None):
    """
    修复 SSH config 文件

    Args:
        config_path: SSH config 文件路径
        json_dir: JSON 配置目录
        output_path: 输出文件路径（如果为 None，则覆盖原文件）
    """
    if output_path is None:
        output_path = config_path

    # 解析现有配置
    blocks = parse_ssh_config(config_path)

    print(f"找到 {len(blocks)} 个 Host 配置")

    # 处理每个块
    new_lines = []
    updated_count = 0
    normalized_count = 0
    password_count = 0

    for block in blocks:
        alias = block['alias']

        if not alias:
            # 无法提取别名，保持原样
            new_lines.extend(block['comments'])
            new_lines.append(block['host_line'])
            new_lines.extend(block['config_lines'])
            continue

        # 查找对应的 JSON 配置
        json_config = find_json_config_by_alias(alias, json_dir)

        if json_config:
            # 提取元数据
            metadata = extract_metadata_from_json(json_config)

            # 生成新注释
            new_comments = generate_updated_comments(alias, metadata)
            new_lines.extend(new_comments)

            updated_count += 1

            if metadata.get('password'):
                password_count += 1
                print(f"  更新 {alias}（包含密码）")
            else:
                print(f"  更新 {alias}")
        else:
            # 没有找到 JSON 配置，保持原注释
            new_lines.extend(block['comments'])
            print(f"  跳过 {alias}（未找到 JSON 配置）")

        # Host 行
        new_lines.append(block['host_line'])

        # 标准化配置行中的证书路径
        normalized_config = normalize_config_lines(block['config_lines'])

        # 检查是否有路径被标准化
        if normalized_config != block['config_lines']:
            normalized_count += 1

        new_lines.extend(normalized_config)

    # 写入新配置
    with open(output_path, 'w', encoding='utf-8') as f:
        f.writelines(new_lines)

    print(f"\n完成:")
    print(f"  更新元数据: {updated_count} 个")
    print(f"  标准化路径: {normalized_count} 个")
    print(f"  添加密码: {password_count} 个")
    print(f"  输出: {output_path}")


if __name__ == '__main__':
    config_path = os.path.expanduser('~/.ssh/config')
    json_dir = os.path.expanduser('~/.ssh/server_config')

    fix_ssh_config(config_path, json_dir)

#!/bin/bash
# 获取系统信息脚本

# 操作系统
if [ -f /etc/os-release ]; then
    OS=$(grep PRETTY_NAME /etc/os-release | cut -d'"' -f2)
else
    OS=$(uname -s)
fi

# CPU核心数
CPU=$(nproc 2>/dev/null || grep -c processor /proc/cpuinfo 2>/dev/null || echo "?")

# 内存（GB）
if [ -f /proc/meminfo ]; then
    MEM_KB=$(grep MemTotal /proc/meminfo | grep -o '[0-9]*')
    MEM_GB=$(echo "scale=1; $MEM_KB/1024/1024" | bc 2>/dev/null || echo "?")
    MEM="${MEM_GB}G"
else
    MEM="?"
fi

# 磁盘总空间
DISK=$(df -h / | tail -1 | tr -s ' ' | cut -d' ' -f2)

# 输出格式化信息
echo "$OS/$CPU核/${MEM}内存/${DISK}磁盘"

"""
SSH Skill - 系统级SSH连接管理工具

基于系统OpenSSH的SSH客户端，提供稳定可靠的SSH连接能力。

核心特性:
- ControlMaster 连接复用（10-100x性能提升）
- ProxyJump 跳板机支持（支持多级）
- 项目级配置管理
- 流式输出和交互式会话
- 批量并发操作

快速开始:
    from ssh_skill import SSHConfigLoader

    # 加载配置
    client = SSHConfigLoader.from_file(".ssh_config/prod.json")

    # 执行命令
    result = client.execute("whoami && hostname")
    print(result.stdout)

    # 批量操作
    from ssh_skill import SSHCluster
    cluster = SSHCluster.from_directory(".ssh_config/cluster/")
    results = cluster.execute_all("uptime", parallel=True)
"""

from .client import SSHClient, SSHResult
from .config import SSHConfigLoader, ServerConfig
from .cluster import SSHCluster, SSHBatchOperations
from .utils import (
    check_ssh_available,
    get_ssh_version,
    validate_key_file
)

__version__ = "0.1.0"

__all__ = [
    # 核心类
    "SSHClient",
    "SSHResult",
    "SSHConfigLoader",
    "ServerConfig",
    "SSHCluster",
    "SSHBatchOperations",

    # 工具函数
    "check_ssh_available",
    "get_ssh_version",
    "validate_key_file",
]

"""
批量服务器操作模块 v3.1

从 SSH config 读取服务器列表，提供多服务器并发操作能力。
支持智能客户端选择（密钥认证用原生 SSH，密码认证用 Paramiko）。
"""

from typing import List, Dict, Optional, Union
from concurrent.futures import ThreadPoolExecutor, as_completed

try:
    from .config_v3 import SSHConfigLoaderV3
    from .native_ssh_client import SSHResult
except ImportError:
    from config_v3 import SSHConfigLoaderV3
    from native_ssh_client import SSHResult


class SSHCluster:
    """SSH集群管理类 v3.1，从 SSH config 读取服务器列表，智能选择客户端类型"""

    def __init__(self, clients: Dict[str, object], max_workers: int = 10):
        """
        初始化集群

        Args:
            clients: {alias: NativeSSHClient 或 ParamikoClient}
            max_workers: 最大并发数
        """
        self.clients = clients
        self.max_workers = max_workers

    @classmethod
    def from_ssh_config(cls, aliases: List[str] = None,
                        environment: str = None,
                        tags: List[str] = None,
                        max_workers: int = 10) -> 'SSHCluster':
        """
        从 SSH config 创建集群（智能选择客户端类型）

        Args:
            aliases: 指定别名列表（可选）
            environment: 按环境过滤（可选）
            tags: 按标签过滤（可选）
            max_workers: 最大并发数
        """
        loader = SSHConfigLoaderV3()

        if aliases:
            # 使用指定的别名列表
            host_list = aliases
        else:
            # 从 SSH config 获取所有 Host
            host_list = cls._list_all_hosts(loader)

        # 创建客户端（使用智能选择）
        clients = {}
        for alias in host_list:
            try:
                params = loader.get_connection_params(alias)
                metadata = params.get('metadata', {})

                # 按环境过滤
                if environment and metadata.get('environment', '') != environment:
                    continue

                # 按标签过滤
                if tags:
                    host_tags = metadata.get('tags', [])
                    if not any(t in host_tags for t in tags):
                        continue

                # 使用智能选择创建客户端
                client = loader.from_alias(alias)
                clients[alias] = client
            except Exception:
                continue  # 跳过无法加载的配置

        return cls(clients, max_workers)

    @staticmethod
    def _list_all_hosts(loader: SSHConfigLoaderV3) -> List[str]:
        """从 SSH config 列出所有 Host 别名"""
        import os
        import re

        config_path = loader.config_path
        if not os.path.exists(config_path):
            return []

        hosts = []
        with open(config_path, 'r', encoding='utf-8') as f:
            for line in f:
                stripped = line.strip()
                if stripped.startswith('Host ') and not stripped.startswith('Host *'):
                    match = re.match(r'Host\s+(.+)', stripped)
                    if match:
                        alias = match.group(1).strip()
                        # 跳过通配符
                        if '*' not in alias and '?' not in alias:
                            hosts.append(alias)
        return hosts

    def execute_all(self, command: str, parallel: bool = True,
                    timeout: Optional[int] = None) -> Dict[str, SSHResult]:
        """在所有服务器上执行命令"""
        if parallel:
            return self._execute_parallel(command, timeout)
        else:
            return self._execute_serial(command, timeout)

    def _execute_serial(self, command: str, timeout: Optional[int]) -> Dict[str, SSHResult]:
        """串行执行命令"""
        results = {}
        for alias, client in self.clients.items():
            try:
                if timeout:
                    original_timeout = client.timeout
                    client.timeout = timeout
                    result = client.execute(command)
                    client.timeout = original_timeout
                else:
                    result = client.execute(command)
                results[alias] = result
            except Exception as e:
                results[alias] = SSHResult(
                    success=False, stdout="",
                    stderr=f"执行异常: {str(e)}", exit_code=-1
                )
        return results

    def _execute_parallel(self, command: str, timeout: Optional[int]) -> Dict[str, SSHResult]:
        """并发执行命令"""
        results = {}

        def execute_on_client(alias, client):
            try:
                if timeout:
                    original_timeout = client.timeout
                    client.timeout = timeout
                    result = client.execute(command)
                    client.timeout = original_timeout
                else:
                    result = client.execute(command)
                return alias, result
            except Exception as e:
                return alias, SSHResult(
                    success=False, stdout="",
                    stderr=f"执行异常: {str(e)}", exit_code=-1
                )

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {
                executor.submit(execute_on_client, alias, client): alias
                for alias, client in self.clients.items()
            }
            for future in as_completed(futures):
                alias, result = future.result()
                results[alias] = result

        return results

    def health_check_all(self, check_command: str = "echo 'OK'",
                         parallel: bool = True,
                         timeout: Optional[int] = None) -> Dict[str, bool]:
        """批量健康检查"""
        results = self.execute_all(check_command, parallel=parallel, timeout=timeout)
        return {name: result.success for name, result in results.items()}

    def upload_all(self, local_path: str, remote_path: str,
                   parallel: bool = True) -> Dict[str, SSHResult]:
        """批量上传文件"""
        def upload_to_client(alias, client):
            try:
                result = client.upload(local_path, remote_path)
                return alias, result
            except Exception as e:
                return alias, SSHResult(
                    success=False, stdout="",
                    stderr=f"上传异常: {str(e)}", exit_code=-1
                )

        results = {}
        if parallel:
            with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
                futures = {
                    executor.submit(upload_to_client, alias, client): alias
                    for alias, client in self.clients.items()
                }
                for future in as_completed(futures):
                    alias, result = future.result()
                    results[alias] = result
        else:
            for alias, client in self.clients.items():
                _, result = upload_to_client(alias, client)
                results[alias] = result

        return results

"""
配置管理模块 v3.0

基于标准 OpenSSH config 格式的配置加载器

新特性：
1. 从 ~/.ssh/config 加载配置
2. 支持别名（Host）
3. 元数据从注释中解析
4. 完全兼容 ProxyJump（跳板机）
5. 支持密码认证（从注释中读取）
"""

import os
import json
import re
from typing import Dict, Optional, List
from pathlib import Path

try:
    import paramiko
except ImportError:
    raise ImportError("需要安装 paramiko 库: pip install paramiko")


class SSHConfigLoaderV3:
    """SSH Config 加载器 v3.0

    从标准 OpenSSH config 文件加载配置
    """

    def __init__(self, config_path: Optional[str] = None,
                 metadata_path: Optional[str] = None):
        """
        初始化加载器

        Args:
            config_path: SSH config 文件路径，默认 ~/.ssh/config
            metadata_path: 元数据文件路径，默认 ~/.ssh/config_metadata.json
        """
        if config_path is None:
            config_path = os.path.expanduser("~/.ssh/config")
        if metadata_path is None:
            metadata_path = os.path.expanduser("~/.ssh/config_metadata.json")

        self.config_path = config_path
        self.metadata_path = metadata_path

    def load_ssh_config(self, alias: str) -> dict:
        """
        从 SSH config 加载指定别名的配置

        Args:
            alias: 主机别名

        Returns:
            配置字典

        Raises:
            FileNotFoundError: 配置文件不存在
            ValueError: 别名不存在
        """
        if not os.path.exists(self.config_path):
            raise FileNotFoundError(f"SSH config 文件不存在: {self.config_path}")

        # 解析 SSH config
        ssh_config = paramiko.SSHConfig()
        with open(self.config_path, 'r', encoding='utf-8') as f:
            ssh_config.parse(f)

        # 获取配置
        try:
            host_config = ssh_config.lookup(alias)
        except Exception as e:
            raise ValueError(f"无法解析别名 '{alias}': {e}")

        # 检查是否真的找到了配置（paramiko 会返回默认值）
        if host_config.get('hostname') == alias and not self._alias_exists(alias):
            raise ValueError(f"别名 '{alias}' 不存在于 SSH config 中")

        return host_config

    def _alias_exists(self, alias: str) -> bool:
        """检查别名是否存在于 SSH config 中"""
        try:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                for line in f:
                    line = line.strip()
                    if line.startswith('Host ') and not line.startswith('Host *'):
                        # 提取 Host 名称
                        import re
                        host_match = re.match(r'Host\s+(.+)', line)
                        if host_match and host_match.group(1).strip() == alias:
                            return True
            return False
        except Exception:
            return False

    def load_metadata(self, alias: str) -> dict:
        """
        从注释中加载元数据

        Args:
            alias: 主机别名

        Returns:
            元数据字典（包括密码）
        """
        metadata = {
            'description': '',
            'environment': 'unknown',
            'tags': [],
            'location': '',
            'password': ''
        }

        # 读取 config 文件，查找该 Host 前的注释
        if not os.path.exists(self.config_path):
            return metadata

        with open(self.config_path, 'r', encoding='utf-8') as f:
            lines = f.readlines()

        # 查找 Host 行
        host_line_index = -1
        for i, line in enumerate(lines):
            stripped = line.strip()
            if stripped.startswith('Host ') and not stripped.startswith('Host *'):
                import re
                match = re.match(r'Host\s+(.+)', stripped)
                if match and match.group(1).strip() == alias:
                    host_line_index = i
                    break

        if host_line_index == -1:
            return metadata

        # 向前查找注释
        comment_lines = []
        i = host_line_index - 1
        while i >= 0:
            line = lines[i]
            stripped = line.strip()
            if stripped.startswith('#') or not stripped:
                comment_lines.insert(0, line)
                i -= 1
            else:
                break

        # 解析注释
        for line in comment_lines:
            line = line.strip()
            if not line.startswith('#'):
                continue

            # 移除开头的 #
            line = line[1:].strip()

            # 跳过分隔线
            if line.startswith('=====') or line == '':
                continue

            # 解析 key: value 格式
            if ':' in line:
                key, value = line.split(':', 1)
                key = key.strip()
                value = value.strip()

                if key == 'description':
                    metadata['description'] = value
                elif key == 'environment':
                    metadata['environment'] = value
                elif key == 'tags':
                    metadata['tags'] = [t.strip() for t in value.split(',') if t.strip()]
                elif key == 'location':
                    metadata['location'] = value
                elif key == 'password':
                    metadata['password'] = value

        return metadata

    def get_connection_params(self, alias: str) -> dict:
        """
        获取连接参数（用于创建 SSH 客户端）

        Args:
            alias: 主机别名

        Returns:
            连接参数字典（包括密码）
        """
        config = self.load_ssh_config(alias)
        metadata = self.load_metadata(alias)

        # 提取连接参数
        params = {
            'hostname': config.get('hostname'),
            'user': config.get('user'),
            'port': int(config.get('port', 22)),
            'timeout': 30,  # 默认超时
        }

        # 密钥文件
        identity_files = config.get('identityfile')
        if identity_files:
            # paramiko 返回的是列表
            if isinstance(identity_files, list):
                params['key_file'] = identity_files[0]
            else:
                params['key_file'] = identity_files

        # 密码（从注释元数据中获取）
        if metadata.get('password'):
            params['password'] = metadata['password']

        # ProxyJump（跳板机）
        proxy_jump = config.get('proxyjump')
        if proxy_jump:
            params['proxy_jump'] = proxy_jump

        # ForwardAgent（SSH agent 转发）
        forward_agent = config.get('forwardagent', 'no').lower()
        params['forward_agent'] = forward_agent in ('yes', 'true', '1')

        # 元数据
        params['metadata'] = metadata
        params['alias'] = alias

        return params

    def from_alias(self, alias: str):
        """
        通过别名创建 SSH 客户端（智能选择）

        策略：
        - 有密钥文件且无密码 → 使用 NativeSSHClient（原生 SSH）
        - 有密码 → 使用 ParamikoClient（Paramiko）

        Args:
            alias: 主机别名

        Returns:
            NativeSSHClient 或 ParamikoClient 实例
        """
        params = self.get_connection_params(alias)

        has_key = params.get('key_file') is not None
        has_password = params.get('password') is not None

        # 智能选择客户端类型
        if has_key and not has_password:
            # 密钥认证 → 使用原生 SSH（支持 agent forwarding）
            try:
                from .native_ssh_client import NativeSSHClient
            except ImportError:
                from native_ssh_client import NativeSSHClient

            client = NativeSSHClient(
                host=params['hostname'],
                user=params['user'],
                port=params['port'],
                key_file=params.get('key_file'),
                timeout=params['timeout'],
                proxy_jump=params.get('proxy_jump'),
                forward_agent=params.get('forward_agent', False),
                alias=alias
            )
        else:
            # 密码认证 → 使用 Paramiko
            try:
                from .paramiko_client import ParamikoClient
            except ImportError:
                from paramiko_client import ParamikoClient

            client = ParamikoClient(
                host=params['hostname'],
                user=params['user'],
                port=params['port'],
                password=params.get('password'),
                key_file=params.get('key_file'),
                timeout=params['timeout']
            )

        # 设置别名（用于守护进程标识）
        client.alias = alias

        return client

    @staticmethod
    def get_default_config_path() -> str:
        """获取默认 SSH config 路径"""
        return os.path.expanduser("~/.ssh/config")

    @staticmethod
    def get_default_metadata_path() -> str:
        """获取默认元数据路径"""
        return os.path.expanduser("~/.ssh/config_metadata.json")


# 向后兼容：提供全局函数接口
def get_config_loader_v3(config_path: Optional[str] = None,
                         metadata_path: Optional[str] = None) -> SSHConfigLoaderV3:
    """
    获取配置加载器实例

    Args:
        config_path: SSH config 文件路径
        metadata_path: 元数据文件路径

    Returns:
        SSHConfigLoaderV3 实例
    """
    return SSHConfigLoaderV3(config_path, metadata_path)