281 lines
11 KiB
Python
281 lines
11 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:
|
|
"""
|
|
Creates a new folder at the specified relative path.
|
|
The relative_path should be the path to the folder relative to the base path,
|
|
and name should be the name of the new folder (without any slashes).
|
|
|
|
Args:
|
|
relative_path (str): The relative path (without base path) where the new folder should be created.
|
|
name (str): The name of the new folder to create (should not contain slashes).
|
|
Returns:
|
|
bool: True if folder was created successfully, False otherwise.
|
|
"""
|
|
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:
|
|
"""
|
|
Creates a new file at the specified relative path.
|
|
The relative_path should be the path to the folder relative to the base path where the file should be created,
|
|
and name should be the name of the new file (without any slashes).
|
|
|
|
Args:
|
|
relative_path (str): The relative path (without base path) where the new file should be created.
|
|
name (str): The name of the new file to create (should not contain slashes).
|
|
Returns:
|
|
bool: True if file was created successfully, False otherwise.
|
|
"""
|
|
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:
|
|
"""
|
|
Reads the content of a file.
|
|
Accepts an absolute Path object (as stored in st.session_state.open_files).
|
|
The path is validated to ensure it stays inside the workspace.
|
|
|
|
Args:
|
|
relative_path (Path): Absolute path to the file to read.
|
|
Returns:
|
|
str: The content of the file, or an empty string if there was an error.
|
|
"""
|
|
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) -> bool:
|
|
"""
|
|
Saves content to a file.
|
|
Accepts an absolute path string (as stored in st.session_state.open_files).
|
|
The path is validated to ensure it stays inside the workspace.
|
|
|
|
Args:
|
|
relative_path (str): Absolute path to the file to save, including the file name.
|
|
content (str): The content to write to the file.
|
|
Returns:
|
|
bool: True if save was successful, False otherwise.
|
|
"""
|
|
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:
|
|
"""
|
|
Renames a file while keeping the same extension.
|
|
The new_name should not include the extension, as it will be preserved from the old name.
|
|
|
|
Args:
|
|
old_relative_path (str): The current relative path (without base path) of the file to rename, including the file name.
|
|
new_name (str): The new name for the file, without extension.
|
|
Returns:
|
|
bool: True if rename was successful, False otherwise.
|
|
"""
|
|
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(self.base_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: str) -> bool:
|
|
"""Deletes a folder and all its contents.
|
|
The relative_path should be the path to the folder relative to the base path.
|
|
|
|
Args:
|
|
relative_path (str): The relative path (without base path) to the folder to delete.
|
|
Returns:
|
|
bool: True if deletion was successful, False otherwise.
|
|
"""
|
|
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: str) -> bool:
|
|
"""Deletes a file.
|
|
The relative_path should be the path to the file relative to the base path.
|
|
|
|
Args:
|
|
relative_path (str): The relative path (without base path) to the file to delete, including the file name.
|
|
Returns:
|
|
bool: True if deletion was successful, False otherwise.
|
|
"""
|
|
file_path = Path(relative_path)
|
|
abs_file_path = (Path(self.base_path) / file_path).resolve()
|
|
|
|
if not str(abs_file_path).startswith(str(self.base_path.resolve())):
|
|
st.error(f"Access denied: {relative_path}")
|
|
return False
|
|
|
|
try:
|
|
abs_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):
|
|
"""
|
|
Builds a nested dictionary representing the file tree starting from the base path.
|
|
Directories are represented as keys with dictionary values,
|
|
and files are represented as keys with None
|
|
|
|
Returns:
|
|
dict: A nested dictionary representing the file tree.
|
|
"""
|
|
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()
|