zcbot/web/userfiles.py

"""user_root 文件操作的安全原语(从 app.py 析出,2026-07-23 拆分)。

所有 files/kb 类端点的路径边界收口:user_root 解析、越界校验(`../` / 绝对路径 /
symlink)、目录枚举、批量 transfer 预检。安全关键 —— 改动需过 tests/test_pptx_render.py
的越界用例。
"""
from __future__ import annotations

from datetime import datetime as _dt
from pathlib import Path
from uuid import UUID

from fastapi import HTTPException
from sqlalchemy import func, or_, select

from core.storage import session_scope
from core.storage.models import Task

from .common import CHANNEL_MIRROR_KINDS


def load_user_root(user_id: UUID) -> Path:
    """user_root = `<workspace>/users/<user_id>/`,所有 files API 的边界。
    若目录尚未存在自动 mkdir(空 user 首次访问也能拿到根)。
    """
    from core.agent_builder import resolve_workspace, user_root
    ws = resolve_workspace(None)
    return user_root(ws, user_id)


def safe_join(root: Path, rel: str) -> Path:
    """归一用户路径到 absolute,并校验仍在 root 内。防 `../` / 绝对 path / symlink 越界。"""
    rel = (rel or "").strip()
    if not rel:
        return root.resolve()
    if rel[0] in ("/", "\\"):
        raise HTTPException(400, f"absolute-style path not allowed: {rel!r}")
    if Path(rel).is_absolute():
        raise HTTPException(400, f"absolute path not allowed: {rel!r}")
    target = (root / rel).resolve()
    try:
        target.relative_to(root.resolve())
    except ValueError:
        raise HTTPException(400, f"path escapes user_root: {rel!r}")
    return target


def rel_to(root: Path, target: Path) -> str:
    try:
        rel = target.resolve().relative_to(root.resolve()).as_posix()
    except ValueError:
        return ""
    return "" if rel == "." else rel


def enumerate_files(
    root: Path,
    current: Path,
    *,
    include_hidden_dirs: bool = False,
) -> tuple[list[dict], list[dict], bool]:
    """枚举 current 下条目 + 拼面包屑。size raw bytes,mtime ISO 串(前端 humanize)。
    dotfile 始终隐藏；调用方仅在已验证的 task 工作目录子树内允许展示 dotdir。
    即使开关开启，`.env` 等点文件仍不展示。
    """
    entries: list[dict] = []
    exists = current.exists()
    if exists and current.is_dir():
        try:
            raw = sorted(current.iterdir(), key=lambda p: (p.is_file(), p.name.lower()))
        except OSError:
            raw = []
        for p in raw:
            if p.name.startswith(".") and not (include_hidden_dirs and p.is_dir()):
                continue
            try:
                st = p.stat()
            except OSError:
                continue
            entries.append({
                "name": p.name,
                "is_dir": p.is_dir(),
                "size": st.st_size if p.is_file() else None,
                "mtime": _dt.fromtimestamp(st.st_mtime).isoformat(timespec="seconds"),
                "rel": rel_to(root, p),
            })
    cur_rel = rel_to(root, current)
    crumbs = [{"label": "/", "rel": ""}]
    # cur_rel == "." 表示当前就在 root(target.relative_to(root) 返 Path(".")),
    # 不该再追加一个无意义的 "." crumb
    if cur_rel and cur_rel != ".":
        acc = ""
        for part in cur_rel.split("/"):
            acc = f"{acc}/{part}" if acc else part
            crumbs.append({"label": part, "rel": acc})
    return entries, crumbs, exists


def system_wd_names(user_id: UUID) -> set[str]:
    """有任务入口的系统工作目录末段名(定时任务执行目录 + 渠道镜像对话目录)。

    这些目录点开对应入口(定时任务运行历史 / 渠道固定卡片)时文件面板会自动跳进去,
    根目录列表隐藏它们只是降噪,不是权限拦截 —— 带 path 直接访问照常放行(任务跳转
    依赖这条路)。孤儿目录(task 已软删 / job 物理删后的漏网)没有别的 UI 入口,
    刻意不隐藏,留在根目录可见。
    """
    with session_scope() as s:
        rows = s.execute(
            select(Task.working_dir).where(
                Task.user_id == user_id,
                Task.deleted_at.is_(None),
                Task.working_dir.isnot(None),
                or_(
                    Task.scheduled_job_id.isnot(None),
                    func.coalesce(Task.channel, "web").in_(CHANNEL_MIRROR_KINDS),
                ),
            )
        ).all()
    return {wd.rstrip("/").rsplit("/", 1)[-1] for (wd,) in rows if wd}


def validate_transfer(
    root: Path, paths: list[str], dest_dir: str,
) -> tuple[list[Path], Path]:
    """预检批量 transfer:解析所有源 + 目标,任意一项不合法即整批 abort(无 FS 副作用)。

    返回 (sources, dest_dir_path)。不区分 copy / move(顶层 working_dir 闸由路由各自加)。
    校验项:
    - paths 非空;每个源在 user_root 内 + 存在;不能是 user_root 本身
    - dest_dir 存在 + 是目录(可以是 user_root)
    - 源不能与 dest_dir 相同(自移动)
    - dest_dir 不能在源的子树内(不能把 a/ 搬进 a/b/)
    - 源不能已是 dest_dir 直接子项(原地移动,no-op)
    - 同批次源 leaf 名不能重复(俩 a.txt 会撞 dest/a.txt)
    - dest_dir/<name> 不能已存在(整批 409,不静默覆盖)
    """
    if not paths:
        raise HTTPException(400, "paths is empty")
    dest = safe_join(root, dest_dir)
    if not dest.exists():
        raise HTTPException(404, f"dest_dir not found: {dest_dir!r}")
    if not dest.is_dir():
        raise HTTPException(400, f"dest_dir is not a directory: {dest_dir!r}")
    dest_r = dest.resolve()

    sources: list[Path] = []
    seen_names: set[str] = set()
    for p in paths:
        src = safe_join(root, p)
        if not src.exists():
            raise HTTPException(404, f"source not found: {p!r}")
        src_r = src.resolve()
        if src_r == root.resolve():
            raise HTTPException(400, "cannot transfer user_root")
        if src_r == dest_r:
            raise HTTPException(400, f"source equals dest_dir: {p!r}")
        # dest 在 src 子树内 → 自嵌套
        try:
            dest_r.relative_to(src_r)
            raise HTTPException(
                400, f"cannot transfer {p!r} into its own subtree"
            )
        except ValueError:
            pass
        # 已是 dest 直接子项 → no-op
        if src.parent.resolve() == dest_r:
            raise HTTPException(
                400, f"{p!r} already directly under dest_dir"
            )
        name = src.name
        if name in seen_names:
            raise HTTPException(400, f"duplicate source leaf name in batch: {name!r}")
        seen_names.add(name)
        target = dest / name
        if target.exists():
            raise HTTPException(
                409, f"target already exists: {rel_to(root, target)!r}"
            )
        sources.append(src)
    return sources, dest