关于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年最后一天把几个月前做的一个小项目总结了一下。。
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)