feat(native): add SFTP client API

Changelog.rst

@@ -7,6 +7,9 @@ Change Log
 Changes
 --------
 
+* Added a native ``SFTPClient`` API with public remote directory, metadata,
+  mutation and transfer operations, plus remote current working directory
+  support.
 * All local file operations now use a thread pool to improve local file I/O performance. This includes loading private
   key files from a local file path, identity authentication using local files as well as SFTP read/write operations on
   local files.

ci/integration_tests/libssh2_clients/test_sftp_client.py

@@ -0,0 +1,43 @@
+#  This file is part of parallel-ssh.
+#  Copyright (C) 2014-2026 Panos Kittenis.
+#  Copyright (C) 2014-2026 parallel-ssh Contributors.
+#
+#  This library is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation, version 2.1.
+
+import os
+import shutil
+import tempfile
+
+from .base_ssh2_case import SSH2TestCase
+
+
+class SFTPClientTest(SSH2TestCase):
+
+    def test_cwd_directory_and_transfer_operations(self):
+        remote_root = tempfile.mkdtemp(prefix='parallel-ssh-sftp-')
+        local_root = tempfile.mkdtemp(prefix='parallel-ssh-local-')
+        local_source = os.path.join(local_root, 'source.txt')
+        local_copy = os.path.join(local_root, 'copy.txt')
+        try:
+            with open(local_source, 'w') as handle:
+                handle.write('parallel-ssh')
+            sftp = self.client.make_sftp_client()
+            self.assertTrue(sftp.getcwd().startswith('/'))
+            sftp.chdir(remote_root)
+            self.assertEqual(sftp.getcwd(), os.path.realpath(remote_root))
+            sftp.mkdir('nested')
+            self.assertIn('nested', sftp.listdir('.'))
+            sftp.put(local_source, 'nested/remote.txt')
+            self.assertIn('remote.txt', sftp.listdir('nested'))
+            sftp.get('nested/remote.txt', local_copy)
+            with open(local_copy) as handle:
+                self.assertEqual(handle.read(), 'parallel-ssh')
+            sftp.rename('nested/remote.txt', 'nested/renamed.txt')
+            sftp.remove('nested/renamed.txt')
+            sftp.rmdir('nested')
+            self.assertNotIn('nested', sftp.listdir('.'))
+        finally:
+            shutil.rmtree(remote_root, ignore_errors=True)
+            shutil.rmtree(local_root, ignore_errors=True)

ci/integration_tests/libssh2_clients/test_single_client.py

@@ -1072,12 +1072,12 @@ def test_interactive_shell_exit_code(self):
         self.assertEqual(shell.exit_code, 1)
 
     def test_sftp_init_exc(self):
-        def _make_sftp():
+        def _make_sftp_eagain():
             raise Exception
         client = SSHClient(self.host, port=self.port,
                            pkey=self.user_key,
                            num_retries=1)
-        client._make_sftp_eagain = _make_sftp
+        client._make_sftp_eagain = _make_sftp_eagain
         self.assertRaises(SFTPError, client._make_sftp)
 
     @patch('pssh.clients.native.single.Session')

doc/api.rst

@@ -6,6 +6,7 @@ API Documentation
 
    native_parallel
    native_single
+   native_sftp
    ssh_parallel
    ssh_single
    base_parallel

doc/native_sftp.rst

@@ -0,0 +1,28 @@
+Native SFTP Client
+==================
+
+The native client can open a user-facing SFTP client that owns one reusable
+SFTP channel and tracks a remote current working directory.
+
+.. code-block:: python
+
+    from pssh.clients import SSHClient
+
+    client = SSHClient('localhost')
+    sftp = client.make_sftp_client()
+    sftp.chdir('/srv/uploads')
+    sftp.mkdir('incoming')
+    sftp.put('local.txt', 'incoming/remote.txt')
+    print(list(sftp.listdir('incoming')))
+    sftp.get('incoming/remote.txt', 'downloaded.txt')
+
+Relative remote paths are joined to ``sftp.getcwd()``; ``.`` and ``..``
+components are resolved by the server. ``listdir`` returns a generator that
+keeps the remote directory handle open until it is exhausted or closed. The SFTP client is bound to its parent ``SSHClient`` connection.
+This API is available for the native ``ssh2-python`` client only; the
+``pssh.clients.ssh`` backend does not currently support SFTP.
+
+.. automodule:: pssh.clients.native.sftp
+    :members:
+    :undoc-members:
+    :member-order: groupwise

