fix symlink traversal for #473

dotdrop/updater.py

@@ -12,6 +12,7 @@
 # local imports
 from dotdrop.logger import Logger
 from dotdrop.ftree import FTreeDir
+from dotdrop.linktypes import LinkTypes
 from dotdrop.templategen import Templategen
 from dotdrop.utils import ignores_to_absolute, removepath, \
     get_unique_tmp_name, write_to_tmpfile, must_ignore, \
@@ -261,6 +262,12 @@ def _handle_file(self, deployed_path, local_path,
     def _handle_dir(self, deployed_path, local_path,
                     dotfile, ignores):
         """sync path (local dir) and local_path (dotdrop dir path)"""
+        if dotfile.link == LinkTypes.LINK_CHILDREN:
+            return self._handle_link_children_dir(deployed_path,
+                                                  local_path,
+                                                  dotfile,
+                                                  ignores)
+
         ret = True
         self.log.dbg(f'handle update for dir {deployed_path} to {local_path}')
 
@@ -345,6 +352,125 @@ def _handle_dir(self, deployed_path, local_path,
             self.log.sub(f'\"{dstpath}\" content updated')
         return ret
 
+    def _handle_link_children_dir(self, deployed_path, local_path,
+                                  dotfile, ignores):
+        """sync link_children dotfile without traversing managed symlinks"""
+        ret = True
+        deployed_path = os.path.expanduser(deployed_path)
+        local_path = os.path.expanduser(local_path)
+
+        self.log.dbg('handling update for link_children dotfile')
+
+        local_children = set(os.listdir(local_path))
+        deployed_children = set(os.listdir(deployed_path))
+
+        local_only = local_children - deployed_children
+        deployed_only = deployed_children - local_children
+        common = local_children & deployed_children
+
+        for child in local_only:
+            path = os.path.join(local_path, child)
+            if self._must_ignore([path], ignores):
+                self.log.sub(f'\"{path}\" ignored')
+                continue
+            if self.dry:
+                self.log.dry(f'would rm -r {path}')
+                continue
+            self.log.dbg(f'rm -r {path}')
+            if not self._confirm_rm_r(path):
+                continue
+            if not removepath(path, logger=self.log):
+                self.log.warn(f'unable to remove {path}, do manually')
+                ret = False
+                continue
+            self.log.sub(f'\"{path}\" removed')
+
+        ignore_missing_in_dotdrop = self.ignore_missing_in_dotdrop or \
+            dotfile.ignore_missing_in_dotdrop
+        if not ignore_missing_in_dotdrop:
+            for child in deployed_only:
+                srcpath = os.path.join(deployed_path, child)
+                dstpath = os.path.join(local_path, child)
+                if self._must_ignore([srcpath, dstpath], ignores):
+                    self.log.sub(f'\"{dstpath}\" ignored')
+                    continue
+                if self.dry:
+                    if os.path.islink(srcpath):
+                        target = os.readlink(srcpath)
+                        self.log.dry(f'would ln -s {target} {dstpath}')
+                    else:
+                        self.log.dry(f'would cp -r {srcpath} {dstpath}')
+                    continue
+                self.log.dbg(f'cp {srcpath} {dstpath}')
+                try:
+                    os.makedirs(os.path.dirname(dstpath), exist_ok=True)
+                    if os.path.islink(srcpath):
+                        target = os.readlink(srcpath)
+                        os.symlink(target, dstpath)
+                    elif not os.path.isdir(srcpath):
+                        shutil.copy2(srcpath, dstpath)
+                except IOError as exc:
+                    msg = f'{srcpath} update right only failed'
+                    msg += f', do manually: {exc}'
+                    self.log.warn(msg)
+                    ret = False
+                    continue
+                self.log.sub(f'\"{dstpath}\" updated')
+
+        for child in common:
+            srcpath = os.path.join(deployed_path, child)
+            dstpath = os.path.join(local_path, child)
+
+            if self._must_ignore([srcpath, dstpath], ignores):
+                self.log.sub(f'\"{dstpath}\" ignored')
+                continue
+
+            if os.path.islink(srcpath):
+                # managed link_children symlink, no update required
+                if os.path.realpath(srcpath) == os.path.realpath(dstpath):
+                    continue
+                if self.dry:
+                    target = os.readlink(srcpath)
+                    msg = f'would replace {dstpath} with symlink {target}'
+                    self.log.dry(msg)
+                    continue
+                if not self._overwrite(srcpath, dstpath):
+                    continue
+                if not removepath(dstpath, logger=self.log):
+                    self.log.warn(f'unable to remove {dstpath}, do manually')
+                    ret = False
+                    continue
+                target = os.readlink(srcpath)
+                os.symlink(target, dstpath)
+                self.log.sub(f'\"{dstpath}\" updated')
+                continue
+
+            if os.path.isdir(srcpath):
+                continue
+
+            if not self._same_rights(dstpath, srcpath):
+                self._mirror_file_perms(srcpath, dstpath)
+            out = diff(modified=dstpath, original=srcpath,
+                       debug=self.debug)
+            if not out:
+                continue
+            if self.dry:
+                msg = f'would update content of {dstpath} from {srcpath}'
+                self.log.dry(msg)
+                continue
+            self.log.dbg(f'cp {srcpath} {dstpath}')
+            try:
+                shutil.copy2(srcpath, dstpath)
+                self._mirror_file_perms(srcpath, dstpath)
+            except IOError as exc:
+                msg = f'{srcpath} update common failed, do manually: {exc}'
+                self.log.warn(msg)
+                ret = False
+                continue
+            self.log.sub(f'\"{dstpath}\" content updated')
+
+        return ret
+
     def _overwrite(self, src, dst):
         """ask for overwritting"""
         msg = f'Overwrite \"{dst}\" with \"{src}\"?'

tests-ng/update-link-children-symlink.sh

@@ -0,0 +1,78 @@
+#!/usr/bin/env bash
+# author: deadc0de6 (https://github.com/deadc0de6)
+# Copyright (c) 2017, deadc0de6
+#
+# ensure link_children update does not traverse managed symlink trees
+# returns 1 in case of error
+#
+
+## start-cookie
+set -eu -o errtrace -o pipefail
+cur=$(cd "$(dirname "${0}")" && pwd)
+ddpath="${cur}/../"
+PPATH="{PYTHONPATH:-}"
+export PYTHONPATH="${ddpath}:${PPATH}"
+altbin="python3 -m dotdrop.dotdrop"
+if hash coverage 2>/dev/null; then
+  mkdir -p coverages/
+  altbin="coverage run -p --data-file coverages/coverage --source=dotdrop -m dotdrop.dotdrop"
+fi
+bin="${DT_BIN:-${altbin}}"
+# shellcheck source=tests-ng/helpers
+source "${cur}"/helpers
+echo -e "$(tput setaf 6)==> RUNNING $(basename "${BASH_SOURCE[0]}") <==$(tput sgr0)"
+## end-cookie
+
+################################################################
+# this is the test
+################################################################
+
+# the dotfile source
+tmps=$(mktemp -d --suffix='-dotdrop-tests' || mktemp -d)
+mkdir -p "${tmps}"/dotfiles
+
+# the dotfile destination
+tmpd=$(mktemp -d --suffix='-dotdrop-tests' || mktemp -d)
+
+clear_on_exit "${tmps}"
+clear_on_exit "${tmpd}"
+
+# create the dotfile and nested hierarchy
+mkdir -p "${tmps}"/dotfiles/dir1/child/sub
+echo 'nested-content' > "${tmps}"/dotfiles/dir1/child/sub/nested
+echo 'root-content' > "${tmps}"/dotfiles/dir1/root-file
+
+# create the config file
+cfg="${tmps}/config.yaml"
+
+cat > "${cfg}" << _EOF
+config:
+  backup: true
+  create: true
+  dotpath: dotfiles
+dotfiles:
+  d_linkchildren:
+    src: dir1
+    dst: ${tmpd}/dir1
+    link: link_children
+profiles:
+  p1:
+    dotfiles:
+    - d_linkchildren
+_EOF
+
+# install
+cd "${ddpath}" | ${bin} install -f -c "${cfg}" -p p1 -V
+
+# direct children are symlinked
+[ ! -h "${tmpd}"/dir1/child ] && echo "child is not a symlink" && exit 1
+[ ! -h "${tmpd}"/dir1/root-file ] && echo "root-file is not a symlink" && exit 1
+
+# update should not walk/traverse managed symlink children
+out=$(cd "${ddpath}" | ${bin} update -f -c "${cfg}" -p p1 -V "${tmpd}"/dir1 2>&1)
+
+echo "${out}" | grep "added file to list of ${tmpd}/dir1" && echo "link_children update traversed destination symlink tree" && exit 1
+echo "${out}" | grep "added dir to list of ${tmpd}/dir1" && echo "link_children update traversed destination symlink tree" && exit 1
+
+echo "OK"
+exit 0

tests/test_update.py

@@ -7,10 +7,13 @@
 
 import unittest
 import os
+from unittest.mock import patch
 
 from dotdrop.dotdrop import cmd_update
 from dotdrop.dotdrop import cmd_importer
+from dotdrop.dotdrop import cmd_install
 from dotdrop.action import Transform
+from dotdrop.linktypes import LinkTypes
 
 from tests.helpers import create_dir, get_string, get_tempdir, clean, \
     create_random_file, create_fake_config, load_options, edit_content
@@ -184,6 +187,55 @@ def test_update(self):
             newcontent = file.read()
         self.assertTrue(newcontent == 'newcontentbykey')
 
+    def test_update_link_children_skips_managed_symlink_trees(self):
+        """Do not recurse managed symlink trees for link_children update."""
+        fold_config = os.path.join(os.path.expanduser('~'), '.config')
+        create_dir(fold_config)
+
+        srcdir = os.path.join(fold_config, get_string(5))
+        create_dir(srcdir)
+        self.addCleanup(clean, srcdir)
+
+        child_dir = os.path.join(srcdir, 'child')
+        create_dir(child_dir)
+        nested_file, _ = create_random_file(child_dir)
+        self.assertTrue(os.path.exists(nested_file))
+
+        root_file, _ = create_random_file(srcdir)
+        self.assertTrue(os.path.exists(root_file))
+
+        dotfilespath = get_tempdir()
+        self.assertTrue(os.path.exists(dotfilespath))
+        self.addCleanup(clean, dotfilespath)
+
+        profile = get_string(5)
+        confpath = create_fake_config(dotfilespath,
+                                      configname=self.CONFIG_NAME,
+                                      dotpath=self.CONFIG_DOTPATH,
+                                      backup=self.CONFIG_BACKUP,
+                                      create=self.CONFIG_CREATE)
+        self.assertTrue(os.path.exists(confpath))
+
+        opt = load_options(confpath, profile)
+        opt.import_path = [srcdir]
+        opt.import_link = LinkTypes.LINK_CHILDREN
+        self.assertTrue(cmd_importer(opt))
+
+        opt = load_options(confpath, profile)
+        opt.safe = False
+        self.assertTrue(cmd_install(opt))
+
+        child_link = os.path.join(srcdir, 'child')
+        self.assertTrue(os.path.islink(child_link))
+
+        opt = load_options(confpath, profile)
+        opt.safe = False
+        opt.update_path = [srcdir]
+
+        with patch('dotdrop.updater.FTreeDir._walk',
+                   side_effect=AssertionError('FTreeDir walk should not run')):
+            self.assertTrue(cmd_update(opt))
+
 
 def main():
     """entry point"""