Add share token support with virtual filesystem and selection toolbar button

cista/api.py

@@ -6,8 +6,9 @@ from sanic import Blueprint, json
 from sanic.exceptions import BadRequest
 from sanic.log import logger
 
-from cista import __version__, auth, config, sso, watching
+from cista import __version__, auth, config, sharefs, sso, watching
 from cista.auth import (
+    create_share_token_handler,
     create_token_handler,
     delete_token_handler,
     list_tokens_handler,
@@ -68,15 +69,27 @@ async def watch(req, ws):
         ).decode()
     )
     uuid = token_bytes(16)
+    share_token = auth.request_share_token(req)
     try:
         q, space, root = await asyncio.get_event_loop().run_in_executor(
             req.app.ctx.threadexec, subscribe, uuid, ws
         )
         await ws.send(space)
-        await ws.send(root)
+        if share_token is None:
+            await ws.send(root)
+        else:
+            await ws.send(watching.format_root(sharefs.build_virtual_root(share_token)))
         # Send updates
         while True:
-            await ws.send(await q.get())
+            msg = await q.get()
+            if share_token is None or (
+                isinstance(msg, str) and msg.startswith('{"space"')
+            ):
+                await ws.send(msg)
+            else:
+                await ws.send(
+                    watching.format_root(sharefs.build_virtual_root(share_token))
+                )
     except RuntimeError as e:
         if str(e) == "cannot schedule new futures after shutdown":
             return  # Server shutting down, drop the WebSocket
@@ -153,3 +166,8 @@ async def create_api_token(request):
 @bp.delete("tokens/<token_id>")
 async def delete_api_token(request, token_id):
     return await delete_token_handler(request, token_id)
+
+
+@bp.post("share-tokens")
+async def create_share_token(request):
+    return await create_share_token_handler(request)

cista/app.py

@@ -16,7 +16,7 @@ from setproctitle import setproctitle
 from stream_zip import ZIP_AUTO, stream_zip
 from zstandard import ZstdCompressor
 
