fix: use custom copy function to ignore extended attributes (#343)

gitbatch/cli.py

@@ -3,17 +3,17 @@
 
 import argparse
 import os
-import shutil
 import tempfile
 from collections import defaultdict
 from pathlib import Path
+from shutil import ignore_patterns
 from urllib.parse import urlparse
 
 import git
 
 from gitbatch import __version__
 from gitbatch.logging import SingleLog
-from gitbatch.utils import normalize_path, to_bool
+from gitbatch.utils import copy, normalize_path, to_bool
 
 
 class GitBatch:
@@ -130,10 +130,10 @@ def _repos_clone(self, repos):
                         if not os.path.isdir(path):
                             raise FileNotFoundError(Path(path).relative_to(tmp))
 
-                    shutil.copytree(
+                    copy.simplecopytree(
                         path,
                         repo["dest"],
-                        ignore=shutil.ignore_patterns(".git"),
+                        ignore=ignore_patterns(".git"),
                         dirs_exist_ok=self.config["ignore_existing"],
                     )
                 except FileExistsError:

gitbatch/utils.py → gitbatch/utils/__init__.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python3
 """Global utility methods and classes."""
 
 import os

gitbatch/utils/copy.py

@@ -0,0 +1,180 @@
+"""
+Copy file utils.
+
+Provides a copy of the shutil.copytree function and its dependencies.
+The copystat function used to preserve extended attributes has side effects
+with SELinux in combination with files copied from temporary directories.
+"""
+
+import contextlib
+import os
+import stat
+import sys
+from shutil import Error, copy
+
+if sys.platform == "win32":
+    import _winapi
+else:
+    _winapi = None
+
+
+def _islink(fn):
+    return fn.is_symlink() if isinstance(fn, os.DirEntry) else os.path.islink(fn)
+
+
+def _copytree(
+    entries,
+    src,
+    dst,
+    symlinks,
+    ignore,
+    ignore_dangling_symlinks,
+    dirs_exist_ok=False,
+):
+    ignored_names = ignore(os.fspath(src), [x.name for x in entries]) if ignore is not None else ()
+
+    os.makedirs(dst, exist_ok=dirs_exist_ok)
+    errors = []
+
+    for srcentry in entries:
+        if srcentry.name in ignored_names:
+            continue
+        srcname = os.path.join(src, srcentry.name)
+        dstname = os.path.join(dst, srcentry.name)
+        try:
+            is_symlink = srcentry.is_symlink()
+            if is_symlink and os.name == "nt":
+                # Special check for directory junctions, which appear as
+                # symlinks but we want to recurse.
+                lstat = srcentry.stat(follow_symlinks=False)
+                if lstat.st_reparse_tag == stat.IO_REPARSE_TAG_MOUNT_POINT:
+                    is_symlink = False
+            if is_symlink:
+                linkto = os.readlink(srcname)
+                if symlinks:
+                    os.symlink(linkto, dstname)
+                    simplecopystat(srcname, dstname, follow_symlinks=(not symlinks))
+                else:
+                    if not os.path.exists(linkto) and ignore_dangling_symlinks:
+                        continue
+
+                    if srcentry.is_dir():
+                        simplecopytree(
+                            srcname,
+                            dstname,
+                            symlinks,
+                            ignore,
+                            ignore_dangling_symlinks,
+                            dirs_exist_ok,
+                        )
+                    else:
+                        simplecopy(srcname, dstname)
+            elif srcentry.is_dir():
+                simplecopytree(
+                    srcname,
+                    dstname,
+                    symlinks,
+                    ignore,
+                    ignore_dangling_symlinks,
+                    dirs_exist_ok,
+                )
+            else:
+                # Will raise a SpecialFileError for unsupported file types
+                simplecopy(srcname, dstname)
+        # catch the Error from the recursive copytree so that we can
+        # continue with other files
+        except Error as err:
+            errors.extend(err.args[0])
+        except OSError as why:
+            errors.append((srcname, dstname, str(why)))
+
+    try:
+        simplecopystat(src, dst)
+    except OSError as why:
+        # Copying file access times may fail on Windows
+        if getattr(why, "winerror", None) is None:
+            errors.append((src, dst, str(why)))
+    if errors:
+        raise Error(errors)
+    return dst
+
+
+def simplecopytree(
+    src,
+    dst,
+    symlinks=False,
+    ignore=None,
+    ignore_dangling_symlinks=False,
+    dirs_exist_ok=False,
+):
+    with os.scandir(src) as itr:
+        entries = list(itr)
+    return _copytree(
+        entries=entries,
+        src=src,
+        dst=dst,
+        symlinks=symlinks,
+        ignore=ignore,
+        ignore_dangling_symlinks=ignore_dangling_symlinks,
+        dirs_exist_ok=dirs_exist_ok,
+    )
+
+
+def simplecopystat(src, dst, *, follow_symlinks=True):
+    def _nop(*args, ns=None, follow_symlinks=None):  # noqa
+        pass
+
+    # follow symlinks (aka don't not follow symlinks)
+    follow = follow_symlinks or not (_islink(src) and os.path.islink(dst))
+    if follow:
+        # use the real function if it exists
+        def lookup(name):
+            return getattr(os, name, _nop)
+    else:
+        # use the real function only if it exists
+        # *and* it supports follow_symlinks
+        def lookup(name):
+            fn = getattr(os, name, _nop)
+            if fn in os.supports_follow_symlinks:
+                return fn
+            return _nop
+
+    if isinstance(src, os.DirEntry):
+        st = src.stat(follow_symlinks=follow)
+    else:
+        st = lookup("stat")(src, follow_symlinks=follow)
+    mode = stat.S_IMODE(st.st_mode)
+    lookup("utime")(dst, ns=(st.st_atime_ns, st.st_mtime_ns), follow_symlinks=follow)
+
+    with contextlib.suppress(NotImplementedError):
+        lookup("chmod")(dst, mode, follow_symlinks=follow)
+
+
+def simplecopy(src, dst, *, follow_symlinks=True):
+    if os.path.isdir(dst):
+        dst = os.path.join(dst, os.path.basename(src))
+
+    if hasattr(_winapi, "CopyFile2"):
+        src_ = os.fsdecode(src)
+        dst_ = os.fsdecode(dst)
+        flags = _winapi.COPY_FILE_ALLOW_DECRYPTED_DESTINATION  # for compat
+        if not follow_symlinks:
+            flags |= _winapi.COPY_FILE_COPY_SYMLINK
+        try:
+            _winapi.CopyFile2(src_, dst_, flags)
+            return dst
+        except OSError as exc:
+            if exc.winerror == _winapi.ERROR_PRIVILEGE_NOT_HELD and not follow_symlinks:
+                # Likely encountered a symlink we aren't allowed to create.
+                # Fall back on the old code
+                pass
+            elif exc.winerror == _winapi.ERROR_ACCESS_DENIED:
+                # Possibly encountered a hidden or readonly file we can't
+                # overwrite. Fall back on old code
+                pass
+            else:
+                raise
+
+    copy(src, dst, follow_symlinks=follow_symlinks)
+    simplecopystat(src, dst, follow_symlinks=follow_symlinks)
+    return dst