@@ -130,12 +130,15 @@ class RemoteFileSystemModel(QAbstractItemModel): |
| 130 | 130 | errorOccurred = pyqtSignal(str) # Emitted on errors |
| 131 | 131 | filesDropped = pyqtSignal(list, str) # Emitted when files are dropped (files, target_directory) |
| 132 | 132 | rootPathChanged = pyqtSignal(str) # Emitted when the root path changes |
| 133 | + accessDenied = pyqtSignal(str, str) # Emitted on permission errors (attempted_path, fallback_path) |
| 133 | 134 | |
| 134 | 135 | def __init__(self, sftp: Optional[SFTPClient] = None, parent: Optional[QObject] = None): |
| 135 | 136 | super().__init__(parent) |
| 136 | 137 | self.sftp = sftp |
| 137 | 138 | self.root_path = "/" |
| 138 | 139 | self.current_root_path = "/" # The path that's currently shown as root |
| 140 | + self.last_valid_path = "/" # Track last successfully accessed path |
| 141 | + self.user_home_path = "/" # Will be set when connection is established |
| 139 | 142 | |
| 140 | 143 | # Create a proper root item with valid attributes |
| 141 | 144 | root_attr = SFTPAttributes() |
@@ -156,6 +159,9 @@ class RemoteFileSystemModel(QAbstractItemModel): |
| 156 | 159 | |
| 157 | 160 | # Track expanded directories |
| 158 | 161 | self._expanded_paths: Set[str] = set() |
| 162 | + |
| 163 | + # Track paths we've had permission errors on |
| 164 | + self._permission_denied_paths: Set[str] = set() |
| 159 | 165 | |
| 160 | 166 | def set_sftp(self, sftp: SFTPClient): |
| 161 | 167 | """Set or update the SFTP connection""" |
@@ -163,11 +169,20 @@ class RemoteFileSystemModel(QAbstractItemModel): |
| 163 | 169 | self.sftp = sftp |
| 164 | 170 | self._path_cache.clear() |
| 165 | 171 | self._expanded_paths.clear() |
| 172 | + self._permission_denied_paths.clear() |
| 166 | 173 | self.root_item.children = None |
| 167 | 174 | self.current_root_path = "/" |
| 175 | + self.last_valid_path = "/" |
| 168 | 176 | self.endResetModel() |
| 169 | 177 | |
| 170 | 178 | if sftp: |
| 179 | + # Try to get user's home directory |
| 180 | + try: |
| 181 | + self.user_home_path = sftp.normalize(".") |
| 182 | + self.last_valid_path = self.user_home_path |
| 183 | + except: |
| 184 | + self.user_home_path = "/" |
| 185 | + |
| 171 | 186 | self.load_directory("/") |
| 172 | 187 | |
| 173 | 188 | def set_root_path(self, path: str): |
@@ -175,12 +190,20 @@ class RemoteFileSystemModel(QAbstractItemModel): |
| 175 | 190 | if not self.sftp: |
| 176 | 191 | return |
| 177 | 192 | |
| 193 | + # Don't try to access paths we know are denied |
| 194 | + if path in self._permission_denied_paths: |
| 195 | + self.errorOccurred.emit(f"Access denied to {path}") |
| 196 | + return |
| 197 | + |
| 178 | 198 | # Don't reset if we're already at this path |
| 179 | 199 | if path == self.current_root_path: |
| 180 | 200 | # Just reload the directory without resetting |
| 181 | 201 | self.load_directory(path) |
| 182 | 202 | return |
| 183 | 203 | |
| 204 | + # Store the current path as last valid before attempting change |
| 205 | + old_path = self.current_root_path |
| 206 | + |
| 184 | 207 | self.beginResetModel() |
| 185 | 208 | self.current_root_path = path |
| 186 | 209 | self._path_cache.clear() |
@@ -194,24 +217,98 @@ class RemoteFileSystemModel(QAbstractItemModel): |
| 194 | 217 | self.root_item.full_path = path |
| 195 | 218 | self.root_item.children = None |
| 196 | 219 | |
| 197 | | - # Load the directory |
| 198 | | - self.load_directory(path) |
| 220 | + # Try to load the directory |
| 221 | + if self.load_directory(path): |
| 222 | + # Success - update last valid path |
| 223 | + self.last_valid_path = path |
| 224 | + else: |
| 225 | + # Failed to load - will be handled by load_directory |
| 226 | + pass |
| 199 | 227 | |
| 228 | + except PermissionError as e: |
| 229 | + self._handle_permission_error(path, old_path, str(e)) |
| 200 | 230 | except Exception as e: |
| 201 | | - self.errorOccurred.emit(f"Failed to change to directory: {str(e)}") |
| 202 | | - # Restore to root on error |
| 203 | | - self.current_root_path = "/" |
| 204 | | - root_attr = SFTPAttributes() |
| 205 | | - root_attr.st_mode = stat.S_IFDIR | 0o755 |
| 206 | | - root_attr.st_size = 0 |
| 207 | | - root_attr.st_mtime = 0 |
| 208 | | - root_attr.filename = "/" |
| 209 | | - self.root_item = RemoteFileInfo(root_attr, "/", "") |
| 210 | | - self.root_item.children = [] |
| 231 | + self._handle_general_error(path, old_path, str(e)) |
| 211 | 232 | |
| 212 | 233 | self.endResetModel() |
| 213 | 234 | self.rootPathChanged.emit(self.current_root_path) |
| 214 | 235 | |
| 236 | + def _handle_permission_error(self, attempted_path: str, fallback_path: str, error_msg: str): |
| 237 | + """Handle permission denied errors""" |
| 238 | + # Track this path as denied |
| 239 | + self._permission_denied_paths.add(attempted_path) |
| 240 | + |
| 241 | + # Determine safe fallback |
| 242 | + safe_path = self._get_safe_fallback_path(fallback_path) |
| 243 | + |
| 244 | + # Emit signal for UI to show message |
| 245 | + self.accessDenied.emit(attempted_path, safe_path) |
| 246 | + |
| 247 | + # Navigate to safe path |
| 248 | + self.current_root_path = safe_path |
| 249 | + try: |
| 250 | + attrs = self.sftp.stat(safe_path) |
| 251 | + self.root_item = RemoteFileInfo(attrs, os.path.basename(safe_path) if safe_path != "/" else "/", |
| 252 | + os.path.dirname(safe_path) if safe_path != "/" else "") |
| 253 | + self.root_item.full_path = safe_path |
| 254 | + self.root_item.children = None |
| 255 | + self.load_directory(safe_path) |
| 256 | + except: |
| 257 | + # Even fallback failed, go to root |
| 258 | + self._navigate_to_root() |
| 259 | + |
| 260 | + def _handle_general_error(self, attempted_path: str, fallback_path: str, error_msg: str): |
| 261 | + """Handle general errors when changing directories""" |
| 262 | + self.errorOccurred.emit(f"Failed to access {attempted_path}: {error_msg}") |
| 263 | + |
| 264 | + # Try to restore previous path |
| 265 | + safe_path = self._get_safe_fallback_path(fallback_path) |
| 266 | + self.current_root_path = safe_path |
| 267 | + |
| 268 | + try: |
| 269 | + attrs = self.sftp.stat(safe_path) |
| 270 | + self.root_item = RemoteFileInfo(attrs, os.path.basename(safe_path) if safe_path != "/" else "/", |
| 271 | + os.path.dirname(safe_path) if safe_path != "/" else "") |
| 272 | + self.root_item.full_path = safe_path |
| 273 | + self.root_item.children = None |
| 274 | + self.load_directory(safe_path) |
| 275 | + except: |
| 276 | + # Even fallback failed, go to root |
| 277 | + self._navigate_to_root() |
| 278 | + |
| 279 | + def _get_safe_fallback_path(self, preferred_path: str) -> str: |
| 280 | + """Get a safe fallback path to navigate to""" |
| 281 | + # Try paths in order: preferred, last valid, user home, root |
| 282 | + paths_to_try = [ |
| 283 | + preferred_path, |
| 284 | + self.last_valid_path, |
| 285 | + self.user_home_path, |
| 286 | + "/" |
| 287 | + ] |
| 288 | + |
| 289 | + for path in paths_to_try: |
| 290 | + if path and path not in self._permission_denied_paths: |
| 291 | + try: |
| 292 | + # Quick check if we can access it |
| 293 | + if self.sftp: |
| 294 | + self.sftp.stat(path) |
| 295 | + return path |
| 296 | + except: |
| 297 | + continue |
| 298 | + |
| 299 | + return "/" # Ultimate fallback |
| 300 | + |
| 301 | + def _navigate_to_root(self): |
| 302 | + """Navigate to root directory as last resort""" |
| 303 | + self.current_root_path = "/" |
| 304 | + root_attr = SFTPAttributes() |
| 305 | + root_attr.st_mode = stat.S_IFDIR | 0o755 |
| 306 | + root_attr.st_size = 0 |
| 307 | + root_attr.st_mtime = 0 |
| 308 | + root_attr.filename = "/" |
| 309 | + self.root_item = RemoteFileInfo(root_attr, "/", "") |
| 310 | + self.root_item.children = [] |
| 311 | + |
| 215 | 312 | def rootPath(self) -> str: |
| 216 | 313 | """Get the current root path""" |
| 217 | 314 | return self.current_root_path |
@@ -253,6 +350,10 @@ class RemoteFileSystemModel(QAbstractItemModel): |
| 253 | 350 | # Normalize path |
| 254 | 351 | path = os.path.normpath(path) |
| 255 | 352 | |
| 353 | + # Check if we've already been denied access |
| 354 | + if path in self._permission_denied_paths: |
| 355 | + return False |
| 356 | + |
| 256 | 357 | # Fetch directory listing |
| 257 | 358 | attrs = self.sftp.listdir_attr(path) |
| 258 | 359 | |
@@ -311,8 +412,17 @@ class RemoteFileSystemModel(QAbstractItemModel): |
| 311 | 412 | self.directoryLoaded.emit(path) |
| 312 | 413 | return True |
| 313 | 414 | |
| 415 | + except PermissionError as e: |
| 416 | + # Track permission denied |
| 417 | + self._permission_denied_paths.add(path) |
| 418 | + # Only emit error if this is the path we're trying to view |
| 419 | + if path == self.current_root_path: |
| 420 | + self.accessDenied.emit(path, self.last_valid_path) |
| 421 | + return False |
| 314 | 422 | except Exception as e: |
| 315 | | - self.errorOccurred.emit(str(e)) |
| 423 | + # Only emit error if this is the path we're trying to view |
| 424 | + if path == self.current_root_path: |
| 425 | + self.errorOccurred.emit(f"Failed to load {path}: {str(e)}") |
| 316 | 426 | return False |
| 317 | 427 | |
| 318 | 428 | def refresh(self, preserve_state: bool = True): |
@@ -422,6 +532,10 @@ class RemoteFileSystemModel(QAbstractItemModel): |
| 422 | 532 | # On error, just do a full reload of this directory |
| 423 | 533 | self.load_directory(path) |
| 424 | 534 | |
| 535 | + def clear_permission_cache(self): |
| 536 | + """Clear the permission denied cache - useful after reconnecting""" |
| 537 | + self._permission_denied_paths.clear() |
| 538 | + |
| 425 | 539 | def _find_item_by_path(self, path: str) -> Optional[RemoteFileInfo]: |
| 426 | 540 | """Find item by path traversing the tree""" |
| 427 | 541 | if path == self.current_root_path: |