🐍 第27课：SSH自动化

—— paramiko：纯 Python SSH 远程操作

📌 本课目标

1️⃣ SSH 连接基础

import paramiko

def create_ssh_client(host, port=22, username=None, password=None, key_path=None):
    """创建 SSH 客户端连接"""
    client = paramiko.SSHClient()
    # 自动添加主机密钥（首次连接不询问）
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    if key_path:
        # 密钥认证
        key = paramiko.RSAKey.from_private_key_file(key_path)
        client.connect(host, port=port, username=username, pkey=key)
    else:
        # 密码认证
        client.connect(host, port=port, username=username, password=password)
    
    return client

# 使用
client = create_ssh_client(
    host="192.168.1.100",
    username="root",
    password="your_password"
)

# 执行命令
stdin, stdout, stderr = client.exec_command("uname -a")
print(stdout.read().decode())

# 关闭连接
client.close()

2️⃣ 远程命令执行

import paramiko

class SSHExecutor:
    """SSH 命令执行器"""
    
    def __init__(self, host, username, password=None, key_path=None, port=22):
        self.host = host
        self.username = username
        self.client = paramiko.SSHClient()
        self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        
        if key_path:
            key = paramiko.RSAKey.from_private_key_file(key_path)
            self.client.connect(host, port=port, username=username, pkey=key)
        else:
            self.client.connect(host, port=port, username=username, password=password)
    
    def run(self, command, timeout=30):
        """执行命令并返回结果"""
        stdin, stdout, stderr = self.client.exec_command(command, timeout=timeout)
        exit_code = stdout.channel.recv_exit_status()
        output = stdout.read().decode()
        error = stderr.read().decode()
        
        return {
            "command": command,
            "exit_code": exit_code,
            "output": output,
            "error": error,
            "success": exit_code == 0,
        }
    
    def run_script(self, script_content):
        """远程执行脚本"""
        # 将脚本写入临时文件并执行
        cmd = f"cat << 'SCRIPT_EOF' | bash\n{script_content}\nSCRIPT_EOF"
        return self.run(cmd)
    
    def close(self):
        self.client.close()
    
    def __enter__(self):
        return self
    
    def __exit__(self, *args):
        self.close()

# 使用上下文管理器
# with SSHExecutor("192.168.1.100", "root", password="xxx") as ssh:
#     result = ssh.run("df -h")
#     print(result["output"])
#     result = ssh.run("free -m")
#     print(result["output"])

3️⃣ SFTP 文件传输

import paramiko
from pathlib import Path

class SFTPManager:
    """SFTP 文件传输管理"""
    
    def __init__(self, ssh_client):
        self.sftp = ssh_client.open_sftp()
    
    def upload(self, local_path, remote_path):
        """上传文件"""
        self.sftp.put(local_path, remote_path)
        print(f"✅ 上传: {local_path} → {remote_path}")
    
    def download(self, remote_path, local_path):
        """下载文件"""
        self.sftp.get(remote_path, local_path)
        print(f"✅ 下载: {remote_path} → {local_path}")
    
    def upload_directory(self, local_dir, remote_dir):
        """上传目录"""
        local_path = Path(local_dir)
        try:
            self.sftp.mkdir(remote_dir)
        except IOError:
            pass  # 目录已存在
        
        for f in local_path.rglob("*"):
            if f.is_file():
                rel = f.relative_to(local_path)
                remote_file = f"{remote_dir}/{rel}"
                # 确保远程目录存在
                remote_parent = str(Path(remote_file).parent)
                try:
                    self.sftp.stat(remote_parent)
                except IOError:
                    self._mkdir_p(remote_parent)
                self.sftp.put(str(f), remote_file)
                print(f"  ✅ {rel}")
    
    def _mkdir_p(self, path):
        """递归创建目录"""
        parts = path.split("/")
        current = ""
        for part in parts:
            current += f"/{part}"
            try:
                self.sftp.stat(current)
            except IOError:
                self.sftp.mkdir(current)
    
    def list_dir(self, remote_path="."):
        """列出远程目录"""
        return self.sftp.listdir(remote_path)
    
    def stat(self, remote_path):
        """获取文件信息"""
        return self.sftp.stat(remote_path)
    
    def close(self):
        self.sftp.close()

