RigShare 1.3.2: access checks without a live room, refuse Windows alias paths, same-origin login redirect

* RigShare 1.3.1: refuse folder paths outside the workspace (Windows)

* Tests: escape test no longer assumes Windows path semantics

CHANGELOG.md

@@ -1,5 +1,30 @@
 # Changelog
 
+## 1.3.2 (2026-10-08)
+
+Access rules tightened, following the review of the ComfyUI Manager listing.
+
+### Fixed
+
+- Snapshots of a workflow could be listed and read by an account with no access to its folder whenever nobody had that workflow open. Snapshot access now follows the workflow's folder (private and restricted shared folders included), open or not
+- Joining a live workflow seeded a new room from the caller's document before the access check. Access is checked first, and only someone who can edit the workflow can open a new room
+- Windows path segments that end in a dot or space, or use an 8.3 short name, are now refused instead of being silently dropped, so a path can no longer be classified as a different folder than the one Windows opens
+- After signing in, the login page only follows its `next` redirect when it resolves to the same origin
+
+## 1.3.1 (2026-10-08)
+
+Security hardening of the workspace folder routes, mainly for Windows servers.
+
+### Fixed
+
+- Folder rename, move, delete and create (and the workspace listing) could be pointed outside the workflows folder with a Windows drive prefix (`C:`), which let `rmtree` / `rename` act on folders such as models or custom_nodes. Every path is now checked before it touches the disk: segments with a colon, backslash tricks or `..` are refused, and the resolved path must stay inside the workflows folder
+- On Windows, case and aliases of a path were not seen by the access checks: `Users/alice`, `users./alice` or `USERS~1/alice` opened Alice's private folder as if it were common, and `Shared/<restricted folder>` skipped its access list. Paths are now matched to the name on disk and trailing dots, spaces and 8.3 short names are dropped before any permission check
+- New usernames and folder names can no longer end in a dot or space, start with a dot, or be a Windows device name (`CON`, `NUL`, `COM1`...). Usernames that differ only by case count as the same name, since they share one folder on Windows. Existing accounts and folders are untouched
+
+### Added
+
+- Tests for the JWT module, the server stats, and the folder routes refusing paths outside the workspace
+
 ## 1.3.0 (2026-10-04)
 
 A Performance panel, a floating monitor that docks, and Manager in its own tab.

README.md

@@ -15,6 +15,14 @@ A custom node that turns a single ComfyUI server into a shared workspace. Everyo
 
 ---
 
+<div align="center">
+
+<img src="docs/images/live-cursors.gif" width="900" alt="Two people editing the same workflow at once, each seeing the other's named cursor and selection">
+
+<sub>Two accounts, one workflow: each person sees the other's cursor, selection and edits live.</sub>
+
+</div>
+
 ## What it does
 
 - asks everyone to sign in before ComfyUI loads, and creates the admin account on first run

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "comfyui-rigshare"
 description = "Turn one ComfyUI server into a shared team workspace. Sign-in with accounts, permissions (edit, queue, Manager, admin) and personal API keys, enforced by the server. A private folder for each person plus shared folders with access lists, with Save asking where. Live co-editing of every shared workflow, with cursors, follow mode and conflict-free merges, in the graph or App view. Per-workflow history with a snapshot on every save, team chat with its whole history, and a CPU, RAM, GPU and queue monitor."
-version = "1.3.0"
+version = "1.3.2"
 license = { file = "LICENSE" }
 dependencies = []
 

rigshare_core/hub.py