pssh/clients/native/__init__.py

@@ -18,3 +18,4 @@
 # flake8: noqa: F401
 from .parallel import ParallelSSHClient
 from .single import SSHClient, logger
+from .sftp import SFTPClient

pssh/clients/native/sftp.py

@@ -0,0 +1,123 @@
+#  This file is part of parallel-ssh.
+#  Copyright (C) 2014-2026 Panos Kittenis.
+#  Copyright (C) 2014-2026 parallel-ssh Contributors.
+#
+#  This library is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation, version 2.1.
+
+import posixpath
+
+from ssh2.error_codes import LIBSSH2_ERROR_EAGAIN
+
+
+class SFTPClient(object):
+    """User-facing SFTP operations bound to one native SSH client."""
+
+    __slots__ = ('_client', '_sftp', '_cwd')
+
+    def __init__(self, client, sftp=None):
+        self._client = client
+        self._sftp = client._make_sftp() if sftp is None else sftp
+        self._cwd = self._canonical_path('.')
+
+    def _canonical_path(self, path):
+        return self._client.eagain(self._sftp.realpath, path)
+
+    def _remote_path(self, path):
+        if not isinstance(path, str):
+            raise TypeError("Remote path must be a string.")
+        if not path:
+            return self._cwd
+        # ``.`` and ``..`` are left for the server to resolve, as a
+        # client-side normalisation would ignore symbolic links.
+        return posixpath.join(self._cwd, path)
+
+    def getcwd(self):
+        """Get the current remote working directory."""
+        return self._cwd
+
+    def chdir(self, path):
+        """Change the current remote working directory."""
+        target = self._canonical_path(self._remote_path(path))
+        with self._client._sftp_openfh(self._sftp.opendir, target):
+            pass
+        self._cwd = target
+        return self._cwd
+
+    def listdir(self, path='.', encoding='utf-8'):
+        """List names in a remote directory.
+
+        Returns a generator. The remote directory handle stays open until
+        the generator is exhausted or closed.
+        """
+        with self._client._sftp_openfh(
+                self._sftp.opendir, self._remote_path(path)) as dir_h:
+            # One readdir entry is one file name, which may contain newlines.
+            for size, entry, _attrs in dir_h.readdir():
+                if size == LIBSSH2_ERROR_EAGAIN:
+                    self._client.poll()
+                    continue
+                name = entry.decode(encoding)
+                if name not in ('.', '..'):
+                    yield name
+
+    def stat(self, path):
+        """Return attributes for a remote path, following symbolic links."""
+        return self._client.eagain(self._sftp.stat, self._remote_path(path))
+
+    def lstat(self, path):
+        """Return attributes for a remote path without following links."""
+        return self._client.eagain(self._sftp.lstat, self._remote_path(path))
+
+    def mkdir(self, path):
+        """Create a remote directory and missing parent directories."""
+        return self._client.mkdir(self._sftp, self._remote_path(path))
+
+    def rmdir(self, path):
+        """Remove an empty remote directory."""
+        return self._client.eagain(self._sftp.rmdir, self._remote_path(path))
+
+    def rename(self, source, destination):
+        """Rename a remote path."""
+        return self._client.eagain(
+            self._sftp.rename,
+            self._remote_path(source),
+            self._remote_path(destination),
+        )
+
+    def remove(self, path):
+        """Remove a remote file."""
+        return self._client.eagain(self._sftp.unlink, self._remote_path(path))
+
+    unlink = remove
+
+    def get(self, remote_file, local_file):
+        """Copy one remote file to a local path."""
+        return self._client.sftp_get(
+            self._sftp, self._remote_path(remote_file), local_file)
+
+    def put(self, local_file, remote_file):
+        """Copy one local file to a remote path."""
+        return self._client.sftp_put(
+            self._sftp, local_file, self._remote_path(remote_file))
+
+    def copy_file(self, local_file, remote_file, recurse=False):
+        """Copy a local file or directory to a remote path."""
+        return self._client.copy_file(
+            local_file,
+            self._remote_path(remote_file),
+            recurse=recurse,
+            sftp=self._sftp,
+        )
+
+    def copy_remote_file(self, remote_file, local_file, recurse=False,
+                         encoding='utf-8'):
+        """Copy a remote file or directory to a local path."""
+        return self._client.copy_remote_file(
+            self._remote_path(remote_file),
+            local_file,
+            recurse=recurse,
+            sftp=self._sftp,
+            encoding=encoding,
+        )

