如何使用python写一个ftp服务器
这段代码是用 Python 搭建一个简单的 FTP 服务器,使用的是 pyftpdlib
库。下面我会逐行详细解释每一部分的含义和作用:
📦 导入模块
from cgitb import handler
from ftplib import FTP
cgitb.handler
:这个模块用于 CGI 脚本的错误处理,在这里其实没有用到,可以删除。ftplib.FTP
:用于客户端连接 FTP 服务器,在这段代码中也没有用到,可以删除。
from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import FTPHandler
from pyftpdlib.servers import FTPServer
这些是 pyftpdlib
的核心模块:
DummyAuthorizer
:用于管理用户权限(用户名、密码、目录、权限等)。FTPHandler
:处理 FTP 请求的类。FTPServer
:创建并运行 FTP 服务器。
🌐 获取本机 IP 地址
import socket
r = socket.gethostbyname(socket.gethostname())
print("本机IP:", r)
- 使用
socket
模块获取当前主机的 IP 地址。 gethostname()
获取主机名,gethostbyname()
将主机名解析为 IP 地址。- 打印出本机 IP,方便你知道 FTP 服务器在哪个地址运行。
👤 设置用户权限
authorizer = DummyAuthorizer()
创建一个用户权限管理器。
authorizer.adduser("user", "123456", "D:\pythonstudy\output\pub", perm="elradfmwMT")
添加一个用户:
- 用户名:
user
- 密码:
123456
- 主目录:
D:\python_study\output\pub
- 权限:
elradfmwMT
,每个字母代表一种权限:e
= 更改目录l
= 列出文件r
= 读取文件a
= 上传文件d
= 删除文件f
= 重命名文件m
= 创建目录w
= 写入文件M
= 更改文件权限T
= 更改时间戳
authorizer.addanonymous("D:\pythonstudy\output\nobody")
添加一个匿名用户(不需要用户名和密码),访问的目录是 nobody
文件夹。
⚙️ 设置处理器和服务器
handler = FTPHandler
handler.authorizer = authorizer
- 创建一个 FTP 请求处理器,并绑定上面设置的用户权限。
handler.passive_ports = range(10000, 20000)
- 设置被动模式端口范围(FTP 的数据传输端口),防止防火墙阻止连接。
server = FTPServer(("", 2121), handler)
- 创建 FTP 服务器:
- 地址:
""
表示监听所有 IP 地址(0.0.0.0) - 端口:
2121
(默认 FTP 是 21,这里用了自定义端口)
- 地址:
server.serve_forever()
- 启动服务器,进入无限循环,等待客户端连接。
✅ 总结
这段代码的作用是:
在本机搭建一个 FTP 服务器,监听 2121 端口,支持一个用户(user)和匿名访问,并设置了详细的权限和端口范围。
🔐 安全提示
- FTP 是明文传输,不安全,建议使用 SFTP(基于 SSH)或 FTPS(加密的 FTP)。
- 密码不要用简单的
123456
,建议使用强密码。 - 端口范围要在防火墙中开放,否则客户端可能连接不上。
下面上完整代码
from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import FTPHandler
from pyftpdlib.servers import FTPServer
import socket
r = socket.gethostbyname(socket.gethostname())
print("本机IP:",r)
authorizer = DummyAuthorizer()
authorizer.add_user("user", "123456", "D:\python_study\output\pub", perm="elradfmwMT")
authorizer.add_anonymous("D:\\python_study\output\\nobody")
handler = FTPHandler
handler.authorizer = authorizer
handler.passive_ports = range(10000, 20000)
server = FTPServer(("", 2121), handler)
server.serve_forever()
使用ftp客户端上传文件一切正常

