关于FTP大家应该不陌生，那么FTPS是什么东西呢？ 简单一句话概括：类似于http 和https的关系。也就是说FTPS就是加密过的FTP。
公司项目需要，需要从使用FTPS来下载装置保存的各种文件，所以研究了下FTPS。
首先介绍一下，python有专门ftp相关的第三方库：ftplib 关于ftplib的信息网上一搜一大把，百度一下就知道其包含哪些方法，这里不做赘述。
ftplib中有两个大类：FTP 和 FTP_TLS,其中FTP_TLS的类，是专门用于FTPS的；该类其实也是继承了FTP类的；也就是FTP的子类，这里大家可以看下源码即可了解。
下面直接上代码吧，边看边解释：
在使用之前需要先导入FTP_TLS
from ftplib import FTP_TLS?由于连接时报错，所以需要在源码中connect方法中添加一行：
self.af = self.sock.family
self.sock = ssl.wrap_socket(self.sock, self.keyfile, self.certfile, ssl_version=ssl.PROTOCOL_TLSv1) #这行
self.file = self.sock.makefile('r', encoding=self.encoding)为了便于项目使用就在项目代码中将该方法重写，代码如下：
class _FTPS(FTP_TLS):
def __init__(self, *args):
super().__init__(*args)
def connect(self, host='', port=0, timeout=-999, source_address=None):
'''The original connect function could not successfully connect our device, so we
reconstructed this function'''
if host != '':
self.host = host
if port > 0:
self.port = port
if timeout != -999:
self.timeout = timeout
if source_address is not None:
self.source_address = source_address
self.sock = socket.create_connection((self.host, self.port), self.timeout, source_address=self.source_address)
self.af = self.sock.family
self.sock = ssl.wrap_socket(self.sock, self.keyfile, self.certfile, ssl_version=ssl.PROTOCOL_TLSv1)
self.file = self.sock.makefile('r', encoding=self.encoding)
self.welcome = self.getresp()
return self.welcome
def nlst(self, *args):
'''The original nlst method in ftplib does not list complete files, so the nlst method is reconstructed here'''
cmd = 'LIST'
templist = []
func = None
if args[-1:] and type(args[-1]) != type(''):
args, func = args[:-1], args[-1]
for arg in args:
if arg:
cmd = cmd + (' ' + arg)
self.retrlines(cmd, templist.append)
return templist在项目代码中重写了_FTPS这个类，继承了FTP_TLS；同时呢重写的connect方法和nlst方法。
重写connect方法是为了加之前提到的那一行代码。
基础工作完成之后就可以封装FTPS相关的函数给项目使用了，大概有以下几个函数：
?
第一个是建立ftps连接的函数：
很简单，没啥可介绍的，其中添加的一个self._acess类属性是为了其他函数使用的，目的是为了验证该账号是否有ftps的登录权限（因为装置根据不同角色做了权限限制）
def open_ftps_connection(self, host=None, username=None, password=None, port=990):
host = host if host is not None else self._device.active_device().ipAddress
username = username if username is not None else self._device.active_device().rbacUsername
password = password if password is not None else self._device.active_device().rbacPassword
self.ftps = _FTPS()
self.ftps.connect(host=host, port=port)
try:
self.ftps.login(username, password)
self.ftps.prot_p()
self._access = True
except Exception as err:
if 'Not logged in' in str(err):
self._access = False
logger.error('Login ftp server error :{}'.format(err))
self.close_ftps_connection()
return self.ftps然后是关闭ftps连接的函数：
更简单，就是调用了ftp已有的方法（其实就是发送了一条‘QUIT’命令）
def close_ftps_connection(self):
self.ftps.quit()
再接下来就是获取文件列表的函数：
这里就是跟大家各自项目相关的了，大家参考下即可
def get_file_list_from_ftps(self, remoteFolder):
self.ftps.cwd(remoteFolder)
if _DeviceKeywords.active_device().deviceType in [SupportedDevice.Q100]:
files = self.ftps.nlst()
fileNames = [file[-12:] for file in files]
elif _DeviceKeywords.active_device().deviceType in [SupportedDevice.Q200]:
pass # TODO: The structure of Q200 file has not been determined yet, this part needs to be changed
return fileNames再接下来就是开始下载文件了：
很简单不多做介绍，就是传入文件名称，下载到什么路径，然后调用retrbinary方法即可
def download_file_from_ftps(self, filePath, destFilePath=None, bufsize=1024):
if destFilePath is None:
targetFolder = self._common.get_output_folder()
filename = pathlib.PurePath(filePath).name
destFilePath = targetFolder / filename
else:
destFilePath = pathlib.Path(destFilePath)
if destFilePath.exists():
raise IOError('File {:s} already exists in test case file folder'.format(destFilePath.name))
try:
with open(destFilePath, 'wb') as f:
self.ftps.retrbinary('RETR {}'.format(filePath), f.write, bufsize)
except Exception as err:
logger.error('Download file error :{}'.format(err))
return destFilePath?最后呢，就是验证用户是否有权限登录ftps；因为我司项目中有多个角色的成员，admin，viewer，manager 等等，不同角色权限不同，有的角色是没有ftp权限的，所以这里写了两个函数，便于再robotframework中写case。
def validate_user_can_access_ftps(self, username, password):
logger.info('Validating user can access ftps with name:{} password:{}'.format(username, password))
self.open_ftps_connection(username=username, password=password)
if self._access:
logger.info('User is able to access ftps server.')
self.close_ftps_connection()
else:
raise AssertionError('User is unable to access ftps server.')
def validate_user_can_not_access_ftps(self, username, password):
logger.info('Validating user can not access ftps with name:{} password:{}'.format(username, password))
self.open_ftps_connection(username=username, password=password)
if self._access:
self.close_ftps_connection()
raise AssertionError('User is able to access ftps server.')
else:
logger.info('User is unable to access ftps server.')总结：其实关于FTPS的操作没有太复杂的东西，因为ftp的方法可以直接看源码，即可直接使用了，我这里无非是根据项目需要做了一层封装，然后便于写case。如果对大家有帮助，那就再好不过了。
哎，拖延症患者终于再2021年最后一天把几个月前做的一个小项目总结了一下。。