diff --git a/server/routes/file_routes.py b/server/routes/file_routes.py new file mode 100755 index 0000000..5d31632 --- /dev/null +++ b/server/routes/file_routes.py @@ -0,0 +1,398 @@ +""" +File persistence routes for Calliope IDE autosave system. +Addresses issue #57. + +Endpoints: + POST /api/files/save — save (create/overwrite) a file + GET /api/files/load — load file contents + GET /api/files/list — list all files in session workspace + DELETE /api/files/delete — delete a file + POST /api/files/mkdir — create a directory +""" + +import os +import re +import logging +from pathlib import Path +from flask import Blueprint, request, jsonify +from server.utils.auth_utils import token_required +from server.utils.monitoring import capture_exception + +try: + from server.models import Session +except Exception: + Session = None # type: ignore + +files_bp = Blueprint("files", __name__, url_prefix="/api/files") +logger = logging.getLogger(__name__) + +# Max file size: 5 MB +_MAX_FILE_SIZE = 5 * 1024 * 1024 +# Allowed extensions (empty = allow all text files) +_BLOCKED_EXTENSIONS = {".exe", ".bin", ".so", ".dll", ".dylib"} + + +# ── Path safety ─────────────────────────────────────────────────────────────── + +def _safe_path(relative_path: str, instance_dir: str) -> str | None: + """ + Resolve relative_path inside instance_dir and validate no path traversal. + Returns absolute path on success, None on traversal attempt. + """ + base = os.path.abspath(instance_dir) + target = os.path.abspath(os.path.join(base, relative_path)) + if not target.startswith(base + os.sep) and target != base: + return None + return target + + +def _validate_filename(path: str) -> bool: + """Block null bytes, absolute paths, and suspicious patterns.""" + if "\x00" in path: + return False + if os.path.isabs(path): + return False + # Block hidden files at root level (e.g. .env, .git) + parts = Path(path).parts + if parts and parts[0].startswith("."): + return False + return True + + +# ── Routes ──────────────────────────────────────────────────────────────────── + +@files_bp.route("/save", methods=["POST"]) +@token_required +def save_file(current_user): + """ + Save (create or overwrite) a file in the session workspace. + + Request JSON: + session_id (int) — active session ID + path (str) — relative path inside the workspace + content (str) — file content (text) + + Response JSON: + success (bool) + path (str) — relative path + size (int) — bytes written + created (bool) — True if new file, False if overwritten + """ + try: + data = request.get_json(silent=True, force=True) + if not data: + return jsonify({"success": False, "error": "No data provided"}), 400 + + session_id = data.get("session_id") + rel_path = (data.get("path") or "").strip() + content = data.get("content", "") + + if not session_id: + return jsonify({"success": False, "error": "session_id is required"}), 400 + if not rel_path: + return jsonify({"success": False, "error": "path is required"}), 400 + if not isinstance(content, str): + return jsonify({"success": False, "error": "content must be a string"}), 400 + if len(content.encode("utf-8")) > _MAX_FILE_SIZE: + return jsonify({"success": False, "error": f"File too large (max {_MAX_FILE_SIZE // 1024}KB)"}), 413 + + # Validate filename + if not _validate_filename(rel_path): + return jsonify({"success": False, "error": "Invalid path"}), 400 + + # Block dangerous extensions + ext = Path(rel_path).suffix.lower() + if ext in _BLOCKED_EXTENSIONS: + return jsonify({"success": False, "error": f"File type '{ext}' not allowed"}), 400 + + # Verify session + session = _get_session(session_id, current_user.id) + if not session: + return jsonify({"success": False, "error": "Session not found or access denied"}), 404 + + instance_dir = session.instance_dir + if not instance_dir or not os.path.isdir(instance_dir): + return jsonify({"success": False, "error": "Session workspace not found"}), 404 + + abs_path = _safe_path(rel_path, instance_dir) + if not abs_path: + return jsonify({"success": False, "error": "Path traversal detected"}), 400 + + # Create parent directories if needed + os.makedirs(os.path.dirname(abs_path), exist_ok=True) + + created = not os.path.exists(abs_path) + with open(abs_path, "w", encoding="utf-8") as f: + f.write(content) + + size = os.path.getsize(abs_path) + logger.info("User %s saved file %s (%d bytes)", current_user.username, rel_path, size) + + return jsonify({ + "success": True, + "path": rel_path, + "size": size, + "created": created, + }), 200 + + except OSError as e: + logger.exception("File save OS error") + return jsonify({"success": False, "error": f"File system error: {e.strerror}"}), 500 + except Exception as e: + logger.exception("Save file error") + capture_exception(e, {"route": "files.save_file", "user_id": current_user.id}) + return jsonify({"success": False, "error": "An error occurred while saving the file"}), 500 + + +@files_bp.route("/load", methods=["GET"]) +@token_required +def load_file(current_user): + """ + Load file contents from the session workspace. + + Query params: + session_id (int) — active session ID + path (str) — relative path inside the workspace + + Response JSON: + success (bool) + path (str) + content (str) + size (int) + """ + try: + session_id = request.args.get("session_id", type=int) + rel_path = (request.args.get("path") or "").strip() + + if not session_id: + return jsonify({"success": False, "error": "session_id is required"}), 400 + if not rel_path: + return jsonify({"success": False, "error": "path is required"}), 400 + + session = _get_session(session_id, current_user.id) + if not session: + return jsonify({"success": False, "error": "Session not found or access denied"}), 404 + + instance_dir = session.instance_dir + if not instance_dir or not os.path.isdir(instance_dir): + return jsonify({"success": False, "error": "Session workspace not found"}), 404 + + abs_path = _safe_path(rel_path, instance_dir) + if not abs_path: + return jsonify({"success": False, "error": "Path traversal detected"}), 400 + + if not os.path.exists(abs_path): + return jsonify({"success": False, "error": "File not found"}), 404 + + if not os.path.isfile(abs_path): + return jsonify({"success": False, "error": "Path is not a file"}), 400 + + size = os.path.getsize(abs_path) + if size > _MAX_FILE_SIZE: + return jsonify({"success": False, "error": "File too large to load"}), 413 + + with open(abs_path, "r", encoding="utf-8", errors="replace") as f: + content = f.read() + + return jsonify({ + "success": True, + "path": rel_path, + "content": content, + "size": size, + }), 200 + + except OSError as e: + return jsonify({"success": False, "error": f"File system error: {e.strerror}"}), 500 + except Exception as e: + logger.exception("Load file error") + capture_exception(e, {"route": "files.load_file", "user_id": current_user.id}) + return jsonify({"success": False, "error": "An error occurred while loading the file"}), 500 + + +@files_bp.route("/list", methods=["GET"]) +@token_required +def list_files(current_user): + """ + List all files in the session workspace (recursive). + + Query params: + session_id (int) — active session ID + subdir (str) — optional subdirectory to list + + Response JSON: + success (bool) + files (list[dict]) — [{path, size, is_dir}] + total (int) + """ + try: + session_id = request.args.get("session_id", type=int) + subdir = (request.args.get("subdir") or "").strip() + + if not session_id: + return jsonify({"success": False, "error": "session_id is required"}), 400 + + session = _get_session(session_id, current_user.id) + if not session: + return jsonify({"success": False, "error": "Session not found or access denied"}), 404 + + instance_dir = session.instance_dir + if not instance_dir or not os.path.isdir(instance_dir): + return jsonify({"success": False, "error": "Session workspace not found"}), 404 + + if subdir: + root = _safe_path(subdir, instance_dir) + if not root: + return jsonify({"success": False, "error": "Path traversal detected"}), 400 + else: + root = os.path.abspath(instance_dir) + + if not os.path.isdir(root): + return jsonify({"success": False, "error": "Directory not found"}), 404 + + files = [] + for item in sorted(Path(root).rglob("*")): + # Skip hidden files and __pycache__ + parts = item.relative_to(root).parts + if any(p.startswith(".") or p == "__pycache__" for p in parts): + continue + rel = str(item.relative_to(instance_dir)) + files.append({ + "path": rel, + "size": item.stat().st_size if item.is_file() else 0, + "is_dir": item.is_dir(), + }) + + return jsonify({ + "success": True, + "files": files, + "total": len(files), + }), 200 + + except Exception as e: + logger.exception("List files error") + capture_exception(e, {"route": "files.list_files", "user_id": current_user.id}) + return jsonify({"success": False, "error": "An error occurred while listing files"}), 500 + + +@files_bp.route("/delete", methods=["DELETE"]) +@token_required +def delete_file(current_user): + """ + Delete a file from the session workspace. + + Request JSON: + session_id (int) — active session ID + path (str) — relative path inside the workspace + + Response JSON: + success (bool) + path (str) + """ + try: + data = request.get_json(silent=True, force=True) + if not data: + return jsonify({"success": False, "error": "No data provided"}), 400 + + session_id = data.get("session_id") + rel_path = (data.get("path") or "").strip() + + if not session_id: + return jsonify({"success": False, "error": "session_id is required"}), 400 + if not rel_path: + return jsonify({"success": False, "error": "path is required"}), 400 + + session = _get_session(session_id, current_user.id) + if not session: + return jsonify({"success": False, "error": "Session not found or access denied"}), 404 + + instance_dir = session.instance_dir + if not instance_dir or not os.path.isdir(instance_dir): + return jsonify({"success": False, "error": "Session workspace not found"}), 404 + + abs_path = _safe_path(rel_path, instance_dir) + if not abs_path: + return jsonify({"success": False, "error": "Path traversal detected"}), 400 + + if not os.path.exists(abs_path): + return jsonify({"success": False, "error": "File not found"}), 404 + + if not os.path.isfile(abs_path): + return jsonify({"success": False, "error": "Path is not a file — use rmdir for directories"}), 400 + + os.remove(abs_path) + logger.info("User %s deleted file %s", current_user.username, rel_path) + + return jsonify({"success": True, "path": rel_path}), 200 + + except OSError as e: + return jsonify({"success": False, "error": f"File system error: {e.strerror}"}), 500 + except Exception as e: + logger.exception("Delete file error") + capture_exception(e, {"route": "files.delete_file", "user_id": current_user.id}) + return jsonify({"success": False, "error": "An error occurred while deleting the file"}), 500 + + +@files_bp.route("/mkdir", methods=["POST"]) +@token_required +def make_directory(current_user): + """ + Create a directory in the session workspace. + + Request JSON: + session_id (int) — active session ID + path (str) — relative path for the new directory + + Response JSON: + success (bool) + path (str) + created (bool) + """ + try: + data = request.get_json(silent=True, force=True) + if not data: + return jsonify({"success": False, "error": "No data provided"}), 400 + + session_id = data.get("session_id") + rel_path = (data.get("path") or "").strip() + + if not session_id: + return jsonify({"success": False, "error": "session_id is required"}), 400 + if not rel_path: + return jsonify({"success": False, "error": "path is required"}), 400 + + if not _validate_filename(rel_path): + return jsonify({"success": False, "error": "Invalid path"}), 400 + + session = _get_session(session_id, current_user.id) + if not session: + return jsonify({"success": False, "error": "Session not found or access denied"}), 404 + + instance_dir = session.instance_dir + if not instance_dir or not os.path.isdir(instance_dir): + return jsonify({"success": False, "error": "Session workspace not found"}), 404 + + abs_path = _safe_path(rel_path, instance_dir) + if not abs_path: + return jsonify({"success": False, "error": "Path traversal detected"}), 400 + + created = not os.path.exists(abs_path) + os.makedirs(abs_path, exist_ok=True) + + return jsonify({"success": True, "path": rel_path, "created": created}), 200 + + except OSError as e: + return jsonify({"success": False, "error": f"File system error: {e.strerror}"}), 500 + except Exception as e: + logger.exception("mkdir error") + capture_exception(e, {"route": "files.make_directory", "user_id": current_user.id}) + return jsonify({"success": False, "error": "An error occurred while creating the directory"}), 500 + + +# ── Helper ──────────────────────────────────────────────────────────────────── + +def _get_session(session_id: int, user_id: int): + if Session is None: + return None + return Session.query.filter_by( + id=session_id, user_id=user_id, is_active=True + ).first() diff --git a/server/start.py b/server/start.py index 6f3f983..b25305f 100644 --- a/server/start.py +++ b/server/start.py @@ -19,6 +19,7 @@ from server.routes.chat_routes import chat_bp from server.routes.soroban_routes import soroban_bp from server.routes.template_routes import templates_bp +from server.routes.file_routes import files_bp from server.routes.project_routes import project_bp from server.routes.soroban_deploy import soroban_deploy_bp from server.routes.soroban_invoke import soroban_invoke_bp @@ -85,6 +86,7 @@ app.register_blueprint(project_bp) app.register_blueprint(soroban_bp) app.register_blueprint(templates_bp) +app.register_blueprint(files_bp) app.register_blueprint(soroban_deploy_bp) app.register_blueprint(soroban_invoke_bp) app.register_blueprint(wallet_bp) diff --git a/server/tests/test_file_routes.py b/server/tests/test_file_routes.py new file mode 100755 index 0000000..617d0b8 --- /dev/null +++ b/server/tests/test_file_routes.py @@ -0,0 +1,297 @@ +"""Tests for server/routes/file_routes.py — autosave & file persistence""" + +import os +import sys +import functools +from unittest.mock import MagicMock + +# Stub deps +def _passthrough(f): + @functools.wraps(f) + def inner(*args, **kwargs): + u = MagicMock(); u.id = 1; u.username = "testuser" + return f(u, *args, **kwargs) + return inner + +_auth_stub = MagicMock() +_auth_stub.token_required = _passthrough +sys.modules["server.utils.auth_utils"] = _auth_stub +sys.modules["server.models"] = MagicMock() +sys.modules["server.utils.monitoring"] = MagicMock() + +import server.routes.file_routes as m +files_bp = m.files_bp + +for mod in ["server.utils.auth_utils", "server.models", "server.utils.monitoring"]: + sys.modules.pop(mod, None) + +import pytest +from flask import Flask + + +@pytest.fixture +def app(): + a = Flask(__name__) + a.config["TESTING"] = True + a.register_blueprint(files_bp) + return a + +@pytest.fixture +def client(app): return app.test_client() + +def yes_session(instance_dir): + s = MagicMock(); s.id = 1; s.user_id = 1; s.is_active = True + s.instance_dir = instance_dir + x = MagicMock(); x.query.filter_by.return_value.first.return_value = s + return x + +def no_session(): + x = MagicMock(); x.query.filter_by.return_value.first.return_value = None + return x + +def bad_workspace(): + s = MagicMock(); s.id = 1; s.user_id = 1; s.is_active = True + s.instance_dir = "/nonexistent/path" + x = MagicMock(); x.query.filter_by.return_value.first.return_value = s + return x + + +# ── _safe_path ──────────────────────────────────────────────────────────────── + +class TestSafePath: + def test_valid_relative_path(self, tmp_path): + result = m._safe_path("src/main.rs", str(tmp_path)) + assert result == str(tmp_path / "src" / "main.rs") + + def test_blocks_parent_traversal(self, tmp_path): + result = m._safe_path("../../etc/passwd", str(tmp_path)) + assert result is None + + def test_blocks_absolute_path(self, tmp_path): + result = m._safe_path("/etc/passwd", str(tmp_path)) + # os.path.join with absolute path replaces base — safe_path should catch it + assert result is None or not result.startswith(str(tmp_path)) + + def test_nested_path_valid(self, tmp_path): + result = m._safe_path("a/b/c/file.txt", str(tmp_path)) + assert result is not None + assert result.startswith(str(tmp_path)) + + +# ── _validate_filename ──────────────────────────────────────────────────────── + +class TestValidateFilename: + def test_valid_filename(self): + assert m._validate_filename("src/lib.rs") is True + + def test_blocks_null_byte(self): + assert m._validate_filename("file\x00name") is False + + def test_blocks_absolute_path(self): + assert m._validate_filename("/etc/passwd") is False + + def test_blocks_hidden_root(self): + assert m._validate_filename(".env") is False + + def test_allows_nested_hidden(self): + # Files in non-root hidden dirs are allowed (e.g. src/.gitkeep) + assert m._validate_filename("src/.gitkeep") is True + + +# ── POST /api/files/save ────────────────────────────────────────────────────── + +class TestSaveFile: + def test_missing_session_id(self, client): + resp = client.post("/api/files/save", json={"path": "x.txt", "content": "hi"}) + assert resp.status_code == 400 + + def test_missing_path(self, client): + resp = client.post("/api/files/save", json={"session_id": 1, "content": "hi"}) + assert resp.status_code == 400 + + def test_no_json_body(self, client): + resp = client.post("/api/files/save") + assert resp.status_code == 400 + + def test_session_not_found(self, client): + m.Session = no_session() + resp = client.post("/api/files/save", json={"session_id": 99, "path": "x.txt", "content": "hi"}) + assert resp.status_code == 404 + + def test_path_traversal_blocked(self, client, tmp_path): + m.Session = yes_session(str(tmp_path)) + resp = client.post("/api/files/save", json={ + "session_id": 1, "path": "../../etc/passwd", "content": "evil" + }) + assert resp.status_code == 400 + + def test_blocked_extension(self, client, tmp_path): + m.Session = yes_session(str(tmp_path)) + resp = client.post("/api/files/save", json={ + "session_id": 1, "path": "malware.exe", "content": "x" + }) + assert resp.status_code == 400 + + def test_saves_file_successfully(self, client, tmp_path): + m.Session = yes_session(str(tmp_path)) + resp = client.post("/api/files/save", json={ + "session_id": 1, "path": "hello.txt", "content": "Hello World" + }) + assert resp.status_code == 200 + data = resp.get_json() + assert data["success"] is True + assert data["created"] is True + assert (tmp_path / "hello.txt").read_text() == "Hello World" + + def test_overwrites_existing_file(self, client, tmp_path): + (tmp_path / "existing.txt").write_text("old content") + m.Session = yes_session(str(tmp_path)) + resp = client.post("/api/files/save", json={ + "session_id": 1, "path": "existing.txt", "content": "new content" + }) + assert resp.status_code == 200 + data = resp.get_json() + assert data["created"] is False + assert (tmp_path / "existing.txt").read_text() == "new content" + + def test_creates_parent_directories(self, client, tmp_path): + m.Session = yes_session(str(tmp_path)) + resp = client.post("/api/files/save", json={ + "session_id": 1, "path": "src/contracts/lib.rs", "content": "// code" + }) + assert resp.status_code == 200 + assert (tmp_path / "src" / "contracts" / "lib.rs").exists() + + def test_returns_file_size(self, client, tmp_path): + m.Session = yes_session(str(tmp_path)) + resp = client.post("/api/files/save", json={ + "session_id": 1, "path": "size.txt", "content": "hello" + }) + data = resp.get_json() + assert data["size"] == 5 + + +# ── GET /api/files/load ─────────────────────────────────────────────────────── + +class TestLoadFile: + def test_missing_session_id(self, client): + resp = client.get("/api/files/load?path=x.txt") + assert resp.status_code == 400 + + def test_missing_path(self, client): + resp = client.get("/api/files/load?session_id=1") + assert resp.status_code == 400 + + def test_session_not_found(self, client): + m.Session = no_session() + resp = client.get("/api/files/load?session_id=99&path=x.txt") + assert resp.status_code == 404 + + def test_file_not_found(self, client, tmp_path): + m.Session = yes_session(str(tmp_path)) + resp = client.get("/api/files/load?session_id=1&path=nonexistent.txt") + assert resp.status_code == 404 + + def test_path_traversal_blocked(self, client, tmp_path): + m.Session = yes_session(str(tmp_path)) + resp = client.get("/api/files/load?session_id=1&path=../../etc/passwd") + assert resp.status_code == 400 + + def test_loads_file_successfully(self, client, tmp_path): + (tmp_path / "hello.txt").write_text("Hello World") + m.Session = yes_session(str(tmp_path)) + resp = client.get("/api/files/load?session_id=1&path=hello.txt") + assert resp.status_code == 200 + data = resp.get_json() + assert data["success"] is True + assert data["content"] == "Hello World" + assert data["size"] == 11 + + +# ── GET /api/files/list ─────────────────────────────────────────────────────── + +class TestListFiles: + def test_missing_session_id(self, client): + resp = client.get("/api/files/list") + assert resp.status_code == 400 + + def test_session_not_found(self, client): + m.Session = no_session() + resp = client.get("/api/files/list?session_id=99") + assert resp.status_code == 404 + + def test_lists_files(self, client, tmp_path): + (tmp_path / "a.txt").write_text("a") + (tmp_path / "b.txt").write_text("b") + m.Session = yes_session(str(tmp_path)) + resp = client.get("/api/files/list?session_id=1") + assert resp.status_code == 200 + data = resp.get_json() + assert data["success"] is True + paths = [f["path"] for f in data["files"]] + assert "a.txt" in paths + assert "b.txt" in paths + + def test_excludes_hidden_files(self, client, tmp_path): + (tmp_path / ".env").write_text("SECRET=x") + (tmp_path / "visible.txt").write_text("ok") + m.Session = yes_session(str(tmp_path)) + resp = client.get("/api/files/list?session_id=1") + data = resp.get_json() + paths = [f["path"] for f in data["files"]] + assert ".env" not in paths + assert "visible.txt" in paths + + def test_returns_total_count(self, client, tmp_path): + (tmp_path / "x.txt").write_text("x") + m.Session = yes_session(str(tmp_path)) + resp = client.get("/api/files/list?session_id=1") + data = resp.get_json() + assert "total" in data + + +# ── DELETE /api/files/delete ────────────────────────────────────────────────── + +class TestDeleteFile: + def test_missing_session_id(self, client): + resp = client.delete("/api/files/delete", json={"path": "x.txt"}) + assert resp.status_code == 400 + + def test_file_not_found(self, client, tmp_path): + m.Session = yes_session(str(tmp_path)) + resp = client.delete("/api/files/delete", json={"session_id": 1, "path": "nope.txt"}) + assert resp.status_code == 404 + + def test_deletes_file(self, client, tmp_path): + (tmp_path / "del.txt").write_text("bye") + m.Session = yes_session(str(tmp_path)) + resp = client.delete("/api/files/delete", json={"session_id": 1, "path": "del.txt"}) + assert resp.status_code == 200 + assert not (tmp_path / "del.txt").exists() + + def test_path_traversal_blocked(self, client, tmp_path): + m.Session = yes_session(str(tmp_path)) + resp = client.delete("/api/files/delete", json={"session_id": 1, "path": "../../etc/hosts"}) + assert resp.status_code == 400 + + +# ── POST /api/files/mkdir ───────────────────────────────────────────────────── + +class TestMkdir: + def test_missing_session_id(self, client): + resp = client.post("/api/files/mkdir", json={"path": "src"}) + assert resp.status_code == 400 + + def test_creates_directory(self, client, tmp_path): + m.Session = yes_session(str(tmp_path)) + resp = client.post("/api/files/mkdir", json={"session_id": 1, "path": "src/contracts"}) + assert resp.status_code == 200 + assert (tmp_path / "src" / "contracts").is_dir() + + def test_idempotent_existing_dir(self, client, tmp_path): + (tmp_path / "existing").mkdir() + m.Session = yes_session(str(tmp_path)) + resp = client.post("/api/files/mkdir", json={"session_id": 1, "path": "existing"}) + assert resp.status_code == 200 + data = resp.get_json() + assert data["created"] is False