pssh/clients/native/single.py

@@ -18,6 +18,7 @@
 import logging
 import os
 from collections import deque
+from time import monotonic
 
 from gevent import sleep, get_hub
 from gevent.fileobject import FileObjectThread
@@ -35,6 +36,7 @@
     LIBSSH2_SFTP_S_IXGRP, LIBSSH2_SFTP_S_IXOTH
 
 from .tunnel import FORWARDER
+from .sftp import SFTPClient
 from ..base.single import BaseSSHClient, PollMixIn
 from ...constants import DEFAULT_RETRIES, RETRY_DELAY
 from ...exceptions import SessionError, SFTPError, \
@@ -94,6 +96,12 @@ class SSHClient(BaseSSHClient):
     """ssh2-python (libssh2) based non-blocking SSH client."""
     # 2MB buffer
     _BUF_SIZE = 2048 * 1024
+    # Keep SCP receives in bounded chunks instead of requesting the whole file.
+    _SCP_RECV_BUF_SIZE = 64 * 1024
+    # Seconds an SCP receive may go without data, EOF or EAGAIN before it
+    # fails. libssh2 also returns 0 bytes when the remote closes the channel
+    # without sending EOF, which would otherwise be retried forever.
+    _SCP_RECV_STALL_TIMEOUT = 30
 
     def __init__(self, host,
                  user=None, password=None, port=None,
@@ -441,6 +449,10 @@ def _make_sftp(self):
             raise SFTPError(ex)
         return sftp
 
+    def make_sftp_client(self):
+        """Open a user-facing SFTP client bound to this SSH session."""
+        return SFTPClient(self)
+
     def _mkdir(self, sftp, directory):
         """Make directory via SFTP channel.
 
@@ -689,11 +701,34 @@ def _scp_recv(self, remote_file, local_file):
         local_fh = FileObjectThread(local_file, 'wb', threadpool=THREAD_POOL)
         try:
             total = 0
+            stalled_since = None
             while total < fileinfo.st_size:
-                size, data = file_chan.read(size=fileinfo.st_size - total)
+                size = min(self._SCP_RECV_BUF_SIZE, fileinfo.st_size - total)
+                size, data = file_chan.read(size=size)
                 if size == LIBSSH2_ERROR_EAGAIN:
+                    stalled_since = None
+                    self.poll()
+                    continue
+                if size == 0:
+                    if file_chan.eof():
+                        raise SCPError(
+                            "Unexpected EOF while receiving %s (%s of %s bytes)" % (
+                                remote_file, total, fileinfo.st_size),
+                            remote_file, self.host)
+                    now = monotonic()
+                    if stalled_since is None:
+                        stalled_since = now
+                    elif now - stalled_since > self._SCP_RECV_STALL_TIMEOUT:
+                        raise SCPError(
+                            "No data received for %s seconds while receiving %s "
+                            "(%s of %s bytes)" % (
+                                self._SCP_RECV_STALL_TIMEOUT, remote_file,
+                                total, fileinfo.st_size),
+                            remote_file, self.host)
                     self.poll()
+                    sleep(.1)
                     continue
+                stalled_since = None
                 total += size
                 local_fh.write(data)
         finally: