File browser fix directory listing (#613)
* fix: filebrowser fix directory listing limit 10000 * fix: handle symlinks in directory listing --------- Co-authored-by: Rafael Uzarowski <uzarowski.rafael@proton.me>
ehl0wr0ld committed
Jul 22, 2025 at 17:09 UTC
935fae558831ee632bcd7bf13fc63698ee75c67b
2 files changed
+135
-51
python/api/get_work_dir_files.py
+2
-1
@@ -24,6 +24,7 @@ class GetWorkDirFiles(ApiHandler):
24
25
return {"data": result}
26
27
+
28
async def get_files(path):
29
browser = FileBrowser()
29
- return browser.get_files(path)
\ No newline at end of file
30
+ return browser.get_files(path)
python/helpers/file_browser.py
+133
-50
@@ -1,16 +1,16 @@
1
import os
2
from pathlib import Path
3
import shutil
4
-import tempfile
4
import base64
6
-from typing import Dict, List, Tuple, Optional, Any
7
-import zipfile
5
+import subprocess
6
+from typing import Dict, List, Tuple, Any
7
from werkzeug.utils import secure_filename
8
from datetime import datetime
9
11
-from python.helpers import files, runtime
10
+from python.helpers import files
11
from python.helpers.print_style import PrintStyle
12
13
+
14
class FileBrowser:
15
ALLOWED_EXTENSIONS = {
16
'image': {'jpg', 'jpeg', 'png', 'bmp'},
@@ -27,7 +27,7 @@ class FileBrowser:
27
# base_dir = "/"
28
base_dir = "/"
29
self.base_dir = Path(base_dir)
30
-
30
+
31
def _check_file_size(self, file) -> bool:
32
try:
33
file.seek(0, os.SEEK_END)
@@ -37,7 +37,7 @@ class FileBrowser:
37
except (AttributeError, IOError):
38
return False
39
40
- def save_file_b64(self, current_path: str, filename:str, base64_content: str):
40
+ def save_file_b64(self, current_path: str, filename: str, base64_content: str):
41
try:
42
# Resolve the target directory path
43
target_file = (self.base_dir / current_path / filename).resolve()
@@ -57,15 +57,15 @@ class FileBrowser:
57
"""Save uploaded files and return successful and failed filenames"""
58
successful = []
59
failed = []
60
-
60
+
61
try:
62
# Resolve the target directory path
63
target_dir = (self.base_dir / current_path).resolve()
64
if not str(target_dir).startswith(str(self.base_dir)):
65
raise ValueError("Invalid target directory")
66
-
66
+
67
os.makedirs(target_dir, exist_ok=True)
68
-
68
+
69
for file in files:
70
try:
71
if file and self._is_allowed_file(file.filename, file):
@@ -79,13 +79,13 @@ class FileBrowser:
79
except Exception as e:
80
PrintStyle.error(f"Error saving file {file.filename}: {e}")
81
failed.append(file.filename)
82
-
82
+
83
return successful, failed
84
-
84
+
85
except Exception as e:
86
PrintStyle.error(f"Error in save_files: {e}")
87
return successful, failed
88
-
88
+
89
def delete_file(self, file_path: str) -> bool:
90
"""Delete a file or empty directory"""
91
try:
@@ -93,35 +93,141 @@ class FileBrowser:
93
full_path = (self.base_dir / file_path).resolve()
94
if not str(full_path).startswith(str(self.base_dir)):
95
raise ValueError("Invalid path")
96
-
96
+
97
if os.path.exists(full_path):
98
if os.path.isfile(full_path):
99
os.remove(full_path)
100
elif os.path.isdir(full_path):
101
shutil.rmtree(full_path)
102
return True
103
-
103
+
104
return False
105
-
105
+
106
except Exception as e:
107
PrintStyle.error(f"Error deleting {file_path}: {e}")
108
return False
109
110
def _is_allowed_file(self, filename: str, file) -> bool:
111
# allow any file to be uploaded in file browser
112
-
112
+
113
# if not filename:
114
# return False
115
# ext = self._get_file_extension(filename)
116
# all_allowed = set().union(*self.ALLOWED_EXTENSIONS.values())
117
# if ext not in all_allowed:
118
# return False
119
-
119
+
120
return True # Allow the file if it passes the checks
121
122
def _get_file_extension(self, filename: str) -> str:
123
return filename.rsplit('.', 1)[1].lower() if '.' in filename else ''
124
-
124
+
125
+ def _get_files_via_ls(self, full_path: Path) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
126
+ """Get files and folders using ls command for better error handling"""
127
+ files: List[Dict[str, Any]] = []
128
+ folders: List[Dict[str, Any]] = []
129
+
130
+ try:
131
+ # Use ls command to get directory listing
132
+ result = subprocess.run(
133
+ ['ls', '-la', str(full_path)],
134
+ capture_output=True,
135
+ text=True,
136
+ timeout=30
137
+ )
138
+
139
+ if result.returncode != 0:
140
+ PrintStyle.error(f"ls command failed: {result.stderr}")
141
+ return files, folders
142
+
143
+ # Parse ls output (skip first line which is "total X")
144
+ lines = result.stdout.strip().split('\n')
145
+ if len(lines) <= 1:
146
+ return files, folders
147
+
148
+ for line in lines[1:]: # Skip the "total" line
149
+ try:
150
+ # Skip current and parent directory entries
151
+ if line.endswith(' .') or line.endswith(' ..'):
152
+ continue
153
+
154
+ # Parse ls -la output format
155
+ parts = line.split()
156
+ if len(parts) < 9:
157
+ continue
158
+
159
+ # Check if this is a symlink (permissions start with 'l')
160
+ permissions = parts[0]
161
+ is_symlink = permissions.startswith('l')
162
+
163
+ if is_symlink:
164
+ # For symlinks, extract the name before the '->' arrow
165
+ full_name_part = ' '.join(parts[8:])
166
+ if ' -> ' in full_name_part:
167
+ filename = full_name_part.split(' -> ')[0]
168
+ symlink_target = full_name_part.split(' -> ')[1]
169
+ else:
170
+ filename = full_name_part
171
+ symlink_target = None
172
+ else:
173
+ filename = ' '.join(parts[8:]) # Handle filenames with spaces
174
+ symlink_target = None
175
+
176
+ if not filename:
177
+ continue
178
+
179
+ # Get full path for this entry
180
+ entry_path = full_path / filename
181
+
182
+ try:
183
+ stat_info = entry_path.stat()
184
+
185
+ entry_data: Dict[str, Any] = {
186
+ "name": filename,
187
+ "path": str(entry_path.relative_to(self.base_dir)),
188
+ "modified": datetime.fromtimestamp(stat_info.st_mtime).isoformat()
189
+ }
190
+
191
+ # Add symlink information if this is a symlink
192
+ if is_symlink and symlink_target:
193
+ entry_data["symlink_target"] = symlink_target
194
+ entry_data["is_symlink"] = True
195
+
196
+ if entry_path.is_file():
197
+ entry_data.update({
198
+ "type": self._get_file_type(filename),
199
+ "size": stat_info.st_size,
200
+ "is_dir": False
201
+ })
202
+ files.append(entry_data)
203
+ elif entry_path.is_dir():
204
+ entry_data.update({
205
+ "type": "folder",
206
+ "size": 0, # Directories show as 0 bytes
207
+ "is_dir": True
208
+ })
209
+ folders.append(entry_data)
210
+
211
+ except (OSError, PermissionError, FileNotFoundError) as e:
212
+ # Log error but continue with other files
213
+ PrintStyle.warning(f"No access to {filename}: {e}")
214
+ continue
215
+
216
+ if len(files) + len(folders) > 10000:
217
+ break
218
+
219
+ except Exception as e:
220
+ # Log error and continue with next line
221
+ PrintStyle.error(f"Error parsing ls line '{line}': {e}")
222
+ continue
223
+
224
+ except subprocess.TimeoutExpired:
225
+ PrintStyle.error("ls command timed out")
226
+ except Exception as e:
227
+ PrintStyle.error(f"Error running ls command: {e}")
228
+
229
+ return files, folders
230
+
231
def get_files(self, current_path: str = "") -> Dict:
232
try:
233
# Resolve the full path while preventing directory traversal
@@ -129,31 +235,8 @@ class FileBrowser:
235
if not str(full_path).startswith(str(self.base_dir)):
236
raise ValueError("Invalid path")
237
132
- files = []
133
- folders = []
134
-
135
- # List all entries in the current directory
136
- for entry in os.scandir(full_path):
137
- entry_data: Dict[str, Any] = {
138
- "name": entry.name,
139
- "path": str(Path(entry.path).relative_to(self.base_dir)),
140
- "modified": datetime.fromtimestamp(entry.stat().st_mtime).isoformat()
141
- }
142
-
143
- if entry.is_file():
144
- entry_data.update({
145
- "type": self._get_file_type(entry.name),
146
- "size": entry.stat().st_size,
147
- "is_dir": False
148
- })
149
- files.append(entry_data)
150
- else:
151
- entry_data.update({
152
- "type": "folder",
153
- "size": 0, # Directories show as 0 bytes
154
- "is_dir": True
155
- })
156
- folders.append(entry_data)
238
+ # Use ls command instead of os.scandir for better error handling
239
+ files, folders = self._get_files_via_ls(full_path)
240
241
# Combine folders and files, folders first
242
all_entries = folders + files
@@ -168,8 +251,8 @@ class FileBrowser:
251
# parent_path is empty only if we're already at root
252
if str(current_abs) != str(self.base_dir):
253
parent_path = str(Path(current_path).parent)
171
-
172
- except Exception as e:
254
+
255
+ except Exception:
256
parent_path = ""
257
258
return {
@@ -181,17 +264,17 @@ class FileBrowser:
264
except Exception as e:
265
PrintStyle.error(f"Error reading directory: {e}")
266
return {"entries": [], "current_path": "", "parent_path": ""}
184
-
267
+
268
def get_full_path(self, file_path: str, allow_dir: bool = False) -> str:
269
"""Get full file path if it exists and is within base_dir"""
187
- full_path = files.get_abs_path(self.base_dir,file_path)
270
+ full_path = files.get_abs_path(self.base_dir, file_path)
271
if not files.exists(full_path):
189
- raise ValueError(f"File {file_path} not found")
272
+ raise ValueError(f"File {file_path} not found")
273
return full_path
191
-
274
+
275
def _get_file_type(self, filename: str) -> str:
276
ext = self._get_file_extension(filename)
277
for file_type, extensions in self.ALLOWED_EXTENSIONS.items():
278
if ext in extensions:
279
return file_type
197
- return 'unknown'
\ No newline at end of file
280
+ return 'unknown'