Normalize protocol-prefixed paths before SFTP operations

sshfs/spec.py

@@ -16,7 +16,7 @@
     sync,
     sync_wrapper,
 )
-from fsspec.utils import infer_storage_options
+from fsspec.utils import infer_storage_options, stringify_path
 
 from sshfs.file import SSHFile
 from sshfs.pools import SFTPSoftChannelPool
@@ -63,6 +63,10 @@ def __init__(
             Options passed to `SSHClientConnection.start_sftp_client`
             (e.g. env, send_env, path_encoding, path_errors,
             sftp_version).
+
+        Note
+        ----
+        when ``sftp_client_kwargs`` [ ``path_encoding`` ] = ``None`` , protocol-prefixed paths are not supported (e.g. ``b"ssh://host/path/to/file"``)
         """
 
         super().__init__(self, **kwargs)
@@ -95,6 +99,12 @@ def __init__(
 
     @classmethod
     def _strip_protocol(cls, path):
+        # Note: when `path_encoding` is None, protocol-prefixed paths are not supported
+        # Raw SFTP paths must retain their encoding (path_encoding=None).
+        if isinstance(path, bytes):
+            return path
+        # convert `pathlib.Path` and `os.PathLike` etc.
+        path = stringify_path(path)
         # Remove components such as host and username from path.
         inferred_path = infer_storage_options(path)["path"]
         stripped = super()._strip_protocol(inferred_path)
@@ -215,6 +225,8 @@ async def _modified(self, path: str, **kwargs) -> datetime:
 
     @wrap_exceptions
     async def _mv(self, lpath, rpath, **kwargs):
+        lpath = self._strip_protocol(lpath)
+        rpath = self._strip_protocol(rpath)
         async with self._pool.get() as channel:
             with suppress(SFTPOpUnsupported):
                 return await channel.posix_rename(lpath, rpath)
@@ -238,6 +250,7 @@ async def _put_file(
         callback=None,
         **kwargs,
     ):
+        rpath = self._strip_protocol(rpath)
         await self._makedirs(self._parent(rpath), exist_ok=True)
         async with self._pool.get() as channel:
             await channel.put(
@@ -251,6 +264,7 @@ async def _put_file(
     async def _get_file(
         self, lpath, rpath, block_size=READ_BLOCK_SIZE, callback=None, **kwargs
     ):
+        lpath = self._strip_protocol(lpath)
         async with self._pool.get() as channel:
             await channel.get(
                 lpath,
@@ -261,11 +275,14 @@ async def _get_file(
 
     @wrap_exceptions
     async def _cp_file(self, lpath, rpath, **kwargs):
+        lpath = self._strip_protocol(lpath)
+        rpath = self._strip_protocol(rpath)
         cmd = f"cp {shlex.quote(lpath)} {shlex.quote(rpath)}"
         await self._execute(cmd)
 
     @wrap_exceptions
     async def _ls(self, path, detail=True, **kwargs):
+        path = self._strip_protocol(path)
         async with self._pool.get() as channel:
             file_attrs = await channel.readdir(path)
 
@@ -287,6 +304,7 @@ async def _ls(self, path, detail=True, **kwargs):
     async def _mkdir(
         self, path, *, create_parents=True, permissions=511, **kwargs
     ):
+        path = self._strip_protocol(path)
         if create_parents:
             return await self._makedirs(path, exist_ok=True)
 
@@ -298,6 +316,7 @@ async def _mkdir(
     async def _makedirs(
         self, path, *, exist_ok=False, permissions=511, **kwargs
     ):
+        path = self._strip_protocol(path)
         attrs = asyncssh.SFTPAttrs(permissions=permissions)
         async with self._pool.get() as channel:
             await channel.makedirs(path, exist_ok=exist_ok, attrs=attrs)
@@ -307,6 +326,7 @@ async def _makedirs(
 
     @wrap_exceptions
     async def _rm_file(self, path, **kwargs):
+        path = self._strip_protocol(path)
         async with self._pool.get() as channel:
             await channel.unlink(path)
 
@@ -319,6 +339,7 @@ async def _rmdir(
         on_error=None,
         **kwargs,
     ):
+        path = self._strip_protocol(path)
         async with self._pool.get() as channel:
             if recursive:
                 await channel.rmtree(
@@ -343,6 +364,7 @@ async def _rm(self, path, recursive=False, **kwargs):
 
     @wrap_exceptions
     async def _checksum(self, path):
+        path = self._strip_protocol(path)
         system = await self._get_system()
         if system == "Linux":
             command = "md5sum"
@@ -381,18 +403,21 @@ async def _execute(self, *args, **kwargs):
     execute = sync_wrapper(_execute)
 
     def _open(self, path, *args, **kwargs):
+        path = self._strip_protocol(path)
         return SSHFile(self, path, *args, **kwargs)
 
     @wrap_exceptions
     async def _cat_file(self, path, **kwargs):
         """Asynchronously fetch the contents of a file"""
+        path = self._strip_protocol(path)
         async with self._pool.get() as channel:
             async with channel.open(path, "rb") as f:
                 return await f.read()
 
     @wrap_exceptions
     async def _pipe_file(self, path, data, chunksize=50 * 2**20, **kwargs):
         """Asynchronously writes the given data to a remote file in chunks."""
+        path = self._strip_protocol(path)
         await self._makedirs(self._parent(path), exist_ok=True)
 
         async with self._pool.get() as channel: