python3实现ftp服务功能（服务端 For Linux）

本文实例为大家分享了python3实现ftp服务功能的具体代码，供大家参考，具体内容如下

功能介绍：

可执行的命令：

ls
pwd
cd
put
rm
get
mkdir

1、用户加密认证

2、允许多用户同时登陆

3、每个用户有自己的家目录，且只可以访问自己的家目录

4、运行在自己家目录下随意切换目录

5、允许上传下载文件，且文件一致

6、传输过程中显示进度条
server main 代码：

# Author by Andy
# _*_ coding:utf-8 _*_
import os, sys, json, hashlib, socketserver, time

base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(base_dir)
from conf import userdb_set
class Ftp_server(socketserver.BaseRequestHandler):
 user_home_dir = ''
 def auth(self, *args):
  '''验证用户名及密码'''
  cmd_dic = args[0]
  username = cmd_dic["username"]
  password = cmd_dic["password"]
  f = open(userdb_set.userdb_set(), 'r')
  user_info = json.load(f)
  if username in user_info.keys():
   if password == user_info[username]:
    self.request.send('0'.encode())
    os.chdir('/home/%s' % username)
    self.user_home_dir = os.popen('pwd').read().strip()
    data = "%s login successed" % username
    self.loging(data)
   else:
    self.request.send('1'.encode())
    data = "%s login failed" % username
    self.loging(data)
    f.close
  else:
   self.request.send('1'.encode())
   data = "%s login failed" % username
   self.loging(data)
   f.close
   ##########################################

 def get(self, *args):
  '''给客户端传输文件'''
  request_code = {
   '0': 'file is ready to get',
   '1': 'file not found!'
  }
  cmd_dic = args[0]
  self.loging(json.dumps(cmd_dic))
  filename = cmd_dic["filename"]
  if os.path.isfile(filename):
   self.request.send('0'.encode('utf-8')) # 确认文件存在
   self.request.recv(1024)
   self.request.send(str(os.stat(filename).st_size).encode('utf-8'))
   self.request.recv(1024)
   m = hashlib.md5()
   f = open(filename, 'rb')
   for line in f:
    m.update(line)
    self.request.send(line)
   self.request.send(m.hexdigest().encode('utf-8'))
   print('From server:Md5 value has been sended!')
   f.close()
  else:
   self.request.send('1'.encode('utf-8'))
   ###########################################

 def cd(self, *args):
  '''执行cd命令'''
  user_current_dir = os.popen('pwd').read().strip()
  cmd_dic = args[0]
  self.loging(json.dumps(cmd_dic))
  path = cmd_dic['path']
  if path.startswith('/'):
   if self.user_home_dir in path:
    os.chdir(path)
    new_dir = os.popen('pwd').read()
    user_current_dir = new_dir
    self.request.send('Change dir successfully!'.encode("utf-8"))
    data = 'Change dir successfully!'
    self.loging(data)
   elif os.path.exists(path):
    self.request.send('Permission Denied!'.encode("utf-8"))
    data = 'Permission Denied!'
    self.loging(data)
   else:
    self.request.send('Directory not found!'.encode("utf-8"))
    data = 'Directory not found!'
    self.loging(data)
  elif os.path.exists(path):
   os.chdir(path)
   new_dir = os.popen('pwd').read().strip()
   if self.user_home_dir in new_dir:
    self.request.send('Change dir successfully!'.encode("utf-8"))
    user_current_dir = new_dir
    data = 'Change dir successfully!'
    self.loging(data)
   else:
    os.chdir(user_current_dir)
    self.request.send('Permission Denied!'.encode("utf-8"))
    data = 'Permission Denied!'
    self.loging(data)
  else:
   self.request.send('Directory not found!'.encode("utf-8"))
   data = 'Directory not found!'
   self.loging(data)
   ###########################################

 def rm(self, *args):
  request_code = {
   '0': 'file exist,and Please confirm whether to rm',
   '1': 'file not found!'
  }
  cmd_dic = args[0]
  self.loging(json.dumps(cmd_dic))
  filename = cmd_dic['filename']
  if os.path.exists(filename):
   self.request.send('0'.encode("utf-8")) # 确认文件存在
   client_response = self.request.recv(1024).decode()
   if client_response == '0':
    os.popen('rm -rf %s' % filename)
    self.request.send(('File %s has been deleted!' % filename).encode("utf-8"))
    self.loging('File %s has been deleted!' % filename)
   else:
    self.request.send(('File %s not deleted!' % filename).encode("utf-8"))
    self.loging('File %s not deleted!' % filename)
  else:
   self.request.send('1'.encode("utf-8"))
   ########################################

 def pwd(self, *args):
  '''执行pwd命令'''
  cmd_dic = args[0]
  self.loging(json.dumps(cmd_dic))
  server_response = os.popen('pwd').read().strip().encode("utf-8")
  self.request.send(server_response)

 #############################################
 def ls(self, *args):
  '''执行ls命名'''
  cmd_dic = args[0]
  self.loging(json.dumps(cmd_dic))
  path = cmd_dic['path']
  cmd = 'ls -l %s' % path
  server_response = os.popen(cmd).read().encode("utf-8")
  self.request.send(server_response)

 ############################################
 def put(self, *args):
  '''接收客户端文件'''
  cmd_dic = args[0]
  self.loging(json.dumps(cmd_dic))
  filename = cmd_dic["filename"]
  filesize = cmd_dic["size"]
  if os.path.isfile(filename):
   f = open(filename + '.new', 'wb')
  else:
   f = open(filename, 'wb')
  request_code = {
   '200': 'Ready to recceive data!',
   '210': 'Not ready to received data!'
  }
  self.request.send('200'.encode())
  receive_size = 0
  while True:
   if receive_size < filesize:
    data = self.request.recv(1024)
    f.write(data)
    receive_size += len(data)
   else:
    data = "File %s has been uploaded successfully!" % filename
    self.loging(data)
    print(data)
    break

    ################################################

 def mkdir(self, *args):
  request_code = {
   '0': 'Directory has been made!',
   '1': 'Directory is aleady exist!'
  }
  cmd_dic = args[0]
  self.loging(json.dumps(cmd_dic))
  dir_name = cmd_dic['dir_name']
  if os.path.exists(dir_name):
   self.request.send('1'.encode("utf-8"))
  else:
   os.popen('mkdir %s' % dir_name)
   self.request.send('0'.encode("utf-8"))

   #############################################

  def loging(self, data):
  '''日志记录'''
  localtime = time.asctime(time.localtime(time.time()))
  log_file = '/root/ftp/ftpserver/log/server.log'
  with open(log_file, 'a', encoding='utf-8') as f:
   f.write('%s-->' % localtime + data + '\n')
   ##############################################

 def handle(self):
  # print("您本次访问使用的IP为:%s" %self.client_address[0])
  # localtime = time.asctime( time.localtime(time.time()))
  # print(localtime)

  while True:
   try:
    self.data = self.request.recv(1024).decode() #
    # print(self.data)
    cmd_dic = json.loads(self.data)
    action = cmd_dic["action"]
    # print("用户请求%s"%action)
    if hasattr(self, action):
     func = getattr(self, action)
     func(cmd_dic)
   except Exception as e:
    self.loging(str(e))
    break

