codex-app-server/cloud/uploads.py

"""任务附件：multipart 解析、相对路径校验、暂存包落盘。

控制台上传流程（见 README）：先 POST /uploads 拿 upload_id，
再 POST /tasks 带 upload_id 绑定；worker 派发时把暂存目录经
host.py --seed-dir 拷进 worktree。图片与普通文件不区分对待，
一律是 worktree 里的文件（模型能否“看懂”图片取决于模型本身，
本模块不承诺多模态理解）。

上限：最多 200 个文件，单文件 20MB，文件合计 100MB。路径
只允许 worktree 内的相对路径，绝不接受绝对路径、..、空段、
反斜杠与 NUL（防止写出暂存目录或 worktree）。
"""

import json
import os
import re
import uuid

MAX_FILES = 200
MAX_FILE_BYTES = 20 * 1024 * 1024
MAX_TOTAL_BYTES = 100 * 1024 * 1024
# multipart 信封本身的开销：413 预检按这个上限放行原始 body，
# 真正的文件合计仍按 MAX_TOTAL_BYTES 卡。
MAX_RAW_BYTES = MAX_TOTAL_BYTES + 10 * 1024 * 1024
BUNDLE_TTL_SECONDS = 86400
DEFAULT_UPLOAD_ROOT = "/opt/codex-app-server/work/uploads"


def upload_root(upload_dir=None):
    """暂存根目录：显式传入 > 环境变量 > 默认值（与 server 侧一致）。"""
    if upload_dir:
        return str(upload_dir)
    return os.environ.get("CODEX_UPLOAD_DIR") or DEFAULT_UPLOAD_ROOT


def parse_multipart(body, content_type):
    """解析 multipart/form-data，返回 (fields, files)。

    fields：name -> 文本（UTF-8 解码失败即 ValueError）。
    files：[(filename, bytes)]，按 part 出现顺序。
    只要结构对不上（缺 boundary、缺头、缺分隔）就 ValueError，
    调用方按 400 处理，不静默丢 part。"""
    if not isinstance(body, (bytes, bytearray)):
        raise ValueError("multipart body 必须是 bytes")
    m = re.search(r"boundary=([^;]+)", content_type or "")
    if not m:
        raise ValueError("multipart 缺 boundary")
    boundary = m.group(1).strip().strip('"').encode("utf-8")
    if not boundary:
        raise ValueError("multipart boundary 为空")
    delimiter = b"--" + boundary
    chunks = body.split(delimiter)
    # 首段是信封前导（空），末段是 -- 收尾；中间每段一个 part。
    fields = {}
    files = []
    for chunk in chunks[1:]:
        if chunk.startswith(b"--"):
            break
        if chunk.startswith(b"\r\n"):
            chunk = chunk[2:]
        elif chunk.startswith(b"\n"):
            chunk = chunk[1:]
        if chunk.endswith(b"\r\n"):
            chunk = chunk[:-2]
        elif chunk.endswith(b"\n"):
            chunk = chunk[:-1]
        head, sep, payload = chunk.partition(b"\r\n\r\n")
        if not sep:
            head, sep, payload = chunk.partition(b"\n\n")
            if not sep:
                raise ValueError("multipart part 缺头部分隔")
        headers = head.decode("utf-8", errors="replace")
        disp = None
        for line in headers.splitlines():
            if line.lower().startswith("content-disposition:"):
                disp = line.split(":", 1)[1].strip()
                break
        if disp is None:
            raise ValueError("multipart part 缺 Content-Disposition")
        name_m = re.search(r'name="([^"]*)"', disp)
        if not name_m:
            raise ValueError("multipart part 缺 name")
        name = name_m.group(1)
        file_m = re.search(r'filename="([^"]*)"', disp)
        if file_m is not None:
            files.append((file_m.group(1), bytes(payload)))
        else:
            try:
                fields[name] = payload.decode("utf-8")
            except UnicodeDecodeError:
                raise ValueError("multipart 文本字段不是 UTF-8: %s" % name)
    return fields, files


def validate_rel_path(path):
    """校验 worktree 内相对路径；非法返回错误文本，合法返回 None。"""
    if not isinstance(path, str) or not path:
        return "附件路径不能为空"
    if len(path) > 512:
        return "附件路径过长: %s" % path[:50]
    if path.startswith("/") or path.startswith("\\"):
        return "附件路径不能是绝对路径: %s" % path
    if re.match(r"^[A-Za-z]:", path):
        return "附件路径不能带盘符: %s" % path
    if "\x00" in path or "\\" in path:
        return "附件路径含非法字符: %s" % path
    parts = path.split("/")
    for part in parts:
        if part in ("", ".", ".."):
            return "附件路径含非法段: %s" % path
    return None


def validate_bundle(paths, sizes):
    """校验一包附件：数量、单文件、合计、路径合法、路径不重复。"""
    if len(paths) > MAX_FILES:
        return "附件最多 %d 个" % MAX_FILES
    if len(paths) != len(sizes):
        return "附件路径与文件数量不一致"
    seen = set()
    total = 0
    for path, size in zip(paths, sizes):
        err = validate_rel_path(path)
        if err:
            return err
        if path in seen:
            return "附件路径重复: %s" % path
        seen.add(path)
        if size > MAX_FILE_BYTES:
            return "单个附件超过 %dMB: %s" % (
                MAX_FILE_BYTES // (1024 * 1024), path)
        total += size
    if total > MAX_TOTAL_BYTES:
        return "附件合计超过 %dMB" % (MAX_TOTAL_BYTES // (1024 * 1024))
    return None


def save_bundle(root, paths, datas):
    """把一包附件写进暂存目录，返回 (bundle_id, [(rel_path, size)])。

    调用方须先过 validate_bundle；这里再逐路径校验一次（纵深
    防御：直接调本函数的调用方漏了校验也不至于把 ../ 写出包外）。
    目录是全新 uuid 子目录，里面不会有符号链接，写盘按相对路径
    逐级建目录。"""
    bundle_id = uuid.uuid4().hex
    bundle_dir = os.path.join(root, bundle_id)
    os.makedirs(bundle_dir, exist_ok=False)
    out = []
    for path, data in zip(paths, datas):
        err = validate_rel_path(path)
        if err:
            raise ValueError(err)
        dest = os.path.join(bundle_dir, path)
        os.makedirs(os.path.dirname(dest), exist_ok=True)
        with open(dest, "wb") as f:
            f.write(data)
        out.append((path, len(data)))
    return bundle_id, out


def bundle_dir_for(root, bundle_id):
    return os.path.join(root, bundle_id)


def parse_paths_field(fields, files):
    """控制台传 paths JSON（与 file part 一一对应）；不传时退回
    各 part 自带的 filename（API 直调可用）。"""
    raw = fields.get("paths")
    if raw is None:
        return [name for name, _data in files]
    try:
        paths = json.loads(raw)
    except ValueError:
        raise ValueError("paths 不是合法 JSON")
    if not isinstance(paths, list) or not all(
            isinstance(p, str) for p in paths):
        raise ValueError("paths 必须是字符串数组")
    if len(paths) != len(files):
        raise ValueError("附件路径与文件数量不一致")
    return paths