| import paramiko
import time
def ssh_connect(host, port=22, username='root', password=None, key_file=None):
"""
建立 SSH 连接。
参数:
host (str): 主机 IP 或域名
port (int): SSH 端口，默认 22
username (str): 用户名
password (str): 密码（与 key_file 二选一）
key_file (str): 私钥文件路径
返回:
paramiko.SSHClient: SSH 客户端对象
"""
# 创建 SSH 客户端
ssh = paramiko.SSHClient()
# 自动添加未知主机密钥（生产环境建议使用 known_hosts）
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
if key_file:
# 使用密钥连接
key = paramiko.RSAKey.from_private_key_file(key_file)
ssh.connect(
hostname=host,
port=port,
username=username,
pkey=key,
timeout=10
)
else:
# 使用密码连接
ssh.connect(
hostname=host,
port=port,
username=username,
password=password,
timeout=10
)
print(f"✅ 成功连接到 {host}")
return ssh
except Exception as e:
print(f"❌ 连接失败: {e}")
return None
def ssh_exec_command(ssh, command, timeout=30):
"""
执行 SSH 命令。
参数:
ssh: paramiko.SSHClient 对象
command (str): 要执行的命令
timeout (int): 超时时间（秒）
返回:
tuple: (stdin, stdout, stderr)
"""
try:
stdin, stdout, stderr = ssh.exec_command(command, timeout=timeout)
# 获取输出
stdout_text = stdout.read().decode('utf-8', errors='ignore')
stderr_text = stderr.read().decode('utf-8', errors='ignore')
return stdout_text, stderr_text
except Exception as e:
return '', str(e)
def ssh_disconnect(ssh):
"""
关闭 SSH 连接
"""
if ssh:
ssh.close()
print("🔌 SSH 连接已关闭")
if __name__ == '__main__':
# 测试连接（请修改为你的服务器信息）
# 这是模拟代码，实际使用需要真实的 SSH 服务器
# 方法1：密码连接
# ssh = ssh_connect('192.168.1.100', username='admin', password='password')
# 方法2：密钥连接
# ssh = ssh_connect('192.168.1.100', username='admin', key_file='id_rsa')
print("SSH 连接代码已准备好，请配置实际参数后使用") |