4️⃣ 批量服务器管理

import paramiko
from concurrent.futures import ThreadPoolExecutor, as_completed
import json

class ServerManager:
    """批量服务器管理"""
    
    def __init__(self, servers_config):
        """
        servers_config = [
            {"host": "192.168.1.100", "username": "root", "password": "xxx"},
            {"host": "192.168.1.101", "username": "root", "key_path": "/path/to/key"},
        ]
        """
        self.servers = servers_config
    
    def _execute_on_server(self, server, command):
        """在单台服务器上执行命令"""
        result = {"host": server["host"], "success": False}
        try:
            client = paramiko.SSHClient()
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            
            kwargs = {
                "hostname": server["host"],
                "port": server.get("port", 22),
                "username": server["username"],
            }
            if "key_path" in server:
                kwargs["pkey"] = paramiko.RSAKey.from_private_key_file(server["key_path"])
            else:
                kwargs["password"] = server["password"]
            
            client.connect(**kwargs)
            stdin, stdout, stderr = client.exec_command(command, timeout=30)
            result["output"] = stdout.read().decode()
            result["error"] = stderr.read().decode()
            result["exit_code"] = stdout.channel.recv_exit_status()
            result["success"] = result["exit_code"] == 0
            client.close()
        except Exception as e:
            result["error"] = str(e)
        return result
    
    def run_on_all(self, command, max_workers=5):
        """在所有服务器上并行执行命令"""
        results = []
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(self._execute_on_server, s, command): s
                for s in self.servers
            }
            for future in as_completed(futures):
                results.append(future.result())
        
        # 打印汇总
        success = sum(1 for r in results if r["success"])
        print(f"\n📊 执行结果: {success}/{len(results)} 成功")
        for r in results:
            status = "✅" if r["success"] else "❌"
            print(f"  {status} {r['host']}")
        
        return results

5️⃣ 验证脚本

#!/usr/bin/env python3
"""第27课 SSH自动化验证"""
import paramiko
import tempfile
import os

def test_ssh_key_generation():
    """SSH密钥生成测试"""
    key = paramiko.RSAKey.generate(2048)
    assert key.get_name() == "ssh-rsa"
    
    with tempfile.TemporaryDirectory() as tmpdir:
        key_path = os.path.join(tmpdir, "test_key")
        key.write_private_key_file(key_path)
        assert os.path.exists(key_path)
        
        # 重新加载
        loaded = paramiko.RSAKey.from_private_key_file(key_path)
        assert loaded.get_name() == "ssh-rsa"
    print("✅ SSH密钥生成测试通过")

def test_ssh_client_creation():
    """SSH客户端创建测试"""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    assert client is not None
    print("✅ SSH客户端创建测试通过")

def test_sftp_local():
    """SFTP本地模拟测试"""
    # 使用本地模拟验证SFTP接口
    key = paramiko.RSAKey.generate(2048)
    assert key.can_sign()
    print("✅ SFTP接口验证通过")

def test_paramiko_import():
    """模块导入测试"""
    assert hasattr(paramiko, "SSHClient")
    assert hasattr(paramiko, "RSAKey")
    assert hasattr(paramiko, "AutoAddPolicy")
    assert hasattr(paramiko, "SFTPClient")
    print("✅ paramiko模块导入测试通过")

if __name__ == "__main__":
    test_ssh_key_generation()
    test_ssh_client_creation()
    test_sftp_local()
    test_paramiko_import()
    print("\n🎉 第27课全部验证通过！")

6️⃣ SSH 配置管理

import paramiko
import json
from pathlib import Path

class SSHConfig:
    """SSH 连接配置管理"""
    
    def __init__(self, config_path="ssh_config.json"):
        self.config_path = Path(config_path)
        self._configs = self._load()
    
    def _load(self):
        if self.config_path.exists():
            with open(self.config_path) as f:
                return json.load(f)
        return {"servers": {}}
    
    def _save(self):
        with open(self.config_path, "w") as f:
            json.dump(self._configs, f, indent=2, ensure_ascii=False)
    
    def add_server(self, name, host, username, port=22, 
                   password=None, key_path=None, tags=None):
        self._configs["servers"][name] = {
            "host": host, "username": username, "port": port,
            "password": password, "key_path": key_path, "tags": tags or [],
        }
        self._save()
    
    def list_servers(self, tag=None):
        servers = self._configs["servers"]
        if tag:
            return {k: v for k, v in servers.items() if tag in v.get("tags", [])}
        return servers
    
    def connect(self, name):
        cfg = self._configs["servers"].get(name)
        if not cfg:
            raise ValueError(f"服务器 {name} 不存在")
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        kwargs = {"hostname": cfg["host"], "port": cfg["port"], "username": cfg["username"]}
        if cfg.get("key_path"):
            kwargs["pkey"] = paramiko.RSAKey.from_private_key_file(cfg["key_path"])
        else:
            kwargs["password"] = cfg.get("password")
        client.connect(**kwargs)
        return client

