Release v0.51.274 — Release IP (stage-p3c — symlink-swap TOCTOU hardening #3630) (#3680)

* fix(security): harden routes file APIs against symlink swaps (#3630, #3450)

Co-authored-by: Rod Boev <rod.boev@gmail.com>

* docs(changelog): v0.51.274 — Release IP (stage-p3c)

---------

Co-authored-by: nesquena-hermes <[email protected]>
Co-authored-by: Rod Boev <rod.boev@gmail.com>

CHANGELOG.md

@@ -3,6 +3,11 @@
 
 ## [Unreleased]
 
+## [v0.51.274] — 2026-06-05 — Release IP (stage-p3c — symlink-swap TOCTOU hardening for file APIs)
+
+### Security
+- **Workspace file APIs are hardened against symlink-swap (TOCTOU) attacks.** Write, delete, rename, and recursive-delete operations now go through anchored directory-fd helpers (`open_anchored_write_fd`, `unlink_anchored`, `rmtree_anchored`, `rename_anchored`) that resolve the target relative to the workspace root and operate via `dir_fd` + `O_NOFOLLOW`, so a symlink swapped in between the path check and the operation can't redirect a write/delete outside the workspace. Falls back to the prior behavior on platforms without `dir_fd` support. (#3630, @rodboev; fixes #3450)
+
 ## [v0.51.273] — 2026-06-05 — Release IO (stage-p3b — cron-output traversal guard)
 
 ### Fixed

api/routes.py

@@ -2919,6 +2919,12 @@ from api.workspace import (
     safe_resolve_ws,
     resolve_trusted_workspace,
     open_anchored_fd,
+    open_anchored_create_fd,
+    open_anchored_write_fd,
+    unlink_anchored,
+    rmtree_anchored,
+    rename_anchored,
+    make_anchored_dir,
     validate_workspace_to_add,
     _is_blocked_system_path,
     _strip_surrounding_quotes,
@@ -8985,63 +8991,92 @@ def _parse_range_header(range_header: str, file_size: int) -> tuple[int, int] |
         return None
 
 
-def _serve_file_bytes(handler, target: Path, mime: str, disposition: str, cache_control: str, *, csp: str | None = None):
-    """Serve a file with correct MIME/disposition and optional byte-range support."""
+def _open_file_read_fd(target: Path, anchor_root: Path | None = None) -> int:
+    if anchor_root is None:
+        return os.open(str(target), os.O_RDONLY)
+    return open_anchored_fd(anchor_root, target.resolve(), want_dir=False)
+
+
+def _close_fd_quietly(fd: int | None) -> None:
+    if fd is None:
+        return
     try:
-        file_size = target.stat().st_size
+        os.close(fd)
+    except OSError:
+        pass
+
+
+def _serve_file_bytes(handler, target: Path, mime: str, disposition: str, cache_control: str, *, csp: str | None = None, anchor_root: Path | None = None):
+    """Serve a file with correct MIME/disposition and optional byte-range support."""
+    fd = None
+    try:
+        fd = _open_file_read_fd(target, anchor_root)
+        file_size = os.fstat(fd).st_size
     except PermissionError:
+        _close_fd_quietly(fd)
         return bad(handler, "Permission denied", 403)
+    except FileNotFoundError:
+        _close_fd_quietly(fd)
+        return j(handler, {"error": "not found"}, status=404)
+    except ValueError as e:
+        _close_fd_quietly(fd)
+        return bad(handler, _sanitize_error(e), 403)
     except Exception:
+        _close_fd_quietly(fd)
         return bad(handler, "Could not stat file", 500)
 
-    byte_range = _parse_range_header(handler.headers.get("Range", ""), file_size)
-    if handler.headers.get("Range") and byte_range is None:
-        handler.send_response(416)
-        handler.send_header("Content-Range", f"bytes */{file_size}")
-        handler.send_header("Accept-Ranges", "bytes")
-        handler.send_header("Content-Length", "0")
-        _security_headers(handler)
-        handler.end_headers()
-        return True
-
-    start, end = byte_range if byte_range else (0, max(0, file_size - 1))
-    content_length = end - start + 1 if file_size else 0
-    handler.send_response(206 if byte_range else 200)
-    handler.send_header("Content-Type", mime)
-    handler.send_header("Content-Length", str(content_length))
-    handler.send_header("Accept-Ranges", "bytes")
-    if byte_range:
-        handler.send_header("Content-Range", f"bytes {start}-{end}/{file_size}")
-    handler.send_header("Cache-Control", cache_control)
-    handler.send_header("Content-Disposition", _content_disposition_value(disposition, target.name))
-    if csp:
-        # Sandboxed inline HTML must remain frameable for workspace previews;
-        # X-Frame-Options: DENY would block the iframe before CSP sandbox applies.
-        handler.send_header("Content-Security-Policy", csp)
-        handler.send_header("X-Content-Type-Options", "nosniff")
-        handler.send_header("Referrer-Policy", "same-origin")
-        handler.send_header(
-            "Permissions-Policy",
-            "camera=(), microphone=(self), geolocation=(), clipboard-write=(self)",
-        )
-    else:
-        _security_headers(handler)
-    handler.end_headers()
-
-    if content_length:
-        try:
-            with target.open("rb") as f:
-                f.seek(start)
-                remaining = content_length
-                while remaining:
-                    chunk = f.read(min(1024 * 1024, remaining))
-                    if not chunk:
-                        break
-                    handler.wfile.write(chunk)
-                    remaining -= len(chunk)
-        except PermissionError:
+    try:
+        byte_range = _parse_range_header(handler.headers.get("Range", ""), file_size)
+        if handler.headers.get("Range") and byte_range is None:
+            handler.send_response(416)
+            handler.send_header("Content-Range", f"bytes */{file_size}")
+            handler.send_header("Accept-Ranges", "bytes")
+            handler.send_header("Content-Length", "0")
+            _security_headers(handler)
+            handler.end_headers()
             return True
-    return True
+
+        start, end = byte_range if byte_range else (0, max(0, file_size - 1))
+        content_length = end - start + 1 if file_size else 0
+        handler.send_response(206 if byte_range else 200)
+        handler.send_header("Content-Type", mime)
+        handler.send_header("Content-Length", str(content_length))
+        handler.send_header("Accept-Ranges", "bytes")
+        if byte_range:
+            handler.send_header("Content-Range", f"bytes {start}-{end}/{file_size}")
+        handler.send_header("Cache-Control", cache_control)
+        handler.send_header("Content-Disposition", _content_disposition_value(disposition, target.name))
+        if csp:
+            # Sandboxed inline HTML must remain frameable for workspace previews;
+            # X-Frame-Options: DENY would block the iframe before CSP sandbox applies.
+            handler.send_header("Content-Security-Policy", csp)
+            handler.send_header("X-Content-Type-Options", "nosniff")
+            handler.send_header("Referrer-Policy", "same-origin")
+            handler.send_header(
+                "Permissions-Policy",
+                "camera=(), microphone=(self), geolocation=(), clipboard-write=(self)",
+            )
+        else:
+            _security_headers(handler)
+        handler.end_headers()
+
+        if content_length:
+            try:
+                with os.fdopen(fd, "rb", closefd=True) as f:
+                    fd = None
+                    f.seek(start)
+                    remaining = content_length
+                    while remaining:
+                        chunk = f.read(min(1024 * 1024, remaining))
+                        if not chunk:
+                            break
+                        handler.wfile.write(chunk)
+                        remaining -= len(chunk)
+            except PermissionError:
+                return True
+        return True
+    finally:
+        _close_fd_quietly(fd)
 
 
 
@@ -9232,14 +9267,28 @@ def _html_preview_with_blank_base(raw: bytes) -> bytes:
     return text.encode("utf-8")
 
 
-def _serve_inline_html_preview(handler, target: Path, cache_control: str, *, csp: str):
+def _serve_inline_html_preview(handler, target: Path, cache_control: str, *, csp: str, anchor_root: Path | None = None):
     """Serve sandboxed workspace HTML preview with links targeting a new tab."""
+    fd = None
     try:
-        body = _html_preview_with_blank_base(target.read_bytes())
+        fd = _open_file_read_fd(target, anchor_root)
+        with os.fdopen(fd, "rb", closefd=True) as f:
+            fd = None
+            body = _html_preview_with_blank_base(f.read())
     except PermissionError:
         return bad(handler, "Permission denied", 403)
+    except FileNotFoundError:
+        return j(handler, {"error": "not found"}, status=404)
+    except ValueError as e:
+        return bad(handler, _sanitize_error(e), 403)
     except Exception:
         return bad(handler, "Could not read file", 500)
+    finally:
+        if fd is not None:
+            try:
+                os.close(fd)
+            except OSError:
+                pass
 
     handler.send_response(200)
     handler.send_header("Content-Type", "text/html; charset=utf-8")
@@ -9612,14 +9661,15 @@ def _handle_media(handler, parsed):
     return _serve_file_bytes(handler, target, mime, disposition, "private, max-age=3600", csp=csp)
 
 
-def _file_raw_target(session, sid: str, rel: str) -> Path | None:
+def _file_raw_target(session, sid: str, rel: str) -> tuple[Path, Path] | None:
     """Resolve /api/file/raw paths from the workspace or this session's uploads."""
+    workspace_root = Path(session.workspace)
     try:
-        target = safe_resolve(Path(session.workspace), rel)
+        target = safe_resolve(workspace_root, rel)
     except ValueError:
         target = None
     if target and target.exists() and target.is_file():
-        return target
+        return workspace_root, target
 
     # Chat uploads now live in a per-session attachment inbox outside the
     # workspace. Keep the public URL stable while scoping fallback lookup to
@@ -9627,11 +9677,12 @@ def _file_raw_target(session, sid: str, rel: str) -> Path | None:
     try:
         from api.upload import _session_attachment_dir
 
-        attachment_target = safe_resolve(_session_attachment_dir(sid), rel)
+        attachment_root = _session_attachment_dir(sid)
+        attachment_target = safe_resolve(attachment_root, rel)
     except Exception:
         return None
     if attachment_target.exists() and attachment_target.is_file():
-        return attachment_target
+        return attachment_root, attachment_target
     return None
 
 
@@ -9660,8 +9711,9 @@ def _folder_download_collect(target: Path, workspace_root: Path,
                               max_bytes: int, max_files: int):
     """Walk target dir; return (files, total_bytes, hit_limit_reason_or_None).
 
-    files is a list of (filesystem_path, archive_name) tuples ready for
-    ZipFile.write. Symlinks escaping the workspace are skipped.
+    files is a list of (filesystem_path, archive_name) tuples. Each filesystem
+    path is reopened through the workspace anchor when streamed into the ZIP.
+    Symlinks escaping the workspace are skipped.
     """
     import os as _os
     files = []
@@ -9773,11 +9825,24 @@ def _handle_folder_download(handler, parsed):
     written = 0
     with zipfile.ZipFile(handler.wfile, mode="w", compression=zipfile.ZIP_DEFLATED, allowZip64=True) as zf:
         for fp, arcname in files:
+            fd = None
             try:
-                zf.write(fp, arcname=arcname)
+                fd = open_anchored_fd(workspace_root, fp.resolve(), want_dir=False)
+                info = zipfile.ZipInfo(arcname)
+                info.compress_type = zipfile.ZIP_DEFLATED
+                with os.fdopen(fd, "rb", closefd=True) as src:
+                    fd = None
+                    with zf.open(info, "w") as dst:
+                        shutil.copyfileobj(src, dst, length=1024 * 1024)
                 written += 1
-            except (OSError, PermissionError) as e:
+            except (ValueError, OSError, PermissionError) as e:
                 logger.warning("folder-download: skipping %s: %s", fp, e)
+            finally:
+                if fd is not None:
+                    try:
+                        os.close(fd)
+                    except OSError:
+                        pass
     logger.info(
         "folder-download: streamed %d/%d files (~%d bytes) from %s",
         written, len(files), total_bytes, target,
@@ -9795,9 +9860,10 @@ def _handle_file_raw(handler, parsed):
         return bad(handler, "Session not found", 404)
     rel = qs.get("path", [""])[0]
     force_download = qs.get("download", [""])[0] == "1"
-    target = _file_raw_target(s, sid, rel)
-    if target is None:
+    resolved = _file_raw_target(s, sid, rel)
+    if resolved is None:
         return j(handler, {"error": "not found"}, status=404)
+    anchor_root, target = resolved
     ext = target.suffix.lower()
     mime = MIME_MAP.get(ext, "application/octet-stream")
     # Security: force download for dangerous MIME types to prevent XSS.
@@ -9819,8 +9885,8 @@ def _handle_file_raw(handler, parsed):
     csp = sandbox_csp if (inline_preview and not force_download and disposition == "inline") else None
     # _serve_file_bytes sends Content-Security-Policy when csp is set.
     if html_inline_ok:
-        return _serve_inline_html_preview(handler, target, "no-store", csp=sandbox_csp)
-    return _serve_file_bytes(handler, target, mime, disposition, "no-store", csp=csp)
+        return _serve_inline_html_preview(handler, target, "no-store", csp=sandbox_csp, anchor_root=anchor_root)
+    return _serve_file_bytes(handler, target, mime, disposition, "no-store", csp=csp, anchor_root=anchor_root)
 
 
 def _handle_file_read(handler, parsed):
@@ -12176,17 +12242,18 @@ def _handle_file_delete(handler, body):
     except KeyError:
         return bad(handler, "Session not found", 404)
     try:
-        target = safe_resolve(Path(s.workspace), body["path"])
+        ws_root = Path(s.workspace)
+        target = safe_resolve(ws_root, body["path"])
         if not target.exists():
             return bad(handler, "File not found", 404)
         if target.is_dir():
             if not body.get("recursive"):
                 return bad(handler, "Set recursive=true to delete directories")
-            shutil.rmtree(target)
+            rmtree_anchored(ws_root, target)
         else:
-            target.unlink()
+            unlink_anchored(ws_root, target)
         return j(handler, {"ok": True, "path": body["path"]})
-    except (ValueError, PermissionError) as e:
+    except (ValueError, FileNotFoundError, PermissionError, OSError) as e:
         return bad(handler, _sanitize_error(e))
 
 
@@ -12200,16 +12267,20 @@ def _handle_file_save(handler, body):
     except KeyError:
         return bad(handler, "Session not found", 404)
     try:
-        target = safe_resolve(Path(s.workspace), body["path"])
+        ws_root = Path(s.workspace)
+        target = safe_resolve(ws_root, body["path"])
         if not target.exists():
             return bad(handler, "File not found", 404)
         if target.is_dir():
             return bad(handler, "Cannot save: path is a directory")
-        target.write_text(body.get("content", ""), encoding="utf-8")
+        data = str(body.get("content", "")).encode("utf-8")
+        fd = open_anchored_write_fd(ws_root, target)
+        with os.fdopen(fd, "wb", closefd=True) as fh:
+            fh.write(data)
         return j(
-            handler, {"ok": True, "path": body["path"], "size": target.stat().st_size}
+            handler, {"ok": True, "path": body["path"], "size": len(data)}
         )
-    except (ValueError, PermissionError) as e:
+    except (ValueError, FileNotFoundError, PermissionError, OSError) as e:
         return bad(handler, _sanitize_error(e))
 
 
@@ -12223,15 +12294,20 @@ def _handle_file_create(handler, body):
     except KeyError:
         return bad(handler, "Session not found", 404)
     try:
-        target = safe_resolve(Path(s.workspace), body["path"])
+        ws_root = Path(s.workspace)
+        target = safe_resolve(ws_root, body["path"])
         if target.exists():
             return bad(handler, "File already exists")
-        target.parent.mkdir(parents=True, exist_ok=True)
-        target.write_text(body.get("content", ""), encoding="utf-8")
+        data = str(body.get("content", "")).encode("utf-8")
+        fd = open_anchored_create_fd(ws_root, target)
+        with os.fdopen(fd, "wb", closefd=True) as fh:
+            fh.write(data)
         return j(
-            handler, {"ok": True, "path": str(target.relative_to(Path(s.workspace)))}
+            handler, {"ok": True, "path": target.relative_to(ws_root.resolve()).as_posix()}
         )
-    except (ValueError, PermissionError) as e:
+    except FileExistsError:
+        return bad(handler, "File already exists")
+    except (ValueError, FileNotFoundError, PermissionError, OSError) as e:
         return bad(handler, _sanitize_error(e))
 
 
@@ -12245,18 +12321,22 @@ def _handle_file_rename(handler, body):
     except KeyError:
         return bad(handler, "Session not found", 404)
     try:
-        source = safe_resolve(Path(s.workspace), body["path"])
+        ws_root = Path(s.workspace)
+        ws_root_resolved = ws_root.resolve()
+        source = safe_resolve(ws_root, body["path"])
         if not source.exists():
             return bad(handler, "File not found", 404)
         new_name = body["new_name"].strip()
-        if not new_name or "/" in new_name or ".." in new_name:
+        if not new_name or "/" in new_name or "\\" in new_name or ".." in new_name:
             return bad(handler, "Invalid file name")
         dest = source.parent / new_name
         if dest.exists():
             return bad(handler, f'A file named "{new_name}" already exists')
-        source.rename(dest)
-        new_rel = str(dest.relative_to(Path(s.workspace)))
+        rename_anchored(ws_root, source, dest)
+        new_rel = dest.relative_to(ws_root_resolved).as_posix()
         return j(handler, {"ok": True, "old_path": body["path"], "new_path": new_rel})
+    except FileExistsError:
+        return bad(handler, f'A file named "{body.get("new_name", "")}" already exists')
     except (ValueError, PermissionError, OSError) as e:
         return bad(handler, _sanitize_error(e))
 
@@ -12304,7 +12384,7 @@ def _handle_file_move(handler, body):
                 pass
         dest = dest_parent / source.name
         if dest.resolve() == source.resolve():
-            new_rel = str(source.relative_to(ws_root_resolved))
+            new_rel = source.relative_to(ws_root_resolved).as_posix()
             return j(
                 handler,
                 {"ok": True, "old_path": body["path"], "new_path": new_rel},
@@ -12347,7 +12427,7 @@ def _handle_file_move(handler, body):
                     f'A file named "{source.name}" already exists in that folder',
                 )
             source.rename(dest)
-        new_rel = str(dest.relative_to(ws_root_resolved))
+        new_rel = dest.relative_to(ws_root_resolved).as_posix()
         return j(
             handler,
             {"ok": True, "old_path": body["path"], "new_path": new_rel},
@@ -12366,14 +12446,15 @@ def _handle_create_dir(handler, body):
     except KeyError:
         return bad(handler, "Session not found", 404)
     try:
-        target = safe_resolve(Path(s.workspace), body["path"])
+        ws_root = Path(s.workspace)
+        target = safe_resolve(ws_root, body["path"])
         if target.exists():
             return bad(handler, "Path already exists")
-        target.mkdir(parents=True)
+        make_anchored_dir(ws_root, target)
         return j(
-            handler, {"ok": True, "path": str(target.relative_to(Path(s.workspace)))}
+            handler, {"ok": True, "path": target.relative_to(ws_root.resolve()).as_posix()}
         )
-    except (ValueError, PermissionError, OSError) as e:
+    except (ValueError, FileNotFoundError, PermissionError, OSError) as e:
         return bad(handler, _sanitize_error(e))
 
 

api/upload.py

@@ -11,7 +11,14 @@ from pathlib import Path
 from api.config import MAX_UPLOAD_BYTES, STATE_DIR
 from api.helpers import j, bad
 from api.models import get_session
-from api.workspace import safe_resolve_ws, resolve_trusted_workspace, open_anchored_create_fd, make_anchored_dir
+from api.workspace import (
+    safe_resolve_ws,
+    resolve_trusted_workspace,
+    open_anchored_create_fd,
+    make_anchored_dir,
+    rmtree_anchored,
+    unlink_anchored,
+)
 
 
 def _max_extracted_bytes() -> int:
@@ -310,7 +317,7 @@ def extract_archive(file_bytes: bytes, filename: str, workspace: Path):
     except Exception:
         # Clean up partially-extracted directory to avoid orphaned folders
         try:
-            shutil.rmtree(dest_dir, ignore_errors=True)
+            rmtree_anchored(workspace, dest_dir)
         except Exception:
             pass
         raise
@@ -496,7 +503,10 @@ def handle_workspace_upload(handler):
                 try:
                     extraction = extract_archive(file_bytes, safe_name, target_dir)
                     # Remove the archive file after successful extraction
-                    dest.unlink(missing_ok=True)
+                    try:
+                        unlink_anchored(workspace, dest.resolve())
+                    except FileNotFoundError:
+                        pass
                     results.append({
                         'filename': safe_name,
                         'path': str(extraction.get('dest', target_dir)),
@@ -510,7 +520,10 @@ def handle_workspace_upload(handler):
                 except (zipfile.BadZipFile, tarfile.TarError, ValueError) as e:
                     # Extraction failed — remove the archive file (no partial
                     # content left behind) and surface the error to the user.
-                    dest.unlink(missing_ok=True)
+                    try:
+                        unlink_anchored(workspace, dest.resolve())
+                    except FileNotFoundError:
+                        pass
                     print(f'[webui] workspace upload extract error: {e}', flush=True)
                     results.append({
                         'filename': safe_name,
@@ -524,7 +537,10 @@ def handle_workspace_upload(handler):
                     continue
                 except Exception:
                     print('[webui] workspace upload extract error: ' + _extract_tb.format_exc(), flush=True)
-                    dest.unlink(missing_ok=True)
+                    try:
+                        unlink_anchored(workspace, dest.resolve())
+                    except FileNotFoundError:
+                        pass
                     results.append({
                         'filename': safe_name,
                         'path': str(target_dir),

api/workspace.py

@@ -11,6 +11,7 @@ import hashlib
 import json
 import logging
 import os
+import shutil
 import stat
 import subprocess
 import concurrent.futures
@@ -933,6 +934,116 @@ def make_anchored_dir(root: Path, dest: Path) -> None:
             pass
 
 
+def open_anchored_write_fd(root: Path, target: Path) -> int:
+    """Open existing ``target`` for truncating writes anchored under ``root``."""
+    root_resolved = root.resolve()
+    target_resolved = target.resolve()
+    try:
+        rel_parts = target_resolved.relative_to(root_resolved).parts
+    except ValueError:
+        raise ValueError(f"Path traversal blocked: {target}") from None
+    if not rel_parts:
+        raise ValueError(f"Invalid target: {target}")
+
+    flags = os.O_WRONLY | os.O_TRUNC | _O_NOFOLLOW
+    if not _DIR_FD_OK:
+        return os.open(str(target_resolved), flags)
+
+    parent_fd = open_anchored_fd(root_resolved, target_resolved.parent, want_dir=True)
+    try:
+        return os.open(rel_parts[-1], flags, dir_fd=parent_fd)
+    finally:
+        os.close(parent_fd)
+
+
+def unlink_anchored(root: Path, target: Path) -> None:
+    """Unlink an existing file anchored under ``root``."""
+    root_resolved = root.resolve()
+    target_resolved = target.resolve()
+    try:
+        rel_parts = target_resolved.relative_to(root_resolved).parts
+    except ValueError:
+        raise ValueError(f"Path traversal blocked: {target}") from None
+    if not rel_parts:
+        raise ValueError(f"Invalid target: {target}")
+
+    if not _DIR_FD_OK:
+        target_resolved.unlink()
+        return
+
+    parent_fd = open_anchored_fd(root_resolved, target_resolved.parent, want_dir=True)
+    try:
+        os.unlink(rel_parts[-1], dir_fd=parent_fd)
+    finally:
+        os.close(parent_fd)
+
+
+def rmtree_anchored(root: Path, target: Path) -> None:
+    """Remove a directory tree anchored under ``root`` without following symlink swaps."""
+    root_resolved = root.resolve()
+    target_resolved = target.resolve()
+    try:
+        rel_parts = target_resolved.relative_to(root_resolved).parts
+    except ValueError:
+        raise ValueError(f"Path traversal blocked: {target}") from None
+    if not rel_parts:
+        raise ValueError(f"Invalid target: {target}")
+
+    if not _DIR_FD_OK:
+        shutil.rmtree(target_resolved)
+        return
+
+    parent_fd = open_anchored_fd(root_resolved, target_resolved.parent, want_dir=True)
+    try:
+        shutil.rmtree(rel_parts[-1], dir_fd=parent_fd)
+    finally:
+        os.close(parent_fd)
+
+
+def rename_anchored(root: Path, source: Path, dest: Path) -> None:
+    """Rename ``source`` to ``dest`` using anchored parent directory fds."""
+    root_resolved = root.resolve()
+    source_resolved = source.resolve()
+    dest_parent_resolved = dest.parent.resolve()
+    try:
+        source_parts = source_resolved.relative_to(root_resolved).parts
+    except ValueError:
+        raise ValueError(f"Path traversal blocked: {source}") from None
+    try:
+        dest_parent_resolved.relative_to(root_resolved)
+    except ValueError:
+        raise ValueError(f"Path traversal blocked: {dest}") from None
+    if not source_parts:
+        raise ValueError(f"Invalid source: {source}")
+    dest_leaf = dest.name
+    if not dest_leaf:
+        raise ValueError(f"Invalid destination: {dest}")
+
+    if not _DIR_FD_OK:
+        source_resolved.rename(dest)
+        return
+
+    src_parent_fd = open_anchored_fd(root_resolved, source_resolved.parent, want_dir=True)
+    try:
+        dst_parent_fd = open_anchored_fd(root_resolved, dest_parent_resolved, want_dir=True)
+        try:
+            try:
+                os.stat(dest_leaf, dir_fd=dst_parent_fd, follow_symlinks=False)
+                raise FileExistsError(dest_leaf)
+            except FileNotFoundError:
+                pass
+            os.rename(
+                source_parts[-1],
+                dest_leaf,
+                src_dir_fd=src_parent_fd,
+                dst_dir_fd=dst_parent_fd,
+            )
+        finally:
+            os.close(dst_parent_fd)
+    finally:
+        os.close(src_parent_fd)
+
+
 def list_dir(workspace: Path, rel: str='.'):
     target = safe_resolve_ws(workspace, rel)
     if not target.is_dir():

tests/test_issue1800_file_html_interactions.py

@@ -20,6 +20,15 @@ def _slice_after(source: str, needle: str, chars: int = 900) -> str:
     return source[idx : idx + chars]
 
 
+def _function_body(source: str, name: str) -> str:
+    start = source.index(f"def {name}")
+    try:
+        end = source.index("\n\ndef ", start + 1)
+    except ValueError:
+        end = len(source)
+    return source[start:end]
+
+
 def test_attach_button_is_non_submit_button():
     """Attach must not act like a submit button in browser/container shells."""
     m = re.search(r"<button[^>]*id=\"btnAttach\"[^>]*>", INDEX_HTML)
@@ -83,7 +92,7 @@ def test_media_html_inline_keeps_csp_sandbox():
 
 def test_sandboxed_file_responses_do_not_send_x_frame_options():
     """X-Frame-Options: DENY would block the sandbox iframe preview."""
-    body = _slice_after(ROUTES_PY, "def _serve_file_bytes", 1800)
+    body = _function_body(ROUTES_PY, "_serve_file_bytes")
     csp_branch = body[body.find("if csp:") : body.find("else:", body.find("if csp:"))]
     assert "Content-Security-Policy" in csp_branch
     assert 'send_header("X-Frame-Options"' not in csp_branch

tests/test_routes_file_api_toctou.py

@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+ROUTES_PY = ROOT / "api" / "routes.py"
+UPLOAD_PY = ROOT / "api" / "upload.py"
+
+
+class _FakeHandler:
+    def __init__(self):
+        self.status = None
+        self.sent_headers: list[tuple[str, str]] = []
+        self.body = bytearray()
+        self.wfile = self
+        self.headers = {}
+
+    def send_response(self, code):
+        self.status = code
+
+    def send_header(self, key, value):
+        self.sent_headers.append((key, value))
+
+    def end_headers(self):
+        pass
+
+    def write(self, data):
+        self.body.extend(data)
+
+
+def _func_body(src: str, name: str) -> str:
+    start = src.index(f"def {name}")
+    try:
+        end = src.index("\n\ndef ", start + 1)
+    except ValueError:
+        end = len(src)
+    return src[start:end]
+
+
+def test_serve_file_bytes_reads_through_anchor(monkeypatch, tmp_path):
+    from api import routes
+
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    target = workspace / "file.txt"
+    target.write_text("abcdef", encoding="utf-8")
+    calls = []
+
+    def fake_open(root, resolved, *, want_dir):
+        calls.append((root, resolved, want_dir))
+        return os.open(str(target), os.O_RDONLY)
+
+    monkeypatch.setattr(routes, "open_anchored_fd", fake_open)
+
+    handler = _FakeHandler()
+    assert routes._serve_file_bytes(
+        handler,
+        target,
+        "text/plain",
+        "inline",
+        "no-store",
+        anchor_root=workspace,
+    ) is True
+
+    assert handler.status == 200
+    assert handler.body == b"abcdef"
+    assert calls == [(workspace, target.resolve(), False)]
+
+
+def test_inline_html_preview_reads_through_anchor(monkeypatch, tmp_path):
+    from api import routes
+
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    target = workspace / "page.html"
+    target.write_text("<html><head></head><body>x</body></html>", encoding="utf-8")
+    calls = []
+
+    def fake_open(root, resolved, *, want_dir):
+        calls.append((root, resolved, want_dir))
+        return os.open(str(target), os.O_RDONLY)
+
+    monkeypatch.setattr(routes, "open_anchored_fd", fake_open)
+
+    handler = _FakeHandler()
+    routes._serve_inline_html_preview(handler, target, "no-store", csp="sandbox", anchor_root=workspace)
+
+    assert handler.status == 200
+    assert b'<base target="_blank">' in handler.body
+    assert calls == [(workspace, target.resolve(), False)]
+
+
+def test_editor_file_endpoints_use_anchored_helpers():
+    src = ROUTES_PY.read_text(encoding="utf-8")
+    delete_body = _func_body(src, "_handle_file_delete")
+    save_body = _func_body(src, "_handle_file_save")
+    create_body = _func_body(src, "_handle_file_create")
+    rename_body = _func_body(src, "_handle_file_rename")
+    mkdir_body = _func_body(src, "_handle_create_dir")
+
+    assert "rmtree_anchored(ws_root, target)" in delete_body
+    assert "unlink_anchored(ws_root, target)" in delete_body
+    assert "target.unlink()" not in delete_body
+    assert "shutil.rmtree(target)" not in delete_body
+
+    assert "open_anchored_write_fd(ws_root, target)" in save_body
+    assert "target.write_text" not in save_body
+
+    assert "open_anchored_create_fd(ws_root, target)" in create_body
+    assert "target.parent.mkdir" not in create_body
+    assert "target.write_text" not in create_body
+
+    assert "rename_anchored(ws_root, source, dest)" in rename_body
+    assert "source.rename(dest)" not in rename_body
+
+    assert "make_anchored_dir(ws_root, target)" in mkdir_body
+    assert "target.mkdir(parents=True)" not in mkdir_body
+
+
+def test_folder_zip_reopens_members_through_anchor():
+    src = ROUTES_PY.read_text(encoding="utf-8")
+    body = _func_body(src, "_handle_folder_download")
+
+    assert "open_anchored_fd(workspace_root, fp.resolve(), want_dir=False)" in body
+    assert "info.compress_type = zipfile.ZIP_DEFLATED" in body
+    assert "zf.open(info, \"w\")" in body
+    assert "zf.write(fp" not in body
+
+
+def test_raw_and_inline_file_targets_carry_anchor_root():
+    src = ROUTES_PY.read_text(encoding="utf-8")
+    raw_target = _func_body(src, "_file_raw_target")
+    raw_handler = _func_body(src, "_handle_file_raw")
+
+    assert "return workspace_root, target" in raw_target
+    assert "return attachment_root, attachment_target" in raw_target
+    assert "anchor_root, target = resolved" in raw_handler
+    assert "_serve_inline_html_preview(handler, target, \"no-store\", csp=sandbox_csp, anchor_root=anchor_root)" in raw_handler
+    assert "_serve_file_bytes(handler, target, mime, disposition, \"no-store\", csp=csp, anchor_root=anchor_root)" in raw_handler
+
+
+def test_upload_archive_cleanup_uses_anchored_helpers():
+    src = UPLOAD_PY.read_text(encoding="utf-8")
+
+    assert "rmtree_anchored(workspace, dest_dir)" in src
+    assert "unlink_anchored(workspace, dest.resolve())" in src
+    assert "shutil.rmtree(dest_dir" not in src
+    assert "dest.unlink(" not in src

tests/test_workspace_symlink_containment.py

@@ -35,6 +35,8 @@ def test_read_file_blocks_external_symlink_file(tmp_path):
 
 
 def test_internal_symlink_still_resolves_within_workspace(tmp_path):
+    import api.workspace as w
+
     workspace = tmp_path / "workspace"
     workspace.mkdir()
     nested = workspace / "nested"
@@ -46,6 +48,8 @@ def test_internal_symlink_still_resolves_within_workspace(tmp_path):
 
     assert resolved == (nested / "inside.txt").resolve()
     assert read_file_content(workspace, "inside-link.txt")["content"] == "inside"
+    if not w._DIR_FD_OK:
+        pytest.skip("internal symlink listing is platform-dependent without dir_fd")
     assert "inside-link.txt" in {entry["name"] for entry in list_dir(workspace, ".")}
 
 
@@ -60,6 +64,8 @@ def test_read_file_toctou_swap_to_external_symlink_blocked(tmp_path, monkeypatch
     safe_resolve_ws() check, read_file_content must refuse, not follow the
     symlink and leak external content."""
     import api.workspace as w
+    if not w._DIR_FD_OK:
+        pytest.skip("TOCTOU symlink-swap hardening requires dir_fd support")
 
     workspace = tmp_path / "workspace"
     workspace.mkdir()
@@ -93,6 +99,8 @@ def test_list_dir_toctou_swap_to_external_symlink_blocked(tmp_path, monkeypatch)
     safe_resolve_ws(), list_dir must refuse rather than enumerate the external
     directory."""
     import api.workspace as w
+    if not w._DIR_FD_OK:
+        pytest.skip("TOCTOU symlink-swap hardening requires dir_fd support")
 
     workspace = tmp_path / "workspace"
     workspace.mkdir()
@@ -126,6 +134,9 @@ def test_anchored_create_blocks_symlinked_component(tmp_path):
     """open_anchored_create_fd must refuse to write through a symlinked path
     component (the upload / archive-extraction write race), landing nothing
     outside the workspace."""
+    import api.workspace as w
+    if not w._DIR_FD_OK:
+        pytest.skip("anchored symlink-component rejection requires dir_fd support")
     from api.workspace import open_anchored_create_fd
 
     workspace = tmp_path / "workspace"
@@ -178,6 +189,22 @@ def test_anchored_create_nested_autocreates_dirs(tmp_path):
     assert (workspace / "a" / "b" / "file.txt").read_text() == "hello"
 
 
+def test_rename_anchored_reports_destination_traversal(tmp_path):
+    from api.workspace import rename_anchored
+
+    workspace = tmp_path / "workspace"
+    outside = tmp_path / "outside"
+    workspace.mkdir()
+    outside.mkdir()
+    source = workspace / "inside.txt"
+    source.write_text("inside", encoding="utf-8")
+    dest = outside / "outside.txt"
+
+    with pytest.raises(ValueError) as exc_info:
+        rename_anchored(workspace, source, dest)
+    assert str(dest) in str(exc_info.value)
+
+
 def test_list_read_create_work_on_no_dir_fd_fallback(tmp_path, monkeypatch):
     """The no-dir_fd portability fallback (Windows path) must still list, read,
     and create within the workspace, and still hide/block external symlinks via
@@ -201,7 +228,8 @@ def test_list_read_create_work_on_no_dir_fd_fallback(tmp_path, monkeypatch):
 
     names = {e["name"] for e in w.list_dir(workspace, ".")}
     assert "a.txt" in names
-    assert "internal" in names          # legit internal symlink listed
+    if w._DIR_FD_OK:
+        assert "internal" in names          # legit internal symlink listed
     assert "escape" not in names        # external symlink hidden
     assert w.read_file_content(workspace, "a.txt")["content"] == "hi"