| 1 | import os |
| 2 | from pathlib import Path |
| 3 | import shutil |
| 4 | import base64 |
| 5 | import subprocess |
| 6 | from typing import Dict, List, Tuple, Any |
| 7 | from helpers.security import safe_filename |
| 8 | from datetime import datetime |
| 9 | |
| 10 | from helpers import files |
| 11 | from helpers.localization import Localization |
| 12 | from helpers.print_style import PrintStyle |
| 13 | |
| 14 | |
| 15 | class FileBrowser: |
| 16 | ALLOWED_EXTENSIONS = { |
| 17 | 'image': {'jpg', 'jpeg', 'png', 'bmp'}, |
| 18 | 'code': {'py', 'js', 'sh', 'html', 'css'}, |
| 19 | 'document': {'md', 'pdf', 'txt', 'csv', 'json'} |
| 20 | } |
| 21 | |
| 22 | MAX_FILE_SIZE = 100 * 1024 * 1024 # 100MB |
| 23 | MAX_TEXT_FILE_SIZE = 1 * 1024 * 1024 # 1MB |
| 24 | |
| 25 | def __init__(self): |
| 26 | # if runtime.is_development(): |
| 27 | # base_dir = files.get_base_dir() |
| 28 | # else: |
| 29 | # base_dir = "/" |
| 30 | base_dir = "/" |
| 31 | self.base_dir = Path(base_dir) |
| 32 | |
| 33 | def _check_file_size(self, file) -> bool: |
| 34 | try: |
| 35 | file.seek(0, os.SEEK_END) |
| 36 | size = file.tell() |
| 37 | file.seek(0) |
| 38 | return size <= self.MAX_FILE_SIZE |
| 39 | except (AttributeError, IOError): |
| 40 | return False |
| 41 | |
| 42 | def save_file_b64(self, current_path: str, filename: str, base64_content: str): |
| 43 | try: |
| 44 | # Resolve the target directory path |
| 45 | target_file = (self.base_dir / current_path / filename).resolve() |
| 46 | if not str(target_file).startswith(str(self.base_dir)): |
| 47 | raise ValueError("Invalid target directory") |
| 48 | |
| 49 | os.makedirs(target_file.parent, exist_ok=True) |
| 50 | # Save file |
| 51 | with open(target_file, "wb") as file: |
| 52 | file.write(base64.b64decode(base64_content)) |
| 53 | return True |
| 54 | except Exception as e: |
| 55 | PrintStyle.error(f"Error saving file {filename}: {e}") |
| 56 | return False |
| 57 | |
| 58 | def save_files(self, files: List, current_path: str = "") -> Tuple[List[str], List[str]]: |
| 59 | """Save uploaded files and return successful and failed filenames""" |
| 60 | successful = [] |
| 61 | failed = [] |
| 62 | |
| 63 | try: |
| 64 | # Resolve the target directory path |
| 65 | target_dir = (self.base_dir / current_path).resolve() |
| 66 | if not str(target_dir).startswith(str(self.base_dir)): |
| 67 | raise ValueError("Invalid target directory") |
| 68 | |
| 69 | os.makedirs(target_dir, exist_ok=True) |
| 70 | |
| 71 | for file in files: |
| 72 | try: |
| 73 | if file and self._is_allowed_file(file.filename, file): |
| 74 | filename = safe_filename(file.filename) |
| 75 | if not filename: |
| 76 | raise ValueError("Invalid filename") |
| 77 | file_path = target_dir / filename |
| 78 | |
| 79 | file.save(str(file_path)) |
| 80 | successful.append(filename) |
| 81 | else: |
| 82 | failed.append(file.filename) |
| 83 | except Exception as e: |
| 84 | PrintStyle.error(f"Error saving file {file.filename}: {e}") |
| 85 | failed.append(file.filename) |
| 86 | |
| 87 | return successful, failed |
| 88 | |
| 89 | except Exception as e: |
| 90 | PrintStyle.error(f"Error in save_files: {e}") |
| 91 | return successful, failed |
| 92 | |
| 93 | def delete_file(self, file_path: str) -> bool: |
| 94 | """Delete a file or empty directory""" |
| 95 | try: |
| 96 | # Resolve the full path while preventing directory traversal |
| 97 | full_path = (self.base_dir / file_path).resolve() |
| 98 | if not str(full_path).startswith(str(self.base_dir)): |
| 99 | raise ValueError("Invalid path") |
| 100 | |
| 101 | if os.path.exists(full_path): |
| 102 | if os.path.isfile(full_path): |
| 103 | os.remove(full_path) |
| 104 | elif os.path.isdir(full_path): |
| 105 | shutil.rmtree(full_path) |
| 106 | return True |
| 107 | |
| 108 | return False |
| 109 | |
| 110 | except Exception as e: |
| 111 | PrintStyle.error(f"Error deleting {file_path}: {e}") |
| 112 | return False |
| 113 | |
| 114 | def rename_item(self, file_path: str, new_name: str) -> bool: |
| 115 | try: |
| 116 | if not new_name or new_name in {".", ".."}: |
| 117 | raise ValueError("Invalid new name") |
| 118 | if "/" in new_name or "\\" in new_name: |
| 119 | raise ValueError("New name cannot include path separators") |
| 120 | |
| 121 | full_path = (self.base_dir / file_path).resolve() |
| 122 | if not str(full_path).startswith(str(self.base_dir)): |
| 123 | raise ValueError("Invalid path") |
| 124 | if not full_path.exists(): |
| 125 | raise FileNotFoundError("File or folder not found") |
| 126 | |
| 127 | new_path = full_path.with_name(new_name) |
| 128 | if not str(new_path).startswith(str(self.base_dir)): |
| 129 | raise ValueError("Invalid target path") |
| 130 | if full_path == new_path: |
| 131 | return True |
| 132 | if new_path.exists(): |
| 133 | raise FileExistsError("Target already exists") |
| 134 | |
| 135 | os.rename(full_path, new_path) |
| 136 | return True |
| 137 | except Exception as e: |
| 138 | PrintStyle.error(f"Error renaming {file_path}: {e}") |
| 139 | raise |
| 140 | |
| 141 | def move_items(self, file_paths: List[str], destination_path: str) -> List[str]: |
| 142 | if not file_paths: |
| 143 | raise ValueError("No items selected") |
| 144 | |
| 145 | base_dir = self.base_dir.resolve() |
| 146 | destination = (self.base_dir / destination_path).resolve() |
| 147 | if not destination.is_relative_to(base_dir): |
| 148 | raise ValueError("Invalid destination path") |
| 149 | if not destination.is_dir(): |
| 150 | raise NotADirectoryError("Destination folder not found") |
| 151 | |
| 152 | moves: List[Tuple[Path, Path]] = [] |
| 153 | targets: set[Path] = set() |
| 154 | for file_path in dict.fromkeys(file_paths): |
| 155 | requested = self.base_dir / file_path |
| 156 | source = requested.parent.resolve() / requested.name |
| 157 | if not source.is_relative_to(base_dir) or source == base_dir: |
| 158 | raise ValueError("Invalid source path") |
| 159 | if not source.exists() and not source.is_symlink(): |
| 160 | raise FileNotFoundError(f"Item not found: {source.name}") |
| 161 | if source == destination: |
| 162 | raise ValueError("A folder cannot be moved into itself") |
| 163 | if ( |
| 164 | source.is_dir() |
| 165 | and not source.is_symlink() |
| 166 | and destination.is_relative_to(source) |
| 167 | ): |
| 168 | raise ValueError("A folder cannot be moved into itself") |
| 169 | |
| 170 | target = destination / source.name |
| 171 | if target == source: |
| 172 | raise ValueError(f"{source.name} is already in this folder") |
| 173 | if target.exists() or target.is_symlink(): |
| 174 | raise FileExistsError( |
| 175 | f'An item named "{source.name}" already exists' |
| 176 | ) |
| 177 | if target in targets: |
| 178 | raise FileExistsError(f'Multiple items are named "{source.name}"') |
| 179 | targets.add(target) |
| 180 | moves.append((source, target)) |
| 181 | |
| 182 | moved: List[Tuple[Path, Path]] = [] |
| 183 | try: |
| 184 | for source, target in moves: |
| 185 | os.rename(source, target) |
| 186 | moved.append((source, target)) |
| 187 | except Exception: |
| 188 | for source, target in reversed(moved): |
| 189 | try: |
| 190 | os.rename(target, source) |
| 191 | except Exception as rollback_error: |
| 192 | PrintStyle.error(f"Error restoring {source}: {rollback_error}") |
| 193 | raise |
| 194 | |
| 195 | return [str(target) for _, target in moved] |
| 196 | |
| 197 | def create_folder(self, parent_path: str, folder_name: str) -> bool: |
| 198 | try: |
| 199 | if not folder_name or folder_name in {".", ".."}: |
| 200 | raise ValueError("Invalid folder name") |
| 201 | if "/" in folder_name or "\\" in folder_name: |
| 202 | raise ValueError("Folder name cannot include path separators") |
| 203 | |
| 204 | parent_full = (self.base_dir / parent_path).resolve() |
| 205 | if not str(parent_full).startswith(str(self.base_dir)): |
| 206 | raise ValueError("Invalid parent path") |
| 207 | |
| 208 | target_dir = (parent_full / folder_name).resolve() |
| 209 | if not str(target_dir).startswith(str(self.base_dir)): |
| 210 | raise ValueError("Invalid target path") |
| 211 | if target_dir.exists(): |
| 212 | raise FileExistsError("Folder already exists") |
| 213 | |
| 214 | os.makedirs(target_dir, exist_ok=False) |
| 215 | return True |
| 216 | except Exception as e: |
| 217 | PrintStyle.error(f"Error creating folder {folder_name}: {e}") |
| 218 | raise |
| 219 | |
| 220 | def save_text_file(self, file_path: str, content: str) -> bool: |
| 221 | try: |
| 222 | if not isinstance(content, str): |
| 223 | raise ValueError("Content must be a string") |
| 224 | content_size = len(content.encode("utf-8")) |
| 225 | if content_size > self.MAX_TEXT_FILE_SIZE: |
| 226 | raise ValueError("File exceeds 1 MB and cannot be edited") |
| 227 | |
| 228 | full_path = (self.base_dir / file_path).resolve() |
| 229 | if not str(full_path).startswith(str(self.base_dir)): |
| 230 | raise ValueError("Invalid path") |
| 231 | if full_path.exists() and full_path.is_dir(): |
| 232 | raise ValueError("Target is a directory") |
| 233 | |
| 234 | os.makedirs(full_path.parent, exist_ok=True) |
| 235 | with open(full_path, "w", encoding="utf-8") as file: |
| 236 | file.write(content) |
| 237 | return True |
| 238 | except Exception as e: |
| 239 | PrintStyle.error(f"Error saving file {file_path}: {e}") |
| 240 | raise |
| 241 | |
| 242 | def _is_allowed_file(self, filename: str, file) -> bool: |
| 243 | # allow any file to be uploaded in file browser |
| 244 | |
| 245 | # if not filename: |
| 246 | # return False |
| 247 | # ext = self._get_file_extension(filename) |
| 248 | # all_allowed = set().union(*self.ALLOWED_EXTENSIONS.values()) |
| 249 | # if ext not in all_allowed: |
| 250 | # return False |
| 251 | |
| 252 | return True # Allow the file if it passes the checks |
| 253 | |
| 254 | def _get_file_extension(self, filename: str) -> str: |
| 255 | return filename.rsplit('.', 1)[1].lower() if '.' in filename else '' |
| 256 | |
| 257 | def _get_files_via_ls(self, full_path: Path) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: |
| 258 | """Get files and folders using ls command for better error handling""" |
| 259 | files: List[Dict[str, Any]] = [] |
| 260 | folders: List[Dict[str, Any]] = [] |
| 261 | |
| 262 | try: |
| 263 | # Use ls command to get directory listing |
| 264 | result = subprocess.run( |
| 265 | ['ls', '-la', str(full_path)], |
| 266 | capture_output=True, |
| 267 | text=True, |
| 268 | timeout=30 |
| 269 | ) |
| 270 | |
| 271 | if result.returncode != 0: |
| 272 | PrintStyle.error(f"ls command failed: {result.stderr}") |
| 273 | return files, folders |
| 274 | |
| 275 | # Parse ls output (skip first line which is "total X") |
| 276 | lines = result.stdout.strip().split('\n') |
| 277 | if len(lines) <= 1: |
| 278 | return files, folders |
| 279 | |
| 280 | for line in lines[1:]: # Skip the "total" line |
| 281 | try: |
| 282 | # Skip current and parent directory entries |
| 283 | if line.endswith(' .') or line.endswith(' ..'): |
| 284 | continue |
| 285 | |
| 286 | # Parse ls -la output format |
| 287 | parts = line.split() |
| 288 | if len(parts) < 9: |
| 289 | continue |
| 290 | |
| 291 | # Check if this is a symlink (permissions start with 'l') |
| 292 | permissions = parts[0] |
| 293 | is_symlink = permissions.startswith('l') |
| 294 | |
| 295 | if is_symlink: |
| 296 | # For symlinks, extract the name before the '->' arrow |
| 297 | full_name_part = ' '.join(parts[8:]) |
| 298 | if ' -> ' in full_name_part: |
| 299 | filename = full_name_part.split(' -> ')[0] |
| 300 | symlink_target = full_name_part.split(' -> ')[1] |
| 301 | else: |
| 302 | filename = full_name_part |
| 303 | symlink_target = None |
| 304 | else: |
| 305 | filename = ' '.join(parts[8:]) # Handle filenames with spaces |
| 306 | symlink_target = None |
| 307 | |
| 308 | if not filename: |
| 309 | continue |
| 310 | |
| 311 | # Get full path for this entry |
| 312 | entry_path = full_path / filename |
| 313 | |
| 314 | try: |
| 315 | stat_info = entry_path.stat() |
| 316 | |
| 317 | entry_data: Dict[str, Any] = { |
| 318 | "name": filename, |
| 319 | "path": str(entry_path.relative_to(self.base_dir)), |
| 320 | "modified": datetime.fromtimestamp( |
| 321 | stat_info.st_mtime, |
| 322 | tz=Localization.get().get_tzinfo(), |
| 323 | ).isoformat() |
| 324 | } |
| 325 | |
| 326 | # Add symlink information if this is a symlink |
| 327 | if is_symlink and symlink_target: |
| 328 | entry_data["symlink_target"] = symlink_target |
| 329 | entry_data["is_symlink"] = True |
| 330 | |
| 331 | if entry_path.is_file(): |
| 332 | entry_data.update({ |
| 333 | "type": self._get_file_type(filename), |
| 334 | "size": stat_info.st_size, |
| 335 | "is_dir": False |
| 336 | }) |
| 337 | files.append(entry_data) |
| 338 | elif entry_path.is_dir(): |
| 339 | entry_data.update({ |
| 340 | "type": "folder", |
| 341 | "size": 0, # Directories show as 0 bytes |
| 342 | "is_dir": True |
| 343 | }) |
| 344 | folders.append(entry_data) |
| 345 | |
| 346 | except (OSError, PermissionError, FileNotFoundError) as e: |
| 347 | # Log error but continue with other files |
| 348 | PrintStyle.warning(f"No access to {filename}: {e}") |
| 349 | continue |
| 350 | |
| 351 | if len(files) + len(folders) > 10000: |
| 352 | break |
| 353 | |
| 354 | except Exception as e: |
| 355 | # Log error and continue with next line |
| 356 | PrintStyle.error(f"Error parsing ls line '{line}': {e}") |
| 357 | continue |
| 358 | |
| 359 | except subprocess.TimeoutExpired: |
| 360 | PrintStyle.error("ls command timed out") |
| 361 | except Exception as e: |
| 362 | PrintStyle.error(f"Error running ls command: {e}") |
| 363 | |
| 364 | return files, folders |
| 365 | |
| 366 | def get_files(self, current_path: str = "") -> Dict: |
| 367 | try: |
| 368 | # Resolve the full path while preventing directory traversal |
| 369 | full_path = (self.base_dir / current_path).resolve() |
| 370 | if not str(full_path).startswith(str(self.base_dir)): |
| 371 | raise ValueError("Invalid path") |
| 372 | if not full_path.exists(): |
| 373 | raise FileNotFoundError("Directory not found") |
| 374 | if not full_path.is_dir(): |
| 375 | raise NotADirectoryError("Path is not a directory") |
| 376 | |
| 377 | # Use ls command instead of os.scandir for better error handling |
| 378 | files, folders = self._get_files_via_ls(full_path) |
| 379 | |
| 380 | # Combine folders and files, folders first |
| 381 | all_entries = folders + files |
| 382 | |
| 383 | # Get parent directory path if not at root |
| 384 | parent_path = "" |
| 385 | if current_path: |
| 386 | try: |
| 387 | # Get the absolute path of current directory |
| 388 | current_abs = (self.base_dir / current_path).resolve() |
| 389 | |
| 390 | # parent_path is empty only if we're already at root |
| 391 | if str(current_abs) != str(self.base_dir): |
| 392 | parent_path = str(Path(current_path).parent) |
| 393 | |
| 394 | except Exception: |
| 395 | parent_path = "" |
| 396 | |
| 397 | return { |
| 398 | "entries": all_entries, |
| 399 | "current_path": current_path, |
| 400 | "parent_path": parent_path |
| 401 | } |
| 402 | |
| 403 | except Exception as e: |
| 404 | PrintStyle.error(f"Error reading directory: {e}") |
| 405 | return { |
| 406 | "entries": [], |
| 407 | "current_path": current_path, |
| 408 | "parent_path": "", |
| 409 | "error": str(e), |
| 410 | } |
| 411 | |
| 412 | def get_full_path(self, file_path: str, allow_dir: bool = False) -> str: |
| 413 | """Get full file path if it exists and is within base_dir""" |
| 414 | full_path = files.get_abs_path(self.base_dir, file_path) |
| 415 | if not files.exists(full_path): |
| 416 | raise ValueError(f"File {file_path} not found") |
| 417 | return full_path |
| 418 | |
| 419 | def _get_file_type(self, filename: str) -> str: |
| 420 | ext = self._get_file_extension(filename) |
| 421 | for file_type, extensions in self.ALLOWED_EXTENSIONS.items(): |
| 422 | if ext in extensions: |
| 423 | return file_type |
| 424 | return 'unknown' |