def run():
 HOST, PORT = '0.0.0.0', 6969
 print("The server is started,and listenning at port 6969")
 server = socketserver.ThreadingTCPServer((HOST, PORT), Ftp_server)
 server.serve_forever()
if __name__ == '__main__':
 run()

设置用户口令代码：

#Author by Andy
#_*_ coding:utf-8 _*_
import os,json,hashlib,sys

base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
userdb_file = base_dir+"\data\\userdb"
# print(userdb_file)
def userdb_set():
 if os.path.isfile(userdb_file):
  # print(userdb_file)
  return userdb_file
 else:
  print('请先为您的服务器创建用户！')
  user_data = {}
  dict={}
  Exit_flags = True
  while Exit_flags:
   username = input("Please input username:")
   if username != 'exit':
    password = input("Please input passwod:")
    if password != 'exit':
      user_data.update({username:password})
      m = hashlib.md5()
      # m.update('hello')
      # print(m.hexdigest())
      for i in user_data:
       # print(i,user_data[i])
       m.update(user_data[i].encode())
       dict.update({i:m.hexdigest()})
    else:
     break
   else:
    break
  f = open(userdb_file,'w')
  json.dump(dict,f)
  f.close()
 return userdb_file

目录结构：

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。

# python3  # ftp  # 服务端  # Linux  # Linux中避免客户端与服务端的端口冲突  # Linux下使用使用socket实现TCP服务端的示例代码  # Linux UDP服务端和客户端程序的实现  # Linux下安装SVN服务端的方法步骤  # linux查看服务端证书方式(keytool和openssl)  # 自己的  # 客户端  # 您的  # 请先  # 大家分享  # 多用户  # 可执行  # 具体内容  # 大家多多  # 过程中  # 进度条  # 传输文件  # 目录下  # open  # password  # cmd_dic  # username  # user_info  # request  # auth 

