Python 实现的 SSH 协议库工具Paramiko
·
Paramiko 是一个纯 Python 实现的 SSHv2 协议库，提供了客户端和服务器功能，支持加密连接、远程命令执行、SFTP 文件传输等。
主要特性
- SSH 客户端功能：连接远程服务器执行命令
- SFTP 客户端/服务器：安全的文件传输
- SSH 服务器功能：创建 SSH 服务器
- 密钥认证支持：RSA、DSA、ECDSA、Ed25519
- 代理支持：通过代理连接
- 端口转发：本地和远程端口转发
安装
# 基础安装
pip install paramiko
# 包含所有依赖（包括加密库）
pip install paramiko[all]
# 安装特定版本
pip install paramiko==3.2.0
核心功能与示例
1. 基础 SSH 连接
密码认证连接
import paramiko
import socket
from typing import Optional, Tuple
def basic_ssh_connection():
"""基础SSH连接示例"""
# SSH连接参数
hostname = 'your_server.com' # 替换为实际主机
port = 22
username = 'your_username'
password = 'your_password'
# 创建SSH客户端
client = paramiko.SSHClient()
try:
# 自动添加主机密钥（生产环境应验证）
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
print(f"正在连接到 {hostname}:{port}...")
# 建立连接
client.connect(
hostname=hostname,
port=port,
username=username,
password=password,
timeout=10
)
print("✓ 连接成功!")
# 获取连接信息
transport = client.get_transport()
if transport:
print(f"加密算法: {transport.get_cipher()}")
print(f"MAC算法: {transport.get_mac()}")
print(f"密钥交换算法: {transport.get_kex()}")
return client
except paramiko.AuthenticationException:
print("✗ 认证失败: 用户名或密码错误")
except paramiko.SSHException as e:
print(f"✗ SSH连接失败: {e}")
except socket.timeout:
print("✗ 连接超时")
except Exception as e:
print(f"✗ 连接错误: {e}")
return None
# 替代示例：本地测试
def local_test_connection():
"""本地测试连接（使用Docker或本地SSH服务器）"""
# 注意：实际使用时需要真实的SSH服务器
print("SSH连接示例需要真实的SSH服务器")
print("以下示例展示代码结构，实际运行时需要修改连接参数")
# 示例代码结构
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.WarningPolicy())
return client
# 运行示例
if __name__ == "__main__":
# 由于需要真实服务器，这里只展示代码结构
print("基础SSH连接示例:")
print("=" * 60)
# basic_ssh_connection() # 需要真实服务器
local_test_connection()
密钥认证连接
import paramiko
import os
from pathlib import Path
def key_based_authentication():
"""密钥认证连接示例"""
# 连接参数
hostname = 'your_server.com'
port = 22
username = 'your_username'
# 密钥文件路径
private_key_path = Path.home() / '.ssh' / 'id_rsa'
client = paramiko.SSHClient()
try:
# 设置主机密钥策略
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
print(f"使用密钥认证连接到 {hostname}...")
# 方法1: 直接使用密钥文件
if private_key_path.exists():
print(f"使用密钥文件: {private_key_path}")
private_key = paramiko.RSAKey.from_private_key_file(str(private_key_path))
client.connect(
hostname=hostname,
port=port,
username=username,
pkey=private_key,
timeout=10
)
# 方法2: 使用密码保护的密钥
elif private_key_path.with_suffix('.pub').exists():
print("尝试使用带密码的密钥...")
# 需要密码来解密密钥
key_password = "your_key_password" # 从安全的地方获取
private_key = paramiko.RSAKey.from_private_key_file(
str(private_key_path),
password=key_password
)
client.connect(
hostname=hostname,
port=port,
username=username,
pkey=private_key
)
else:
# 方法3: 使用SSH代理 (ssh-agent)
print("尝试使用SSH代理...")
# 连接到SSH代理
agent = paramiko.Agent()
agent_keys = agent.get_keys()
if len(agent_keys) > 0:
print(f"找到 {len(agent_keys)} 个代理密钥")
# 尝试每个密钥
for key in agent_keys:
try:
client.connect(
hostname=hostname,
port=port,
username=username,
pkey=key
)
print("✓ 使用代理密钥连接成功")
break
except paramiko.AuthenticationException:
continue
else:
print("✗ 所有代理密钥都认证失败")
return None
else:
print("✗ 未找到SSH代理或代理中没有密钥")
return None
print("✓ 密钥认证成功!")
# 显示连接信息
transport = client.get_transport()
if transport:
remote_version = transport.remote_version
print(f"远程SSH版本: {remote_version}")
print(f"会话ID: {transport.get_hex_session_id()}")
return client
except paramiko.AuthenticationException:
print("✗ 密钥认证失败")
except FileNotFoundError:
print(f"✗ 密钥文件未找到: {private_key_path}")
except paramiko.SSHException as e:
print(f"✗ SSH错误: {e}")
except Exception as e:
print(f"✗ 错误: {e}")
return None
def generate_ssh_keypair():
"""生成SSH密钥对示例"""
from paramiko.rsakey import RSAKey
import io
print("生成RSA密钥对...")
try:
# 生成2048位RSA密钥
key = RSAKey.generate(bits=2048)
# 获取私钥
private_key_io = io.StringIO()
key.write_private_key(private_key_io)
private_key = private_key_io.getvalue()
# 获取公钥
public_key = f"{key.get_name()} {key.get_base64()}"
print("✓ 密钥对生成成功!")
print(f"密钥类型: {key.get_name()}")
print(f"密钥长度: {key.get_bits()} 位")
print(f"公钥指纹: {key.get_fingerprint().hex()}")
# 保存到文件（示例）
print("\n公钥 (添加到 ~/.ssh/authorized_keys):")
print(public_key)
print("\n私钥 (保存到安全位置):")
# 注意：实际应用中不要打印私钥
# print(private_key[:100] + "...")
return key
except Exception as e:
print(f"✗ 生成密钥失败: {e}")
return None
def test_multiple_auth_methods():
"""测试多种认证方法"""
hostname = 'your_server.com'
port = 22
username = 'your_username'
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 尝试多种认证方法
auth_methods = []
# 1. 密钥认证
private_key_path = Path.home() / '.ssh' / 'id_rsa'
if private_key_path.exists():
try:
key = paramiko.RSAKey.from_private_key_file(str(private_key_path))
auth_methods.append({'pkey': key})
except:
pass
# 2. 密码认证（最后尝试）
# 从安全的地方获取密码
password = "your_password" # 示例，实际应从安全的地方获取
auth_methods.append({'password': password})
# 3. SSH代理
try:
agent = paramiko.Agent()
agent_keys = agent.get_keys()
for key in agent_keys:
auth_methods.append({'pkey': key})
except:
pass
print(f"尝试 {len(auth_methods)} 种认证方法...")
for i, auth in enumerate(auth_methods, 1):
method_name = list(auth.keys())[0]
print(f"\n尝试方法 {i}: {method_name}")
try:
client.connect(
hostname=hostname,
port=port,
username=username,
**auth,
timeout=5
)
print(f"✓ 认证成功! 使用方法: {method_name}")
return client
except paramiko.AuthenticationException:
print(f"✗ 认证失败: {method_name}")
continue
except Exception as e:
print(f"✗ 连接错误: {e}")
continue
print("\n✗ 所有认证方法都失败")
return None
# 运行示例
if __name__ == "__main__":
print("SSH密钥认证示例:")
print("=" * 60)
# 生成密钥对示例
print("1. 生成SSH密钥对:")
key = generate_ssh_keypair()
print("\n" + "=" * 60)
print("2. 密钥认证连接:")
# key_based_authentication() # 需要真实服务器
print("\n" + "=" * 60)
print("3. 多认证方法测试:")
# test_multiple_auth_methods() # 需要真实服务器
2. 远程命令执行
基本命令执行
import paramiko
import time
from typing import List, Tuple, Optional
class RemoteCommandExecutor:
"""远程命令执行器"""
def __init__(self, hostname: str, username: str,
password: str = None, pkey=None, port: int = 22):
self.hostname = hostname
self.port = port
self.username = username
self.password = password
self.pkey = pkey
self.client = None
self.transport = None
def connect(self) -> bool:
"""建立SSH连接"""
try:
self.client = paramiko.SSHClient()
self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
connect_kwargs = {
'hostname': self.hostname,
'port': self.port,
'username': self.username,
'timeout': 10
}
if self.password:
connect_kwargs['password'] = self.password
if self.pkey:
connect_kwargs['pkey'] = self.pkey
self.client.connect(**connect_kwargs)
self.transport = self.client.get_transport()
print(f"✓ 连接到 {self.hostname}:{self.port}")
return True
except Exception as e:
print(f"✗ 连接失败: {e}")
return False
def execute_command(self, command: str,
timeout: int = 30,
get_pty: bool = False) -> Tuple[int, str, str]:
"""
执行远程命令
Args:
command: 要执行的命令
timeout: 超时时间（秒）
get_pty: 是否获取伪终端
Returns:
(退出码, 标准输出, 标准错误)
"""
if not self.client:
raise RuntimeError("未建立连接，请先调用connect()")
stdin, stdout, stderr = None, None, None
try:
print(f"执行命令: {command}")
# 执行命令
stdin, stdout, stderr = self.client.exec_command(
command,
timeout=timeout,
get_pty=get_pty
)
# 读取输出
output = stdout.read().decode('utf-8', errors='ignore')
error = stderr.read().decode('utf-8', errors='ignore')
# 获取退出码
exit_status = stdout.channel.recv_exit_status()
return exit_status, output, error
except socket.timeout:
print(f"✗ 命令执行超时: {command}")
return -1, "", "命令执行超时"
except Exception as e:
print(f"✗ 命令执行错误: {e}")
return -1, "", str(e)
finally:
# 清理
for stream in [stdin, stdout, stderr]:
if stream:
stream.close()
def execute_command_with_sudo(self, command: str,
sudo_password: str = None,
timeout: int = 30) -> Tuple[int, str, str]:
"""
使用sudo执行命令
Args:
command: 要执行的命令（不需要包含sudo）
sudo_password: sudo密码
timeout: 超时时间
Returns:
(退出码, 标准输出, 标准错误)
"""
sudo_command = f"sudo -S {command}"
stdin, stdout, stderr = None, None, None
try:
print(f"使用sudo执行: {command}")
stdin, stdout, stderr = self.client.exec_command(
sudo_command,
timeout=timeout,
get_pty=True
)
# 如果需要，提供sudo密码
if sudo_password:
stdin.write(sudo_password + '\n')
stdin.flush()
# 读取输出
output = stdout.read().decode('utf-8', errors='ignore')
error = stderr.read().decode('utf-8', errors='ignore')
# 获取退出码
exit_status = stdout.channel.recv_exit_status()
return exit_status, output, error
except Exception as e:
print(f"✗ sudo命令执行错误: {e}")
return -1, "", str(e)
finally:
for stream in [stdin, stdout, stderr]:
if stream:
stream.close()
def execute_interactive_command(self, command: str) -> Tuple[int, str]:
"""
执行交互式命令
Args:
command: 要执行的命令
Returns:
(退出码, 输出)
"""
print(f"执行交互式命令: {command}")
# 创建交互式会话
channel = self.transport.open_session()
channel.get_pty()
channel.exec_command(command)
output = []
try:
while True:
if channel.recv_ready():
data = channel.recv(1024).decode('utf-8', errors='ignore')
output.append(data)
print(data, end='')
if channel.recv_stderr_ready():
data = channel.recv_stderr(1024).decode('utf-8', errors='ignore')
output.append(f"[STDERR] {data}")
print(f"[STDERR] {data}", end='')
if channel.exit_status_ready():
break
time.sleep(0.1)
except Exception as e:
print(f"✗ 交互命令错误: {e}")
exit_status = channel.recv_exit_status()
channel.close()
return exit_status, ''.join(output)
def execute_long_running_command(self, command: str,
callback=None,
timeout: int = 3600) -> Tuple[int, str]:
"""
执行长时间运行的命令
Args:
command: 要执行的命令
callback: 回调函数，用于处理实时输出
timeout: 超时时间
Returns:
(退出码, 输出)
"""
print(f"执行长时间命令: {command}")
channel = self.transport.open_session()
channel.settimeout(timeout)
channel.exec_command(command)
output = []
try:
while True:
if channel.recv_ready():
data = channel.recv(1024).decode('utf-8', errors='ignore')
output.append(data)
if callback:
callback(data)
else:
print(data, end='')
if channel.exit_status_ready():
break
# 检查通道是否关闭
if channel.closed:
break
time.sleep(0.1)
except socket.timeout:
print("✗ 命令执行超时")
channel.close()
return -1, ''.join(output)
except Exception as e:
print(f"✗ 命令执行错误: {e}")
channel.close()
return -1, ''.join(output)
exit_status = channel.recv_exit_status()
channel.close()
return exit_status, ''.join(output)
def get_system_info(self) -> dict:
"""获取系统信息"""
info = {}
# 系统信息
exit_code, output, error = self.execute_command('uname -a')
if exit_code == 0:
info['uname'] = output.strip()
# 内存信息
exit_code, output, error = self.execute_command('free -h')
if exit_code == 0:
info['memory'] = output.strip()
# 磁盘信息
exit_code, output, error = self.execute_command('df -h')
if exit_code == 0:
info['disk'] = output.strip()
# CPU信息
exit_code, output, error = self.execute_command('lscpu')
if exit_code == 0:
info['cpu'] = output.strip()
# 负载信息
exit_code, output, error = self.execute_command('uptime')
if exit_code == 0:
info['uptime'] = output.strip()
# 网络信息
exit_code, output, error = self.execute_command('ip addr')
if exit_code == 0:
info['network'] = output.strip()
return info
def close(self):
"""关闭连接"""
if self.client:
self.client.close()
print(f"✓ 关闭到 {self.hostname} 的连接")
self.client = None
self.transport = None
def __enter__(self):
"""上下文管理器入口"""
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""上下文管理器出口"""
self.close()
# 使用示例
def command_execution_examples():
"""命令执行示例"""
# 创建执行器（需要真实服务器）
executor = RemoteCommandExecutor(
hostname='your_server.com',
username='your_username',
password='your_password' # 或使用密钥
)
try:
# 连接
if not executor.connect():
return
print("\n1. 基本命令执行:")
print("-" * 40)
# 执行简单命令
exit_code, output, error = executor.execute_command('ls -la')
if exit_code == 0:
print(f"输出:\n{output}")
else:
print(f"错误: {error}")
print("\n2. 系统信息收集:")
print("-" * 40)
# 获取系统信息
info = executor.get_system_info()
for key, value in info.items():
print(f"\n{key.upper()}:\n{value[:200]}...")
print("\n3. 文件操作:")
print("-" * 40)
# 创建测试文件
test_content = "Hello from Paramiko!\nThis is a test file."
exit_code, output, error = executor.execute_command(
f'echo "{test_content}" > /tmp/paramiko_test.txt'
)
# 读取文件
exit_code, output, error = executor.execute_command(
'cat /tmp/paramiko_test.txt'
)
if exit_code == 0:
print(f"文件内容:\n{output}")
# 检查文件信息
exit_code, output, error = executor.execute_command(
'ls -la /tmp/paramiko_test.txt'
)
if exit_code == 0:
print(f"文件信息:\n{output}")
print("\n4. 进程检查:")
print("-" * 40)
# 检查运行中的进程
exit_code, output, error = executor.execute_command('ps aux | head -10')
if exit_code == 0:
print(f"前10个进程:\n{output}")
print("\n5. 网络检查:")
print("-" * 40)
# 检查网络连接
exit_code, output, error = executor.execute_command('netstat -tulpn | head -10')
if exit_code == 0:
print(f"网络连接:\n{output}")
print("\n6. 服务状态:")
print("-" * 40)
# 检查服务状态（如果有systemd）
exit_code, output, error = executor.execute_command(
'systemctl list-units --type=service --state=running | head -5'
)
if exit_code == 0:
print(f"运行中的服务:\n{output}")
print("\n7. 使用sudo:")
print("-" * 40)
# 清理测试文件（需要sudo）
sudo_password = input("输入sudo密码: ") if input("需要sudo密码? (y/n): ").lower() == 'y' else None
if sudo_password:
exit_code, output, error = executor.execute_command_with_sudo(
'rm /tmp/paramiko_test.txt',
sudo_password=sudo_password
)
if exit_code == 0:
print("✓ 测试文件已删除")
else:
print(f"✗ 删除失败: {error}")
print("\n8. 长时间运行命令监控:")
print("-" * 40)
# 定义回调函数处理实时输出
def output_callback(data):
print(f"[实时输出] {data}", end='')
# 执行长时间命令
exit_code, output = executor.execute_long_running_command(
'for i in {1..5}; do echo "Iteration $i"; sleep 1; done',
callback=output_callback
)
print(f"\n命令退出码: {exit_code}")
finally:
# 关闭连接
executor.close()
# 使用上下文管理器
def context_manager_example():
"""上下文管理器使用示例"""
with RemoteCommandExecutor(
hostname='your_server.com',
username='your_username',
password='your_password'
) as executor:
if executor.client:
print("在上下文管理器中执行命令...")
# 执行命令
exit_code, output, error = executor.execute_command('hostname')
if exit_code == 0:
print(f"主机名: {output.strip()}")
# 自动关闭连接
print("连接将在退出上下文时自动关闭")
# 运行示例
if __name__ == "__main__":
print("远程命令执行示例:")
print("=" * 60)
# 由于需要真实服务器，这里只展示代码结构
print("注意: 需要真实SSH服务器来运行这些示例")
print("取消注释以下代码并提供正确的连接信息:")
print("\n# command_execution_examples()")
print("# context_manager_example()")
交互式Shell
import paramiko
import select
import sys
import threading
import time
from typing import Optional, Callable
class InteractiveSSHShell:
"""交互式SSH Shell"""
def __init__(self, hostname: str, username: str,
password: str = None, pkey=None, port: int = 22):
self.hostname = hostname
self.port = port
self.username = username
self.password = password
self.pkey = pkey
self.client = None
self.channel = None
self.is_connected = False
def connect(self) -> bool:
"""建立连接并创建交互式Shell"""
try:
self.client = paramiko.SSHClient()
self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
connect_kwargs = {
'hostname': self.hostname,
'port': self.port,
'username': self.username,
'timeout': 10
}
if self.password:
connect_kwargs['password'] = self.password
if self.pkey:
connect_kwargs['pkey'] = self.pkey
print(f"连接到 {self.hostname}:{self.port}...")
self.client.connect(**connect_kwargs)
# 创建交互式Shell通道
self.channel = self.client.invoke_shell()
# 设置终端类型和大小
self.channel.get_pty(
term='xterm',
width=80,
height=24
)
# 激活交互模式
self.channel.invoke_shell()
# 等待欢迎信息
time.sleep(1)
self._read_initial_output()
self.is_connected = True
print("✓ 交互式Shell已就绪")
return True
except Exception as e:
print(f"✗ 连接失败: {e}")
return False
def _read_initial_output(self):
"""读取初始输出"""
if self.channel.recv_ready():
initial_output = self.channel.recv(4096).decode('utf-8', errors='ignore')
print(initial_output, end='')
def start_interactive_session(self,
local_echo: bool = True,
timeout: float = 0.1):
"""
启动交互式会话
Args:
local_echo: 是否本地回显输入
timeout: select超时时间
"""
if not self.is_connected:
print("未连接，请先调用connect()")
return
print("\n" + "="*60)
print("交互式SSH Shell")
print("输入 'exit' 或 Ctrl+D 退出")
print("="*60 + "\n")
try:
while True:
# 检查是否有数据可读
rlist, _, _ = select.select([self.channel, sys.stdin], [], [], timeout)
for r in rlist:
if r is self.channel:
# 有来自远程的数据
if self.channel.recv_ready():
data = self.channel.recv(4096).decode('utf-8', errors='ignore')
if data:
sys.stdout.write(data)
sys.stdout.flush()
elif r is sys.stdin:
# 本地输入
local_input = sys.stdin.readline()
if not local_input: # Ctrl+D
print("\n检测到EOF，退出...")
return
command = local_input.strip()
# 检查退出命令
if command.lower() in ['exit', 'quit', 'logout']:
print("退出交互式Shell...")
return
# 发送命令到远程
if local_echo:
# 如果本地已经回显了输入，就不需要发送了
# 但有时需要确保命令被正确发送
pass
# 发送命令（需要加上换行符）
self.channel.send(command + '\n')
# 小延迟，让远程有时间处理
time.sleep(0.1)
# 检查通道是否关闭
if self.channel.closed:
print("\n远程连接已关闭")
break
except KeyboardInterrupt:
print("\n\n检测到Ctrl+C，退出...")
except Exception as e:
print(f"\n会话错误: {e}")
finally:
print("交互式会话结束")
def execute_commands_interactively(self, commands: list):
"""以交互方式执行一系列命令"""
if not self.is_connected:
self.connect()
if not self.is_connected:
return
print(f"执行 {len(commands)} 个命令...")
for i, cmd in enumerate(commands, 1):
print(f"\n[{i}/{len(commands)}] 执行: {cmd}")
# 发送命令
self.channel.send(cmd + '\n')
# 等待并读取输出
output = self._read_command_output(timeout=5)
print(f"输出:\n{output}")
# 检查是否有错误
if "command not found" in output.lower():
print(f"警告: 命令可能不存在: {cmd}")
# 命令间的小延迟
time.sleep(0.5)
def _read_command_output(self, timeout: float = 2.0) -> str:
"""读取命令输出"""
output = []
start_time = time.time()
while time.time() - start_time < timeout:
if self.channel.recv_ready():
data = self.channel.recv(4096).decode('utf-8', errors='ignore')
output.append(data)
# 检查是否还有更多数据
time.sleep(0.1)
# 如果一段时间没有新数据，认为输出结束
if not self.channel.recv_ready():
# 再等待一小会儿确认
time.sleep(0.2)
if not self.channel.recv_ready():
break
return ''.join(output)
def upload_and_execute_script(self, script_content: str,
interpreter: str = "bash"):
"""上传并执行脚本"""
if not self.is_connected:
self.connect()
if not self.is_connected:
return
# 创建临时脚本文件
temp_script = f"/tmp/script_{int(time.time())}.sh"
# 发送创建脚本的命令
self.channel.send(f"cat > {temp_script} << 'EOF'\n")
time.sleep(0.1)
# 发送脚本内容
self.channel.send(script_content + "\nEOF\n")
time.sleep(0.1)
# 设置执行权限
self.channel.send(f"chmod +x {temp_script}\n")
time.sleep(0.1)
# 执行脚本
self.channel.send(f"{interpreter} {temp_script}\n")
# 读取输出
output = self._read_command_output(timeout=10)
# 清理脚本文件
self.channel.send(f"rm -f {temp_script}\n")
return output
def start_background_monitor(self, callback: Callable[[str], None],
interval: float = 1.0):
"""启动后台监控线程"""
def monitor_thread():
while self.is_connected and not self.channel.closed:
if self.channel.recv_ready():
data = self.channel.recv(4096).decode('utf-8', errors='ignore')
if data and callback:
callback(data)
time.sleep(interval)
thread = threading.Thread(target=monitor_thread, daemon=True)
thread.start()
return thread
def send_control_sequence(self, control_char: str):
"""发送控制序列"""
control_sequences = {
'ctrl_c': '\x03',
'ctrl_d': '\x04',
'ctrl_z': '\x1a',
'tab': '\t',
'enter': '\n',
'backspace': '\x7f',
'esc': '\x1b',
}
if control_char in control_sequences:
self.channel.send(control_sequences[control_char])
print(f"发送控制序列: {control_char}")
def resize_terminal(self, width: int, height: int):
"""调整终端大小"""
if self.channel:
self.channel.resize_pty(width=width, height=height)
print(f"终端大小调整为: {width}x{height}")
def close(self):
"""关闭连接"""
if self.channel:
self.channel.close()
if self.client:
self.client.close()
self.is_connected = False
print("连接已关闭")
def __enter__(self):
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
# 使用示例
def interactive_shell_example():
"""交互式Shell示例"""
# 创建Shell实例
shell = InteractiveSSHShell(
hostname='your_server.com',
username='your_username',
password='your_password'
)
try:
# 连接
if not shell.connect():
print("连接失败")
return
print("\n选择模式:")
print("1. 完全交互模式")
print("2. 执行预定义命令")
print("3. 上传并执行脚本")
choice = input("\n请选择 (1-3): ").strip()
if choice == '1':
# 完全交互模式
shell.start_interactive_session()
elif choice == '2':
# 执行预定义命令
commands = [
'pwd',
'whoami',
'ls -la',
'df -h',
'free -h',
'uptime',
'echo "命令执行完成"'
]
shell.execute_commands_interactively(commands)
elif choice == '3':
# 上传并执行脚本
script_content = """#!/bin/bash
echo "=== 系统信息 ==="
echo "主机名: $(hostname)"
echo "用户: $(whoami)"
echo "日期: $(date)"
echo ""
echo "=== 磁盘使用 ==="
df -h | grep -v tmpfs
echo ""
echo "=== 内存使用 ==="
free -h
echo ""
echo "脚本执行完成!"
"""
print("脚本内容:")
print(script_content)
print("\n执行脚本...")
output = shell.upload_and_execute_script(script_content)
print(f"脚本输出:\n{output}")
else:
print("无效选择")
except KeyboardInterrupt:
print("\n用户中断")
except Exception as e:
print(f"错误: {e}")
finally:
shell.close()
# 高级示例：带颜色和格式的交互式终端
class EnhancedInteractiveShell(InteractiveSSHShell):
"""增强的交互式Shell"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.command_history = []
self.current_dir = None
self.prompt_pattern = None
def start_enhanced_session(self):
"""启动增强的交互会话"""
if not self.is_connected:
if not self.connect():
return
print("\n" + "="*60)
print("增强交互式SSH Shell")
print("功能:")
print(" - 命令历史 (上下箭头)")
print(" - 自动补全 (Tab)")
print(" - 当前目录显示")
print(" - 语法高亮")
print("="*60 + "\n")
# 启动后台线程监控输出
import threading
def output_monitor():
while self.is_connected and self.channel and not self.channel.closed:
if self.channel.recv_ready():
data = self.channel.recv(4096).decode('utf-8', errors='ignore')
# 在这里可以添加输出处理逻辑
sys.stdout.write(data)
sys.stdout.flush()
# 尝试提取当前目录
self._extract_current_dir(data)
time.sleep(0.01)
monitor_thread = threading.Thread(target=output_monitor, daemon=True)
monitor_thread.start()
# 简单的前端循环
try:
while self.is_connected and self.channel and not self.channel.closed:
# 显示自定义提示符
prompt = self._get_prompt()
command = input(prompt)
if not command.strip():
continue
# 添加到历史
self.command_history.append(command)
# 处理特殊命令
if command.lower() in ['exit', 'quit']:
break
elif command.lower() == 'history':
self._show_history()
continue
elif command.lower() == 'clear':
print("\033[H\033[J", end='') # 清屏
continue
# 发送命令
self.channel.send(command + '\n')
# 等待命令执行
time.sleep(0.1)
except KeyboardInterrupt:
print("\n\n退出...")
except EOFError:
print("\n\n检测到EOF，退出...")
finally:
print("会话结束")
def _extract_current_dir(self, output: str):
"""从输出中提取当前目录"""
# 查找常见的提示符模式
import re
patterns = [
r'\[?([^\s@]+@[^\s]+)[:\s]([^\s>#]+)[>#]', # user@host:dir>
r'([^\s>#]+)[>#]\s*$' # dir>
]
for pattern in patterns:
match = re.search(pattern, output)
if match:
if len(match.groups()) > 1:
self.current_dir = match.group(2)
else:
self.current_dir = match.group(1)
break
def _get_prompt(self) -> str:
"""获取提示符"""
if self.current_dir:
# 显示当前目录
dir_name = self.current_dir.split('/')[-1] if '/' in self.current_dir else self.current_dir
return f"\033[32m{dir_name}\033[0m $ "
else:
return "\033[36mssh>\033[0m "
def _show_history(self):
"""显示命令历史"""
print("\n命令历史:")
for i, cmd in enumerate(self.command_history[-10:], 1):
print(f" {i:2}. {cmd}")
# 运行示例
if __name__ == "__main__":
print("交互式SSH Shell示例:")
print("=" * 60)
print("注意: 需要真实SSH服务器来运行这些示例")
print("提供正确的连接信息后取消注释以下代码:")
print("\n# interactive_shell_example()")
# 模拟演示
print("\n模拟演示代码结构:")
print("-" * 40)
# 创建模拟Shell类
class MockShell:
def __init__(self):
print("初始化SSH Shell...")
def connect(self):
print("模拟连接服务器...")
return True
def start_interactive_session(self):
print("模拟交互式会话:")
print(" 输入 'ls -la' 查看文件")
print(" 输入 'pwd' 查看当前目录")
print(" 输入 'exit' 退出")
def close(self):
print("关闭连接")
# 演示用法
shell = MockShell()
if shell.connect():
shell.start_interactive_session()
shell.close()
3. SFTP 文件传输
基础文件传输
import paramiko
import os
from pathlib import Path
import stat
import time
from typing import Optional, List, Tuple
import hashlib
class SFTPClient:
"""SFTP客户端"""
def __init__(self, hostname: str, username: str,
password: str = None, pkey=None, port: int = 22):
self.hostname = hostname
self.port = port
self.username = username
self.password = password
self.pkey = pkey
self.ssh_client = None
self.sftp_client = None
self.is_connected = False
def connect(self) -> bool:
"""建立SFTP连接"""
try:
self.ssh_client = paramiko.SSHClient()
self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
connect_kwargs = {
'hostname': self.hostname,
'port': self.port,
'username': self.username,
'timeout': 10
}
if self.password:
connect_kwargs['password'] = self.password
if self.pkey:
connect_kwargs['pkey'] = self.pkey
print(f"连接到 {self.hostname}:{self.port}...")
self.ssh_client.connect(**connect_kwargs)
# 创建SFTP客户端
self.sftp_client = self.ssh_client.open_sftp()
self.is_connected = True
print("✓ SFTP连接成功")
return True
except Exception as e:
print(f"✗ 连接失败: {e}")
return False
def upload_file(self, local_path: str, remote_path: str,
callback=None, confirm: bool = True) -> bool:
"""
上传文件到远程服务器
Args:
local_path: 本地文件路径
remote_path: 远程文件路径
callback: 进度回调函数
confirm: 是否确认覆盖
Returns:
是否成功
"""
if not self.is_connected:
print("未连接")
return False
local_path = Path(local_path)
if not local_path.exists():
print(f"✗ 本地文件不存在: {local_path}")
return False
if not local_path.is_file():
print(f"✗ 不是文件: {local_path}")
return False
try:
# 检查远程文件是否存在
remote_exists = False
try:
self.sftp_client.stat(remote_path)
remote_exists = True
except IOError:
pass
if remote_exists and confirm:
# 确认是否覆盖
response = input(f"远程文件已存在: {remote_path}\n是否覆盖? (y/n): ")
if response.lower() != 'y':
print("取消上传")
return False
# 获取文件大小用于进度显示
file_size = local_path.stat().st_size
print(f"上传文件: {local_path} -> {remote_path}")
print(f"文件大小: {file_size:,} 字节")
# 上传文件
if callback:
# 使用带进度回调的上传
self._upload_with_progress(str(local_path), remote_path,
callback, file_size)
else:
# 简单上传
self.sftp_client.put(str(local_path), remote_path)
print(f"✓ 上传成功: {remote_path}")
return True
except Exception as e:
print(f"✗ 上传失败: {e}")
return False
def _upload_with_progress(self, local_path: str, remote_path: str,
callback, total_size: int):
"""带进度显示的上传"""
# 创建自定义文件对象来跟踪进度
class ProgressFile:
def __init__(self, filepath, callback, total):
self.file = open(filepath, 'rb')
self.callback = callback
self.total = total
self.transferred = 0
self.start_time = time.time()
def read(self, size):
data = self.file.read(size)
if data:
self.transferred += len(data)
if self.callback:
self.callback(self.transferred, self.total)
return data
def close(self):
self.file.close()
progress_file = ProgressFile(local_path, callback, total_size)
try:
self.sftp_client.putfo(progress_file, remote_path)
finally:
progress_file.close()
def download_file(self, remote_path: str, local_path: str,
callback=None, confirm: bool = True) -> bool:
"""
从远程服务器下载文件
Args:
remote_path: 远程文件路径
local_path: 本地文件路径
callback: 进度回调函数
confirm: 是否确认覆盖
Returns:
是否成功
"""
if not self.is_connected:
print("未连接")
return False
local_path = Path(local_path)
try:
# 检查远程文件是否存在
remote_stat = self.sftp_client.stat(remote_path)
if not stat.S_ISREG(remote_stat.st_mode):
print(f"✗ 不是文件: {remote_path}")
return False
# 检查本地文件是否存在
if local_path.exists() and confirm:
response = input(f"本地文件已存在: {local_path}\n是否覆盖? (y/n): ")
if response.lower() != 'y':
print("取消下载")
return False
# 确保本地目录存在
local_path.parent.mkdir(parents=True, exist_ok=True)
print(f"下载文件: {remote_path} -> {local_path}")
print(f"文件大小: {remote_stat.st_size:,} 字节")
# 下载文件
if callback:
self._download_with_progress(remote_path, str(local_path),
callback, remote_stat.st_size)
else:
self.sftp_client.get(remote_path, str(local_path))
print(f"✓ 下载成功: {local_path}")
return True
except FileNotFoundError:
print(f"✗ 远程文件不存在: {remote_path}")
return False
except Exception as e:
print(f"✗ 下载失败: {e}")
return False
def _download_with_progress(self, remote_path: str, local_path: str,
callback, total_size: int):
"""带进度显示的下载"""
# 创建自定义文件对象来跟踪进度
class ProgressFile:
def __init__(self, filepath, callback, total):
self.file = open(filepath, 'wb')
self.callback = callback
self.total = total
self.transferred = 0
self.start_time = time.time()
def write(self, data):
self.file.write(data)
self.transferred += len(data)
if self.callback:
self.callback(self.transferred, self.total)
def close(self):
self.file.close()
progress_file = ProgressFile(local_path, callback, total_size)
try:
self.sftp_client.getfo(remote_path, progress_file)
finally:
progress_file.close()
def list_directory(self, remote_path: str = '.',
detailed: bool = False) -> List[dict]:
"""
列出远程目录内容
Args:
remote_path: 远程目录路径
detailed: 是否显示详细信息
Returns:
目录内容列表
"""
if not self.is_connected:
print("未连接")
return []
try:
items = []
for item in self.sftp_client.listdir_attr(remote_path):
item_info = {
'name': item.filename,
'size': item.st_size,
'uid': item.st_uid,
'gid': item.st_gid,
'mode': item.st_mode,
'atime': item.st_atime,
'mtime': item.st_mtime,
'is_dir': stat.S_ISDIR(item.st_mode),
'is_file': stat.S_ISREG(item.st_mode),
'is_link': stat.S_ISLNK(item.st_mode)
}
# 权限字符串
mode = item.st_mode
permissions = ''
# 文件类型
if stat.S_ISDIR(mode):
permissions += 'd'
elif stat.S_ISLNK(mode):
permissions += 'l'
else:
permissions += '-'
# 用户权限
permissions += 'r' if mode & stat.S_IRUSR else '-'
permissions += 'w' if mode & stat.S_IWUSR else '-'
permissions += 'x' if mode & stat.S_IXUSR else '-'
# 组权限
permissions += 'r' if mode & stat.S_IRGRP else '-'
permissions += 'w' if mode & stat.S_IWGRP else '-'
permissions += 'x' if mode & stat.S_IXGRP else '-'
# 其他权限
permissions += 'r' if mode & stat.S_IROTH else '-'
permissions += 'w' if mode & stat.S_IWOTH else '-'
permissions += 'x' if mode & stat.S_IXOTH else '-'
item_info['permissions'] = permissions
items.append(item_info)
return items
except Exception as e:
print(f"✗ 列出目录失败: {e}")
return []
def create_directory(self, remote_path: str,
parents: bool = True) -> bool:
"""
创建远程目录
Args:
remote_path: 远程目录路径
parents: 是否创建父目录
Returns:
是否成功
"""
if not self.is_connected:
print("未连接")
return False
try:
if parents:
# 递归创建目录
parts = remote_path.strip('/').split('/')
current_path = ''
for part in parts:
current_path = f"{current_path}/{part}" if current_path else part
try:
self.sftp_client.stat(current_path)
except FileNotFoundError:
self.sftp_client.mkdir(current_path)
else:
# 直接创建目录
self.sftp_client.mkdir(remote_path)
print(f"✓ 创建目录: {remote_path}")
return True
except Exception as e:
print(f"✗ 创建目录失败: {e}")
return False
def delete_file(self, remote_path: str) -> bool:
"""删除远程文件"""
if not self.is_connected:
print("未连接")
return False
try:
self.sftp_client.remove(remote_path)
print(f"✓ 删除文件: {remote_path}")
return True
except Exception as e:
print(f"✗ 删除文件失败: {e}")
return False
def delete_directory(self, remote_path: str,
recursive: bool = True) -> bool:
"""
删除远程目录
Args:
remote_path: 远程目录路径
recursive: 是否递归删除
Returns:
是否成功
"""
if not self.is_connected:
print("未连接")
return False
try:
if recursive:
# 递归删除目录内容
self._delete_directory_recursive(remote_path)
else:
# 只删除空目录
self.sftp_client.rmdir(remote_path)
print(f"✓ 删除目录: {remote_path}")
return True
except Exception as e:
print(f"✗ 删除目录失败: {e}")
return False
def _delete_directory_recursive(self, remote_path: str):
"""递归删除目录"""
for item in self.sftp_client.listdir_attr(remote_path):
item_path = f"{remote_path}/{item.filename}"
if stat.S_ISDIR(item.st_mode):
# 递归删除子目录
self._delete_directory_recursive(item_path)
else:
# 删除文件
self.sftp_client.remove(item_path)
# 删除空目录
self.sftp_client.rmdir(remote_path)
def rename(self, old_path: str, new_path: str) -> bool:
"""重命名文件或目录"""
if not self.is_connected:
print("未连接")
return False
try:
self.sftp_client.rename(old_path, new_path)
print(f"✓ 重命名: {old_path} -> {new_path}")
return True
except Exception as e:
print(f"✗ 重命名失败: {e}")
return False
def get_file_info(self, remote_path: str) -> Optional[dict]:
"""获取文件信息"""
if not self.is_connected:
print("未连接")
return None
try:
stat_result = self.sftp_client.stat(remote_path)
info = {
'path': remote_path,
'size': stat_result.st_size,
'uid': stat_result.st_uid,
'gid': stat_result.st_gid,
'mode': stat_result.st_mode,
'atime': stat_result.st_atime,
'mtime': stat_result.st_mtime,
'ctime': stat_result.st_ctime if hasattr(stat_result, 'st_ctime') else None,
'is_dir': stat.S_ISDIR(stat_result.st_mode),
'is_file': stat.S_ISREG(stat_result.st_mode),
'is_link': stat.S_ISLNK(stat_result.st_mode)
}
return info
except Exception as e:
print(f"✗ 获取文件信息失败: {e}")
return None
def calculate_md5(self, remote_path: str) -> Optional[str]:
"""计算远程文件的MD5哈希"""
if not self.is_connected:
print("未连接")
return None
try:
# 创建临时本地文件
import tempfile
import hashlib
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
temp_path = temp_file.name
try:
# 下载文件
self.sftp_client.get(remote_path, temp_path)
# 计算MD5
md5_hash = hashlib.md5()
with open(temp_path, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b''):
md5_hash.update(chunk)
return md5_hash.hexdigest()
finally:
# 清理临时文件
if os.path.exists(temp_path):
os.unlink(temp_path)
except Exception as e:
print(f"✗ 计算MD5失败: {e}")
return None
def upload_directory(self, local_dir: str, remote_dir: str,
callback=None) -> Tuple[int, int]:
"""
上传整个目录
Args:
local_dir: 本地目录
remote_dir: 远程目录
callback: 进度回调函数
Returns:
(成功数, 总数)
"""
local_dir = Path(local_dir)
if not local_dir.exists() or not local_dir.is_dir():
print(f"✗ 本地目录不存在: {local_dir}")
return 0, 0
# 创建远程目录
self.create_directory(remote_dir, parents=True)
total_files = 0
success_count = 0
# 遍历本地目录
for root, dirs, files in os.walk(local_dir):
# 计算相对路径
rel_path = Path(root).relative_to(local_dir)
remote_path = f"{remote_dir}/{rel_path}" if str(rel_path) != '.' else remote_dir
# 确保远程目录存在
self.create_directory(remote_path, parents=True)
# 上传文件
for file in files:
total_files += 1
local_file = Path(root) / file
remote_file = f"{remote_path}/{file}"
print(f"上传 [{total_files}]: {local_file} -> {remote_file}")
if self.upload_file(str(local_file), remote_file, callback=callback):
success_count += 1
print(f"✓ 目录上传完成: {success_count}/{total_files} 个文件")
return success_count, total_files
def download_directory(self, remote_dir: str, local_dir: str,
callback=None) -> Tuple[int, int]:
"""
下载整个目录
Args:
remote_dir: 远程目录
local_dir: 本地目录
callback: 进度回调函数
Returns:
(成功数, 总数)
"""
local_dir = Path(local_dir)
# 创建本地目录
local_dir.mkdir(parents=True, exist_ok=True)
total_files = 0
success_count = 0
def _download_recursive(remote_path, local_path):
nonlocal total_files, success_count
try:
items = self.sftp_client.listdir_attr(remote_path)
for item in items:
item_remote = f"{remote_path}/{item.filename}"
item_local = local_path / item.filename
if stat.S_ISDIR(item.st_mode):
# 递归下载子目录
item_local.mkdir(exist_ok=True)
_download_recursive(item_remote, item_local)
elif stat.S_ISREG(item.st_mode):
# 下载文件
total_files += 1
print(f"下载 [{total_files}]: {item_remote} -> {item_local}")
if self.download_file(item_remote, str(item_local), callback=callback):
success_count += 1
except Exception as e:
print(f"下载目录时出错 {remote_path}: {e}")
_download_recursive(remote_dir, local_dir)
print(f"✓ 目录下载完成: {success_count}/{total_files} 个文件")
return success_count, total_files
def sync_directory(self, local_dir: str, remote_dir: str,
direction: str = 'both') -> dict:
"""
同步目录
Args:
local_dir: 本地目录
remote_dir: 远程目录
direction: 同步方向 ('up', 'down', 'both')
Returns:
同步统计
"""
stats = {
'uploaded': 0,
'downloaded': 0,
'skipped': 0,
'errors': 0
}
# 这里可以实现更复杂的同步逻辑
# 包括比较文件修改时间、大小等
print("目录同步功能需要更复杂的实现")
print("可以考虑使用rsync或实现自定义同步逻辑")
return stats
def close(self):
"""关闭连接"""
if self.sftp_client:
self.sftp_client.close()
if self.ssh_client:
self.ssh_client.close()
self.is_connected = False
print("✓ SFTP连接已关闭")
def __enter__(self):
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
# 使用示例
def sftp_examples():
"""SFTP使用示例"""
# 创建SFTP客户端
sftp = SFTPClient(
hostname='your_server.com',
username='your_username',
password='your_password'
)
try:
# 连接
if not sftp.connect():
return
print("\n1. 列出目录内容:")
print("-" * 40)
# 列出当前目录
items = sftp.list_directory('.', detailed=True)
for item in items[:10]: # 显示前10个
size_str = f"{item['size']:,}" if item['is_file'] else "DIR"
print(f"{item['permissions']} {size_str:>10} {item['name']}")
print("\n2. 创建目录:")
print("-" * 40)
# 创建测试目录
test_dir = '/tmp/paramiko_test'
if sftp.create_directory(test_dir):
print(f"创建目录: {test_dir}")
print("\n3. 上传文件:")
print("-" * 40)
# 创建本地测试文件
local_test_file = 'test_upload.txt'
with open(local_test_file, 'w') as f:
f.write("This is a test file for SFTP upload.\n")
f.write("Created by Paramiko SFTP client.\n")
# 定义进度回调函数
def upload_progress(transferred, total):
percent = (transferred / total) * 100
print(f"\r进度: {transferred:,}/{total:,} 字节 ({percent:.1f}%)", end='')
# 上传文件
remote_test_file = f'{test_dir}/uploaded_file.txt'
if sftp.upload_file(local_test_file, remote_test_file,
callback=upload_progress):
print("\n上传完成!")
print("\n4. 下载文件:")
print("-" * 40)
# 定义下载进度回调
def download_progress(transferred, total):
percent = (transferred / total) * 100
print(f"\r进度: {transferred:,}/{total:,} 字节 ({percent:.1f}%)", end='')
# 下载文件
local_downloaded = 'downloaded_file.txt'
if sftp.download_file(remote_test_file, local_downloaded,
callback=download_progress):
print("\n下载完成!")
# 显示下载的文件内容
with open(local_downloaded, 'r') as f:
print(f"文件内容:\n{f.read()}")
print("\n5. 获取文件信息:")
print("-" * 40)
# 获取文件信息
file_info = sftp.get_file_info(remote_test_file)
if file_info:
print(f"文件: {file_info['path']}")
print(f"大小: {file_info['size']:,} 字节")
print(f"修改时间: {time.ctime(file_info['mtime'])}")
print(f"权限: {oct(file_info['mode'])[-3:]}")
print("\n6. 计算文件哈希:")
print("-" * 40)
# 计算MD5
md5_hash = sftp.calculate_md5(remote_test_file)
if md5_hash:
print(f"MD5哈希: {md5_hash}")
# 也计算本地文件的MD5进行对比
import hashlib
with open(local_downloaded, 'rb') as f:
local_md5 = hashlib.md5(f.read()).hexdigest()
print(f"本地MD5: {local_md5}")
print(f"匹配: {md5_hash == local_md5}")
print("\n7. 目录操作:")
print("-" * 40)
# 测试目录上传下载
print("创建测试目录结构...")
# 创建本地测试目录
local_test_dir = 'test_directory'
os.makedirs(local_test_dir, exist_ok=True)
# 创建一些测试文件
for i in range(3):
file_path = os.path.join(local_test_dir, f'file_{i}.txt')
with open(file_path, 'w') as f:
f.write(f"Test file {i}\n" * 10)
# 创建子目录
sub_dir = os.path.join(local_test_dir, 'subdir')
os.makedirs(sub_dir, exist_ok=True)
with open(os.path.join(sub_dir, 'subfile.txt'), 'w') as f:
f.write("Subdirectory file\n")
# 上传整个目录
print(f"上传目录: {local_test_dir}")
remote_test_dir = f'{test_dir}/uploaded_dir'
success, total = sftp.upload_directory(local_test_dir, remote_test_dir)
print(f"上传结果: {success}/{total} 文件")
# 下载整个目录
print(f"\n下载目录: {remote_test_dir}")
local_downloaded_dir = 'downloaded_directory'
success, total = sftp.download_directory(remote_test_dir, local_downloaded_dir)
print(f"下载结果: {success}/{total} 文件")
print("\n8. 清理:")
print("-" * 40)
# 清理远程文件
if sftp.delete_file(remote_test_file):
print(f"删除远程文件: {remote_test_file}")
# 递归删除远程目录
if sftp.delete_directory(test_dir, recursive=True):
print(f"删除远程目录: {test_dir}")
# 清理本地文件
for file in [local_test_file, local_downloaded]:
if os.path.exists(file):
os.remove(file)
print(f"删除本地文件: {file}")
# 清理本地目录
import shutil
for dir_path in [local_test_dir, local_downloaded_dir]:
if os.path.exists(dir_path):
shutil.rmtree(dir_path)
print(f"删除本地目录: {dir_path}")
print("\n✓ 所有测试完成!")
except Exception as e:
print(f"✗ 发生错误: {e}")
finally:
sftp.close()
# 高级示例：SFTP监控和实时同步
class SFTPMonitor:
"""SFTP文件监控器"""
def __init__(self, sftp_client: SFTPClient):
self.sftp = sftp_client
self.monitored_files = {}
self.is_monitoring = False
def start_monitoring(self, remote_path: str, interval: float = 5.0):
"""开始监控文件变化"""
print(f"开始监控: {remote_path}")
self.is_monitoring = True
import threading
def monitor_thread():
while self.is_monitoring:
try:
self._check_for_changes(remote_path)
time.sleep(interval)
except Exception as e:
print(f"监控错误: {e}")
time.sleep(interval)
thread = threading.Thread(target=monitor_thread, daemon=True)
thread.start()
return thread
def _check_for_changes(self, remote_path: str):
"""检查文件变化"""
try:
# 获取当前文件状态
current_files = {}
def _collect_files(path):
try:
items = self.sftp.sftp_client.listdir_attr(path)
for item in items:
item_path = f"{path}/{item.filename}"
if stat.S_ISDIR(item.st_mode):
_collect_files(item_path)
else:
current_files[item_path] = {
'size': item.st_size,
'mtime': item.st_mtime
}
except:
pass
_collect_files(remote_path)
# 检查变化
for path, info in current_files.items():
if path in self.monitored_files:
old_info = self.monitored_files[path]
if info['mtime'] != old_info['mtime']:
print(f"文件修改: {path}")
print(f" 大小: {old_info['size']:,} -> {info['size']:,}")
print(f" 时间: {time.ctime(old_info['mtime'])} -> {time.ctime(info['mtime'])}")
else:
print(f"新文件: {path}")
# 检查删除的文件
for path in list(self.monitored_files.keys()):
if path not in current_files:
print(f"文件删除: {path}")
# 更新监控记录
self.monitored_files = current_files
except Exception as e:
print(f"检查变化时出错: {e}")
def stop_monitoring(self):
"""停止监控"""
self.is_monitoring = False
print("监控已停止")
# 运行示例
if __name__ == "__main__":
print("SFTP文件传输示例:")
print("=" * 60)
print("注意: 需要真实SSH服务器来运行这些示例")
print("提供正确的连接信息后取消注释以下代码:")
print("\n# sftp_examples()")
# 模拟演示
print("\n模拟演示代码结构:")
print("-" * 40)
class MockSFTP:
def connect(self):
print("模拟SFTP连接...")
return True
def list_directory(self, path='.'):
print(f"列出目录: {path}")
return [
{'name': 'file1.txt', 'size': 1024, 'is_dir': False},
{'name': 'folder1', 'size': 0, 'is_dir': True},
{'name': 'file2.log', 'size': 2048, 'is_dir': False}
]
def upload_file(self, local, remote):
print(f"上传: {local} -> {remote}")
return True
def download_file(self, remote, local):
print(f"下载: {remote} -> {local}")
return True
def close(self):
print("关闭连接")
# 演示用法
sftp = MockSFTP()
if sftp.connect():
items = sftp.list_directory()
for item in items:
print(f" {item['name']} ({'DIR' if item['is_dir'] else 'FILE'})")
sftp.upload_file('local.txt', 'remote.txt')
sftp.download_file('remote.txt', 'local_copy.txt')
sftp.close()
4. SSH服务器
基础SSH服务器
import paramiko
import socket
import threading
import time
import logging
from typing import Optional, Dict, Any
from pathlib import Path
class SimpleSSHServer(paramiko.ServerInterface):
"""简单SSH服务器接口"""
def __init__(self):
self.event = threading.Event()
self.username = None
self.password = None
def check_channel_request(self, kind: str, chanid: int) -> int:
"""检查通道请求"""
if kind == 'session':
return paramiko.OPEN_SUCCEEDED
return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
def check_auth_password(self, username: str, password: str) -> int:
"""检查密码认证"""
print(f"认证尝试: {username}")
# 简单的认证逻辑
if username == 'admin' and password == 'password':
self.username = username
self.password = password
print(f"✓ 用户 {username} 认证成功")
return paramiko.AUTH_SUCCESSFUL
else:
print(f"✗ 用户 {username} 认证失败")
return paramiko.AUTH_FAILED
def check_auth_publickey(self, username: str, key: paramiko.PKey) -> int:
"""检查公钥认证"""
print(f"公钥认证尝试: {username}")
print(f"密钥类型: {key.get_name()}")
print(f"密钥指纹: {key.get_fingerprint().hex()}")
# 这里可以添加公钥验证逻辑
# 例如，检查公钥是否在授权列表中
return paramiko.AUTH_FAILED # 暂时禁用公钥认证
def check_channel_shell_request(self, channel) -> bool:
"""检查Shell请求"""
return True
def check_channel_pty_request(self, channel, term, width, height,
pixelwidth, pixelheight, modes) -> bool:
"""检查PTY请求"""
return True
def get_allowed_auths(self, username: str) -> str:
"""返回允许的认证方法"""
return "password,publickey"
def check_channel_exec_request(self, channel, command: bytes) -> bool:
"""检查执行命令请求"""
print(f"执行命令: {command.decode('utf-8')}")
# 处理命令
response = self._handle_command(command.decode('utf-8'))
# 发送响应
channel.send(response)
channel.send_exit_status(0)
return True
def _handle_command(self, command: str) -> str:
"""处理命令"""
cmd = command.strip().lower()
if cmd == 'help':
return """可用命令:
help - 显示此帮助
time - 显示当前时间
whoami - 显示当前用户
exit - 退出
"""
elif cmd == 'time':
return f"当前时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n"
elif cmd == 'whoami':
return f"当前用户: {self.username}\n"
elif cmd == 'exit':
return "再见!\n"
else:
return f"未知命令: {command}\n输入 'help' 查看可用命令\n"
class SSHServer:
"""SSH服务器"""
def __init__(self, host: str = '0.0.0.0', port: int = 2222,
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
30
0- 0
已为社区贡献6条内容
所有评论(0)