diff --git a/src/portkeydrop/protocols.py b/src/portkeydrop/protocols.py index 4efa4fc..15855c9 100644 --- a/src/portkeydrop/protocols.py +++ b/src/portkeydrop/protocols.py @@ -21,6 +21,10 @@ logger = logging.getLogger(__name__) +# asyncssh SFTP v4+ file type constants (avoids import at module level) +_SFTP_TYPE_DIRECTORY = 2 +_SFTP_TYPE_SYMLINK = 3 + ProgressCallback = Callable[[int, int], None] # (bytes_transferred, total_bytes) @@ -569,8 +573,61 @@ def list_dir(self, path: str = ".") -> list[RemoteFile]: target = (path if path != "." else self._cwd).rstrip("/") or "/" files: list[RemoteFile] = [] logger.debug("list_dir: requesting entries for '%s'", target) + + async def _readdir_safe(): + """Readdir loop that treats consecutive empty responses as EOF. + + Some SFTP servers (e.g. Bitvise on the .ssh directory) return + FXP_NAME with count=0 indefinitely instead of FX_EOF. asyncssh + loops forever in this case; we break after 3 consecutive empty + batches — matching WinSCP behaviour. + """ + import asyncssh as _asyncssh + + _MAX_EMPTY = 3 + dirpath = sftp.compose_path(target) + handle = await sftp._handler.opendir(dirpath) + result = [] + consecutive_empty = 0 + at_end = False + try: + while not at_end: + names, at_end = await sftp._handler.readdir(handle) + if not names: + consecutive_empty += 1 + logger.debug( + "_readdir_safe: empty batch %d for '%s'", consecutive_empty, target + ) + if consecutive_empty >= _MAX_EMPTY: + logger.warning( + "_readdir_safe: treating %d consecutive empty batches as EOF for '%s'", + _MAX_EMPTY, + target, + ) + break + else: + consecutive_empty = 0 + # Decode filenames from bytes → str (same as asyncssh scandir does + # internally when called with a str path). + for entry in names: + if entry.filename and isinstance(entry.filename, (bytes, bytearray)): + entry.filename = sftp.decode(entry.filename) + if entry.longname and isinstance(entry.longname, (bytes, bytearray)): + entry.longname = sftp.decode(entry.longname) + result.extend(names) + except _asyncssh.SFTPEOFError: + pass + finally: + await sftp._handler.close(handle) + return result + try: - entries = self._run(sftp.readdir(target)) + entries = self._run(_readdir_safe()) + logger.debug( + "list_dir: readdir returned %d raw entries for '%s'", + len(entries) if entries is not None else -1, + target, + ) except PermissionError: raise except OSError as e: @@ -579,6 +636,21 @@ def list_dir(self, path: str = ".") -> list[RemoteFile]: if e.errno in (_errno.EACCES, _errno.EPERM): raise PermissionError(f"Permission denied: cannot list '{target}'") from e raise + except Exception as e: + # asyncssh raises SFTPError (not OSError) for server-side errors. + # Map permission errors; surface everything else. + import asyncssh as _asyncssh + + if isinstance(e, _asyncssh.SFTPError): + logger.warning( + "list_dir: SFTPError for '%s': code=%s msg=%s", + target, + getattr(e, "code", "?"), + e, + ) + if getattr(e, "code", None) in (3, 4): # FX_PERMISSION_DENIED=3, FX_FAILURE=4 + raise PermissionError(f"Permission denied: cannot list '{target}'") from e + raise logger.debug("list_dir: got %d entries for '%s'", len(entries), target) for entry in entries: name = entry.filename @@ -597,7 +669,13 @@ def list_dir(self, path: str = ".") -> list[RemoteFile]: continue is_dir = bool(mode is not None and stat.S_ISDIR(mode)) full_path = f"{target.rstrip('/')}/{name}" + # SFTP v4+ file type field (separate from permissions) — used by + # strict servers like Bitvise that may not embed the type in the + # permission bits. + sftp_type = getattr(attrs, "type", None) is_link = bool(mode is not None and stat.S_ISLNK(mode)) + if not is_link and sftp_type == _SFTP_TYPE_SYMLINK: + is_link = True if is_link: try: target_attrs = self._run(sftp.stat(full_path)) @@ -605,8 +683,13 @@ def list_dir(self, path: str = ".") -> list[RemoteFile]: target_attrs.permissions ): is_dir = True + elif getattr(target_attrs, "type", None) == _SFTP_TYPE_DIRECTORY: + is_dir = True except Exception: pass + # Fallback: use SFTP v4+ type field when permissions lack type bits + if not is_dir and sftp_type == _SFTP_TYPE_DIRECTORY: + is_dir = True longname = getattr(entry, "longname", "") if not is_dir and longname and longname.startswith("d"): is_dir = True @@ -639,8 +722,34 @@ def chdir(self, path: str) -> str: logger.debug("chdir: '%s'", path) sftp = self._ensure_connected() try: - self._cwd = self._run(sftp.realpath(path)) + resolved = self._run(sftp.realpath(path)) + # Validate the target is a directory (strict servers like Bitvise + # require an explicit stat check — realpath only canonicalises). + attrs = self._run(sftp.stat(resolved)) + logger.debug( + "chdir stat: path='%s' permissions=%s type=%s", + resolved, + attrs.permissions, + getattr(attrs, "type", None), + ) + is_dir = False + if attrs.permissions is not None and stat.S_ISDIR(attrs.permissions): + is_dir = True + elif getattr(attrs, "type", None) == _SFTP_TYPE_DIRECTORY: + is_dir = True + elif attrs.permissions is None and getattr(attrs, "type", None) is None: + # Server returned no type info (e.g. Bitvise on .ssh) — + # assume directory and let the server reject if wrong. + logger.debug( + "chdir: no type info from server, assuming directory for '%s'", resolved + ) + is_dir = True + if not is_dir: + raise NotADirectoryError(f"Not a directory: '{path}'") + self._cwd = resolved logger.debug("chdir: done, cwd='%s'", self._cwd) + except (NotADirectoryError, PermissionError): + raise except OSError as e: import errno as _errno @@ -749,6 +858,8 @@ def stat(self, path: str) -> RemoteFile: attrs = self._run(sftp.stat(path)) mode = attrs.permissions is_dir = stat.S_ISDIR(mode) if mode else False + if not is_dir and getattr(attrs, "type", None) == _SFTP_TYPE_DIRECTORY: + is_dir = True modified = datetime.fromtimestamp(attrs.mtime) if attrs.mtime else None perms = stat.filemode(mode) if mode else "" name = PurePosixPath(path).name diff --git a/tests/test_protocols.py b/tests/test_protocols.py index 7be6d81..8acc1b7 100644 --- a/tests/test_protocols.py +++ b/tests/test_protocols.py @@ -427,7 +427,15 @@ def test_list_dir_maps_file_attributes(self, mock_connect): dot_entry = MagicMock() dot_entry.filename = "." - mock_sftp.readdir.return_value = [dot_entry, file_entry, dir_entry] + + # list_dir uses _readdir_safe which calls sftp._handler.opendir/readdir directly. + mock_handler = AsyncMock() + mock_sftp._handler = mock_handler + mock_sftp.compose_path.return_value = b"/home/user" + # First call returns entries, second call returns empty (EOF signal) + mock_handler.readdir.side_effect = [ + ([dot_entry, file_entry, dir_entry], True), + ] client = SFTPClient(ConnectionInfo(protocol=Protocol.SFTP, host="example.com")) client.connect() @@ -445,6 +453,11 @@ def test_chdir_download_upload_and_file_ops(self, mock_connect): mock_conn = AsyncMock() mock_sftp = AsyncMock() mock_sftp.realpath.side_effect = ["/", "/uploads"] + # chdir now validates with stat — return directory attributes + chdir_stat_attrs = MagicMock() + chdir_stat_attrs.permissions = stat_mod.S_IFDIR | 0o755 + chdir_stat_attrs.type = None + mock_sftp.stat.return_value = chdir_stat_attrs mock_conn.start_sftp_client.return_value = mock_sftp mock_connect.return_value = mock_conn diff --git a/tests/test_sftp_client.py b/tests/test_sftp_client.py index de02671..4aa26f9 100644 --- a/tests/test_sftp_client.py +++ b/tests/test_sftp_client.py @@ -31,6 +31,28 @@ def _make_mock_conn() -> tuple[MagicMock, MagicMock]: return mock_conn, mock_sftp +def _setup_readdir(mock_sftp: AsyncMock, entries: list) -> None: + """Wire up _handler mocks so _readdir_safe returns *entries* in one batch. + + _readdir_safe uses sftp._handler.opendir / readdir / close instead of the + high-level sftp.readdir wrapper. + """ + mock_sftp.compose_path.side_effect = lambda p: p + mock_handler = AsyncMock() + mock_handler.opendir.return_value = "fake-handle" + mock_handler.readdir.return_value = (entries, True) # (names, at_end) + mock_handler.close.return_value = None + mock_sftp._handler = mock_handler + + +def _setup_readdir_error(mock_sftp: AsyncMock, error: Exception) -> None: + """Wire up _handler mocks so _readdir_safe raises *error* on opendir.""" + mock_sftp.compose_path.side_effect = lambda p: p + mock_handler = AsyncMock() + mock_handler.opendir.side_effect = error + mock_sftp._handler = mock_handler + + class TestSFTPClientInit: def test_creates_conn_attribute(self, sftp_info: ConnectionInfo) -> None: client = SFTPClient(sftp_info) @@ -350,7 +372,7 @@ def test_list_dir_returns_files(self, sftp_info: ConnectionInfo) -> None: entry.attrs.uid = 1000 entry.attrs.gid = 1000 entry.longname = "-rw-r--r-- 1 user group 100 Jan 1 file.txt" - mock_sftp.readdir.return_value = [entry] + _setup_readdir(mock_sftp, [entry]) files = client.list_dir() assert len(files) == 1 @@ -363,7 +385,7 @@ def test_list_dir_permission_error(self, sftp_info: ConnectionInfo) -> None: client, mock_sftp = self._make_connected(sftp_info) err = IOError() err.errno = errno.EACCES - mock_sftp.readdir.side_effect = err + _setup_readdir_error(mock_sftp, err) with pytest.raises(PermissionError, match="Permission denied"): client.list_dir("/restricted") @@ -374,7 +396,7 @@ def test_list_dir_reraises_other_oserror(self, sftp_info: ConnectionInfo) -> Non client, mock_sftp = self._make_connected(sftp_info) err = IOError() err.errno = errno.ENOENT - mock_sftp.readdir.side_effect = err + _setup_readdir_error(mock_sftp, err) with pytest.raises(IOError): client.list_dir("/gone") @@ -385,6 +407,10 @@ def _make_connected(self, sftp_info: ConnectionInfo) -> tuple[SFTPClient, AsyncM client = SFTPClient(sftp_info) mock_sftp = AsyncMock() mock_sftp.realpath.return_value = "/home/user/subdir" + stat_attrs = MagicMock() + stat_attrs.permissions = stat_mod.S_IFDIR | 0o755 + stat_attrs.type = None + mock_sftp.stat.return_value = stat_attrs client._conn = AsyncMock() client._sftp = mock_sftp client._cwd = "/home/user" @@ -396,6 +422,22 @@ def test_chdir_updates_cwd(self, sftp_info: ConnectionInfo) -> None: assert result == "/home/user/subdir" assert client._cwd == "/home/user/subdir" + def test_chdir_validates_directory_with_stat(self, sftp_info: ConnectionInfo) -> None: + client, mock_sftp = self._make_connected(sftp_info) + client.chdir("/home/user/subdir") + mock_sftp.stat.assert_awaited_once_with("/home/user/subdir") + + def test_chdir_rejects_non_directory(self, sftp_info: ConnectionInfo) -> None: + client, mock_sftp = self._make_connected(sftp_info) + stat_attrs = MagicMock() + stat_attrs.permissions = stat_mod.S_IFREG | 0o644 + stat_attrs.type = None + mock_sftp.stat.return_value = stat_attrs + + with pytest.raises(NotADirectoryError, match="Not a directory"): + client.chdir("/home/user/file.txt") + assert client._cwd == "/home/user" # unchanged + def test_chdir_permission_error(self, sftp_info: ConnectionInfo) -> None: import errno @@ -425,7 +467,7 @@ def test_socket_file_skipped(self, sftp_info: ConnectionInfo) -> None: entry.attrs = MagicMock() entry.attrs.permissions = stat_mod.S_IFSOCK | 0o600 entry.longname = "srw------- 1 user group 0 Jan 1 agent.sock" - mock_sftp.readdir.return_value = [entry] + _setup_readdir(mock_sftp, [entry]) files = client.list_dir() assert files == [] @@ -437,7 +479,7 @@ def test_fifo_file_skipped(self, sftp_info: ConnectionInfo) -> None: entry.attrs = MagicMock() entry.attrs.permissions = stat_mod.S_IFIFO | 0o644 entry.longname = "prw-r--r-- 1 user group 0 Jan 1 mypipe" - mock_sftp.readdir.return_value = [entry] + _setup_readdir(mock_sftp, [entry]) files = client.list_dir() assert files == [] @@ -448,7 +490,334 @@ def test_oserror_reraises(self, sftp_info: ConnectionInfo) -> None: client, mock_sftp = self._make_connected(sftp_info) err = IOError() err.errno = errno.ENOENT - mock_sftp.readdir.side_effect = err + _setup_readdir_error(mock_sftp, err) with pytest.raises(IOError): client.list_dir("/gone") + + +# SFTP v4+ file type constant (matches protocols._SFTP_TYPE_DIRECTORY) +_SFTP_TYPE_DIRECTORY = 2 +_SFTP_TYPE_SYMLINK = 3 +_SFTP_TYPE_REGULAR = 1 +_SFTP_TYPE_UNKNOWN = 5 + + +class TestSFTPBitviseCompliance: + """Tests for strict SFTP servers (Bitvise) that send file type in the + SFTP v4+ ``type`` field rather than (or in addition to) permission bits.""" + + def _make_connected(self, sftp_info: ConnectionInfo) -> tuple[SFTPClient, AsyncMock]: + client = SFTPClient(sftp_info) + mock_sftp = AsyncMock() + mock_sftp.realpath.return_value = "/home/user" + client._conn = AsyncMock() + client._sftp = mock_sftp + client._cwd = "/home/user" + return client, mock_sftp + + # ---- list_dir: type field fallback ---- + + def test_list_dir_detects_dir_via_sftp_type_when_permissions_none( + self, sftp_info: ConnectionInfo + ) -> None: + """Bitvise may return type=DIRECTORY with permissions=None.""" + client, mock_sftp = self._make_connected(sftp_info) + entry = MagicMock() + entry.filename = ".ssh" + entry.attrs = MagicMock() + entry.attrs.permissions = None + entry.attrs.type = _SFTP_TYPE_DIRECTORY + entry.attrs.size = 0 + entry.attrs.mtime = 0 + entry.attrs.uid = 1000 + entry.attrs.gid = 1000 + entry.longname = "" + _setup_readdir(mock_sftp, [entry]) + + files = client.list_dir() + assert len(files) == 1 + assert files[0].name == ".ssh" + assert files[0].is_dir is True + + def test_list_dir_detects_dir_via_sftp_type_when_permissions_lack_type_bits( + self, sftp_info: ConnectionInfo + ) -> None: + """Bitvise may return permissions=0o700 without S_IFDIR bits + type=DIRECTORY.""" + client, mock_sftp = self._make_connected(sftp_info) + entry = MagicMock() + entry.filename = ".ssh" + entry.attrs = MagicMock() + entry.attrs.permissions = 0o700 # No S_IFDIR prefix + entry.attrs.type = _SFTP_TYPE_DIRECTORY + entry.attrs.size = 0 + entry.attrs.mtime = 0 + entry.attrs.uid = 1000 + entry.attrs.gid = 1000 + entry.longname = "" + _setup_readdir(mock_sftp, [entry]) + + files = client.list_dir() + assert len(files) == 1 + assert files[0].name == ".ssh" + assert files[0].is_dir is True + + def test_list_dir_file_with_type_regular_stays_file(self, sftp_info: ConnectionInfo) -> None: + """File with type=REGULAR and no permission type bits stays a file.""" + client, mock_sftp = self._make_connected(sftp_info) + entry = MagicMock() + entry.filename = "readme.txt" + entry.attrs = MagicMock() + entry.attrs.permissions = 0o644 + entry.attrs.type = _SFTP_TYPE_REGULAR + entry.attrs.size = 100 + entry.attrs.mtime = 0 + entry.attrs.uid = 1000 + entry.attrs.gid = 1000 + entry.longname = "" + _setup_readdir(mock_sftp, [entry]) + + files = client.list_dir() + assert len(files) == 1 + assert files[0].is_dir is False + + def test_list_dir_symlink_to_dir_via_sftp_type(self, sftp_info: ConnectionInfo) -> None: + """Symlink with type=SYMLINK that resolves to a dir via stat type field.""" + client, mock_sftp = self._make_connected(sftp_info) + entry = MagicMock() + entry.filename = "link-to-dir" + entry.attrs = MagicMock() + entry.attrs.permissions = None + entry.attrs.type = _SFTP_TYPE_SYMLINK + entry.attrs.size = 0 + entry.attrs.mtime = 0 + entry.attrs.uid = 1000 + entry.attrs.gid = 1000 + entry.longname = "" + + target_attrs = MagicMock() + target_attrs.permissions = None + target_attrs.type = _SFTP_TYPE_DIRECTORY + mock_sftp.stat.return_value = target_attrs + _setup_readdir(mock_sftp, [entry]) + + files = client.list_dir() + assert len(files) == 1 + assert files[0].is_dir is True + + # ---- chdir: directory validation ---- + + def test_chdir_accepts_dir_via_sftp_type(self, sftp_info: ConnectionInfo) -> None: + """chdir succeeds when stat returns type=DIRECTORY with permissions=None.""" + client, mock_sftp = self._make_connected(sftp_info) + mock_sftp.realpath.return_value = "/home/user/.ssh" + stat_attrs = MagicMock() + stat_attrs.permissions = None + stat_attrs.type = _SFTP_TYPE_DIRECTORY + mock_sftp.stat.return_value = stat_attrs + + result = client.chdir("/home/user/.ssh") + assert result == "/home/user/.ssh" + assert client._cwd == "/home/user/.ssh" + + def test_chdir_accepts_dir_with_permissions_only(self, sftp_info: ConnectionInfo) -> None: + """chdir succeeds when stat returns S_IFDIR in permissions (standard Linux).""" + client, mock_sftp = self._make_connected(sftp_info) + mock_sftp.realpath.return_value = "/home/user/docs" + stat_attrs = MagicMock() + stat_attrs.permissions = stat_mod.S_IFDIR | 0o755 + stat_attrs.type = _SFTP_TYPE_UNKNOWN + mock_sftp.stat.return_value = stat_attrs + + result = client.chdir("/home/user/docs") + assert result == "/home/user/docs" + + def test_chdir_stat_permission_error_surfaces(self, sftp_info: ConnectionInfo) -> None: + """chdir raises PermissionError when stat fails with EACCES.""" + import errno + + client, mock_sftp = self._make_connected(sftp_info) + mock_sftp.realpath.return_value = "/home/user/.ssh" + err = IOError() + err.errno = errno.EACCES + mock_sftp.stat.side_effect = err + + with pytest.raises(PermissionError, match="Permission denied"): + client.chdir("/home/user/.ssh") + assert client._cwd == "/home/user" # unchanged + + # ---- stat: type field fallback ---- + + def test_stat_detects_dir_via_sftp_type(self, sftp_info: ConnectionInfo) -> None: + """stat() returns is_dir=True when type=DIRECTORY and permissions=None.""" + client, mock_sftp = self._make_connected(sftp_info) + stat_attrs = MagicMock() + stat_attrs.permissions = None + stat_attrs.type = _SFTP_TYPE_DIRECTORY + stat_attrs.size = 0 + stat_attrs.mtime = 0 + mock_sftp.stat.return_value = stat_attrs + + remote = client.stat("/home/user/.ssh") + assert remote.is_dir is True + assert remote.name == ".ssh" + + # ---- _readdir_safe: Bitvise count=0 EOF quirk ---- + + def test_readdir_safe_treats_consecutive_empty_batches_as_eof( + self, sftp_info: ConnectionInfo + ) -> None: + """_readdir_safe stops after 3 consecutive empty readdir responses. + + Bitvise may return FXP_NAME with count=0 (empty list, at_end=False) + indefinitely instead of FX_EOF. _readdir_safe must break out after + _MAX_EMPTY (3) consecutive empty batches — matching WinSCP behaviour. + """ + client, mock_sftp = self._make_connected(sftp_info) + + entry = MagicMock() + entry.filename = "hello.txt" + entry.attrs = MagicMock() + entry.attrs.permissions = stat_mod.S_IFREG | 0o644 + entry.attrs.type = _SFTP_TYPE_REGULAR + entry.attrs.size = 42 + entry.attrs.mtime = 0 + entry.attrs.uid = 1000 + entry.attrs.gid = 1000 + entry.longname = "" + + mock_sftp.compose_path.side_effect = lambda p: p + mock_handler = AsyncMock() + mock_handler.opendir.return_value = "fake-handle" + # Batch 1: real entry, at_end=False + # Batches 2-4: empty, at_end=False (Bitvise quirk — never sends EOF) + mock_handler.readdir.side_effect = [ + ([entry], False), + ([], False), + ([], False), + ([], False), + ] + mock_handler.close.return_value = None + mock_sftp._handler = mock_handler + + files = client.list_dir() + + assert len(files) == 1 + assert files[0].name == "hello.txt" + # 1 real batch + 3 empty batches = 4 readdir calls + assert mock_handler.readdir.call_count == 4 + mock_handler.close.assert_awaited_once() + + # ---- _readdir_safe: SFTPEOFError handling ---- + + def test_readdir_safe_handles_sftp_eof_error(self, sftp_info: ConnectionInfo) -> None: + """_readdir_safe catches SFTPEOFError and returns collected results.""" + import asyncssh + + client, mock_sftp = self._make_connected(sftp_info) + + entry = MagicMock() + entry.filename = "data.csv" + entry.longname = "" + entry.attrs = MagicMock() + entry.attrs.permissions = stat_mod.S_IFREG | 0o644 + entry.attrs.type = _SFTP_TYPE_REGULAR + entry.attrs.size = 99 + entry.attrs.mtime = 0 + entry.attrs.uid = 1000 + entry.attrs.gid = 1000 + + mock_sftp.compose_path.side_effect = lambda p: p + mock_handler = AsyncMock() + mock_handler.opendir.return_value = "fake-handle" + # First call returns an entry; second raises SFTPEOFError + mock_handler.readdir.side_effect = [ + ([entry], False), + asyncssh.SFTPEOFError(), + ] + mock_handler.close.return_value = None + mock_sftp._handler = mock_handler + + files = client.list_dir() + assert len(files) == 1 + assert files[0].name == "data.csv" + mock_handler.close.assert_awaited_once() + + # ---- _readdir_safe: bytes filename decode ---- + + def test_readdir_safe_decodes_bytes_filenames(self, sftp_info: ConnectionInfo) -> None: + """_readdir_safe decodes bytes filenames/longnames via sftp.decode.""" + client, mock_sftp = self._make_connected(sftp_info) + + entry = MagicMock() + entry.filename = b"report.txt" + entry.longname = b"-rw-r--r-- 1 user user 42 Jan 1 00:00 report.txt" + entry.attrs = MagicMock() + entry.attrs.permissions = stat_mod.S_IFREG | 0o644 + entry.attrs.type = _SFTP_TYPE_REGULAR + entry.attrs.size = 42 + entry.attrs.mtime = 0 + entry.attrs.uid = 1000 + entry.attrs.gid = 1000 + + mock_sftp.decode = MagicMock(side_effect=lambda b: b.decode("utf-8")) + mock_sftp.compose_path.side_effect = lambda p: p + mock_handler = AsyncMock() + mock_handler.opendir.return_value = "fake-handle" + mock_handler.readdir.return_value = ([entry], True) + mock_handler.close.return_value = None + mock_sftp._handler = mock_handler + + files = client.list_dir() + assert len(files) == 1 + assert files[0].name == "report.txt" + assert mock_sftp.decode.call_count == 2 + + # ---- list_dir: SFTPError exception mapping ---- + + def test_list_dir_sftp_error_permission_denied(self, sftp_info: ConnectionInfo) -> None: + """list_dir maps SFTPError code=3 (FX_PERMISSION_DENIED) to PermissionError.""" + import asyncssh + + client, mock_sftp = self._make_connected(sftp_info) + _setup_readdir_error(mock_sftp, asyncssh.SFTPError(3, "Permission denied")) + + with pytest.raises(PermissionError, match="Permission denied"): + client.list_dir("/secret") + + def test_list_dir_sftp_error_failure_mapped_to_permission( + self, sftp_info: ConnectionInfo + ) -> None: + """list_dir maps SFTPError code=4 (FX_FAILURE) to PermissionError.""" + import asyncssh + + client, mock_sftp = self._make_connected(sftp_info) + _setup_readdir_error(mock_sftp, asyncssh.SFTPError(4, "Failure")) + + with pytest.raises(PermissionError, match="Permission denied"): + client.list_dir("/secret") + + def test_list_dir_sftp_error_non_permission_reraises(self, sftp_info: ConnectionInfo) -> None: + """list_dir re-raises SFTPError with non-permission code unchanged.""" + import asyncssh + + client, mock_sftp = self._make_connected(sftp_info) + _setup_readdir_error(mock_sftp, asyncssh.SFTPError(7, "Connection lost")) + + with pytest.raises(asyncssh.SFTPError): + client.list_dir("/data") + + # ---- chdir: no type info fallback ---- + + def test_chdir_assumes_dir_when_no_type_info(self, sftp_info: ConnectionInfo) -> None: + """chdir assumes directory when stat returns permissions=None and type=None.""" + client, mock_sftp = self._make_connected(sftp_info) + mock_sftp.realpath.return_value = "/home/user/.ssh" + stat_attrs = MagicMock() + stat_attrs.permissions = None + stat_attrs.type = None + mock_sftp.stat.return_value = stat_attrs + + result = client.chdir("/home/user/.ssh") + assert result == "/home/user/.ssh" + assert client._cwd == "/home/user/.ssh"