/usr/lib/python2.7/dist-packages/jnpr/junos/utils/ftp.py is in python-junos-eznc 2.0.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | """
FTP utility
"""
import re
import ftplib
import os
import logging
logger = logging.getLogger("jnpr.junos.utils.ftp")
class FTP(ftplib.FTP):
"""
FTP utility can be used to transfer files to and from device.
"""
def __init__(self, junos, **ftpargs):
"""
:param Device junos: Device object
:param kvargs ftpargs: any additional args to be passed to ftplib FTP
Supports python *context-manager* pattern. For example::
from jnpr.junos.utils.ftp import FTP
with FTP(dev) as ftp:
ftp.put(package, remote_path)
"""
self._junos = junos
self._ftpargs = ftpargs
ftplib.FTP.__init__(self, self._junos._hostname, self._junos._auth_user,
self._junos._auth_password)
# dummy function, as connection is created by ftb lib in __init__ only
def open(self):
return self
def put(self, local_file, remote_path=None):
"""
This function is used to upload file to the router from local
execution server/shell.
:param local_file: Full path along with filename which has to be
copied to router
:param remote_path: path in which to receive the files on the remote
host. If ignored FILE will be copied to "tmp"
:returns: True if the transfer succeeds, else False
"""
try:
mat = re.search('^.*/(.*)$', local_file)
if mat:
if not remote_path:
remote_file = '/tmp/' + mat.group(1)
else:
if re.search('^.*/(.*)$', remote_path) and \
re.search('\.\w+$', remote_path):
remote_file = remote_path
# Looks like remote path is given as file location
else:
remote_file = os.path.join(remote_path, mat.group(1))
else:
if not remote_path:
remote_file = os.path.join('/tmp/', local_file)
else:
remote_file = os.path.join(remote_path, local_file)
self.storbinary('STOR ' + remote_file, open(local_file, 'rb'))
except Exception as ex:
logger.error(ex)
return False
return True
def get(self, remote_file, local_path=os.getcwd()):
"""
This function is used to download file from router to local execution
server/shell.
:param local_path: path in which to receive files locally
:param remote_file: Full path along with filename on the router. If
ignored FILE will be copied to "tmp"
:returns: True if the transfer succeeds, else False
"""
if os.path.isdir(local_path):
mat = re.search('^.*/(.*)$', remote_file)
if mat:
local_file=os.path.join(local_path, mat.group(1))
else:
local_file=local_path
else:
local_file = local_path
try:
self.retrbinary('RETR ' + remote_file,
open(local_file, 'wb').write)
except Exception as ex:
logger.error(ex)
return False
return True
# -------------------------------------------------------------------------
# CONTEXT MANAGER
# -------------------------------------------------------------------------
def __enter__(self):
# return self.open(**self._ftpargs)
return self
def __exit__(self, exc_ty, exc_val, exc_tb):
return self.close()
|