Source code for duplicity.path

# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4; encoding:utf-8 -*-
# Copyright 2002 Ben Escoto <>
# Copyright 2007 Kenneth Loafman <>
# This file is part of duplicity.
# Duplicity is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation; either version 2 of the License, or (at your
# option) any later version.
# Duplicity is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with duplicity; if not, write to the Free Software Foundation,
# Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

"""Wrapper class around a file like "/usr/bin/env"

This class makes certain file operations more convenient and
associates stat information with filenames


import errno
import gzip
import re
import shutil
import socket
import stat
import time

from duplicity import cached_ops
from duplicity import config
from duplicity import dup_time
from duplicity import file_naming
from duplicity import gpg
from duplicity import librsync
from duplicity import dup_tarfile
from duplicity.lazy import *  # pylint: disable=unused-wildcard-import,redefined-builtin

_tmp_path_counter = 1

[docs]class StatResult(object): """Used to emulate the output of os.stat() and related""" # st_mode is required by the TarInfo class, but it's unclear how # to generate it from file permissions. st_mode = 0
[docs]class PathException(Exception): pass
[docs]class ROPath(object): """Read only Path Objects of this class doesn't represent real files, so they don't have a name. They are required to be indexed though. """
[docs] def __init__(self, index, stat=None): # pylint: disable=unused-argument """ROPath initializer""" self.opened, self.fileobj = None, None self.index = index self.stat, self.type = None, None self.mode, self.devnums = None, None
[docs] def set_from_stat(self): """Set the value of self.type, self.mode from self.stat""" if not self.stat: self.type = None st_mode = self.stat.st_mode if stat.S_ISREG(st_mode): self.type = "reg" elif stat.S_ISDIR(st_mode): self.type = "dir" elif stat.S_ISLNK(st_mode): self.type = "sym" elif stat.S_ISFIFO(st_mode): self.type = "fifo" elif stat.S_ISSOCK(st_mode): raise PathException(os.fsdecode(self.get_relative_path()) + "is a socket, unsupported by tar") self.type = "sock" # pylint: disable=unreachable elif stat.S_ISCHR(st_mode): self.type = "chr" elif stat.S_ISBLK(st_mode): self.type = "blk" else: raise PathException("Unknown type") self.mode = stat.S_IMODE(st_mode) if self.type in ("chr", "blk"): try: self.devnums = ( os.major(self.stat.st_rdev), os.minor(self.stat.st_rdev), ) except Exception as e: log.Warn( _("Warning: %s invalid devnums (0x%X), treating as (0, 0).") % (os.fsdecode(self.get_relative_path()), self.stat.st_rdev) ) self.devnums = (0, 0)
[docs] def blank(self): """Black out self - set type and stat to None""" self.type, self.stat = None, None
[docs] def exists(self): """True if corresponding file exists""" return self.type
[docs] def isreg(self): """True if self corresponds to regular file""" return self.type == "reg"
[docs] def isdir(self): """True if self is dir""" return self.type == "dir"
[docs] def issym(self): """True if self is sym""" return self.type == "sym"
[docs] def isfifo(self): """True if self is fifo""" return self.type == "fifo"
[docs] def issock(self): """True is self is socket""" return self.type == "sock"
[docs] def isdev(self): """True is self is a device file""" return self.type == "chr" or self.type == "blk"
[docs] def getdevloc(self): """Return device number path resides on""" return self.stat.st_dev
[docs] def getsize(self): """Return length in bytes from stat object""" return self.stat.st_size
[docs] def getmtime(self): """Return mod time of path in seconds""" return int(self.stat.st_mtime)
[docs] def get_relative_path(self): """Return relative path, created from index""" if self.index: return b"/".join(self.index) else: return b"."
[docs] def getperms(self): """Return permissions mode, owner and group""" s1 = self.stat return f"{s1.st_uid}:{s1.st_gid} {self.mode:o}"
[docs] def open(self, mode): """Return fileobj associated with self""" assert mode == "rb" and self.fileobj and not self.opened, f"{mode} {self.fileobj} {self.opened}" self.opened = 1 return self.fileobj
[docs] def get_data(self): """Return contents of associated fileobj in string""" fin ="rb") buf = assert not fin.close() return buf
[docs] def setfileobj(self, fileobj): """Set file object returned by open()""" assert not self.fileobj self.fileobj = fileobj self.opened = None
[docs] def init_from_tarinfo(self, tarinfo): """Set data from tarinfo object (part of dup_tarfile module)""" # Set the typepp type = tarinfo.type # pylint: disable=redefined-builtin if type == dup_tarfile.REGTYPE or type == dup_tarfile.AREGTYPE: self.type = "reg" elif type == dup_tarfile.LNKTYPE: raise PathException("Hard links not supported yet") elif type == dup_tarfile.SYMTYPE: self.type = "sym" self.symtext = tarinfo.linkname if isinstance(self.symtext, str): self.symtext = os.fsencode(self.symtext) elif type == dup_tarfile.CHRTYPE: self.type = "chr" self.devnums = (tarinfo.devmajor, tarinfo.devminor) elif type == dup_tarfile.BLKTYPE: self.type = "blk" self.devnums = (tarinfo.devmajor, tarinfo.devminor) elif type == dup_tarfile.DIRTYPE: self.type = "dir" elif type == dup_tarfile.FIFOTYPE: self.type = "fifo" else: raise PathException(f"Unknown tarinfo type {type}") self.mode = tarinfo.mode self.stat = StatResult() """ If do_not_restore_owner is False, set user and group id use numeric id if name lookup fails OR --numeric-owner is set """ try: if config.numeric_owner: raise KeyError self.stat.st_uid = cached_ops.getpwnam(tarinfo.uname)[2] except KeyError: self.stat.st_uid = tarinfo.uid try: if config.numeric_owner: raise KeyError self.stat.st_gid = cached_ops.getgrnam(tarinfo.gname)[2] except KeyError: self.stat.st_gid = tarinfo.gid self.stat.st_mtime = int(tarinfo.mtime) if self.stat.st_mtime < 0: log.Warn(_("Warning: %s has negative mtime, treating as 0.") % tarinfo.uc_name) self.stat.st_mtime = 0 self.stat.st_size = tarinfo.size
[docs] def get_ropath(self): """Return ropath copy of self""" new_ropath = ROPath(self.index, self.stat) new_ropath.type, new_ropath.mode = self.type, self.mode if self.issym(): new_ropath.symtext = self.symtext elif self.isdev(): new_ropath.devnums = self.devnums if self.exists(): new_ropath.stat = self.stat return new_ropath
[docs] def get_tarinfo(self): """Generate a dup_tarfile.TarInfo object based on self Doesn't set size based on stat, because we may want to replace data wiht other stream. Size should be set separately by calling function. """ ti = dup_tarfile.TarInfo() if self.index: = os.fsdecode(b"/".join(self.index)) else: = "." if self.isdir(): += "/" # tar dir naming convention ti.size = 0 if self.type: # Lots of this is specific to, hope it doesn't # change much... if self.isreg(): ti.type = dup_tarfile.REGTYPE ti.size = self.stat.st_size elif self.isdir(): ti.type = dup_tarfile.DIRTYPE elif self.isfifo(): ti.type = dup_tarfile.FIFOTYPE elif self.issym(): ti.type = dup_tarfile.SYMTYPE ti.linkname = self.symtext if isinstance(ti.linkname, bytes): ti.linkname = os.fsdecode(ti.linkname) elif self.isdev(): if self.type == "chr": ti.type = dup_tarfile.CHRTYPE else: ti.type = dup_tarfile.BLKTYPE ti.devmajor, ti.devminor = self.devnums else: raise PathException(f"Unrecognized type {str(self.type)}") ti.mode = self.mode ti.uid, ti.gid = self.stat.st_uid, self.stat.st_gid if self.stat.st_mtime < 0: log.Warn(_("Warning: %s has negative mtime, treating as 0.") % (os.fsdecode(self.get_relative_path()))) ti.mtime = 0 else: ti.mtime = int(self.stat.st_mtime) try: ti.uname = cached_ops.getpwuid(ti.uid)[0] except KeyError: ti.uname = "" try: ti.gname = cached_ops.getgrgid(ti.gid)[0] except KeyError: ti.gname = "" if ti.type in (dup_tarfile.CHRTYPE, dup_tarfile.BLKTYPE): if hasattr(os, "major") and hasattr(os, "minor"): ti.devmajor, ti.devminor = self.devnums else: # Currently we depend on an uninitiliazed tarinfo file to # already have appropriate headers. Still, might as well # make sure mode and size set. ti.mode, ti.size = 0, 0 return ti
def __eq__(self, other): """Used to compare two ROPaths. Doesn't look at fileobjs""" if not self.type and not other.type: return 1 # neither exists if not self.stat and other.stat or not other.stat and self.stat: return 0 if self.type != other.type: return 0 if self.isreg() or self.isdir() or self.isfifo(): # Don't compare sizes, because we might be comparing # signature size to size of file. if not self.perms_equal(other): return 0 if int(self.stat.st_mtime) == int(other.stat.st_mtime): return 1 # Below, treat negative mtimes as equal to 0 return self.stat.st_mtime <= 0 and other.stat.st_mtime <= 0 elif self.issym(): # here only symtext matters return self.symtext == other.symtext elif self.isdev(): return self.perms_equal(other) and self.devnums == other.devnums assert 0 def __ne__(self, other): return not self.__eq__(other)
[docs] def compare_verbose(self, other, include_data=0): """Compare ROPaths like __eq__, but log reason if different This is placed in a separate function from __eq__ because __eq__ should be very time sensitive, and logging statements would slow it down. Used when verifying. Only run if include_data is true. """ def log_diff(log_string): log_str = f"{_('Difference found:')} {log_string}" log.Notice(log_str % (os.fsdecode(self.get_relative_path()))) if include_data is False: return True if not self.type and not other.type: return 1 if not self.stat and other.stat: log_diff(_("New file %s")) return 0 if not other.stat and self.stat: log_diff(_("File %s is missing")) return 0 if self.type != other.type: log_diff(_("File %%s has type %s, expected %s") % (other.type, self.type)) return 0 if self.isreg() or self.isdir() or self.isfifo(): if not self.perms_equal(other): log_diff(_("File %%s has permissions %s, expected %s") % (other.getperms(), self.getperms())) return 0 if (int(self.stat.st_mtime) != int(other.stat.st_mtime)) and ( self.stat.st_mtime > 0 or other.stat.st_mtime > 0 ): log_diff( _("File %%s has mtime %s, expected %s") % ( dup_time.timetopretty(int(other.stat.st_mtime)), dup_time.timetopretty(int(self.stat.st_mtime)), ) ) return 0 if self.isreg(): if self.compare_data(other): return 1 else: log_diff(_("Data for file %s is different")) return 0 else: return 1 elif self.issym(): if self.symtext == other.symtext or self.symtext + os.fsencode(os.sep) == other.symtext: return 1 else: log_diff(_("Symlink %%s points to %s, expected %s") % (other.symtext, self.symtext)) return 0 elif self.isdev(): if not self.perms_equal(other): log_diff(_("File %%s has permissions %s, expected %s") % (other.getperms(), self.getperms())) return 0 if self.devnums != other.devnums: log_diff(_("Device file %%s has numbers %s, expected %s") % (other.devnums, self.devnums)) return 0 return 1 assert 0
[docs] def compare_data(self, other): """Compare data from two regular files, return true if same""" f1 ="rb") f2 ="rb") def close(): assert not f1.close() assert not f2.close() while True: buf1 = buf2 = if buf1 != buf2: close() return 0 if not buf1: close() return 1
[docs] def perms_equal(self, other): """True if self and other have same permissions and ownership""" s1, s2 = self.stat, other.stat return self.mode == other.mode and s1.st_gid == s2.st_gid and s1.st_uid == s2.st_uid
[docs] def copy(self, other): """Copy self to other. Also copies data. Other must be Path""" if self.isreg(): other.writefileobj("rb")) elif self.isdir(): os.mkdir( elif self.issym(): os.symlink(self.symtext, if config.restore_ownership: os.lchown(, self.stat.st_uid, self.stat.st_gid) other.setdata() return # no need to copy symlink attributes elif self.isfifo(): os.mkfifo( elif self.issock(): socket.socket(socket.AF_UNIX).bind( elif self.isdev(): if self.type == "chr": devtype = "c" else: devtype = "b" other.makedev(devtype, *self.devnums) self.copy_attribs(other)
[docs] def copy_attribs(self, other): """Only copy attributes from self to other""" if isinstance(other, Path): if not self.issym(): if self.stat and config.restore_ownership: util.maybe_ignore_errors(lambda: os.chown(, self.stat.st_uid, self.stat.st_gid)) util.maybe_ignore_errors(lambda: os.chmod(, self.mode)) util.maybe_ignore_errors(lambda: os.utime(, (time.time(), self.stat.st_mtime))) other.setdata() else: # write results to fake stat object assert isinstance(other, ROPath) stat = StatResult() stat.st_uid, stat.st_gid = self.stat.st_uid, self.stat.st_gid stat.st_mtime = int(self.stat.st_mtime) other.stat = stat other.mode = self.mode
def __str__(self): """Return string representation""" return f"({util.uindex(self.index)} {self.type})"
[docs]class Path(ROPath): """ Path class - wrapper around ordinary local files Besides caching stat() results, this class organizes various file code. """ regex_chars_to_quote = re.compile('[\\\\\\"\\$`]')
[docs] def rename_index(self, index): if not config.rename or not index: return index # early exit path = os.path.normcase(os.path.join(*index)) tail = [] while path and path not in config.rename: path, extra = os.path.split(path) tail.insert(0, extra) if path: return config.rename[path].split(os.fsencode(os.sep)) + tail else: return index # no rename found
[docs] def __init__(self, base, index=()): """Path initializer""" # self.opened should be true if the file has been opened, and # self.fileobj can override returned fileobj super().__init__(index) self.opened, self.fileobj = None, None if isinstance(base, str): base = os.fsencode(base) self.base = base # Create self.index, which is the path as a tuple self.index = self.rename_index(index) = os.path.join(base, *self.index) # We converted any unicode base to filesystem encoding, so should # be in filesystem encoding already and does not need to change self.uc_name = os.fsdecode( self.setdata()
[docs] def setdata(self): """Refresh stat cache""" try: # We may be asked to look at the target of symlinks rather than # the link itself. if config.copy_links: self.stat = os.stat( else: self.stat = os.lstat( except OSError as e: err_string = errno.errorcode[e.errno] if err_string in ["ENOENT", "ENOTDIR", "ELOOP", "ENOTCONN", "ENODEV"]: self.stat, self.type = None, None # file doesn't exist self.mode = None else: raise else: self.set_from_stat() if self.issym(): self.symtext = os.readlink(
[docs] def append(self, ext): """Return new Path with ext added to index""" if isinstance(ext, str): ext = os.fsencode(ext) return self.__class__(self.base, self.index + (ext,))
[docs] def new_index(self, index): """Return new Path with index index""" return self.__class__(self.base, index)
[docs] def listdir(self): """Return list generated by os.listdir""" return os.listdir(
[docs] def isemptydir(self): """Return true if path is a directory and is empty""" return self.isdir() and not self.listdir()
[docs] def contains(self, child): """Return true if path is a directory and contains child""" if isinstance(child, str): child = os.fsencode(child) # We don't use append(child).exists() here because that requires exec # permissions as well as read. listdir() just needs read permissions. return self.isdir() and child in self.listdir()
[docs] def open(self, mode="rb"): """ Return fileobj associated with self Usually this is just the file data on disk, but can be replaced with arbitrary data using the setfileobj method. """ assert not self.opened if self.fileobj: result = self.fileobj else: result = open(, mode) return result
[docs] def makedev(self, type, major, minor): # pylint: disable=redefined-builtin """Make a device file with specified type, major/minor nums""" cmdlist = ["mknod",, type, str(major), str(minor)] if os.spawnvp(os.P_WAIT, "mknod", cmdlist) != 0: raise PathException(f"Error running {cmdlist}") self.setdata()
[docs] def mkdir(self): """Make directory(s) at specified path""" log.Debug(_("Making directory %s") % self.uc_name) try: os.makedirs( except OSError: if not config.force: raise PathException(f"Error creating directory {self.uc_name}", 7) self.setdata()
[docs] def delete(self): """Remove this file""" log.Debug(_("Deleting %s") % self.uc_name) if self.isdir(): util.ignore_missing(os.rmdir, else: util.ignore_missing(os.unlink, self.setdata()
[docs] def touch(self): """Open the file, write 0 bytes, close""" log.Debug(_("Touching %s") % self.uc_name) fp ="wb") fp.close()
[docs] def deltree(self): """Remove self by recursively deleting files under it""" from duplicity import selection # TODO: avoid circ. dep. issue log.Debug(_("Deleting tree %s") % self.uc_name) itr = IterTreeReducer(PathDeleter, []) for path in selection.Select(self).set_iter(): itr(path.index, path) itr.Finish() self.setdata()
[docs] def get_parent_dir(self): """Return directory that self is in""" if self.index: return Path(self.base, self.index[:-1]) else: components = self.base.split(b"/") if len(components) == 2 and not components[0]: return Path(b"/") # already in root directory else: return Path(b"/".join(components[:-1]))
[docs] def writefileobj(self, fin): """Copy file object fin to self. Close both when done.""" fout ="wb") while True: buf = if not buf: break fout.write(buf) if fin.close() or fout.close(): raise PathException("Error closing file object") self.setdata()
[docs] def rename(self, new_path): """Rename file at current path to new_path.""" shutil.move(, self.setdata() new_path.setdata()
[docs] def move(self, new_path): """Like rename but destination may be on different file system""" self.copy(new_path) self.delete()
[docs] def chmod(self, mode): """Change permissions of the path""" os.chmod(, mode) self.setdata()
[docs] def patch_with_attribs(self, diff_ropath): """Patch self with diff and then copy attributes over""" assert self.isreg() and diff_ropath.isreg() temp_path = self.get_temp_in_same_dir() fbase ="rb") fdiff ="rb") patch_fileobj = librsync.PatchedFile(fbase, fdiff) temp_path.writefileobj(patch_fileobj) assert not fbase.close() assert not fdiff.close() diff_ropath.copy_attribs(temp_path) temp_path.rename(self)
[docs] def get_temp_in_same_dir(self): """Return temp non existent path in same directory as self""" global _tmp_path_counter parent_dir = self.get_parent_dir() while True: temp_path = parent_dir.append(f"duplicity_temp.{str(_tmp_path_counter)}") if not temp_path.type: return temp_path _tmp_path_counter += 1 assert _tmp_path_counter < 10000, f"Warning too many temp files created for {self.uc_name}"
[docs] def compare_recursive(self, other, verbose=None): """Compare self to other Path, descending down directories""" from duplicity import selection # TODO: avoid circ. dep. issue selfsel = selection.Select(self).set_iter() othersel = selection.Select(other).set_iter() return Iter.equal(selfsel, othersel, verbose)
def __repr__(self): """Return string representation""" return f"({self.index} {self.uc_name} {self.type})"
[docs] def quote(self, s=None): """ Return quoted version of s (defaults to The output is meant to be interpreted with shells, so can be used with os.system. """ if not s: s = self.uc_name return '"%s"' % self.regex_chars_to_quote.sub(lambda m: f"\\{}", s)
[docs] def unquote(self, s): """Return unquoted version of string s, as quoted by above quote()""" assert s[0] == s[-1] == '"' # string must be quoted by above result = "" i = 1 while i < len(s) - 1: if s[i] == "\\": result += s[i + 1] i += 2 else: result += s[i] i += 1 return result
[docs] def get_filename(self): """Return filename of last component""" components ="/") assert components and components[-1] return components[-1]
[docs] def get_canonical(self): """ Return string of canonical version of path Remove ".", and trailing slashes where possible. Note that it's harder to remove "..", as "foo/bar/.." is not necessarily "foo", so we can't use path.normpath() """ newpath = b"/".join([x for x in"/") if x and x != b"."]) if self.uc_name[0] == "/": return b"/" + newpath elif newpath: return newpath else: return b"."
[docs]class DupPath(Path): """ Represent duplicity data files Based on the file name, files that are compressed or encrypted will have different open() methods. """
[docs] def __init__(self, base, index=(), parseresults=None): """ DupPath initializer The actual filename (no directory) must be the single element of the index, unless parseresults is given. """ if parseresults: = parseresults else: assert len(index) == 1 = file_naming.parse(index[0]) assert, "must be a recognizable duplicity file" Path.__init__(self, base, index)
[docs] def filtered_open(self, mode="rb", gpg_profile=None): """ Return fileobj with appropriate encryption/compression If encryption is specified but no gpg_profile, use config.default_profile. """ assert not self.opened and not self.fileobj assert not ( and if gpg_profile: assert if return gzip.GzipFile(, mode) elif if not gpg_profile: gpg_profile = config.gpg_profile if mode == "rb": return gpg.GPGFile(False, self, gpg_profile) elif mode == "wb": return gpg.GPGFile(True, self, gpg_profile) else: return
[docs]class PathDeleter(ITRBranch): """Delete a directory. Called by Path.deltree"""
[docs] def start_process(self, index, path): # pylint: disable=unused-argument self.path = path
[docs] def end_process(self): self.path.delete()
[docs] def can_fast_process(self, index, path): # pylint: disable=unused-argument return not path.isdir()
[docs] def fast_process(self, index, path): # pylint: disable=unused-argument path.delete()