Paramiko:SFTP跨平台传输文件时Windows路径报错问题
我来分享下针对你这套跨平台服务器-客户端架构的实操思路和关键代码片段,毕竟我之前也搭过类似的多服务器分工的环境,踩过不少坑:
一、基于 Paramiko 的 SSH 监听器(部署在 macOS/Ubuntu)
首先得确保服务器上安装了 Paramiko:
pip install paramiko
核心实现思路是启动一个 SSH 服务端,监听指定端口,处理客户端的连接和命令执行请求。这里给你一个极简但可用的示例代码:
import paramiko import socket from paramiko.server import SSHServerInterface from paramiko.channel import Channel class MySSHServer(SSHServerInterface): def check_auth_password(self, username, password): # 这里可以替换成你的用户认证逻辑,比如从数据库读取 if username == "your_user" and password == "your_pass": return paramiko.AUTH_SUCCESSFUL return paramiko.AUTH_FAILED def check_channel_request(self, kind, chanid): if kind == "session": return paramiko.OPEN_SUCCEEDED return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED def check_channel_exec_request(self, channel, command): # 处理客户端的命令执行请求 try: import subprocess result = subprocess.check_output(command.decode(), shell=True, stderr=subprocess.STDOUT) channel.send(result) except Exception as e: channel.send(str(e).encode()) channel.close() return True def start_ssh_server(host, port): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((host, port)) sock.listen(100) print(f"SSH server listening on {host}:{port}") while True: client_sock, client_addr = sock.accept() print(f"New connection from {client_addr}") try: transport = paramiko.Transport(client_sock) host_key = paramiko.RSAKey.generate(2048) transport.add_server_key(host_key) server = MySSHServer() transport.start_server(server=server) except Exception as e: print(f"Connection failed: {e}") client_sock.close() if __name__ == "__main__": start_ssh_server("0.0.0.0", 2222) # 用自定义端口避免和系统SSH冲突
注意事项:
- 生产环境一定要替换掉硬编码的密码,改用密钥认证(可以用
check_auth_publickey方法实现) - 监听
0.0.0.0允许所有子网内机器访问,如果你只想让特定IP连接,可以设置对应的host - 记得在服务器防火墙开放你指定的端口(比如2222),macOS用
pfctl配置,Ubuntu用ufw allow 2222/tcp
二、独立 SFTP 服务器(部署在另一台 Unix 机器)
同样用 Paramiko 实现专门的 SFTP 服务,这样能更灵活控制文件操作权限:
先安装依赖:
pip install paramiko
核心代码示例:
import paramiko import os import socket from paramiko.sftp_server import SFTPServerInterface from paramiko.sftp_attr import SFTPAttributes class MySFTPServer(SFTPServerInterface): def __init__(self, server, channel, name, root="/path/to/sftp/root"): super().__init__(server, channel, name) self.root = os.path.abspath(root) # 确保根目录存在 os.makedirs(self.root, exist_ok=True) def _real_path(self, path): # 防止客户端访问根目录外的文件(路径遍历防护) real_path = os.path.abspath(os.path.join(self.root, path.lstrip("/"))) if not real_path.startswith(self.root): raise paramiko.SFTPError(paramiko.SFTP_PERMISSION_DENIED, "Access denied") return real_path def list_folder(self, path): real_path = self._real_path(path) if not os.path.isdir(real_path): raise paramiko.SFTPError(paramiko.SFTP_NO_SUCH_FILE, "Not a directory") files = [] for f in os.listdir(real_path): full_path = os.path.join(real_path, f) attr = SFTPAttributes.from_stat(os.stat(full_path)) attr.filename = f files.append(attr) return files def stat(self, path): real_path = self._real_path(path) if not os.path.exists(real_path): raise paramiko.SFTPError(paramiko.SFTP_NO_SUCH_FILE, "File not found") return SFTPAttributes.from_stat(os.stat(real_path)) def open(self, path, flags, attr): real_path = self._real_path(path) mode = "" if flags & os.O_RDWR: mode += "r+" elif flags & os.O_WRONLY: mode += "w" else: mode += "r" if flags & os.O_CREAT: mode += "x" if flags & os.O_EXCL else "a" try: file_obj = open(real_path, mode) return file_obj except Exception as e: raise paramiko.SFTPError(paramiko.SFTP_PERMISSION_DENIED, str(e)) def start_sftp_server(host, port, username, password): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((host, port)) sock.listen(100) print(f"SFTP server listening on {host}:{port}") while True: client_sock, client_addr = sock.accept() print(f"New SFTP connection from {client_addr}") try: transport = paramiko.Transport(client_sock) host_key = paramiko.RSAKey.generate(2048) transport.add_server_key(host_key) # 启动服务并等待认证 transport.start_server(server=None) channel = transport.accept(20) if not channel: print("No channel received") transport.close() continue # 检查密码认证 if transport.auth_password(username, password) != paramiko.AUTH_SUCCESSFUL: print("Authentication failed") channel.close() transport.close() continue # 初始化SFTP服务 sftp_server = MySFTPServer(transport, channel, "sftp", root="/var/sftp/files") sftp_server.start() except Exception as e: print(f"SFTP connection failed: {e}") client_sock.close() if __name__ == "__main__": start_sftp_server("0.0.0.0", 2223, "sftp_user", "sftp_pass")
关键提醒:
- 一定要实现
_real_path的路径校验,不然客户端能遍历服务器任意目录,这是严重的安全漏洞 - 可以把根目录设置为无执行权限的目录,进一步提升安全性
- 同样,生产环境建议改用密钥认证,避免密码泄露
三、Windows 10 客户端实现
Windows 上用 Python 写客户端很方便,先安装 Paramiko:
pip install paramiko
SSH 客户端(执行命令)
import paramiko def ssh_client(host, port, username, password, command): ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: ssh.connect(host, port, username, password) stdin, stdout, stderr = ssh.exec_command(command) print("Command output:") print(stdout.read().decode()) if stderr.read(): print("Error:") print(stderr.read().decode()) except Exception as e: print(f"SSH connection failed: {e}") finally: ssh.close() # 示例调用 ssh_client("192.168.1.100", 2222, "your_user", "your_pass", "ls -l")
SFTP 客户端(上传/下载文件)
import paramiko def sftp_client(host, port, username, password, local_path, remote_path, action="upload"): transport = paramiko.Transport((host, port)) try: transport.connect(username=username, password=password) sftp = paramiko.SFTPClient.from_transport(transport) if action == "upload": sftp.put(local_path, remote_path) print(f"Uploaded {local_path} to {remote_path} successfully") elif action == "download": sftp.get(remote_path, local_path) print(f"Downloaded {remote_path} to {local_path} successfully") except Exception as e: print(f"SFTP operation failed: {e}") finally: transport.close() # 示例:上传文件到SFTP服务器 sftp_client("192.168.1.101", 2223, "sftp_user", "sftp_pass", "C:\\Users\\User\\test.txt", "/test.txt") # 示例:下载文件到本地 # sftp_client("192.168.1.101", 2223, "sftp_user", "sftp_pass", "C:\\Users\\User\\downloaded.txt", "/remote_file.txt", action="download")
客户端注意点:
- 如果遇到连接超时,先检查Windows防火墙是否允许Python程序出站访问对应的端口
- 要是用密钥认证,可以把
connect方法里的password换成key_filename="C:\\Users\\User\\.ssh\\id_rsa"
内容的提问来源于stack exchange,提问作者npc