-from cista import auth, config, fileserver, preview, session, sso, watching
+from cista import auth, config, fileserver, preview, session, sharefs, sso, watching
 from cista.api import bp
 from cista.preview import shutdown_preview_workers, start_preview_workers
 from cista.sanic_logging import (
@@ -240,25 +240,43 @@ async def favicon(req):
     return redirect("/assets/logo-ctv8tVwU.svg", status=308)
 
 
-def get_files(wanted: set) -> list[tuple[PurePosixPath, Path]]:
+def get_files(req, wanted: set) -> list[tuple[PurePosixPath, Path]]:
     loc = PurePosixPath()
     idx = 0
     ret = []
     level: int | None = None
     parent: PurePosixPath | None = None
-    with watching.state.lock:
+    token = auth.request_share_token(req)
-        root = watching.state.root
+
-        while idx < len(root):
+    if token is None:
-            f = root[idx]
+        with watching.state.lock:
-            loc = PurePosixPath(*loc.parts[: f.level - 1]) / f.name
+            root = watching.state.root
-            if parent is not None and f.level <= level:
+            while idx < len(root):
-                level = parent = None
+                f = root[idx]
-            if f.key in wanted:
+                loc = PurePosixPath(*loc.parts[: f.level - 1]) / f.name
-                level, parent = f.level, loc.parent
+                if parent is not None and f.level <= level:
-            if parent is not None:
+                    level = parent = None
-                wanted.discard(f.key)
+                if f.key in wanted:
-                ret.append((loc.relative_to(parent), watching.rootpath / loc))
+                    level, parent = f.level, loc.parent
-            idx += 1
+                if parent is not None:
+                    wanted.discard(f.key)
+                    ret.append((loc.relative_to(parent), watching.rootpath / loc))
+                idx += 1
+        return ret
+
+    root = sharefs.build_virtual_root(token)
+    while idx < len(root):
+        f = root[idx]
+        loc = PurePosixPath(*loc.parts[: f.level - 1]) / f.name
+        if parent is not None and f.level <= level:
+            level = parent = None
+        if f.key in wanted:
+            level, parent = f.level, loc.parent
+        if parent is not None:
+            wanted.discard(f.key)
+            real_path = sharefs.resolve_virtual_rel_to_real(token, loc)
+            ret.append((loc.relative_to(parent), real_path))
+        idx += 1
     return ret
 
 
@@ -268,7 +286,7 @@ async def zip_download(req, keys, zipfile, ext):
     await auth.verify(req)
 
     wanted = set(keys.split("+"))
-    files = get_files(wanted)
+    files = get_files(req, wanted)
 
     if not files:
         raise NotFound(

cista/auth.py

@@ -5,6 +5,7 @@ import hmac
 import re
 import secrets
 import struct
+from pathlib import PurePosixPath
 from time import time
 from unicodedata import normalize
 
@@ -15,8 +16,9 @@ from sanic import Blueprint, html, json, redirect
 from sanic.exceptions import BadRequest, Forbidden, Unauthorized
 from sanic.log import logger
 
-from cista import config, session
+from cista import config, session, sharefs
 from cista.util import pwgen
+from cista.util.filename import sanitize
 
 _LOGIN_PAGE_CSS = """\
 /* ===========================================
@@ -611,6 +613,8 @@ def _basic_auth_login(request):
                 request.ctx.session = None
                 request.ctx.username = token.username
                 request.ctx.user = user
+                request.ctx.auth_token_id = password
+                request.ctx.auth_token = token
                 user.lastSeen = int(time())
                 return user
         raise Unauthorized("Invalid token", quiet=True)
@@ -654,6 +658,9 @@ async def _token_auth_login(request, *, privileged=False):
     if not token:
         return False
 
+    request.ctx.auth_token_id = password
+    request.ctx.auth_token = token
+
     sso = _get_sso()
     if sso.paskia_enabled() and token.sso_user_id:
         perm = "cista:admin" if privileged else "cista:login"
@@ -823,6 +830,8 @@ async def _ntlm_auth_login(request, *, privileged=False):
                     try:
                         data = await sso.check_permissions(token.sso_user_id, perm)
                         request.ctx.sso_user = data
+                        request.ctx.auth_token_id = tid
+                        request.ctx.auth_token = token
                         ctx = data.get("ctx", {}) if isinstance(data, dict) else {}
                         user_info = ctx.get("user", {}) if isinstance(ctx, dict) else {}
                         request.ctx.username = user_info.get("display_name", "")
@@ -854,6 +863,8 @@ async def _ntlm_auth_login(request, *, privileged=False):
                         request.ctx.session = None
                         request.ctx.username = token.username
                         request.ctx.user = user
+                        request.ctx.auth_token_id = tid
+                        request.ctx.auth_token = token
                         request.ctx._create_session_username = token.username
                         logger.debug(
                             "NTLM auth success for local user %s (token=%s...)",
@@ -1294,6 +1305,26 @@ def _token_belongs_to_user(token, username, sso_user_id):
     return bool(sso_user_id is not None and token.sso_user_id == sso_user_id)
 
 
+def request_token(request) -> config.Token | None:
+    token = getattr(request.ctx, "auth_token", None)
+    return token if isinstance(token, config.Token) else None
+
+
+def request_share_token(request) -> config.Token | None:
+    token = request_token(request)
+    if token is None:
+        return None
+    return token if sharefs.is_share_token(token) else None
+
+
+def ensure_write_allowed(request) -> None:
+    token = request_share_token(request)
+    if token is None:
+        return
+    if token.mode != "rw":
+        raise Forbidden("Share token is read-only", quiet=True)
+
+
 # Token management handlers (shared between /auth and /api blueprints)
 
 
@@ -1310,6 +1341,8 @@ async def list_tokens_handler(request):
                     "sso_user_id": t.sso_user_id,
                     "name": t.name,
                     "created": t.created,
+                    "kind": t.kind,
+                    "mode": t.mode,
                 }
             )
     return json({"tokens": tokens})
@@ -1358,6 +1391,9 @@ async def create_token_handler(request):
         "sso_user_id": sso_user_id or "",
         "name": name,
         "created": int(time()),
+        "kind": "api",
+        "mode": "rw",
+        "share_paths": [],
     }
     config.update_token(token, changes)
     scheme = request.scheme
@@ -1371,6 +1407,100 @@ async def create_token_handler(request):
             "username": username or "",
             "sso_user_id": sso_user_id or "",
             "name": name,
+            "kind": "api",
+            "mode": "rw",
+        }
+    )
+
+
+async def create_share_token_handler(request):
+    await verify(request)
+    current_username, current_sso_user_id = _current_user_id(request)
+    try:
+        if request.headers.content_type == "application/json":
+            paths = request.json.get("paths")
+            mode = request.json.get("mode", "ro")
+            name = request.json.get("name", "")
+        else:
+            paths = request.form.get("paths", [])
+            mode = request.form.get("mode", ["ro"])[0]
+            name = request.form.get("name", [""])[0]
+    except (KeyError, IndexError):
+        raise BadRequest("Missing fields") from None
+
+    if not isinstance(paths, list) or not paths:
+        raise BadRequest("paths must be a non-empty array")
+    if mode not in ("ro", "rw"):
+        raise BadRequest("mode must be ro or rw")
+
+    clean_paths: list[str] = []
+    seen: set[str] = set()
+    base = config.config.path.resolve()
+    for raw_path in paths:
+        if not isinstance(raw_path, str):
+            raise BadRequest("paths must contain strings")
+        try:
+            clean = sanitize(raw_path)
+        except ValueError as e:
+            raise BadRequest(f"Invalid path: {e}") from e
+        if not clean:
+            continue
+        rel = PurePosixPath(clean)
+        resolved = (base / rel).resolve()
+        if not resolved.is_relative_to(base):
+            raise BadRequest("Invalid path")
+        if not resolved.exists():
+            raise BadRequest(f"Path does not exist: {clean}")
+        key = rel.as_posix()
+        if key in seen:
+            continue
+        seen.add(key)
+        clean_paths.append(key)
+
+    if not clean_paths:
+        raise BadRequest("No valid paths selected")
+
+    sso = _get_sso()
+    username = ""
+    sso_user_id = ""
+    if sso.paskia_enabled():
+        sso_user_id = current_sso_user_id or ""
+        if not sso_user_id:
+            raise BadRequest("Could not determine SSO user")
+    else:
+        username = current_username or ""
+        if not username:
+            raise BadRequest("Could not determine user")
+        if username not in config.config.users:
+            raise BadRequest("User does not exist")
+
+    token = secrets.token_urlsafe(12)
+    changes = {
+        "key": token,
+        "username": username,
+        "sso_user_id": sso_user_id,
+        "name": name,
+        "created": int(time()),
+        "kind": "share",
+        "mode": mode,
+        "share_paths": clean_paths,
+    }
+    config.update_token(token, changes)
+
+    scheme = request.scheme
+    host = request.host or "localhost"
+    share_url = f"{scheme}://token:{token}@{host}/#/"
+    return json(
+        {
+            "id": token,
+            "key": token,
+            "url": share_url,
+            "username": username,
+            "sso_user_id": sso_user_id,
+            "name": name,
+            "kind": "share",
+            "mode": mode,
+            "paths": clean_paths,
         }
     )
 

cista/config.py

@@ -51,6 +51,9 @@ class Token(msgspec.Struct, omit_defaults=True):
     sso_user_id: str = ""  # set in SSO mode
     name: str = ""
     created: int = 0
+    kind: str = "api"  # api | share
+    mode: str = "rw"  # ro | rw
+    share_paths: list[str] = []
 
 
 # Global variables - initialized during application startup

cista/fileserver.py

@@ -14,7 +14,7 @@ from wsgiref.handlers import format_date_time
 from sanic import Blueprint, HTTPResponse, empty, json
 from sanic.exceptions import BadRequest, NotFound
 
-from cista import auth, config, watching
+from cista import auth, config, sharefs, watching
 from cista.api import fileserver
 from cista.util import filename
 
@@ -40,6 +40,7 @@ async def verify_fileserver(request):
 
 @bp.put("/<name:path>")
 async def upload_file_chunk(request, name):
+    auth.ensure_write_allowed(request)
     body = request.body
     header = request.headers.get("content-range")
     if header:
@@ -49,7 +50,7 @@ async def upload_file_chunk(request, name):
         end = len(body)
         total = end
 
-    rel, _ = _safe_relpath(name)
+    rel, path = _safe_relpath(name, request=request)
     rel_name = rel.as_posix()
     upload_info = await asyncio.to_thread(
         fileserver.upload_info,
@@ -76,7 +77,8 @@ async def upload_file_chunk(request, name):
     if size_before is not None and size_after is not None and size_before != size_after:
         extras.append("resized")
     request.ctx._log_extra = " ".join(extras) if extras else None
-    watching.notify_change(rel, *rel.parents)
+    real_rel = PurePosixPath(path.relative_to(config.config.path.resolve()).as_posix())
+    watching.notify_change(real_rel, *real_rel.parents)
     return json(
         {
             "status": "ack",
@@ -92,7 +94,8 @@ async def upload_file_chunk(request, name):
 
 @bp.delete("/<name:path>")
 async def delete_file(request, name):
-    rel, path = _safe_relpath(name)
+    auth.ensure_write_allowed(request)
+    rel, path = _safe_relpath(name, request=request)
     if not rel.parts:
         raise BadRequest("Refusing to delete root folder")
 
@@ -105,23 +108,27 @@ async def delete_file(request, name):
             path.unlink()
 
     await asyncio.to_thread(_delete)
-    watching.notify_change(rel, *rel.parents)
+    real_rel = PurePosixPath(path.relative_to(config.config.path.resolve()).as_posix())
+    watching.notify_change(real_rel, *real_rel.parents)
     return empty(status=204)
 
 
 @bp.route("/<name:path>", methods=["MKCOL"])
 async def create_folder(request, name):
-    rel, path = _safe_relpath(name)
+    auth.ensure_write_allowed(request)
+    rel, path = _safe_relpath(name, request=request)
     if not rel.parts:
         raise BadRequest("Refusing to create root folder")
     await asyncio.to_thread(path.mkdir, parents=True, exist_ok=False)
-    watching.notify_change(rel, *rel.parents)
+    real_rel = PurePosixPath(path.relative_to(config.config.path.resolve()).as_posix())
+    watching.notify_change(real_rel, *real_rel.parents)
     return empty(status=201)
 
 
 @bp.post("/", name="post_root", strict_slashes=False)
 @bp.post("/<name:path>", name="post_path")
 async def copy_or_move(request, name=""):
+    auth.ensure_write_allowed(request)
     provided_args = set(request.args.keys())
     if not provided_args:
         raise BadRequest("No query arguments passed")
@@ -145,13 +152,13 @@ async def copy_or_move(request, name=""):
     if not mv_keys and not cp_keys:
         raise BadRequest("No keys given")
 
-    dst_rel, dst_abs = _safe_relpath(name)
+    dst_rel, dst_abs = _safe_relpath(name, request=request)
 
     dst_exists = dst_abs.exists()
     dst_is_dir = dst_exists and dst_abs.is_dir()
 
     ordered_keys = cp_keys + mv_keys
-    key_paths = _get_key_paths(set(ordered_keys))
+    key_paths = _get_key_paths(request, set(ordered_keys))
     missing = [key for key in ordered_keys if key not in key_paths]
     if missing:
         raise NotFound("Files not found", context={"missing": missing})
@@ -194,7 +201,7 @@ async def copy_or_move(request, name=""):
             for key in op_keys:
                 try:
                     src_rel = key_paths[key]
-                    src_abs = _resolve_from_relpath(src_rel)
+                    src_abs = _resolve_from_relpath(src_rel, request=request)
 
                     if op_multi:
                         if not dst_is_dir:
@@ -224,7 +231,7 @@ async def copy_or_move(request, name=""):
                             )
                         dst_item_rel = dst_rel
 
-                    dst_item_abs = _resolve_from_relpath(dst_item_rel)
+                    dst_item_abs = _resolve_from_relpath(dst_item_rel, request=request)
 
                     if op_name == "mv":
                         # A no-op rename should still return success.
@@ -263,7 +270,15 @@ async def copy_or_move(request, name=""):
 
     notify_paths = [p for p in changed if p.parts]
     if notify_paths:
-        watching.notify_change(*notify_paths)
+        real_notify_paths: list[PurePosixPath] = []
+        for p in notify_paths:
+            real_abs = _resolve_from_relpath(p, request=request)
+            real_notify_paths.append(
+                PurePosixPath(
+                    real_abs.relative_to(config.config.path.resolve()).as_posix()
+                )
+            )
+        watching.notify_change(*real_notify_paths)
 
     return json(
         {
@@ -299,7 +314,29 @@ async def dav_options(request, name=""):
 @bp.route("/", methods=["PROPFIND"], name="propfind_root", strict_slashes=False)
 @bp.route("/<name:path>", methods=["PROPFIND"], name="propfind_path")
 async def dav_propfind(request, name=""):
-    rel, path = _safe_relpath(name)
+    rel, path = _safe_relpath(name, request=request)
+    token = auth.request_share_token(request)
+    if token is not None and not rel.parts:
+        base = config.config.path.resolve()
+        entries = [_propfind_entry(PurePosixPath(), base)]
+        depth = request.headers.get("depth", "1").strip()
+        if depth == "infinity":
+            return HTTPResponse(status=403)
+        if depth == "1":
+            for root in sharefs.build_share_roots(token):
+                child_abs = (base / root.real_rel).resolve()
+                if not child_abs.exists() or not child_abs.is_relative_to(base):
+                    continue
+                with contextlib.suppress(OSError):
+                    entries.append(
+                        _propfind_entry(PurePosixPath(root.alias), child_abs)
+                    )
+        return HTTPResponse(
+            body=_build_propfind_xml(entries),
+            status=207,
+            content_type='application/xml; charset="utf-8"',
+        )
+
     if not path.exists():
         raise NotFound(f"Not found: {name}")
     depth = request.headers.get("depth", "1").strip()
@@ -316,12 +353,15 @@ async def dav_propfind(request, name=""):
 @bp.route("/", methods=["COPY"], name="copy_root", strict_slashes=False)
 @bp.route("/<name:path>", methods=["COPY"], name="copy_path")
 async def dav_copy(request, name=""):
+    auth.ensure_write_allowed(request)
     dest_header = request.headers.get("destination")
     if not dest_header:
         raise BadRequest("Missing Destination header")
     overwrite = request.headers.get("overwrite", "T").strip().upper() != "F"
-    _src_rel, src_abs = _safe_relpath(name)
+    _src_rel, src_abs = _safe_relpath(name, request=request)
-    dst_rel, dst_abs = _parse_webdav_destination(dest_header)
+    dst_rel, dst_abs = _parse_webdav_destination(dest_header, request=request)
+    if auth.request_share_token(request) is not None and not dst_rel.parts:
+        raise BadRequest("Destination cannot be virtual root")
     request.ctx._log_extra = f"→ {dst_rel}"
     if not src_abs.exists():
         raise NotFound(f"Source not found: {name}")
@@ -342,19 +382,25 @@ async def dav_copy(request, name=""):
             shutil.copy2(src_abs, dst_abs)
 
     await asyncio.to_thread(_do_copy)
-    watching.notify_change(dst_rel, *dst_rel.parents)
+    real_dst_rel = PurePosixPath(
+        dst_abs.relative_to(config.config.path.resolve()).as_posix()
+    )
+    watching.notify_change(real_dst_rel, *real_dst_rel.parents)
     return HTTPResponse(status=201 if not dst_existed else 204)
 
 
 @bp.route("/", methods=["MOVE"], name="move_root", strict_slashes=False)
 @bp.route("/<name:path>", methods=["MOVE"], name="move_path")
 async def dav_move(request, name=""):
+    auth.ensure_write_allowed(request)
     dest_header = request.headers.get("destination")
     if not dest_header:
         raise BadRequest("Missing Destination header")
     overwrite = request.headers.get("overwrite", "T").strip().upper() != "F"
-    src_rel, src_abs = _safe_relpath(name)
+    _src_rel, src_abs = _safe_relpath(name, request=request)
-    dst_rel, dst_abs = _parse_webdav_destination(dest_header)
+    dst_rel, dst_abs = _parse_webdav_destination(dest_header, request=request)
+    if auth.request_share_token(request) is not None and not dst_rel.parts:
+        raise BadRequest("Destination cannot be virtual root")
     request.ctx._log_extra = f"→ {dst_rel}"
     if not src_abs.exists():
         raise NotFound(f"Source not found: {name}")
@@ -372,7 +418,15 @@ async def dav_move(request, name=""):
         shutil.move(src_abs, dst_abs)
 
     await asyncio.to_thread(_do_move)
-    watching.notify_change(src_rel, *src_rel.parents, dst_rel, *dst_rel.parents)
+    real_src_rel = PurePosixPath(
+        src_abs.relative_to(config.config.path.resolve()).as_posix()
+    )
+    real_dst_rel = PurePosixPath(
+        dst_abs.relative_to(config.config.path.resolve()).as_posix()
+    )
+    watching.notify_change(
+        real_src_rel, *real_src_rel.parents, real_dst_rel, *real_dst_rel.parents
+    )
     return HTTPResponse(status=201 if not dst_existed else 204)
 
 
@@ -399,8 +453,15 @@ def _to_mib_int(value_bytes: int) -> int:
     return round(value_bytes / (1 << 20))
 
 
-def _safe_relpath(path: str) -> tuple[PurePosixPath, Path]:
+def _safe_relpath(path: str, *, request=None) -> tuple[PurePosixPath, Path]:
     """Resolve a user path under storage root and enforce containment."""
+    token = auth.request_share_token(request) if request is not None else None
+    if token is not None:
+        vrel, _rrel, resolved, is_root = sharefs.resolve_virtual_path(token, path)
+        if is_root:
+            return vrel, config.config.path.resolve()
+        return vrel, resolved
+
     base = config.config.path.resolve()
     try:
         sanitized = filename.sanitize(unquote(path))
@@ -413,8 +474,12 @@ def _safe_relpath(path: str) -> tuple[PurePosixPath, Path]:
     return rel, resolved
 
 
-def _resolve_from_relpath(rel: PurePosixPath) -> Path:
+def _resolve_from_relpath(rel: PurePosixPath, *, request=None) -> Path:
     """Resolve a relative path under storage root and enforce containment."""
+    token = auth.request_share_token(request) if request is not None else None
+    if token is not None:
+        return sharefs.resolve_virtual_rel_to_real(token, rel)
+
     base = config.config.path.resolve()
     resolved = (base / rel).resolve()
     if not resolved.is_relative_to(base):
@@ -423,9 +488,12 @@ def _resolve_from_relpath(rel: PurePosixPath) -> Path:
 
 
 async def _send_static_file(request, name: str, *, head_only: bool):
-    _, path = _safe_relpath(name)
+    _, path = _safe_relpath(name, request=request)
 
-    st = await asyncio.to_thread(path.stat)
+    try:
+        st = await asyncio.to_thread(path.stat)
+    except FileNotFoundError:
+        raise NotFound(f"File not found: {name}") from None
     if path.is_dir():
         raise NotFound(f"Not a file: {name}")
 
@@ -513,8 +581,12 @@ def _parse_range_header(header: str, size: int) -> tuple[int, int] | None:
     return start, size
 
 
-def _get_key_paths(wanted: set[str]) -> dict[str, PurePosixPath]:
+def _get_key_paths(request, wanted: set[str]) -> dict[str, PurePosixPath]:
     """Map file keys to their current relative filesystem paths."""
+    token = auth.request_share_token(request)
+    if token is not None:
+        return sharefs.key_paths_for_token(token, wanted)
+
     loc = PurePosixPath()
     ret: dict[str, PurePosixPath] = {}
     with watching.state.lock:
@@ -533,7 +605,9 @@ def _get_key_paths(wanted: set[str]) -> dict[str, PurePosixPath]:
 # ---------------------------------------------------------------------------
 
 
-def _parse_webdav_destination(dest_header: str) -> tuple[PurePosixPath, Path]:
+def _parse_webdav_destination(
+    dest_header: str, *, request=None
+) -> tuple[PurePosixPath, Path]:
     """Parse a WebDAV Destination header and resolve it to a storage path."""
     parsed = urlparse(dest_header)
     raw_path = parsed.path  # still percent-encoded
@@ -544,7 +618,7 @@ def _parse_webdav_destination(dest_header: str) -> tuple[PurePosixPath, Path]:
         rel_str = raw_path[len(prefix) + 1 :]
     else:
         raise BadRequest("Destination must be within /files")
-    return _safe_relpath(rel_str)
+    return _safe_relpath(rel_str, request=request)
 
 
 def _rel_to_href(rel: PurePosixPath, *, is_dir: bool) -> str:

cista/preview.py

@@ -25,7 +25,7 @@ from sanic import Blueprint, empty, raw, redirect
 from sanic.exceptions import NotFound
 from sanic.log import logger
 
-from cista import auth, config
+from cista import auth, config, sharefs
 from cista.preview_worker import PreviewRequest, PreviewResponse
 from cista.util.filename import sanitize
 
@@ -388,8 +388,16 @@ async def preview(req, path):
     maxsize = int(req.args.get("px", 1024))
     maxzoom = float(req.args.get("zoom", 2.0))
     quality = int(req.args.get("q", 60))
-    rel = PurePosixPath(sanitize(unquote(path)))
+    share_token = auth.request_share_token(req)
-    filepath = config.config.path / rel
+    if share_token is not None:
+        rel, _real_rel, filepath, is_root = sharefs.resolve_virtual_path(
+            share_token, path
+        )
+        if is_root:
+            raise NotFound from None
+    else:
+        rel = PurePosixPath(sanitize(unquote(path)))
+        filepath = config.config.path / rel
     try:
         stat = filepath.lstat()
     except FileNotFoundError:

cista/sharefs.py

@@ -0,0 +1,230 @@
+from __future__ import annotations
+
+from pathlib import Path, PurePosixPath
+from stat import S_ISDIR, S_ISREG
+from time import time
+from typing import NamedTuple
+
+from natsort import humansorted
+from sanic.exceptions import BadRequest, NotFound
+
+from cista import config, watching
+from cista.fileio import fuid
+from cista.protocol import FileEntry
+from cista.util.filename import sanitize
+
+
+class ShareRootEntry(NamedTuple):
+    alias: str
+    real_rel: PurePosixPath
+
+
+def _token_is_share(token: config.Token) -> bool:
+    return token.kind == "share" and bool(token.share_paths)
+
+
+def is_share_token(token: config.Token | None) -> bool:
+    return bool(token and _token_is_share(token))
+
+
+def build_share_roots(token: config.Token) -> list[ShareRootEntry]:
+    if not _token_is_share(token):
+        return []
+
+    base = config.config.path.resolve()
+    roots: list[ShareRootEntry] = []
+    used_aliases: set[str] = set()
+
+    for raw_path in token.share_paths:
+        try:
+            clean = sanitize(raw_path)
+        except ValueError:
+            continue
+        if not clean:
+            continue
+
+        rel = PurePosixPath(clean)
+        resolved = (base / rel).resolve()
+        if not resolved.is_relative_to(base) or not resolved.exists():
+            continue
+
+        display = rel.name or config.config.path.name
+        alias = display
+        suffix = 2
+        while alias in used_aliases:
+            alias = f"{display} ({suffix})"
+            suffix += 1
+        used_aliases.add(alias)
+        roots.append(ShareRootEntry(alias=alias, real_rel=rel))
+
+    return roots
+
+
+def resolve_virtual_path(
+    token: config.Token,
+    raw_path: str,
+) -> tuple[PurePosixPath, PurePosixPath, Path, bool]:
+    """Resolve a share-virtual path to real path.
+
+    Returns (virtual_rel, real_rel, real_abs, is_virtual_root).
+    """
+    base = config.config.path.resolve()
+    if raw_path.strip("/") == "":
+        return PurePosixPath(), PurePosixPath(), base, True
+
+    try:
+        clean = sanitize(raw_path)
+    except ValueError as e:
+        raise BadRequest(f"Invalid path: {e}") from e
+
+    if not clean:
+        return PurePosixPath(), PurePosixPath(), base, True
+
+    virtual_rel = PurePosixPath(clean)
+    roots = build_share_roots(token)
+    if not roots:
+        raise NotFound("Share token has no visible files")
+
+    root_by_alias = {r.alias: r.real_rel for r in roots}
+    first = virtual_rel.parts[0]
+    real_root = root_by_alias.get(first)
+    if real_root is None:
+        raise NotFound(f"Not found: {raw_path}")
+
+    rest = virtual_rel.parts[1:]
+    real_rel = real_root.joinpath(*rest) if rest else real_root
+    resolved = (base / real_rel).resolve()
+    if not resolved.is_relative_to(base):
+        raise BadRequest("Invalid path")
+    return virtual_rel, real_rel, resolved, False
+
+
+def real_to_virtual_aliases(token: config.Token) -> dict[PurePosixPath, str]:
+    return {entry.real_rel: entry.alias for entry in build_share_roots(token)}
+
+
+def _walk_virtual_entry(path: Path, name: str, level: int) -> list[FileEntry]:
+    st = path.lstat()
+    is_dir = S_ISDIR(st.st_mode)
+    is_file = S_ISREG(st.st_mode)
+    if not is_dir and not is_file:
+        return []
+
+    if is_file:
+        try:
+            allocated = watching.get_allocated_size(path, st)
+        except Exception:
+            allocated = st.st_size
+        return [
+            FileEntry(
+                level=level,
+                name=name,
+                key=fuid(st),
+                mtime=int(st.st_mtime),
+                size=st.st_size,
+                allocated=allocated,
+                isfile=1,
+            )
+        ]
+
+    children: list[tuple[int, str, object]] = []
+    for child in path.iterdir():
+        if child.name.startswith("."):
+            continue
+        try:
+            cst = child.lstat()
+        except FileNotFoundError:
+            continue
+        c_is_file = S_ISREG(cst.st_mode)
+        c_is_dir = S_ISDIR(cst.st_mode)
+        if not c_is_file and not c_is_dir:
+            continue
+        children.append((int(c_is_file), child.name, cst))
+
+    entries: list[FileEntry] = []
+    agg_mtime = int(st.st_mtime)
+    agg_size = 0
+    agg_alloc = 0
+
+    for _, child_name, _ in humansorted(children):
+        child_path = path / child_name
+        child_entries = _walk_virtual_entry(child_path, child_name, level + 1)
+        if not child_entries:
+            continue
+        head = child_entries[0]
+        agg_mtime = max(agg_mtime, head.mtime)
+        agg_size += head.size
+        agg_alloc += head.allocated
+        entries.extend(child_entries)
+
+    head = FileEntry(
+        level=level,
+        name=name,
+        key=fuid(st),
+        mtime=agg_mtime,
+        size=agg_size,
+        allocated=agg_alloc,
+        isfile=0,
+    )
+    return [head, *entries]
+
+
+def build_virtual_root(token: config.Token) -> list[FileEntry]:
+    roots = build_share_roots(token)
+    now = int(time())
+    root_key = config.derived_secret("share-root", token.key or "", token.created).hex()
+
+    entries: list[FileEntry] = []
+    total_size = 0
+    total_alloc = 0
+    root_mtime = 0
+
+    base = config.config.path.resolve()
+    for entry in roots:
+        real_abs = (base / entry.real_rel).resolve()
+        if not real_abs.is_relative_to(base) or not real_abs.exists():
+            continue
+        try:
+            subtree = _walk_virtual_entry(real_abs, entry.alias, 1)
+        except OSError:
+            continue
+        if not subtree:
+            continue
+        head = subtree[0]
+        total_size += head.size
+        total_alloc += head.allocated
+        root_mtime = max(root_mtime, head.mtime)
+        entries.extend(subtree)
+
+    root = FileEntry(
+        level=0,
+        name="",
+        key=root_key,
+        mtime=root_mtime or now,
+        size=total_size,
+        allocated=total_alloc,
+        isfile=0,
+    )
+    return [root, *entries]
+
+
+def key_paths_for_token(
+    token: config.Token, wanted: set[str]
+) -> dict[str, PurePosixPath]:
+    ret: dict[str, PurePosixPath] = {}
+    loc = PurePosixPath()
+    root = build_virtual_root(token)
+    for f in root:
+        loc = PurePosixPath(*loc.parts[: f.level - 1]) / f.name
+        if f.key in wanted and f.key not in ret:
+            ret[f.key] = loc
+            if len(ret) == len(wanted):
+                break
+    return ret
+
+
+def resolve_virtual_rel_to_real(token: config.Token, rel: PurePosixPath) -> Path:
+    _vrel, _rrel, real_abs, is_root = resolve_virtual_path(token, rel.as_posix())
+    if is_root:
+        raise BadRequest("Virtual root is not a writable filesystem path")
+    return real_abs

frontend/src/components/SelectionToolbar.vue

@@ -15,6 +15,11 @@
     </div>
     <span class="select-size">{{ selectionDisplay.size }}</span>
     <DownloadButton />
+    <button
+      class="action-button"
+      title="Copy share link (Alt-click for read/write)"
+      @click="copyShareLink"
+    >share</button>
     <SvgButton name="copy" tooltip="Copy here" @click="op('cp', dst)" />
     <SvgButton name="paste" tooltip="Move here" @click="op('mv', dst)" />
     <SvgButton name="trash" tooltip="Delete ⚠️" @click="op('rm')" />
@@ -30,6 +35,8 @@
 
 <script setup lang="ts">
 import { apiFetch } from '@/repositories/Client'
+import type { ISimpleError } from '@/repositories/Client'
+import { createShareToken } from '@/repositories/User'
 import router from '@/router'
 import { useMainStore } from '@/stores/main'
 import { formatSize } from '@/utils'
@@ -170,6 +177,35 @@ const op = async (opName: string, dst?: string) => {
     }
   }
 }
+
+const copyShareLink = async (ev: MouseEvent) => {
+  const mode: 'ro' | 'rw' = ev.altKey ? 'rw' : 'ro'
+  const sel = store.selectedFiles
+  const paths = sel.keys
+    .map(key => {
+      const doc = sel.docs[key]
+      if (!doc) return ''
+      if (doc.loc === '/' || !doc.loc) return doc.name
+      return `${doc.loc}/${doc.name}`
+    })
+    .filter(Boolean)
+
+  if (!paths.length) {
+    store.showToast('No selected files')
+    return
+  }
+
+  try {
+    const token = await createShareToken(paths, mode)
+    await navigator.clipboard.writeText(token.url)
+    store.showToast(
+      mode === 'rw' ? 'Copied read/write share link' : 'Copied share link'
+    )
+  } catch (e) {
+    const httpError = e as ISimpleError
+    store.showToast(httpError.message || 'Failed to create share link')
+  }
+}
 </script>
 
 <style>

frontend/src/repositories/User.ts

@@ -93,3 +93,14 @@ export async function deleteToken(tokenId: string) {
   const data = await Client.delete(`${url_tokens}/${tokenId}`)
   return data
 }
+
+export async function createShareToken(paths: string[], mode: 'ro' | 'rw' = 'ro') {
+  const data = await Client.post('/api/share-tokens', { paths, mode })
+  return data as {
+    id: string
+    key: string
+    url: string
+    mode: 'ro' | 'rw'
+    paths: string[]
+  }
+}

tests/test_files_auth.py

@@ -106,16 +106,38 @@ def setup_storage(tmp_path: Path):
     user = config.User()
     auth.set_password(user, "secret")
     token = config.Token(key="test_token_123", username="alice")
+    share_ro = config.Token(
+        key="share_ro_123",
+        username="alice",
+        kind="share",
+        mode="ro",
+        share_paths=["hello.txt", "docs"],
+    )
+    share_rw = config.Token(
+        key="share_rw_123",
+        username="alice",
+        kind="share",
+        mode="rw",
+        share_paths=["docs"],
+    )
     config.config = config.Config(
         path=tmp_path,
         listen=":0",
         public=False,
         users={"alice": user},
-        tokens={"test_token_123": token},
+        tokens={
+            "test_token_123": token,
+            "share_ro_123": share_ro,
+            "share_rw_123": share_rw,
+        },
     )
     watching.state.root = []
     watching.rootpath = tmp_path
     (tmp_path / "hello.txt").write_text("hello", encoding="utf-8")
+    (tmp_path / "secret.txt").write_text("secret", encoding="utf-8")
+    (tmp_path / "docs").mkdir()
+    (tmp_path / "docs" / "a.txt").write_text("A", encoding="utf-8")
+    (tmp_path / "docs" / "b.txt").write_text("B", encoding="utf-8")
     yield tmp_path
     watching.state.root = []
 
@@ -222,3 +244,49 @@ async def test_ntlm_auth_with_token(client):
     )
     assert res3.status_code == 200
     assert res3.body == b"hello"
+
+
+@pytest.mark.asyncio
+async def test_share_token_limits_visible_paths(client):
+    _, res = await client.get(
+        "/files/docs/a.txt", headers=_basic_auth("token", "share_ro_123")
+    )
+    assert res.status_code == 200
+    assert res.body == b"A"
+
+    _, res = await client.get(
+        "/files/hello.txt", headers=_basic_auth("token", "share_ro_123")
+    )
+    assert res.status_code == 200
+    assert res.body == b"hello"
+
+    _, res = await client.get(
+        "/files/secret.txt", headers=_basic_auth("token", "share_ro_123")
+    )
+    assert res.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_share_token_read_only_blocks_writes(client):
+    _, res = await client.delete(
+        "/files/hello.txt", headers=_basic_auth("token", "share_ro_123")
+    )
+    assert res.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_share_token_rw_allows_writes_in_scope_only(client):
+    _, res = await client.delete(
+        "/files/docs/a.txt", headers=_basic_auth("token", "share_rw_123")
+    )
+    assert res.status_code == 204
+
+    _, res = await client.get(
+        "/files/docs/a.txt", headers=_basic_auth("token", "share_rw_123")
+    )
+    assert res.status_code == 404
+
+    _, res = await client.delete(
+        "/files/secret.txt", headers=_basic_auth("token", "share_rw_123")
+    )
+    assert res.status_code == 404

tests/test_tokens.py

@@ -7,6 +7,7 @@ import pytest_asyncio
 from sanic import Sanic
 
 from cista import auth, config, watching
+from cista.api import bp as api_bp
 from cista.auth import bp as auth_bp
 
 
@@ -41,6 +42,9 @@ def setup_storage(tmp_path: Path):
     _persist_config()
     watching.state.root = []
     watching.rootpath = tmp_path
+    (tmp_path / "hello.txt").write_text("hello", encoding="utf-8")
+    (tmp_path / "docs").mkdir()
+    (tmp_path / "docs" / "a.txt").write_text("A", encoding="utf-8")
     yield tmp_path
     watching.state.root = []
 
@@ -56,6 +60,7 @@ async def client(setup_storage: Path):
         "PROPFIND",
     )
     app.blueprint(auth_bp)
+    app.blueprint(api_bp)
     yield app.asgi_client
 
 
@@ -189,3 +194,27 @@ async def test_token_user_scoped(client):
     # Actually the token key lookup will fail, and since there's no session fallback...
     # With auth header present but invalid, it should return 401
     assert res.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_create_share_token(client):
+    _, res = await client.post(
+        "/api/share-tokens",
+        json={"paths": ["hello.txt", "docs"], "mode": "ro", "name": "selection"},
+        headers={"Authorization": _basic_auth("alice", "secret")},
+    )
+    assert res.status_code == 200
+    data = res.json
+    assert data["kind"] == "share"
+    assert data["mode"] == "ro"
+    assert data["paths"] == ["hello.txt", "docs"]
+    assert "token:" in data["url"]
+
+    _, res = await client.get(
+        "/auth/tokens",
+        headers={"Authorization": _basic_auth("alice", "secret")},
+    )
+    assert res.status_code == 200
+    share_tokens = [t for t in res.json["tokens"] if t.get("kind") == "share"]
+    assert len(share_tokens) == 1
+    assert share_tokens[0]["mode"] == "ro"