JezK
Edit File: put.py
import os import sys from .fstab import Fstab from .fstab import volume_of from .trash import EX_OK, EX_IOERR from .trash import TrashDirectories from .trash import backup_file_path_from from .trash import logger from .trash import version from datetime import datetime def main(): return TrashPutCmd( sys.stdout, sys.stderr, os.environ, volume_of, parent_path, os.path.realpath ).run(sys.argv) def parent_path(path): return os.path.realpath(os.path.dirname(path)) class TrashPutCmd: def __init__(self, stdout, stderr, environ, volume_of, parent_path, realpath): self.stdout = stdout self.stderr = stderr self.environ = environ self.volume_of = volume_of self.fs = RealFs() self.getuid = os.getuid self.now = datetime.now self.parent_path = parent_path self.realpath = realpath def run(self, argv): program_name = os.path.basename(argv[0]) logger = MyLogger(self.stderr, program_name) parser = self.get_option_parser(program_name) try: (options, args) = parser.parse_args(argv[1:]) if options.verbose: logger.be_verbose() if len(args) <= 0: parser.error("Please specify the files to trash.") except SystemExit as e: return e.code else: reporter = TrashPutReporter(logger) trashcan = GlobalTrashCan(reporter = reporter, volume_of = self.volume_of, environ = self.environ, fs = self.fs, getuid = self.getuid, now = self.now, parent_path = self.parent_path, realpath = self.realpath) trashcan.trash_all(args) return reporter.exit_code() def get_option_parser(self, program_name): from optparse import OptionParser parser = OptionParser(prog=program_name, usage="%prog [OPTION]... FILE...", description="Put files in trash", version="%%prog %s" % version, formatter=NoWrapFormatter(), epilog=epilog) parser.add_option("-d", "--directory", action="store_true", help="ignored (for GNU rm compatibility)") parser.add_option("-f", "--force", action="store_true", help="ignored (for GNU rm compatibility)") parser.add_option("-i", "--interactive", action="store_true", help="ignored (for GNU rm compatibility)") parser.add_option("-r", "-R", "--recursive", action="store_true", help="ignored (for GNU rm compatibility)") parser.add_option("-v", "--verbose", action="store_true", help="explain what is being done", dest="verbose") original_print_help = parser.print_help def patched_print_help(): original_print_help(self.stdout) def patched_error(msg): parser.print_usage(self.stderr) parser.exit(2, "%s: error: %s\n" % (program_name, msg)) def patched_exit(status=0, msg=None): if msg: self.stderr.write(msg) import sys sys.exit(status) parser.print_help = patched_print_help parser.error = patched_error parser.exit = patched_exit return parser epilog="""\ To remove a file whose name starts with a '-', for example '-foo', use one of these commands: trash -- -foo trash ./-foo Report bugs to https://github.com/andreafrancia/trash-cli/issues""" class MyLogger: def __init__(self, stderr, program_name): self.program_name = program_name self.stderr=stderr self.verbose = False def be_verbose(self): self.verbose = True def info(self,message): if self.verbose: self.emit(message) def warning(self,message): self.emit(message) def emit(self, message): self.stderr.write("%s: %s\n" % (self.program_name,message)) from optparse import IndentedHelpFormatter class NoWrapFormatter(IndentedHelpFormatter) : def _format_text(self, text) : "[Does not] format a text, return the text as it is." return text class NullObject: def __getattr__(self, name): return lambda *argl,**args:None class TrashPutReporter: def __init__(self, logger): self.logger = logger self.some_file_has_not_be_trashed = False self.no_argument_specified = False def unable_to_trash_dot_entries(self,file): self.logger.warning("cannot trash %s '%s'" % (describe(file), file)) def unable_to_trash_file(self,f): self.logger.warning("cannot trash %s '%s'" % (describe(f), f)) self.some_file_has_not_be_trashed = True def file_has_been_trashed_in_as(self, trashee, trash_directory, destination): self.logger.info("'%s' trashed in %s" % (trashee, shrinkuser(trash_directory))) def found_unsercure_trash_dir_symlink(self, trash_dir_path): self.logger.info("found unsecure .Trash dir (should not be a symlink): %s" % trash_dir_path) def invalid_top_trash_is_not_a_dir(self, trash_dir_path): self.logger.info("found unusable .Trash dir (should be a dir): %s" % trash_dir_path) def found_unsecure_trash_dir_unsticky(self, trash_dir_path): self.logger.info("found unsecure .Trash dir (should be sticky): %s" % trash_dir_path) def unable_to_trash_file_in_because(self, file_to_be_trashed, trash_directory, error): self.logger.info("Failed to trash %s in %s, because :%s" % ( file_to_be_trashed, shrinkuser(trash_directory), error)) def trash_dir_with_volume(self, trash_dir_path, volume_path): self.logger.info("Trash-dir: %s from volume: %s" % (trash_dir_path, volume_path)) def exit_code(self): if not self.some_file_has_not_be_trashed: return EX_OK else: return EX_IOERR def volume_of_file(self,volume): self.logger.info("Volume of file: %s" % volume) def describe(path): """ Return a textual description of the file pointed by this path. Options: - "symbolic link" - "directory" - "'.' directory" - "'..' directory" - "regular file" - "regular empty file" - "non existent" - "entry" """ if os.path.islink(path): return 'symbolic link' elif os.path.isdir(path): if path == '.': return 'directory' elif path == '..': return 'directory' else: if os.path.basename(path) == '.': return "'.' directory" elif os.path.basename(path) == '..': return "'..' directory" else: return 'directory' elif os.path.isfile(path): if os.path.getsize(path) == 0: return 'regular empty file' else: return 'regular file' elif not os.path.exists(path): return 'non existent' else: return 'entry' class RealFs: def __init__(self): import os from . import fs self.move = fs.move self.atomic_write = fs.atomic_write self.remove_file = fs.remove_file self.ensure_dir = fs.ensure_dir self.isdir = os.path.isdir self.islink = os.path.islink self.has_sticky_bit = fs.has_sticky_bit class GlobalTrashCan: class NullReporter: def __getattr__(self,name): return lambda *argl,**args:None def __init__(self, environ, volume_of, reporter, fs, getuid, now, parent_path, realpath): self.getuid = getuid self.reporter = reporter self.volume_of = volume_of self.now = now self.fs = fs self.environ = environ self.parent_path = parent_path self.realpath = realpath def trash_all(self, args): for arg in args : self.trash(arg) def trash(self, file) : """ Trash a file in the appropriate trash directory. If the file belong to the same volume of the trash home directory it will be trashed in the home trash directory. Otherwise it will be trashed in one of the relevant volume trash directories. Each volume can have two trash directories, they are - $volume/.Trash/$uid - $volume/.Trash-$uid Firstly the software attempt to trash the file in the first directory then try to trash in the second trash directory. """ if self._should_skipped_by_specs(file): self.reporter.unable_to_trash_dot_entries(file) return volume_of_file_to_be_trashed = self.volume_of_parent(file) self.reporter.volume_of_file(volume_of_file_to_be_trashed) candidates = self._possible_trash_directories_for(volume_of_file_to_be_trashed) file_has_been_trashed = False for trash_dir in candidates.trash_dirs: if self._is_trash_dir_secure(trash_dir): volume_of_trash_dir = self.volume_of(self.realpath(trash_dir.path)) self.reporter.trash_dir_with_volume(trash_dir.path, volume_of_trash_dir) if self._file_could_be_trashed_in(volume_of_file_to_be_trashed, volume_of_trash_dir): try: trashed_file = trash_dir.trash(file) self.reporter.file_has_been_trashed_in_as( file, trashed_file['trash_directory'], trashed_file['where_file_was_stored']) file_has_been_trashed = True except (IOError, OSError) as error: self.reporter.unable_to_trash_file_in_because( file, trash_dir.path, str(error)) if file_has_been_trashed: break if not file_has_been_trashed: self.reporter.unable_to_trash_file(file) def volume_of_parent(self, file): return self.volume_of(self.parent_path(file)) def _is_trash_dir_secure(self, trash_dir): class ValidationOutput: def __init__(self): self.valid = True def not_valid_should_be_a_dir(_): self.reporter.invalid_top_trash_is_not_a_dir( os.path.dirname(trash_dir.path)) _.valid = False def not_valid_parent_should_not_be_a_symlink(_): self.reporter.found_unsercure_trash_dir_symlink( os.path.dirname(trash_dir.path)) _.valid = False def not_valid_parent_should_be_sticky(_): self.reporter.found_unsecure_trash_dir_unsticky( os.path.dirname(trash_dir.path)) _.valid = False def is_valid(self): self.valid = True output = ValidationOutput() trash_dir.checker.fs = self.fs trash_dir.checker.valid_to_be_written(trash_dir.path, output) return output.valid def _should_skipped_by_specs(self, file): basename = os.path.basename(file) return (basename == ".") or (basename == "..") def _file_could_be_trashed_in(self, volume_of_file_to_be_trashed, volume_of_trash_dir): return volume_of_trash_dir == volume_of_file_to_be_trashed def _possible_trash_directories_for(self, volume): candidates = PossibleTrashDirectories(self.fs) trash_directories = TrashDirectories(self.volume_of, self.getuid, self.environ) trash_directories.home_trash_dir(candidates.add_home_trash) trash_directories.volume_trash_dir1(volume, candidates.add_top_trash_dir) trash_directories.volume_trash_dir2(volume, candidates.add_alt_top_trash_dir) return candidates class PossibleTrashDirectories: def __init__(self, fs): self.trash_dirs = [] self.fs = fs def add_home_trash(self, path, volume): trash_dir = self._make_trash_dir(path, volume) trash_dir.store_absolute_paths() self.trash_dirs.append(trash_dir) def add_top_trash_dir(self, path, volume): trash_dir = self._make_trash_dir(path, volume) trash_dir.store_relative_paths() trash_dir.checker = TopTrashDirWriteRules(None) self.trash_dirs.append(trash_dir) def add_alt_top_trash_dir(self, path, volume): trash_dir = self._make_trash_dir(path, volume) trash_dir.store_relative_paths() self.trash_dirs.append(trash_dir) def _make_trash_dir(self, path, volume): return TrashDirectoryForPut(path, volume, datetime.now, self.fs) def parent_realpath(path): parent = os.path.dirname(path) return os.path.realpath(parent) class TrashDirectoryForPut: from datetime import datetime def __init__(self, path, volume, now, fs): self.path = os.path.normpath(path) self.volume = volume self.logger = logger self.info_dir = os.path.join(self.path, 'info') self.files_dir = os.path.join(self.path, 'files') class all_is_ok_checker: def valid_to_be_written(self, a, b): pass def check(self, a):pass self.checker = all_is_ok_checker() self.now = now self.move = fs.move self.atomic_write = fs.atomic_write self.remove_file = fs.remove_file self.ensure_dir = fs.ensure_dir self.path_for_trash_info = OriginalLocation(parent_realpath) def store_absolute_paths(self): self.path_for_trash_info.make_absolutes_paths() def store_relative_paths(self): self.path_for_trash_info.make_paths_relatives_to(self.volume) def trash(self, path): path = os.path.normpath(path) original_location = self.path_for_trash_info.for_file(path) basename = os.path.basename(original_location) content = self.format_trashinfo(original_location, self.now()) trash_info_file = self.persist_trash_info( self.info_dir, basename, content) where_to_store_trashed_file = backup_file_path_from(trash_info_file) self.ensure_files_dir_exists() try : self.move(path, where_to_store_trashed_file) except IOError as e : self.remove_file(trash_info_file) raise e result = dict() result['trash_directory'] = self.path result['where_file_was_stored'] = where_to_store_trashed_file return result def format_trashinfo(self, original_location, deletion_date): def format_date(deletion_date): return deletion_date.strftime("%Y-%m-%dT%H:%M:%S") def format_original_location(original_location): try: from urllib import quote except ImportError: from urllib.parse import quote return quote(original_location,'/') content = ("[Trash Info]\n" + "Path=%s\n" % format_original_location(original_location) + "DeletionDate=%s\n" % format_date(deletion_date)).encode('utf-8') return content def ensure_files_dir_exists(self): self.ensure_dir(self.files_dir, 0o700) def persist_trash_info(self, info_dir, basename, content) : """ Create a .trashinfo file in the $trash/info directory. returns the created TrashInfoFile. """ self.ensure_dir(info_dir, 0o700) # write trash info index = 0 while True : if index == 0 : suffix = "" elif index < 100: suffix = "_%d" % index else : import random suffix = "_%d" % random.randint(0, 65535) base_id = basename trash_id = base_id + suffix trash_info_basename = trash_id+".trashinfo" dest = os.path.join(info_dir, trash_info_basename) try : self.atomic_write(dest, content) self.logger.debug(".trashinfo created as %s." % dest) return dest except OSError: self.logger.debug("Attempt for creating %s failed." % dest) index += 1 raise IOError() def shrinkuser(path, environ=os.environ): import posixpath import re try: home_dir = environ['HOME'] home_dir = posixpath.normpath(home_dir) if home_dir != '': path = re.sub('^' + re.escape(home_dir + os.path.sep), '~' + os.path.sep, path) except KeyError: pass return path class TopTrashDirWriteRules: def __init__(self, fs): self.fs = fs def valid_to_be_written(self, trash_dir_path, output): parent = os.path.dirname(trash_dir_path) if not self.fs.isdir(parent): output.not_valid_should_be_a_dir() return if self.fs.islink(parent): output.not_valid_parent_should_not_be_a_symlink() return if not self.fs.has_sticky_bit(parent): output.not_valid_parent_should_be_sticky() return output.is_valid() class OriginalLocation: def __init__(self, parent_realpath): self.parent_realpath = parent_realpath self.make_absolutes_paths() def make_paths_relatives_to(self, topdir): self.path_maker = TopDirRelativePaths(topdir) def make_absolutes_paths(self): self.path_maker = AbsolutePaths() def for_file(self, path): self.normalized_path = os.path.normpath(path) basename = os.path.basename(self.normalized_path) parent = self.parent_realpath(self.normalized_path) parent = self.path_maker.calc_parent_path(parent) return os.path.join(parent, basename) class TopDirRelativePaths: def __init__(self, topdir): self.topdir = topdir def calc_parent_path(self, parent): if (parent == self.topdir) or parent.startswith(self.topdir+os.path.sep) : parent = parent[len(self.topdir+os.path.sep):] return parent class AbsolutePaths: def calc_parent_path(self, parent): return parent