Use the copy-data extension for cp_file when the server supports it

sshfs/spec.py

@@ -8,7 +8,17 @@
 from typing import Optional
 
 import asyncssh
-from asyncssh.sftp import SFTPOpUnsupported
+from asyncssh.sftp import (
+    FILEXFER_TYPE_DIRECTORY,
+    FILEXFER_TYPE_REGULAR,
+    FILEXFER_TYPE_SYMLINK,
+    FILEXFER_TYPE_UNKNOWN,
+    SFTPError,
+    SFTPFailure,
+    SFTPFileAlreadyExists,
+    SFTPOpUnsupported,
+    SFTPPermissionDenied,
+)
 from fsspec.asyn import (
     AsyncFileSystem,
     FSTimeoutError,
@@ -29,6 +39,12 @@
 
 async_methods.append("_mv")
 
+_FILE_TYPES = {
+    FILEXFER_TYPE_REGULAR: "file",
+    FILEXFER_TYPE_DIRECTORY: "directory",
+    FILEXFER_TYPE_SYMLINK: "link",
+}
+
 # Always allocate 2 channels for shell operations
 # and the rest (generally 8) for SFTP.
 _SHELL_CHANNELS = 2
@@ -79,6 +95,8 @@ def __init__(
 
         self._stack = AsyncExitStack()
         self.active_executors = 0
+        # None means "not probed yet"; resolved on first _cp_file call.
+        self._supports_remote_copy = None
         self._client, self._pool = self.connect(
             host,
             pool_type,
@@ -166,7 +184,13 @@ def client(self):
         return self._client
 
     def _decode_attributes(self, attributes):
-        if stat.S_ISDIR(attributes.permissions):
+        # SFTP v4 and later carry the file type in its own field and
+        # leave only the permission bits in `permissions`, so the type
+        # has to be read from there when the server reports one.
+        file_type = getattr(attributes, "type", FILEXFER_TYPE_UNKNOWN)
+        if file_type != FILEXFER_TYPE_UNKNOWN:
+            kind = _FILE_TYPES.get(file_type, "unknown")
+        elif stat.S_ISDIR(attributes.permissions):
             kind = "directory"
         elif stat.S_ISREG(attributes.permissions):
             kind = "file"
@@ -216,18 +240,35 @@ async def _modified(self, path: str, **kwargs) -> datetime:
     @wrap_exceptions
     async def _mv(self, lpath, rpath, **kwargs):
         async with self._pool.get() as channel:
+            # posix-rename is an extension to the original SFTP
+            # protocol, but it is the only form that can replace an
+            # existing destination atomically.
             with suppress(SFTPOpUnsupported):
                 return await channel.posix_rename(lpath, rpath)
 
-        # Some systems doesn't natively support posix_rename
-        # which is an extension to the original SFTP protocol.
-        # In that case we are going to copy the file and delete
-        # it.
-
-        try:
-            await self._cp_file(lpath, rpath)
-        finally:
-            await self._rm_file(lpath)
+            # The standard rename keeps the object's identity
+            # (symlinks stay symlinks, hardlinks keep their inode and
+            # their special bits) and cannot lose data. Only the
+            # statuses that a rename returns for "cannot rename these
+            # operands" -- an existing destination, a cross-device
+            # move, an unimplemented request -- fall through to the
+            # shell below; permission, missing-file and transport
+            # errors are the caller's answer.
+            with suppress(
+                SFTPFailure, SFTPFileAlreadyExists, SFTPOpUnsupported
+            ):
+                return await channel.rename(lpath, rpath)
+
+        # Neither rename applies, so let the remote mv do it. Copying
+        # and deleting the source here would be unsafe: a successful
+        # copy only proves that the bytes reached an open handle, not
+        # that the destination path still names it, so the source could
+        # be removed after its data ended up somewhere unreachable. On
+        # a server without shell access this raises instead, leaving
+        # the source untouched.
+        lpath, rpath = self._shell_paths(lpath, rpath)
+        cmd = f"mv -- {shlex.quote(lpath)} {shlex.quote(rpath)}"
+        await self._execute(cmd)
 
     @wrap_exceptions
     async def _put_file(
@@ -259,9 +300,144 @@ async def _get_file(
                 progress_handler=as_progress_handler(callback),
             )
 
+    @staticmethod
+    def _shell_paths(*paths):
+        # Shell commands are text. Byte paths are decoded as UTF-8;
+        # paths that are not valid UTF-8 can only be used with the
+        # operations that stay on the SFTP channel.
+        decoded = []
+        for path in paths:
+            if isinstance(path, bytes):
+                try:
+                    path = path.decode("utf-8")
+                except UnicodeDecodeError as exc:
+                    raise ValueError(
+                        f"{path!r} is not valid UTF-8 and cannot be used "
+                        "in a shell command"
+                    ) from exc
+            decoded.append(path)
+        return decoded
+
+    async def _remote_copy_file(self, channel, lpath, rpath):
+        """Copy over the copy-data extension. Returns False when the
+        copy must be handled by the shell fallback instead."""
+        # The remote copy only ever CREATES the destination (FXF_EXCL)
+        # and copies until the source's end of file. Everything an
+        # in-place overwrite would need is unsolvable over SFTP: a
+        # stale pre-copy size corrupts sources whose stat lies (procfs)
+        # or that change while copying, a post-copy truncate can be
+        # denied (fsetstat policy) after the destination was already
+        # modified, a pre-copy truncate destroys sources aliased behind
+        # an undetectable hardlink, and replacing the directory entry
+        # loses the inode. Existing destinations therefore go to the
+        # shell fallback, whose cp implements those semantics natively.
+        # FXF_EXCL also refuses to create through a dangling symlink.
+        #
+        # A failed or interrupted copy may leave a partial destination
+        # behind, exactly like an interrupted cp. It is deliberately
+        # never unlinked: FXF_EXCL proves this client created the
+        # inode, but over SFTP the pathname cannot be re-verified to
+        # still name that inode at cleanup time, so removing it could
+        # delete an unrelated file that raced onto the same name.
+
+        # A directory destination means "copy into": like cp, resolve
+        # it against the source's basename. isdir() checks the file
+        # type on every SFTP version (v4+ attributes carry no type bits
+        # in `permissions`) and is False for missing paths.
+        if await channel.isdir(rpath):
+            rpath = posixpath.join(
+                channel.encode(rpath),
+                posixpath.basename(channel.encode(lpath)),
+            )
+
+        # The source is opened before the destination is created so
+        # that a missing source cannot leave an empty destination.
+        src_file = await channel.open(lpath, "rb", block_size=0)
+        try:
+            # Like cp for new files: the source's mode without the
+            # special bits, which the server then filters through its
+            # umask. A mode of 0 is a valid mode, so only a genuinely
+            # missing one is replaced -- and it is replaced with 0600,
+            # never with the server's default: servers may deny fstat
+            # (OpenSSH -P fstat) while allowing the copy, and defaulting
+            # to a world-readable mode would publish the contents of a
+            # private source.
+            mode = 0o600
+            with suppress(OSError, SFTPError):
+                src_attrs = await src_file.stat()
+                if src_attrs.permissions is not None:
+                    mode = src_attrs.permissions & 0o777
+            attrs = asyncssh.SFTPAttrs(permissions=mode)
+            try:
+                dst_file = await channel.open(
+                    rpath,
+                    asyncssh.FXF_WRITE
+                    | asyncssh.FXF_CREAT
+                    | asyncssh.FXF_EXCL,
+                    attrs,
+                    block_size=0,
+                )
+            except (OSError, SFTPError):
+                # Existing destination (any alias of the source is one),
+                # dangling symlink, missing parent, trailing slash on a
+                # file: the shell fallback owns these forms.
+                return False
+
+            try:
+                await channel.remote_copy(src_file, dst_file)
+            except SFTPOpUnsupported:
+                # Advertised but not actually implemented: stop trying
+                # the extension on this connection and let the shell cp
+                # overwrite the empty file just created.
+                self._supports_remote_copy = False
+                with suppress(OSError, SFTPError):
+                    await dst_file.close()
+                return False
+            except SFTPPermissionDenied:
+                # Denied for these operands (an OpenSSH allow/deny
+                # policy can depend on the paths involved), which says
+                # nothing about the next copy: fall back for this one
+                # without disabling the extension connection-wide.
+                with suppress(OSError, SFTPError):
+                    await dst_file.close()
+                return False
+            except BaseException:
+                with suppress(OSError, SFTPError):
+                    await dst_file.close()
+                raise
+            # A close failure after writing must surface: the data may
+            # not be durable.
+            await dst_file.close()
+        finally:
+            with suppress(OSError, SFTPError):
+                await src_file.close()
+        return True
+
     @wrap_exceptions
     async def _cp_file(self, lpath, rpath, **kwargs):
-        cmd = f"cp {shlex.quote(lpath)} {shlex.quote(rpath)}"
+        # Server-side copy through the copy-data extension (asyncssh >=
+        # 2.19 with an OpenSSH >= 9.0 server) needs no shell access and
+        # keeps the data on the server. The capability is
+        # per-connection, so it is cached after the first probe and the
+        # shell fallback never touches the channel pool again. The
+        # extension path only creates new destinations; everything else
+        # falls back to a shell cp.
+        if self._supports_remote_copy is not False:
+            async with self._pool.get() as channel:
+                if self._supports_remote_copy is None:
+                    self._supports_remote_copy = getattr(
+                        channel, "supports_remote_copy", False
+                    )
+                if (
+                    self._supports_remote_copy
+                    and await self._remote_copy_file(channel, lpath, rpath)
+                ):
+                    return
+
+        lpath, rpath = self._shell_paths(lpath, rpath)
+        # `--` keeps a relative path starting with a dash from being
+        # parsed as an option.
+        cmd = f"cp -- {shlex.quote(lpath)} {shlex.quote(rpath)}"
         await self._execute(cmd)
 
     @wrap_exceptions

sshfs/utils.py

@@ -4,7 +4,7 @@
 
 from asyncssh import ProcessError
 from asyncssh.misc import PermissionDenied
-from asyncssh.sftp import SFTPFailure, SFTPNoSuchFile
+from asyncssh.sftp import SFTPFailure, SFTPNoSuchFile, SFTPPermissionDenied
 from fsspec.asyn import sync_wrapper
 
 _NOT_FOUND = os.strerror(errno.ENOENT)
@@ -27,6 +27,8 @@ async def wrapper(*args, **kwargs):
             return await func(*args, **kwargs)
         except PermissionDenied as exc:
             raise PermissionError(exc.reason) from exc
+        except SFTPPermissionDenied as exc:
+            raise PermissionError(errno.EACCES, exc.reason) from exc
         except SFTPNoSuchFile as exc:
             raise FileNotFoundError(errno.ENOENT, _NOT_FOUND) from exc
         except ProcessError as exc:

tests/conftest.py

@@ -1,7 +1,112 @@
+import asyncio
 import threading
+from contextlib import suppress
+from pathlib import Path
 from queue import Queue
 
+import asyncssh
 import mockssh.server
+import pytest
+
+_STATIC = (Path(__file__).parent / "static").resolve()
+_USER_KEY = asyncssh.read_private_key(str(_STATIC / "user.key"))
+
+
+class _TestSSHServer(asyncssh.SSHServer):
+    def begin_auth(self, username):
+        return True
+
+    def public_key_auth_supported(self):
+        return True
+
+    def validate_public_key(self, username, key):
+        return key == _USER_KEY.convert_to_public()
+
+
+class _ZeroSizeSFTPServer(asyncssh.SFTPServer):
+    """Reports every file as empty, like procfs and sysfs do, while
+    still serving the real content."""
+
+    def _zero(self, result):
+        attrs = asyncssh.SFTPAttrs.from_local(result)
+        attrs.size = 0
+        return attrs
+
+    def stat(self, path):
+        return self._zero(super().stat(path))
+
+    def lstat(self, path):
+        return self._zero(super().lstat(path))
+
+    def fstat(self, file_obj):
+        return self._zero(super().fstat(file_obj))
+
+
+def _serve(root, sftp_server=asyncssh.SFTPServer, sftp_version=3):
+    """Start an authenticated SFTP server chrooted to `root` on its own
+    event loop thread. Yields (host, port, root, sftp_version)."""
+    if not hasattr(asyncssh.SFTPClient, "supports_remote_copy"):
+        pytest.skip("asyncssh without copy-data support (< 2.19)")
+    loop = asyncio.new_event_loop()
+    thread = threading.Thread(target=loop.run_forever, daemon=True)
+    thread.start()
+
+    async def _listen():
+        return await asyncssh.listen(
+            "127.0.0.1",
+            0,
+            server_host_keys=[_USER_KEY],
+            server_factory=_TestSSHServer,
+            sftp_factory=lambda chan: sftp_server(chan, chroot=str(root)),
+            sftp_version=sftp_version,
+        )
+
+    server = asyncio.run_coroutine_threadsafe(_listen(), loop).result(30)
+    try:
+        yield "127.0.0.1", server.get_port(), root, sftp_version
+    finally:
+
+        async def _shutdown():
+            server.close()
+            await server.wait_closed()
+            tasks = [
+                task
+                for task in asyncio.all_tasks()
+                if task is not asyncio.current_task()
+            ]
+            for task in tasks:
+                task.cancel()
+            if tasks:
+                await asyncio.wait(tasks, timeout=5)
+
+        with suppress(Exception):
+            asyncio.run_coroutine_threadsafe(_shutdown(), loop).result(30)
+        loop.call_soon_threadsafe(loop.stop)
+        thread.join(timeout=5)
+        loop.close()
+
+
+@pytest.fixture(scope="session", params=[3, 4], ids=["sftpv3", "sftpv4"])
+def asyncssh_server(tmp_path_factory, request):
+    """SFTP server that, unlike the paramiko-based mockssh fixture,
+    implements the copy-data and limits extensions. Authenticated with
+    the test user key and chrooted to a fresh directory; yields
+    (host, port, root, version) where the remote "/" maps to root.
+    Runs once per
+    SFTP protocol generation: v4+ moves the file type out of the
+    permission bits."""
+    yield from _serve(
+        tmp_path_factory.mktemp(f"asyncssh-root-v{request.param}"),
+        sftp_version=request.param,
+    )
+
+
+@pytest.fixture(scope="session")
+def zero_size_server(tmp_path_factory):
+    """Like asyncssh_server, but every stat reports size 0."""
+    yield from _serve(
+        tmp_path_factory.mktemp("zero-size-root"), _ZeroSizeSFTPServer
+    )
 
 
 def _handler_run(self):