You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

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

火山引擎 最新活动