相关文章： c++23 std::expected怎么用 c++优雅处理函数错误返回【详解】  建站之星后台管理系统如何操作？  贸易公司网站制作流程,出口贸易网站设计怎么做？  网站制作中优化长尾关键字挖掘的技巧,建一个视频网站需要多少钱？  岳西云建站教程与模板下载_一站式快速建站系统操作指南  XML的“混合内容”是什么 怎么用DTD或XSD定义  高防服务器：AI智能防御DDoS攻击与数据安全保障  名字制作网站免费,所有小说网站的名字？  盐城做公司网站,江苏电子版退休证办理流程？  建站为何优先选择香港服务器？  微信h5制作网站有哪些,免费微信H5页面制作工具？  Android自定义listview布局实现上拉加载下拉刷新功能  微课制作网站有哪些,微课网怎么进？  电视网站制作tvbox接口,云海电视怎样自定义添加电视源？  香港服务器WordPress建站指南：SEO优化与高效部署策略  建站之星后台密码遗忘如何找回？  建站org新手必看：2024最新搭建流程与模板选择技巧  沈阳个人网站制作公司,哪个网站能考到沈阳事业编招聘的信息？  西安专业网站制作公司有哪些,陕西省建行官方网站？  矢量图网站制作软件,用千图网的一张矢量图做公司app首页，该网站并未说明版权等问题，这样做算不算侵权？应该如何解决？  建站之星如何优化SEO以实现高效排名？  网站制作公司,橙子建站是合法的吗？  建站之星云端配置指南：模板选择与SEO优化一键生成  如何在橙子建站中快速调整背景颜色？  广德云建站网站建设方案与建站流程优化指南  c++怎么编写动态链接库dll_c++ __declspec(dllexport)导出与调用【方法】  建站一年半SEO优化实战指南：核心词挖掘与长尾流量提升策略  成都响应式网站开发,dw怎么把手机适应页面变成网页？  如何快速重置建站主机并恢复默认配置？  可靠的网站设计制作软件,做网站设计需要什么样的电脑配置？  子杰智能建站系统｜零代码开发与AI生成SEO优化指南  宝塔建站后网页无法访问如何解决？  小建面朝正北，A点实际方位是否存在偏差？  建站主机如何选？性能与价格怎样平衡？  html制作网站的步骤有哪些,iapp如何添加网页？  如何快速配置高效服务器建站软件？  如何在腾讯云服务器快速搭建个人网站？  如何用花生壳三步快速搭建专属网站？  韩国服务器如何优化跨境访问实现高效连接？  高端网站建设与定制开发一站式解决方案 中企动力  建站主机选哪种环境更利于SEO优化？  成都品牌网站制作公司,成都营业执照年报网上怎么办理？  招商网站制作流程,网站招商广告语？  婚礼视频制作网站,学习*后期制作的网站有哪些？  如何通过FTP服务器快速搭建网站？  昆明高端网站制作公司,昆明公租房申请网上登录入口？  七夕网站制作视频,七夕大促活动怎么报名？  如何通过虚拟主机快速搭建个人网站？  学校为何禁止电信移动建设网站？  北京营销型网站制作公司,可以用python做一个营销推广网站吗？