7️⃣ 远程部署脚本

import paramiko
from pathlib import Path

class RemoteDeployer:
    """远程部署工具"""
    
    def __init__(self, ssh_client):
        self.client = ssh_client
    
    def deploy(self, local_dir, remote_dir, exclude=None):
        exclude = exclude or [".git", "__pycache__", ".venv", "*.pyc"]
        local = Path(local_dir)
        sftp = self.client.open_sftp()
        
        stdin, stdout, stderr = self.client.exec_command(f"mkdir -p {remote_dir}")
        
        uploaded = 0
        for f in local.rglob("*"):
            if f.is_file():
                rel = str(f.relative_to(local))
                if any(ex in rel for ex in exclude):
                    continue
                remote_path = f"{remote_dir}/{rel}"
                remote_parent = str(Path(remote_path).parent)
                stdin, stdout, stderr = self.client.exec_command(f"mkdir -p {remote_parent}")
                sftp.put(str(f), remote_path)
                uploaded += 1
                print(f"  ✅ {rel}")
        
        sftp.close()
        print(f"部署完成: {uploaded} 个文件")
    
    def restart_service(self, service_name):
        stdin, stdout, stderr = self.client.exec_command(f"systemctl restart {service_name}")
        exit_code = stdout.channel.recv_exit_status()
        return exit_code == 0

8️⃣ SSH 安全最佳实践

"""SSH 安全最佳实践：
1. 禁用密码登录，只用密钥
2. 禁用 root 直接登录
3. 修改默认端口 22
4. 使用 fail2ban 防暴力破解
5. 使用跳板机/堡垒机
6. 定期轮换密钥
7. 使用 ssh-agent 管理密钥
8. 配置 AllowUsers 限制用户
"""

# 生成 SSH 密钥对
import paramiko
import os

def generate_ssh_key(key_path, key_type="rsa", bits=4096, passphrase=None):
    """生成 SSH 密钥对"""
    if key_type == "rsa":
        key = paramiko.RSAKey.generate(bits)
    elif key_type == "ed25519":
        key = paramiko.Ed25519Key.generate()
    
    # 保存私钥
    key.write_private_key_file(key_path, password=passphrase)
    
    # 保存公钥
    pub_path = key_path + ".pub"
    with open(pub_path, "w") as f:
        f.write(f"{key.get_name()} {key.get_base64()}")
    
    os.chmod(key_path, 0o600)  # 私钥权限必须 600
    os.chmod(pub_path, 0o644)
    
    print(f"✅ 密钥已生成:")
    print(f"  私钥: {key_path}")
    print(f"  公钥: {pub_path}")

# SSH 配置检查
def check_ssh_config(ssh_client):
    """检查远程 SSH 配置安全性"""
    checks = [
        ("PasswordAuthentication", "no", "密码认证"),
        ("PermitRootLogin", "no", "Root登录"),
        ("Port", "22", "SSH端口"),
        ("PubkeyAuthentication", "yes", "密钥认证"),
    ]
    
    results = []
    for directive, expected, desc in checks:
        cmd = f"grep -E '^{directive}' /etc/ssh/sshd_config"
        stdin, stdout, stderr = ssh_client.exec_command(cmd)
        actual = stdout.read().decode().strip()
        is_ok = expected in actual
        results.append({
            "directive": directive,
            "expected": expected,
            "actual": actual,
            "ok": is_ok,
            "desc": desc,
        })
        status = "✅" if is_ok else "⚠️"
        print(f"  {status} {desc}: {actual or '未配置'}")
    
    return results

🔑 本课要点