@@ -334,8 +334,12 @@ class Hub:
         Access follows the workflow's folder (restricted shared folders), within
         the person's account permissions.
         """
-        if self.workspace and room.key.startswith("file:workflows/"):
-            return self.workspace.role(room.key[len("file:workflows/"):], kind, key, perms)
+        return self.access_for_key(room.key, kind, key, perms)
+
+    def access_for_key(self, room_key, kind, key, perms):
+        """Same as access(), from a room key: works before (or without) a live room."""
+        if self.workspace and room_key.startswith("file:workflows/"):
+            return self.workspace.role(room_key[len("file:workflows/"):], kind, key, perms)
         return "edit" if (perms.get("edit") or perms.get("admin")) else "view"
 
     def client_access(self, room, client):
@@ -556,16 +560,25 @@ class Hub:
             else:
                 room = self.rooms.get(key)
                 created = False
-                if room is None:
-                    doc = data.get("doc") if isinstance(data.get("doc"), dict) else {}
-                    room = Room(key, clean_text(data.get("name"), 120) or key, "file", doc)
-                    room.dirty = True
-                    self.rooms[key] = room
-                    created = True
-                if self.client_access(room, client) is None:
+                # Check access before anything is created: only someone who may edit the
+                # workflow can open a room and seed it with their document.
+                role = self.access_for_key(key, client.kind, client.key, client.perms)
+                if role is None:
                     client.room = None
-                    await self.send(client, {"type": "room_denied", "room": key, "name": room.name})
+                    await self.send(client, {"type": "room_denied", "room": key,
+                                             "name": room.name if room else clean_text(data.get("name"), 120) or key})
+                elif room is None and role != "edit":
+                    client.room = None
+                    await self.send(client, {"type": "room_denied", "room": key,
+                                             "name": clean_text(data.get("name"), 120) or key,
+                                             "reason": "Nobody has opened this workflow for editing yet."})
                 else:
+                    if room is None:
+                        doc = data.get("doc") if isinstance(data.get("doc"), dict) else {}
+                        room = Room(key, clean_text(data.get("name"), 120) or key, "file", doc)
+                        room.dirty = True
+                        self.rooms[key] = room
+                        created = True
                     client.room = key
                     room.updated = time.time()
                     await self.send(client, {"type": "room", "room": key, "name": room.name, "version": room.version,

rigshare_core/login.html

@@ -84,6 +84,13 @@
   const err = document.getElementById("e");
   const btn = document.getElementById("b");
   const next = new URLSearchParams(location.search).get("next") || "/";
+  // Follow `next` only when it resolves to this same origin (blocks //host, /\host, scheme tricks).
+  const safeNext = () => {
+    try {
+      const url = new URL(next, location.origin);
+      return url.origin === location.origin ? url.pathname + url.search + url.hash : "/";
+    } catch { return "/"; }
+  };
   if (SETUP) {
     document.title = "Set up · {{SERVER_NAME}}";
     document.getElementById("sub").textContent = "First-time setup";
@@ -112,7 +119,7 @@
       if (res.status === 409) { location.reload(); return; } // someone finished setup meanwhile
       if (!res.ok) throw new Error(data.error || "Sign in failed");
       try { localStorage.setItem("rigshare.token", data.token); } catch {}
-      location.replace(next.startsWith("/") && !next.startsWith("//") ? next : "/");
+      location.replace(safeNext());
     } catch (e) {
       err.textContent = e.message;
       btn.disabled = false;

rigshare_core/routes.py

@@ -14,7 +14,7 @@ from collections import defaultdict
 
 from aiohttp import web
 
-from .workspace import Workspace, norm
+from .workspace import InvalidPath, Workspace, norm as _norm
 
 log = logging.getLogger("ComfyUI-RigShare")
 
@@ -148,6 +148,13 @@ def setup(server, store, hub):
     def error(message, status=400):
         return web.json_response({"error": message}, status=status)
 
+    def norm(path):
+        """workspace.norm, answering 403 for a path that is refused instead of adjusted."""
+        try:
+            return _norm(path)
+        except InvalidPath:
+            raise web.HTTPForbidden(text='{"error": "Invalid path"}', content_type="application/json")
+
     def set_cookie(request, response, token):
         secure = request.secure or request.headers.get("X-Forwarded-Proto") == "https"
         response.set_cookie(COOKIE, token, max_age=30 * 24 * 3600, path="/", httponly=True,
@@ -580,8 +587,7 @@ def setup(server, store, hub):
         if not key:
             require_admin(request)
             return web.json_response(store.list_snapshots())
-        room = hub.rooms.get(key)
-        if room and not room_access(request, room):
+        if hub.access_for_key(key, *who(request)) is None:
             return error("No such room", 404)
         return web.json_response(store.list_snapshots(key))
 
@@ -605,8 +611,9 @@ def setup(server, store, hub):
         snap = store.get_snapshot(request.match_info["snap_id"])
         if not snap:
             return error("Not found", 404)
-        room = hub.rooms.get(snap["meta"].get("room", ""))
-        if room and not room_access(request, room):
+        # Access follows the workflow's folder whether or not anyone has it open right now.
+        room_key = snap["meta"].get("room", "")
+        if room_key and hub.access_for_key(room_key, *who(request)) is None:
             return error("Not found", 404)
         return web.json_response(snap)
 

rigshare_core/store.py

@@ -24,7 +24,7 @@ SESSION_TTL = 30 * 24 * 3600      # a login lasts at most this long
 TOKEN_TTL = 12 * 3600              # each JWT is short-lived...
 TOKEN_REFRESH = 6 * 3600           # ...and re-issued while in use once older than this
 ISSUER = "rigshare"
-USERNAME_RE = re.compile(r"^[A-Za-z0-9_.-]{2,32}$")
+USERNAME_RE = re.compile(r"^(?!(?i:con|prn|aux|nul|com\d|lpt\d)(?:\.|$))[A-Za-z0-9_][A-Za-z0-9_.-]{0,30}[A-Za-z0-9_-]\Z")  # no leading/trailing dot, no Windows device names
 PERMS = ("edit", "queue", "manager", "admin")
 API_KEY_PREFIX = "rs_"
 
@@ -216,10 +216,14 @@ class Store:
         _verify_password(password, _hash_password("x", b"0" * 16))
         return None
 
+    def _name_taken(self, name, ignore=None):
+        """Names differing only by case share one folder on Windows, so they count as the same."""
+        return any(u.casefold() == name.casefold() for u in self.users if u != ignore)
+
     def create_user(self, username, password, display_name=None, perms=None):
         if not USERNAME_RE.match(username or ""):
             raise ValueError("Username must be 2-32 characters: letters, digits, _ . -")
-        if username in self.users:
+        if self._name_taken(username):
             raise ValueError("Username already exists")
         if len(password or "") < 8:
             raise ValueError("Password must be at least 8 characters")
@@ -239,7 +243,7 @@ class Store:
             raise KeyError(username)
         if not USERNAME_RE.match(new_username):
             raise ValueError("Username must be 2-32 characters: letters, digits, _ . -")
-        if new_username in self.users:
+        if self._name_taken(new_username, ignore=username):
             raise ValueError("Username already exists")
         with self._lock:
             user = self.users.pop(username)

rigshare_core/workspace.py

@@ -20,7 +20,7 @@ import time
 from urllib.parse import unquote
 
 WORKFLOWS = "workflows"
-FOLDER_NAME_RE = re.compile(r"^[\w][\w .()-]{0,63}$")
+FOLDER_NAME_RE = re.compile(r"^(?!(?i:con|prn|aux|nul|com\d|lpt\d)(?:\.|$))\w(?:[\w .()-]{0,62}[\w()-])?\Z")  # no trailing dot/space, no Windows device names
 ROLE_RANK = {None: 0, "view": 1, "edit": 2}
 
 
@@ -35,10 +35,66 @@ def full_unquote(value):
     return value
 
 
+_WIN = os.name == "nt"
+_SHORT_NAME_RE = re.compile(r"~\d")  # Windows 8.3 alias of a long name (USERS~1)
+_case_root = None  # workflows directory used to restore on-disk casing (Windows only)
+
+
+def _on_disk_case(parts, base):
+    """Windows ignores case: map each segment to the casing it has on disk, so 'Users/Bob' is
+    judged as the same path it will open."""
+    out = []
+    for part in parts:
+        try:
+            names = os.listdir(base) if base else []
+        except OSError:
+            names = []
+        match = part if part in names else next((n for n in names if n.casefold() == part.casefold()), part)
+        out.append(match)
+        base = os.path.join(base, match) if base and match in names else None
+    return out
+
+
+class InvalidPath(PermissionError):
+    """A path Windows would read as a different one than the checks do: refused, never adjusted."""
+
+
 def norm(path):
-    """Decode and normalise a userdata-relative path ('' for the root)."""
+    """Decode and normalise a userdata-relative path ('' for the root).
+
+    A path with a segment Windows would read as something else is refused (InvalidPath), not
+    adjusted: a colon (drive prefix 'C:', NTFS stream), a 8.3 short name, or trailing dots or
+    spaces (Windows strips them, so 'users./bob' would open 'users/bob'). Casing is restored from
+    disk on Windows. Anything that still reaches the disk goes through safe_join().
+    """
     path = full_unquote(str(path or "")).replace("\\", "/")
-    return posixpath.normpath("/" + path).lstrip("/") if path else ""
+    if not path:
+        return ""
+    clean = posixpath.normpath("/" + path).lstrip("/")
+    parts = [p for p in clean.split("/") if p]
+    for p in parts:
+        if ":" in p or (_WIN and (_SHORT_NAME_RE.search(p) or p != p.rstrip(" ."))):
+            raise InvalidPath("Invalid path")
+    if _WIN and _case_root and parts:
+        lead = [WORKFLOWS] if parts[0].casefold() == WORKFLOWS else []
+        parts = lead + _on_disk_case(parts[len(lead):], _case_root)
+    return "/".join(parts)
+
+
+def safe_join(root, *parts):
+    """Join relative parts under root; refuse anything that resolves outside it."""
+    segments = [p for part in parts for p in str(part).replace("\\", "/").split("/") if p]
+    if any(":" in p or p in (".", "..") or (_WIN and p != p.rstrip(" .")) for p in segments):
+        raise PermissionError("Invalid path")
+    full = os.path.join(root, *segments)
+    base = os.path.realpath(root)
+    try:
+        inside = os.path.commonpath([base, os.path.realpath(full)]) == base
+    except ValueError:  # different drives
+        inside = False
+    if not inside:
+        raise PermissionError("Invalid path")
+    return full
 
 
 def has_app(doc):
@@ -56,6 +112,8 @@ class Workspace:
     def __init__(self, store, root):
         self.store = store
         self.root = root  # absolute path of the workflows directory
+        global _case_root
+        _case_root = root
         self.folders = store._read("folders.json", {})  # "shared/<name>" -> {owner, acl, created}
         os.makedirs(os.path.join(root, "users"), exist_ok=True)
         os.makedirs(os.path.join(root, "shared"), exist_ok=True)
@@ -147,7 +205,7 @@ class Workspace:
         return self.role(rel, kind, key, perms) is not None
 
     def _files(self, rel_dir):
-        base = os.path.join(self.root, *rel_dir.split("/")) if rel_dir else self.root
+        base = safe_join(self.root, rel_dir) if rel_dir else self.root
         out = []
         if not os.path.isdir(base):
             return out
@@ -198,7 +256,7 @@ class Workspace:
     # ----- browsing (Files tab) -------------------------------------------
 
     def _abs(self, rel):
-        return os.path.join(self.root, *[p for p in rel.split("/") if p])
+        return safe_join(self.root, rel)
 
     def _file_has_app(self, full):
         """Read a workflow file only when it changed since the last check."""
@@ -372,7 +430,7 @@ class Workspace:
         name = (name or "").strip()
         if not FOLDER_NAME_RE.match(name) or name in (".", ".."):
             raise ValueError("Folder names: up to 64 letters, digits, spaces and . _ - ( )")
-        path = os.path.join(self.root, "shared", name)
+        path = safe_join(self.root, "shared", name)
         if os.path.exists(path):
             raise ValueError("A folder with that name already exists")
         os.makedirs(path)
@@ -385,8 +443,8 @@ class Workspace:
         if not FOLDER_NAME_RE.match(new_name):
             raise ValueError("Folder names: up to 64 letters, digits, spaces and . _ - ( )")
         new_folder = f"shared/{new_name}"
-        src = os.path.join(self.root, *folder.split("/"))
-        dst = os.path.join(self.root, "shared", new_name)
+        src = safe_join(self.root, folder)
+        dst = safe_join(self.root, "shared", new_name)
         if os.path.exists(dst):
             raise ValueError("A folder with that name already exists")
         os.rename(src, dst)
@@ -395,7 +453,7 @@ class Workspace:
         return new_folder
 
     def delete_folder(self, folder):
-        path = os.path.join(self.root, *folder.split("/"))
+        path = safe_join(self.root, folder)
         if self._files(folder):
             raise ValueError("Move or delete the workflows inside it first")
         shutil.rmtree(path, ignore_errors=True)

tests/python/test_jwt.py

@@ -0,0 +1,60 @@
+"""HS256 tokens: round trip, tampering, expiry, issuer, algorithm confusion."""
+
+import base64
+import json
+import time
+
+import pytest
+
+from rigshare_core import jwt
+
+SECRET = b"s3cret"
+
+
+def claims(**extra):
+    return {"iss": "rig", "exp": time.time() + 60, **extra}
+
+
+def b64(obj):
+    return base64.urlsafe_b64encode(json.dumps(obj).encode()).rstrip(b"=").decode()
+
+
+def test_round_trip():
+    token = jwt.encode(claims(sub="bob"), SECRET)
+    assert jwt.decode(token, SECRET, "rig")["sub"] == "bob"
+
+
+def test_wrong_secret_and_tampered_payload():
+    token = jwt.encode(claims(sub="bob"), SECRET)
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(token, b"other", "rig")
+    head, _, sig = token.split(".")
+    forged = f"{head}.{b64(claims(sub='admin'))}.{sig}"
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(forged, SECRET, "rig")
+
+
+def test_expiry_leeway_and_not_before():
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(jwt.encode({"iss": "rig", "exp": time.time() - 3600}, SECRET), SECRET, "rig")
+    jwt.decode(jwt.encode({"iss": "rig", "exp": time.time() - 5}, SECRET), SECRET, "rig")  # inside leeway
+    assert jwt.decode(jwt.encode({"iss": "rig", "exp": time.time() - 3600}, SECRET), SECRET, "rig", verify_exp=False)
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(jwt.encode(claims(nbf=time.time() + 3600), SECRET), SECRET, "rig")
+
+
+def test_wrong_issuer():
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(jwt.encode(claims(), SECRET), SECRET, "someone-else")
+
+
+def test_alg_none_is_refused():
+    token = f"{b64({'alg': 'none', 'typ': 'JWT'})}.{b64(claims())}."
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(token, SECRET, "rig")
+
+
+@pytest.mark.parametrize("junk", ["", "a.b", "a.b.c.d", "!!!.???.***", "e30.e30.e30"])
+def test_malformed_tokens(junk):
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(junk, SECRET, "rig")

tests/python/test_server.py

@@ -646,3 +646,62 @@ def test_performance_run_log_eta_history_and_clear(run):
             assert (await r.json())["cleared"] == 3
             assert hub.etas() == {"video": {"ms": 90000, "runs": 1}}, "the ETA starts over"
     run(scenario())
+
+
+def test_folder_routes_refuse_paths_outside_the_workspace(run, tmp_path):
+    async def scenario():
+        async with rig_server() as rig:
+            await rig.setup_admin()
+            await rig.add_user("alice", edit=True)
+            outside = tmp_path / "models"
+            outside.mkdir()
+            (outside / "keep.bin").write_bytes(b"x")
+            H, h = rig.headers, rig.http
+            targets = [str(outside), "C:" + str(outside)[2:], r"..\..\models", "shared/C:/evil"]
+            for route, extra in (("delete-folder", {}), ("rename-folder", {"name": "x"}), ("move-folder", {"dest": "@shared"})):
+                for target in targets:
+                    r = await h.post(rig.url(f"/rigshare/api/workspace/{route}"), json={"path": target, **extra}, headers=H["alice"])
+                    assert r.status in (400, 403, 404), (route, target, r.status)
+            r = await h.post(rig.url("/rigshare/api/workspace/mkdir"), json={"parent": "C:/", "name": "x"}, headers=H["alice"])
+            assert r.status in (400, 403, 404)
+            assert (outside / "keep.bin").exists()
+    run(scenario())
+
+
+def test_access_is_checked_without_a_live_room(run):
+    async def scenario():
+        async with rig_server() as rig:
+            await rig.setup_admin()
+            alice = await rig.add_user("alice", edit=True)
+            bob = await rig.add_user("bob", edit=True)
+            await rig.add_user("viewer")
+            h = rig.http
+            r = await h.post(rig.url("/rigshare/api/workspace/mkdir"), json={"parent": "@shared", "name": "Team"}, headers=alice)
+            assert r.status == 200
+            rig.hub.workspace.set_acl("shared/Team", {"members": {"alice": "edit"}}, "alice")
+            path = "workflows/shared/Team/plan.json"
+            doc = json.dumps({"nodes": [{"id": 1}], "links": []}).encode()
+            assert (await h.post(rig.url(f"/api/userdata/{enc(path)}"), data=doc, headers=alice)).status == 200
+            key = "file:" + path
+            assert key not in rig.hub.rooms, "nobody has it open"
+            snap_id = rig.store.list_snapshots(key)[0]["id"]
+
+            for who_, status in ((alice, 200), (bob, 404)):
+                r = await h.get(rig.url(f"/rigshare/api/snapshots?room={enc(key)}"), headers=who_)
+                assert r.status == status, who_
+                r = await h.get(rig.url(f"/rigshare/api/snapshots/{snap_id}"), headers=who_)
+                assert r.status == status, who_
+
+            # Joining a workflow you cannot open must not create (or seed) a room.
+            sock = await rig.ws("bob")
+            await sock.send({"type": "join", "room": key, "name": "plan", "doc": BASE_DOC})
+            await sock.until("room_denied")
+            assert key not in rig.hub.rooms
+
+            # A viewer cannot be the one who opens a room, so cannot seed it with their own document.
+            open_key = "file:workflows/common.json"
+            viewer_sock = await rig.ws("viewer")
+            await viewer_sock.send({"type": "join", "room": open_key, "name": "common", "doc": BASE_DOC})
+            assert "editing" in (await viewer_sock.until("room_denied"))["reason"]
+            assert open_key not in rig.hub.rooms
+    run(scenario())

tests/python/test_stats.py

@@ -0,0 +1,45 @@
+"""Server stats: every metric is optional and never raises."""
+
+from rigshare_core import stats
+
+
+class Queue:
+    def __init__(self, running, pending):
+        self.running, self.pending = running, pending
+
+    def get_current_queue_volatile(self):
+        return self.running, self.pending
+
+
+class Server:
+    def __init__(self, queue):
+        self.prompt_queue = queue
+
+
+def test_queue_items_order_and_node_count():
+    server = Server(Queue([(1, "r1", {"1": {}, "2": {}})], [(5, "p5", {"1": {}}), (3, "p3", {})]))
+    items = stats.queue_items(server)
+    assert [(i["id"], i["status"], i["nodes"]) for i in items] == [("r1", "running", 2), ("p3", "pending", 0), ("p5", "pending", 1)]
+
+
+def test_queue_items_survive_odd_entries_and_missing_queue():
+    assert stats.queue_items(Server(Queue([(1,)], [(9,)]))) == []
+    assert stats.queue_items(object()) is None
+
+
+def test_queue_counts_with_fallback():
+    assert stats._queue(Server(Queue([1], [2, 3]))) == {"running": 1, "pending": 2}
+
+    class Old:
+        def get_tasks_remaining(self):
+            return 4
+    assert stats._queue(Server(Old())) == {"running": 0, "pending": 4}
+    assert stats._queue(object()) is None
+
+
+def test_collect_stats_without_psutil_or_gpu(monkeypatch):
+    monkeypatch.setattr(stats, "psutil", None)
+    monkeypatch.setattr(stats, "pynvml", None)
+    out = stats.collect_stats(Server(Queue([], [])))
+    assert out["gpus"] == [] and "cpu" not in out and out["queue"] == {"running": 0, "pending": 0}
+    assert stats.collect_stats() == {"gpus": []}

tests/python/test_workspace.py

@@ -131,3 +131,68 @@ def test_rename_user_moves_private_folder(ws):
     ws.rename_user("alice", "alicia")
     assert os.path.exists(os.path.join(ws.root, "users", "alicia", "keep.json"))
     assert ws.folders["shared/Team"]["owner"] == "alicia"
+
+
+@pytest.mark.parametrize("raw", ["C:/models", "C:", "a/D:x/b", "c%3A%5Cmodels"])
+def test_norm_refuses_drive_segments(raw):
+    with pytest.raises(PermissionError):
+        norm(raw)
+
+
+def test_norm_still_resolves_dotdot_inside_the_tree():
+    assert norm(r"..\..\x") == "x"
+    assert norm("a/../b") == "b"
+
+
+@pytest.mark.parametrize("bad", ["C:/models", "C:", "..", "a/../..", "x:y"])
+def test_safe_join_refuses_escapes(ws, bad):
+    from rigshare_core.workspace import safe_join
+    with pytest.raises(PermissionError):
+        safe_join(ws.root, bad)
+
+
+def test_folder_ops_cannot_leave_the_workspace(ws, tmp_path):
+    outside = tmp_path / "models"
+    outside.mkdir()
+    (outside / "keep.txt").write_text("x")
+    # A colon segment (drive prefix) is refused outright; a plain absolute path is only ever
+    # read relative to the workflows root (a leading slash is dropped), so it can't reach `outside`.
+    for bad in ("C:" + str(outside)[2:], "shared/C:evil"):
+        with pytest.raises(PermissionError):
+            ws.delete_folder(bad)
+        with pytest.raises(PermissionError):
+            ws.rename_folder(bad, "renamed")
+    for plain in (str(outside), str(outside).replace("\\", "/")):
+        try:
+            ws.delete_folder(plain)
+        except (PermissionError, ValueError):
+            pass
+    assert (outside / "keep.txt").exists()
+
+
+@pytest.mark.skipif(os.name != "nt", reason="trailing dots, spaces and short names only alias on Windows")
+@pytest.mark.parametrize("raw", ["users./bob/a.json", "users /bob", "USERS~1/bob", "users/bob./a.json"])
+def test_norm_refuses_windows_aliases(raw):
+    with pytest.raises(PermissionError):
+        norm(raw)
+
+
+@pytest.mark.skipif(os.name != "nt", reason="case-insensitive filesystem")
+def test_case_variants_cannot_bypass_private_folders(ws):
+    touch(ws, "users/alice/a.json")
+    assert norm("Users/Alice/a.json") == "users/alice/a.json"
+    assert norm("Workflows/USERS/ALICE") == "workflows/users/alice"
+    assert ws.role(norm("USERS/alice/a.json"), "user", "bob", EDITOR) is None
+
+
+def test_name_rules():
+    from rigshare_core.workspace import FOLDER_NAME_RE
+    from rigshare_core.store import USERNAME_RE
+    for bad in ("con", "NUL", "com1.txt", "art.", "art ", "bad\n"):
+        assert not FOLDER_NAME_RE.match(bad), bad
+    for bad in ("..", "bob.", ".bob", "con", "aux", "ab\n", "a"):
+        assert not USERNAME_RE.match(bad), bad
+    for ok in ("Art (v2)", "a", "my-folder"):
+        assert FOLDER_NAME_RE.match(ok), ok
+    for ok in ("bob", "al.ice", "x_y-z", "ab"):
+        assert USERNAME_RE.match(ok), ok