File manager

File manager - Edit - /usr/share/l.v.e-manager/copy_directory.py

Back
# -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from __future__ import print_function from __future__ import division from __future__ import absolute_import import errno import glob import os import shutil import stat def _safe_open_nofollow_for_write(path): """Open path for writing, refusing to follow a final-component symlink. Returned object is a binary file handle the caller is responsible for closing. Used so a planted symlink at the destination cannot redirect the root-context copy into /etc/cron.d, /usr/lib/systemd, etc. """ flags = os.O_WRONLY \| os.O_CREAT \| os.O_TRUNC \| os.O_NOFOLLOW fd = os.open(path, flags, 0o644) return os.fdopen(fd, 'wb') def _safe_copyfile_nofollow(source, destination): """shutil.copyfile replacement that refuses to follow a destination symlink. Also refuses to copy a file onto itself: the destination open uses O_TRUNC, so if source and destination resolve to the same inode the source is zeroed before ``copyfileobj`` reads a single byte. The upstream ``shutil.copyfile`` raises ``SameFileError`` for this case; the NOFOLLOW writer must too — otherwise CopyDirectory silently truncates every file when its source and destination roots resolve to the same directory (CLOS-6789). """ try: same = os.path.samefile(source, destination) except OSError: same = False if same: return with open(source, 'rb') as fsrc: with _safe_open_nofollow_for_write(destination) as fdst: shutil.copyfileobj(fsrc, fdst) class CopyDirectory: def __init__(self, source, destination, pattern="", directorypattern="", overwriteolder=0, copyfile=None, copy_file_mode=False, rewrite_plugin_conf=False): """ ATTENTION: values other than '' for parameter 'directorypattern' are not supported """ self.sourcedirectory = source self.destinationdirectory = destination self.pattern = pattern self.directorypattern = directorypattern self.overwriteolder = overwriteolder self.directorylist = [] self.filelist = [] # Default to the NOFOLLOW-safe writer. Callers may still pass a # custom copyfile, but the bare shutil.copyfile default would # silently follow a destination symlink and clobber whatever the # symlink points at — exactly the primitive the symlink-following # finding flags. The installer (plugins/install-lvemanager-plugin.py) # runs from %post as root, so this matters under any operator # deployment that places base_path on a tree any non-root user can # write into. self.copyfile = copyfile if copyfile is not None else _safe_copyfile_nofollow self.copy_file_mode = copy_file_mode # specify if permissions for files (files only, not directories) should be copied self.rewrite_plugin_conf = rewrite_plugin_conf def process(self): print("Gathering file list for {}".format(self.destinationdirectory)) self.walk() print("Building directory tree and copying files") self.go() def walk(self): ''' Gather the list of directories and files to copy ''' for dir, subdirs, files in os.walk(self.sourcedirectory): self._copydir(self, dir, subdirs + files) def go(self): ''' Perform the primary function of copying a directory given an already constructed directory and file list (generally created by walk) ''' for directory in self.directorylist: # following is built in to this class self.copydirectory(directory) print("Copy/update files...") for source, destination in self.filelist: # following is an argument to the constructor try: filepath = os.path.dirname(destination) if not os.path.exists(filepath): os.makedirs(filepath) # Refuse to overwrite a destination whose final component # is a symlink — the helper would otherwise follow it and # the privileged %post-time copy would write through to # whatever the symlink target points at. try: if stat.S_ISLNK(os.lstat(destination).st_mode): print("Warning! Refusing to overwrite symlinked path: {}".format(destination)) continue except OSError as oe: if oe.errno != errno.ENOENT: raise self.copyfile(source, destination) if self.copy_file_mode: shutil.copymode(source, destination) except Exception as e: print("Warning! Can not update fle: {}".format(destination)) print(e) def _copydir(self, arg, directory, files): ''' Perform copying for a particular directory ''' # print directory # collect files, use files if we have '' pattern workingfiles = self._files(directory, files) # filter subdirectories, modifies files in-place self._subdirectories(directory, files) destinationdirectory = self._directorysetup(directory) # should provide option for not overwriting files # possibly with date checks etc. # For extra marks, just collect this information # and return a list of copies to be made, # so the user can review before copying. # Do the copying # print workingfiles for file in workingfiles: source = os.path.join(directory, file) destination = os.path.join(destinationdirectory, file) if source.endswith('plugin.conf') and os.path.exists(destination) and \ not self.rewrite_plugin_conf: continue if self.overwriteolder and os.path.exists(destination): if os.stat(source)[stat.ST_MTIME] > os.stat(destination)[stat.ST_MTIME]: # we don't care if it's older, or it is older (hopefully) self.filelist.append((source, destination)) else: print("skip %s (source not newer)" % (source)) else: self.filelist.append((source, destination)) def _subdirectories(self, directory, default): ''' Filters the list of subdirectories which will be copied ''' # following modifies which directories are copied # does so by modifying file list in place (see os.path.walk) # Note: this does not work because os.path.walk is removed in python 3 if self.directorypattern != "": # default[:] = glob.glob(os.path.join(directory, self.directorypattern)) raise Exception('Filtering of directories not supported') def _files(self, directory, default): ''' create the filtered list of files ''' def dirfilter(values, directory=directory): result = [] for value in values: if not os.path.isdir(os.path.join(directory, value)): result.append(value) return result # following is for local processing if self.pattern != "": return dirfilter(glob.glob(os.path.join(directory, self.pattern))) else: return dirfilter(default) def _directorysetup(self, directory): ''' Ensure that the destination directory is available build it if it is not ''' # print "setup directory", directory # should make this manipulation more robust # currently assumes all sorts of things :( prefix = os.path.commonprefix((self.sourcedirectory, directory)) extendeddirectory = directory[len(prefix):] # get rid of preceding /, so won't be interpreted as root if extendeddirectory and extendeddirectory[0] == os.sep: extendeddirectory = extendeddirectory[1:] destinationdirectory = os.path.join(self.destinationdirectory, extendeddirectory) # print " destinationdirectory", destinationdirectory self.directorylist.append(destinationdirectory) return destinationdirectory def copydirectory(self, directory): ''' Called after tree building, creates needed directories ''' # create directory if not already there # if not os.path.exists( directory ): try: os.mkdir(directory) print("made directory", directory) except os.error: pass # If the path already existed as a symlink, refuse to traverse it # for subsequent writes — otherwise self.copyfile(...) would follow # the symlink target out of self.destinationdirectory. Caller code # in go() does its own per-file check; this is the directory-level # guard for the same primitive. try: if stat.S_ISLNK(os.lstat(directory).st_mode): raise OSError( errno.ELOOP, 'Refusing to traverse symlinked destination directory: %s' % directory, ) except OSError as oe: if oe.errno == errno.ENOENT: # The mkdir above failed for some other reason and the path # does not exist; let the per-file copy raise the real error. return raise

| ver. 1.4 | Github | . | PHP 8.1.34 | Generation time: 0.14 | proxy | phpinfo | Settings