下面将改进一下代码,会使ftp服务器更加安全,以及加入日志和安全通知等功能
为了让你的 FTP 服务器更安全,并具备日志记录和上传通知功能,我们可以对原始代码进行以下几个方面的改进:
✅ 安全性增强
- 使用 FTPS(FTP over SSL):
pyftpdlib
支持 TLS 加密,可以防止用户名、密码和文件内容被明文传输。- 你需要一个 SSL 证书(可以使用自签名证书用于测试)。
- 限制匿名访问:
- 如果不需要匿名用户,可以移除
add_anonymous()
,避免潜在的安全风险。
- 如果不需要匿名用户,可以移除
- 限制用户权限:
- 根据实际需求只开放必要的权限,比如只允许上传和下载,不允许删除或修改。
📋 日志记录功能
pyftpdlib
支持标准 Python 的logging
模块,可以记录:- 用户登录/退出
- 文件上传/下载
- 错误信息
🔔 上传通知功能
- 可以通过重写
on_file_received()
方法,在文件上传完成后执行自定义操作,比如:- 打印日志
- 发送邮件或消息通知
- 调用其他系统接口
🛠️ 改进后的示例代码
下面是一个更安全、带日志和上传通知的 FTP 服务器示例:
from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import TLS_FTPHandler
from pyftpdlib.servers import FTPServer
import logging
# 设置日志记录
logging.basicConfig(filename='ftp_server.log', level=logging.INFO, format='%(asctime)s - %(message)s')
# 创建用户权限
authorizer = DummyAuthorizer()
authorizer.add_user("user", "strong_password", "D:/python_study/output/pub", perm="elradfmw") # 权限更严格
# 不添加匿名用户以增强安全性
# 创建处理器
class CustomHandler(TLS_FTPHandler):
def on_connect(self):
logging.info(f"连接:{self.remote_ip}")
def on_login(self, username):
logging.info(f"登录:{username}")
def on_logout(self, username):
logging.info(f"退出:{username}")
def on_file_received(self, file):
logging.info(f"文件上传成功:{file}")
# 你可以在这里添加通知逻辑,比如发送邮件或调用 API
handler = CustomHandler
handler.authorizer = authorizer
# 设置 TLS(需要证书)
handler.certfile = "cert.pem" # 自签名证书路径
handler.tls_control_required = True
handler.tls_data_required = True
# 设置被动端口范围
handler.passive_ports = range(10000, 10100)
# 启动服务器
server = FTPServer(("0.0.0.0", 2121), handler)
server.serve_forever()
🔐 如何生成自签名证书(用于测试)
你可以使用 OpenSSL 命令生成一个简单的证书:
继续改进加入webhook通知
from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import TLS_FTPHandler
from pyftpdlib.servers import FTPServer
import logging
import requests
# 设置日志记录
logging.basicConfig(filename='ftp_server.log', level=logging.INFO, format='%(asctime)s - %(message)s')
# Webhook URL(请替换为你的实际地址)
WEBHOOK_URL = "https://your-webhook-url.com/notify"
# 创建用户权限
authorizer = DummyAuthorizer()
authorizer.add_user("user", "strong_password", "D:/python_study/output/pub", perm="elradfmw")
# 自定义处理器
class CustomHandler(TLS_FTPHandler):
def on_connect(self):
logging.info(f"连接:{self.remote_ip}")
def on_login(self, username):
logging.info(f"登录:{username}")
def on_logout(self, username):
logging.info(f"退出:{username}")
def on_file_received(self, file):
logging.info(f"文件上传成功:{file}")
try:
response = requests.post(WEBHOOK_URL, json={
"event": "file_uploaded",
"filename": file,
"user": self.username,
"ip": self.remote_ip
})
if response.status_code == 200:
logging.info("Webhook 通知成功")
else:
logging.warning(f"Webhook 通知失败:{response.status_code}")
except Exception as e:
logging.error(f"Webhook 请求异常:{e}")
# 设置处理器
handler = CustomHandler
handler.authorizer = authorizer
handler.certfile = "cert.pem" # TLS 证书路径
handler.tls_control_required = True
handler.tls_data_required = True
handler.passive_ports = range(10000, 10100)
# 启动服务器
server = FTPServer(("0.0.0.0", 2121), handler)
server.serve_forever()