diff --git a/l4d2web/l4d2web/routes/files_routes.py b/l4d2web/l4d2web/routes/files_routes.py index 44a3c2c..2d4340c 100644 --- a/l4d2web/l4d2web/routes/files_routes.py +++ b/l4d2web/l4d2web/routes/files_routes.py @@ -10,8 +10,6 @@ Read-only endpoints (any overlay): is refused. Mutating endpoints (only `overlay.type == 'files'`, owner or admin): -- `GET /overlays//files/content?path=` — JSON `{path, content}` for - an editable text file, 415 if not editable. - `POST /overlays//files/save` — JSON `{path, content, new_path?}`, text-mode write with optional atomic rename. - `POST /overlays//files/replace` — multipart `path`, `file`, @@ -243,24 +241,6 @@ def _load_file_for_editing(overlay_id: int, sub_path: str, user): return overlay, target, content, False, len(content.encode("utf-8")) -@bp.get("/overlays//files/content") -@require_login -def overlay_file_content(overlay_id: int): - """Return `{path, content}` for an editable text file.""" - user = current_user() - assert user is not None - sub_path = request.args.get("path", "") - - result = _load_file_for_editing(overlay_id, sub_path, user) - if isinstance(result, Response): - return result - _overlay, _target, content, is_binary, _byte_count = result - - if is_binary: - return Response("not editable", status=415) - return jsonify({"path": sub_path, "content": content}) - - @bp.get("/overlays//files/edit") @require_login def overlay_file_edit_page(overlay_id: int): @@ -355,6 +335,40 @@ def _validate_save_content(content: str) -> Response | None: return None +def _apply_optional_rename(overlay, path: str, new_path: str | None): + """Resolve the write target for save / replace, atomically renaming + when `new_path` differs from `path`. Returns: + * (write_target, echo_path) on success — write_target is the + resolved Path the caller should write to; echo_path is the + canonical relative path to echo back in the response + * a Flask Response on failure: 422 (path validation) or 409 + (rename destination already exists) + + The atomic rename happens INSIDE this helper. Save's "destination + is not a file" 409 (creation branch) stays in the caller, since + /replace doesn't enforce that check.""" + if new_path is not None and new_path != path: + try: + src_path, dst_path = safe_resolve_for_move(overlay.path, path, new_path) + except ValueError as exc: + return Response(str(exc), status=422) + if dst_path.exists(): + return Response("destination already exists", status=409) + dst_path.parent.mkdir(parents=True, exist_ok=True) + os.rename(src_path, dst_path) + return dst_path, new_path + try: + write_target = safe_resolve_for_write(overlay.path, path) + except ValueError as exc: + return Response(str(exc), status=422) + write_target.parent.mkdir(parents=True, exist_ok=True) + return write_target, path + + +def _is_existing_file(path) -> bool: + return path.is_file() and not path.is_symlink() + + @bp.post("/overlays//files/save") @require_login def overlay_file_save(overlay_id: int): @@ -380,38 +394,23 @@ def overlay_file_save(overlay_id: int): return result overlay = result - try: - write_target = safe_resolve_for_write(overlay.path, new_path or path) - except ValueError as exc: - return Response(str(exc), status=422) + rename_result = _apply_optional_rename(overlay, path, new_path) + if isinstance(rename_result, Response): + return rename_result + write_target, echo_path = rename_result - # Rename branch: source must exist, dst must not collide. - if new_path is not None and new_path != path: - try: - src_path, dst_path = safe_resolve_for_move(overlay.path, path, new_path) - except ValueError as exc: - return Response(str(exc), status=422) - if dst_path.exists(): - return Response("destination already exists", status=409) - dst_path.parent.mkdir(parents=True, exist_ok=True) - os.rename(src_path, dst_path) - write_target = dst_path - else: - write_target.parent.mkdir(parents=True, exist_ok=True) - # Creation branch: must not collide with an existing path. - if write_target.exists() and not _is_existing_file(write_target): - return Response("destination is not a file", status=409) + # Creation branch (no rename): must not collide with an existing + # non-file (e.g. a directory at the same path). + if (new_path is None or new_path == path) \ + and write_target.exists() and not _is_existing_file(write_target): + return Response("destination is not a file", status=409) try: write_target.write_text(content, encoding="utf-8") except OSError as exc: return Response(f"write failed: {exc}", status=500) - return jsonify({"path": new_path or path}) - - -def _is_existing_file(path) -> bool: - return path.is_file() and not path.is_symlink() + return jsonify({"path": echo_path}) @bp.post("/overlays//files/replace") @@ -433,24 +432,10 @@ def overlay_file_replace(overlay_id: int): return result overlay = result - if new_path and new_path != path: - try: - src_path, dst_path = safe_resolve_for_move(overlay.path, path, new_path) - except ValueError as exc: - return Response(str(exc), status=422) - if dst_path.exists(): - return Response("destination already exists", status=409) - dst_path.parent.mkdir(parents=True, exist_ok=True) - os.rename(src_path, dst_path) - write_target = dst_path - echo_path = new_path - else: - try: - write_target = safe_resolve_for_write(overlay.path, path) - except ValueError as exc: - return Response(str(exc), status=422) - write_target.parent.mkdir(parents=True, exist_ok=True) - echo_path = path + rename_result = _apply_optional_rename(overlay, path, new_path) + if isinstance(rename_result, Response): + return rename_result + write_target, echo_path = rename_result return _stream_upload_into(upload, write_target, echo_path) diff --git a/l4d2web/tests/test_overlay_files_routes.py b/l4d2web/tests/test_overlay_files_routes.py index b17c967..d6b90b3 100644 --- a/l4d2web/tests/test_overlay_files_routes.py +++ b/l4d2web/tests/test_overlay_files_routes.py @@ -582,47 +582,6 @@ def _csrf_headers(): return {"X-CSRF-Token": "test-token"} -# ---- /content ------------------------------------------------------------- - - -def test_content_returns_text(app, left4me_root: Path) -> None: - user_id = _make_user() - overlay_id = _make_files_overlay(left4me_root, user_id=user_id, name="files") - overlay_dir = left4me_root / "overlays" / str(overlay_id) - (overlay_dir / "motd.txt").write_text("welcome") - - client = _client_for(app, user_id) - r = client.get(f"/overlays/{overlay_id}/files/content?path=motd.txt") - - assert r.status_code == 200 - assert r.get_json() == {"path": "motd.txt", "content": "welcome"} - - -def test_content_returns_415_for_binary(app, left4me_root: Path) -> None: - user_id = _make_user() - overlay_id = _make_files_overlay(left4me_root, user_id=user_id, name="files") - overlay_dir = left4me_root / "overlays" / str(overlay_id) - (overlay_dir / "image.bin").write_bytes(b"\x00\x01\x02") - - client = _client_for(app, user_id) - r = client.get(f"/overlays/{overlay_id}/files/content?path=image.bin") - - assert r.status_code == 415 - - -def test_content_404_for_non_files_overlay(app, left4me_root: Path) -> None: - """`content` is gated to type=='files' to keep the new editor scoped.""" - user_id = _make_user() - overlay_id = _make_overlay(left4me_root, user_id=user_id, name="ws") # type='script' helper - overlay_dir = left4me_root / "overlays" / str(overlay_id) - (overlay_dir / "motd.txt").write_text("hi") - - client = _client_for(app, user_id) - r = client.get(f"/overlays/{overlay_id}/files/content?path=motd.txt") - - assert r.status_code == 404 - - # ---- /save --------------------------------------------------------------- @@ -1064,7 +1023,6 @@ def test_mutating_endpoints_404_for_workshop_overlay(app, left4me_root: Path) -> ("post", f"/overlays/{overlay_id}/files/delete", {"data": {"path": "x", "csrf_token": "test-token"}}), ("get", f"/overlays/{overlay_id}/files/download_zip?path=", {}), - ("get", f"/overlays/{overlay_id}/files/content?path=x.txt", {}), ] for method, url, kwargs in paths: kwargs = dict(kwargs)