230 lines
8.4 KiB
Python
230 lines
8.4 KiB
Python
"""Manages all file and folder operations inside the workspace directory.
|
|
|
|
Every method validates that the target path stays inside the workspace before
|
|
touching the filesystem, preventing path-traversal attacks.
|
|
"""
|
|
|
|
import streamlit as st
|
|
from pathlib import Path
|
|
|
|
# The workspace folder is created at module load so it always exists.
|
|
WORKSPACE = Path("workspace")
|
|
WORKSPACE.mkdir(exist_ok=True)
|
|
|
|
class FileManager:
|
|
def __init__(self, base_path=Path("workspace")) -> None:
|
|
self.base_path = Path(base_path)
|
|
self.base_path.mkdir(exist_ok=True)
|
|
|
|
def create_folder(self, relative_path: str, name: str) -> bool:
|
|
"""Create a new sub-folder at relative_path/name inside the workspace.
|
|
Returns True on success, False on any error.
|
|
"""
|
|
if not name:
|
|
st.error(f"Invalid folder name: {name}")
|
|
return False
|
|
|
|
# Slashes in the name would silently create nested paths — reject them.
|
|
if "/" in name or "\\" in name:
|
|
st.error(f"Invalid folder name (no slashes allowed): {name}")
|
|
return False
|
|
|
|
name = Path(name)
|
|
if relative_path:
|
|
relative_path = Path(relative_path)
|
|
else:
|
|
relative_path = Path()
|
|
|
|
folder_path = (self.base_path / relative_path / name).resolve()
|
|
|
|
# Ensure the resolved path is still inside the workspace (prevents path traversal).
|
|
if not str(folder_path).startswith(str(self.base_path.resolve())):
|
|
st.error(f"Access denied: {relative_path}")
|
|
return False
|
|
|
|
try:
|
|
folder_path.mkdir(exist_ok=False)
|
|
return True
|
|
except FileExistsError:
|
|
st.warning(f"Folder already exists: {relative_path}")
|
|
return False
|
|
except Exception as e:
|
|
st.error(f"Error creating folder {relative_path}: {str(e)}")
|
|
return False
|
|
|
|
def create_file(self, relative_path: str, name: str) -> bool:
|
|
"""Create a new empty file at relative_path/name inside the workspace.
|
|
If name has no extension, .txt is used as the default.
|
|
Returns True on success, False on any error.
|
|
"""
|
|
if not name or name.strip() == "" :
|
|
st.error(f"Invalid file name: {name}")
|
|
return False
|
|
|
|
name = Path(name)
|
|
if not name.suffix:
|
|
name = name.with_suffix(".txt") # Default to .txt if no extension provided
|
|
|
|
if relative_path:
|
|
relative_path = Path(relative_path)
|
|
else:
|
|
relative_path = Path()
|
|
|
|
file_path = (self.base_path / relative_path / name).resolve()
|
|
|
|
# Ensure the resolved path is still inside the workspace (prevents path traversal).
|
|
if not str(file_path).startswith(str(self.base_path.resolve())):
|
|
st.error(f"Access denied: {relative_path}")
|
|
return False
|
|
|
|
try:
|
|
file_path.touch(exist_ok=False)
|
|
return True
|
|
except FileExistsError:
|
|
st.warning(f"File already exists: {relative_path}")
|
|
return False
|
|
except Exception as e:
|
|
st.error(f"Error creating file {relative_path}: {str(e)}")
|
|
return False
|
|
|
|
def read_file(self, relative_path: Path) -> str:
|
|
"""Read and return the text content of a file.
|
|
Returns an empty string and shows a Streamlit error on failure.
|
|
"""
|
|
file_path = (relative_path).resolve()
|
|
|
|
if not file_path.exists():
|
|
st.error(f"File not found: {relative_path}")
|
|
return ""
|
|
if not file_path.is_file():
|
|
st.error(f"Path is not a file: {relative_path}")
|
|
return ""
|
|
# Ensure the resolved path is still inside the workspace (prevents path traversal).
|
|
if not str(file_path).startswith(str(self.base_path.resolve())):
|
|
st.error(f"Access denied: {relative_path}")
|
|
return ""
|
|
|
|
try:
|
|
with open(file_path, "r") as f:
|
|
return f.read()
|
|
except FileNotFoundError:
|
|
st.error(f"File not found: {relative_path}")
|
|
return ""
|
|
except Exception as e:
|
|
st.error(f"Error reading file {relative_path}: {str(e)}")
|
|
return ""
|
|
|
|
def save_file(self, relative_path: str, content: str):
|
|
"""Overwrite a file with new content.
|
|
Returns True on success, False on any error.
|
|
"""
|
|
file_path = (Path(relative_path)).resolve()
|
|
|
|
# Ensure the resolved path is still inside the workspace (prevents path traversal).
|
|
if not str(file_path).startswith(str(self.base_path.resolve())):
|
|
st.error(f"Access denied: {relative_path}")
|
|
return False
|
|
|
|
try:
|
|
with open(file_path, "w") as f:
|
|
f.write(content)
|
|
return True
|
|
except Exception as e:
|
|
st.error(f"Error saving file {relative_path}: {str(e)}")
|
|
return False
|
|
|
|
def rename_file(self, old_relative_path: str, new_name: str) -> bool:
|
|
"""Rename a file while preserving its original extension.
|
|
If new_name carries a different extension it is ignored; the original
|
|
suffix is always kept. Returns True on success, False on any error.
|
|
"""
|
|
if not new_name or new_name.strip() == "":
|
|
st.error(f"Invalid file name: {new_name}")
|
|
return False
|
|
|
|
file_type = Path(old_relative_path).suffix
|
|
new_name = Path(new_name)
|
|
|
|
# Force the original extension so the file type cannot be changed by renaming.
|
|
if not Path(new_name).suffix == file_type:
|
|
new_name = Path(new_name).with_suffix(file_type) # Ensure the file extension remains the same
|
|
|
|
old_file_path = (Path(old_relative_path)).resolve()
|
|
new_file_path = old_file_path.parent / new_name
|
|
|
|
# Both old and new paths must stay inside the workspace.
|
|
if not str(old_file_path).startswith(str(self.base_path.resolve())) or not str(new_file_path).startswith(str(self.base_path.resolve())):
|
|
st.error(f"Access denied: {old_relative_path}")
|
|
return False
|
|
|
|
try:
|
|
old_file_path.rename(new_file_path)
|
|
return True
|
|
except FileNotFoundError:
|
|
st.error(f"File not found: {old_relative_path}")
|
|
return False
|
|
|
|
def delete_folder(self, relative_path) -> bool:
|
|
"""Delete a folder and all its contents recursively.
|
|
Returns True on success, False on any error.
|
|
"""
|
|
folder_path = (self.base_path / relative_path).resolve()
|
|
|
|
# Ensure the resolved path is still inside the workspace (prevents path traversal).
|
|
if not str(folder_path).startswith(str(self.base_path.resolve())):
|
|
st.error(f"Access denied: {relative_path}")
|
|
return False
|
|
|
|
if not folder_path.exists():
|
|
st.error(f"Folder not found: {relative_path}")
|
|
return False
|
|
|
|
try:
|
|
import shutil
|
|
shutil.rmtree(folder_path)
|
|
return True
|
|
except Exception as e:
|
|
st.error(f"Error deleting folder {relative_path}: {str(e)}")
|
|
return False
|
|
|
|
def delete_file(self, relative_path):
|
|
"""Delete a single file from the workspace.
|
|
Returns True on success, False on any error.
|
|
"""
|
|
file_path = Path(relative_path).resolve()
|
|
|
|
# Ensure the resolved path is still inside the workspace (prevents path traversal).
|
|
if not str(file_path).startswith(str(self.base_path.resolve())):
|
|
st.error(f"Access denied: {relative_path}")
|
|
return False
|
|
|
|
try:
|
|
file_path.unlink()
|
|
return True
|
|
except FileNotFoundError:
|
|
st.error(f"File not found: {relative_path}")
|
|
return False
|
|
except Exception as e:
|
|
st.error(f"Error deleting file {relative_path}: {str(e)}")
|
|
return False
|
|
|
|
def get_file_tree(self):
|
|
"""Return a nested dict representing the workspace directory tree.
|
|
Folders are dicts; files are None values.
|
|
Example: {"src": {"main.py": None}, "README.txt": None}
|
|
"""
|
|
def build_tree(path: Path):
|
|
|
|
tree = {}
|
|
|
|
for item in sorted(path.iterdir()):
|
|
if item.is_dir():
|
|
tree[item.name] = build_tree(item) # recurse into sub-folders
|
|
else:
|
|
tree[item.name] = None # leaf node for files
|
|
return tree
|
|
return build_tree(self.base_path)
|
|
|
|
if __name__ == "__main__":
|
|
FileManager()
|