technicalkiwi/python-editor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add admin invites and user workspace management tools.

Browse Source 
Implement invite-token registration with optional email delivery, add admin UI actions for creating invites and opening user workspaces, and support superuser workspace override while preserving per-user code isolation with shared read-only lib.

Made-with: Cursor
This commit is contained in:
jimmy
2026-05-01 21:13:13 +12:00
parent e4c811f51d
commit 7d682cce8d
15 changed files with 683 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
src/editor_app/services/filesystem.py
View File

@@ -10,6 +10,14 @@ LIB_DIR_NAME = "lib"
WRITABLE_ROOTS = {"code"}
def _workspace_root(workspace_root: Path | None = None) -> Path:
return (workspace_root or config.WORKSPACE_ROOT).resolve()
def _shared_lib_root() -> Path:
return (config.WORKSPACE_ROOT.resolve() / LIB_DIR_NAME).resolve()
def normalize_relative_path(relative_path: str) -> str:
cleaned = (relative_path or "").strip().lstrip("/")
if not cleaned:
@@ -22,33 +30,48 @@ def normalize_relative_path(relative_path: str) -> str:
return "/".join(parts)
def resolve_workspace_path(relative_path: str) -> Path:
def resolve_workspace_path(relative_path: str, workspace_root: Path | None = None) -> Path:
relative_path = normalize_relative_path(relative_path)
target_path = (config.WORKSPACE_ROOT / relative_path).resolve()
root = _workspace_root(workspace_root)
if relative_path == LIB_DIR_NAME or relative_path.startswith(f"{LIB_DIR_NAME}/"):
suffix = relative_path[len(LIB_DIR_NAME) :].lstrip("/")
target_path = (_shared_lib_root() / suffix).resolve()
try:
target_path.relative_to(_shared_lib_root())
except ValueError as exc:
raise HTTPException(status_code=400, detail="Path escapes shared lib") from exc
return target_path
target_path = (root / relative_path).resolve()
try:
target_path.relative_to(config.WORKSPACE_ROOT.resolve())
target_path.relative_to(root)
except ValueError as exc:
raise HTTPException(status_code=400, detail="Path escapes workspace") from exc
return target_path
def _is_path_in_lib(target_path: Path) -> bool:
workspace = config.WORKSPACE_ROOT.resolve()
def _is_path_in_lib(target_path: Path, workspace_root: Path | None = None) -> bool:
workspace = _workspace_root(workspace_root)
lib_root = (workspace / LIB_DIR_NAME).resolve()
shared_lib_root = _shared_lib_root()
try:
target_path.resolve().relative_to(lib_root)
return True
except ValueError:
pass
try:
target_path.resolve().relative_to(shared_lib_root)
return True
except ValueError:
return False
def _ensure_not_lib_path(target_path: Path) -> None:
if _is_path_in_lib(target_path):
def _ensure_not_lib_path(target_path: Path, workspace_root: Path | None = None) -> None:
if _is_path_in_lib(target_path, workspace_root):
raise HTTPException(status_code=403, detail="lib is read-only")
def _is_writable_path(target_path: Path) -> bool:
workspace = config.WORKSPACE_ROOT.resolve()
def _is_writable_path(target_path: Path, workspace_root: Path | None = None) -> bool:
workspace = _workspace_root(workspace_root)
resolved = target_path.resolve()
try:
relative = resolved.relative_to(workspace)
@@ -59,17 +82,18 @@ def _is_writable_path(target_path: Path) -> bool:
return relative.parts[0] in WRITABLE_ROOTS
def _ensure_writable_path(target_path: Path) -> None:
if not _is_writable_path(target_path):
def _ensure_writable_path(target_path: Path, workspace_root: Path | None = None) -> None:
if not _is_writable_path(target_path, workspace_root):
raise HTTPException(
status_code=403,
detail="Only code/ is writable (lib is read-only)",
)
def list_files(path: str = "") -> list[FileInfo]:
def list_files(path: str = "", workspace_root: Path | None = None) -> list[FileInfo]:
path = normalize_relative_path(path)
target_path = config.WORKSPACE_ROOT / path if path else config.WORKSPACE_ROOT
root = _workspace_root(workspace_root)
target_path = resolve_workspace_path(path, root) if path else root
if not target_path.exists() or not target_path.is_dir():
raise HTTPException(status_code=404, detail="Directory not found")
@@ -77,6 +101,8 @@ def list_files(path: str = "") -> list[FileInfo]:
for item in sorted(target_path.iterdir()):
if item.name.startswith("."):
continue
if not path and item.name == "users":
continue
files.append(
FileInfo(
name=item.name,
@@ -84,11 +110,15 @@ def list_files(path: str = "") -> list[FileInfo]:
size=item.stat().st_size if item.is_file() else None,
)
)
if not path:
shared_lib = _shared_lib_root()
if shared_lib.exists() and not any(f.name == LIB_DIR_NAME for f in files):
files.append(FileInfo(name=LIB_DIR_NAME, is_directory=True, size=None))
return files
def read_text_file(file_path: str) -> tuple[str, str]:
target_path = resolve_workspace_path(file_path)
def read_text_file(file_path: str, workspace_root: Path | None = None) -> tuple[str, str]:
target_path = resolve_workspace_path(file_path, workspace_root)
if not target_path.exists():
raise HTTPException(status_code=404, detail="File not found")
if target_path.is_dir():
@@ -100,19 +130,19 @@ def read_text_file(file_path: str) -> tuple[str, str]:
return content, target_path.name
def save_text_file(file_path: str, content: str) -> str:
target_path = resolve_workspace_path(file_path)
_ensure_not_lib_path(target_path)
_ensure_writable_path(target_path)
def save_text_file(file_path: str, content: str, workspace_root: Path | None = None) -> str:
target_path = resolve_workspace_path(file_path, workspace_root)
_ensure_not_lib_path(target_path, workspace_root)
_ensure_writable_path(target_path, workspace_root)
target_path.parent.mkdir(parents=True, exist_ok=True)
target_path.write_text(content, encoding="utf-8")
return target_path.name
def delete_file(file_path: str) -> None:
target_path = resolve_workspace_path(file_path)
_ensure_not_lib_path(target_path)
_ensure_writable_path(target_path)
def delete_file(file_path: str, workspace_root: Path | None = None) -> None:
target_path = resolve_workspace_path(file_path, workspace_root)
_ensure_not_lib_path(target_path, workspace_root)
_ensure_writable_path(target_path, workspace_root)
if not target_path.exists():
raise HTTPException(status_code=404, detail="File not found")
if target_path.is_dir():
@@ -120,20 +150,21 @@ def delete_file(file_path: str) -> None:
target_path.unlink()
def move_path(source_path: str, destination_folder: str) -> tuple[str, str]:
source = resolve_workspace_path(source_path)
_ensure_not_lib_path(source)
_ensure_writable_path(source)
def move_path(source_path: str, destination_folder: str, workspace_root: Path | None = None) -> tuple[str, str]:
root = _workspace_root(workspace_root)
source = resolve_workspace_path(source_path, root)
_ensure_not_lib_path(source, root)
_ensure_writable_path(source, root)
if not source.exists():
raise HTTPException(status_code=404, detail="Source path not found")
destination_dir = (
resolve_workspace_path(destination_folder)
resolve_workspace_path(destination_folder, root)
if destination_folder
else config.WORKSPACE_ROOT
else root
)
_ensure_not_lib_path(destination_dir)
_ensure_writable_path(destination_dir)
_ensure_not_lib_path(destination_dir, root)
_ensure_writable_path(destination_dir, root)
if not destination_dir.exists() or not destination_dir.is_dir():
raise HTTPException(status_code=404, detail="Destination folder not found")
@@ -157,23 +188,23 @@ def move_path(source_path: str, destination_folder: str) -> tuple[str, str]:
destination.parent.mkdir(parents=True, exist_ok=True)
source.rename(destination)
moved_type = "folder" if destination.is_dir() else "file"
return str(destination.relative_to(config.WORKSPACE_ROOT)), moved_type
return str(destination.relative_to(root)), moved_type
def create_folder(folder_path: str) -> str:
target_path = resolve_workspace_path(folder_path)
_ensure_not_lib_path(target_path)
_ensure_writable_path(target_path)
def create_folder(folder_path: str, workspace_root: Path | None = None) -> str:
target_path = resolve_workspace_path(folder_path, workspace_root)
_ensure_not_lib_path(target_path, workspace_root)
_ensure_writable_path(target_path, workspace_root)
if target_path.exists():
raise HTTPException(status_code=400, detail="Folder already exists")
target_path.mkdir(parents=True, exist_ok=False)
return target_path.name
def delete_folder(folder_path: str) -> None:
target_path = resolve_workspace_path(folder_path)
_ensure_not_lib_path(target_path)
_ensure_writable_path(target_path)
def delete_folder(folder_path: str, workspace_root: Path | None = None) -> None:
target_path = resolve_workspace_path(folder_path, workspace_root)
_ensure_not_lib_path(target_path, workspace_root)
_ensure_writable_path(target_path, workspace_root)
if not target_path.exists():
raise HTTPException(status_code=404, detail="Folder not found")
if not target_path.is_dir():
@@ -181,10 +212,10 @@ def delete_folder(folder_path: str) -> None:
shutil.rmtree(target_path)
def collect_python_sources() -> dict[str, str]:
def collect_python_sources(workspace_root: Path | None = None) -> dict[str, str]:
"""Return all UTF-8 .py files under the workspace for browser-side Pyodide sync."""
result: dict[str, str] = {}
workspace = config.WORKSPACE_ROOT.resolve()
workspace = _workspace_root(workspace_root)
if not workspace.exists():
return result
for path in workspace.rglob("*.py"):
@@ -199,4 +230,18 @@ def collect_python_sources() -> dict[str, str]:
result[key] = path.read_text(encoding="utf-8")
except (UnicodeDecodeError, OSError):
continue
shared_lib = _shared_lib_root()
if shared_lib.exists() and shared_lib.is_dir() and shared_lib != (workspace / LIB_DIR_NAME).resolve():
for path in shared_lib.rglob("*.py"):
try:
rel = path.relative_to(shared_lib)
except ValueError:
continue
if any(part.startswith(".") for part in rel.parts):
continue
try:
key = str(Path(LIB_DIR_NAME) / rel).replace("\\", "/")
result[key] = path.read_text(encoding="utf-8")
except (UnicodeDecodeError, OSError):
continue
return result