CloudDrive2 API 调用指南.md 12 KB


title: CloudDrive2 API 调用指南 date: 2026-08-01

tags: [clouddrive2, api, nas, grpc, webdav]

CloudDrive2 API 调用指南

基本信息

  • 服务器:[[VPS/服务器清单|ZLSH_NAS]](192.168.6.23,公司白群晖 DS923+)
  • CloudDrive2 版本:v1.0.13(Synology 套件安装)
  • 安装路径/volume1/@appstore/CloudDrive2/
  • 数据路径/volume1/@appdata/CloudDrive2/Config/
  • Web UIhttp://192.168.6.23:19798 / https://192.168.6.23:19799
  • API 类型:gRPC-web(HTTP/1.1 承载的 gRPC)
  • gRPC 服务名clouddrive.CloudDriveFileSrv

当前已挂载的云盘

云盘 路径 账号
115 网盘 /115open ID: 22499506
OneDrive /OneDrive zhensolid@vip.qq.com
阿里云盘Open /阿里云盘Open c370ce6b7e
天翼云盘 /{4130419372} 18184109687@189.cn
123 云盘 /{123-1821196620}

API Token

属性
Token 9b194899-23fc-4db9-bebe-b2eba5a8f07a
名称 hermes
根目录 /
过期 永不过期
权限 全部 45 项

鉴权方式

Authorization: Bearer 9b194899-23fc-4db9-bebe-b2eba5a8f07a

错误码:

  • grpc-status: 16 Invalid auth token — token 格式错误
  • grpc-status: 16 No valid auth token — 未提供 token

gRPC-web 协议

请求格式

POST http://192.168.6.23:19798/clouddrive.CloudDriveFileSrv/<方法名>
Content-Type: application/grpc-web+proto
X-Grpc-Web: 1
Authorization: Bearer <token>

Body: gRPC-web 帧

帧结构

请求帧: | 1B flags(0x00) | 4B 大端长度 | protobuf 消息体 |
响应帧: | 1B flags | 4B 大端长度 | protobuf 消息体 |

响应帧 flags 含义

  • 0x00 = 普通数据帧
  • 0x80 = 最后一帧,包含 gRPC trailer(grpc-status 等)在 protobuf 数据之后

gRPC 状态码

含义
0 OK
5 NOT_FOUND
7 PERMISSION_DENIED
12 UNIMPLEMENTED
13 INTERNAL
16 UNAUTHENTICATED

✅ 完整 Python 客户端(已验证可用)

以下客户端已通过实测,支持列出任意云盘目录。

#!/usr/bin/env python3
"""
CloudDrive2 gRPC-web API 客户端
依赖: 无(仅标准库)
用法: python3 clouddrive_client.py
"""
import struct
import subprocess
import tempfile
import os
import re

BASE = "http://192.168.6.23:19798"
SVC = "clouddrive.CloudDriveFileSrv"
TOKEN = "9b194899-23fc-4db9-bebe-b2eba5a8f07a"


def make_frame(proto_body: bytes) -> bytes:
    """构建 gRPC-web 请求帧"""
    return b'\x00' + struct.pack('>I', len(proto_body)) + proto_body


def call(method: str, proto_body: bytes = b"") -> bytes:
    """
    调用 gRPC 方法,返回纯 protobuf 响应体(已跳过帧头和 trailer)
    抛出 RuntimeError 如果 grpc-status != 0
    """
    data = make_frame(proto_body)
    with tempfile.NamedTemporaryFile(delete=False) as f:
        f.write(data)
        tmp = f.name

    try:
        result = subprocess.run([
            'curl', '-s', '--connect-timeout', '15', '-X', 'POST',
            f'{BASE}/{SVC}/{method}',
            '-H', 'Content-Type: application/grpc-web+proto',
            '-H', 'X-Grpc-Web: 1',
            '-H', f'Authorization: Bearer {TOKEN}',
            '--data-binary', f'@{tmp}',
            '-o', '/tmp/grpc_body.bin', '-D', '/tmp/grpc_hdr.txt'
        ], capture_output=True, text=True, timeout=20)

        with open('/tmp/grpc_body.bin', 'rb') as fb:
            raw = fb.read()

        # 解析 gRPC-web 响应帧
        bodies = []
        i = 0
        grpc_status = None
        while i + 5 <= len(raw):
            flags = raw[i]; i += 1
            msg_len = struct.unpack('>I', raw[i:i+4])[0]; i += 4
            if i + msg_len > len(raw):
                break
            chunk = raw[i:i+msg_len]
            i += msg_len

            if flags & 0x80:
                # trailer 帧:解析 grpc-status
                for line in chunk.decode('utf-8', errors='replace').split('\r\n'):
                    if line.startswith('grpc-status:'):
                        grpc_status = int(line.split(':')[1].strip())
            else:
                bodies.append(chunk)

        if grpc_status and grpc_status != 0:
            raise RuntimeError(f"gRPC error: status={grpc_status}")

        return b"".join(bodies)
    finally:
        os.unlink(tmp)


def encode_path(path: str) -> bytes:
    """编码 path 字段(field 1, wire type 2 = string)"""
    data = path.encode('utf-8')
    return bytes([0x0a, len(data)]) + data


def extract_strings(body: bytes, min_len: int = 3) -> list[str]:
    """从 protobuf 中提取所有可读 ASCII 字符串"""
    result = []
    current = b""
    for byte in body:
        if 32 <= byte < 127:
            current += bytes([byte])
        else:
            if len(current) >= min_len:
                result.append(current.decode('ascii', errors='replace'))
            current = b""
    if len(current) >= min_len:
        result.append(current.decode('ascii', errors='replace'))
    return result


def list_dir(path: str) -> list[str]:
    """
    列出目录内容,返回文件名列表。
    自动过滤系统噪音(ID、路径、URL 等)。
    """
    body = call("GetSubFiles", encode_path(path))
    strings = extract_strings(body, min_len=3)

    skip = ['grpc-status', 'token=', 'https://', '/static/',
            'CloudDrive', '<br>', '115open', 'OneDrive', 'Open']
    seen = set()
    names = []

    for s in strings:
        if any(k in s for k in skip):
            continue
        if re.match(r'^\d{10,}$', s):          # 纯数字 ID
            continue
        if re.match(r'^[A-F0-9]{25,}$', s):    # 哈希值
            continue
        if s.startswith('/') or s.startswith('b/'):
            continue
        if '@' in s:
            continue

        name = s.strip()
        if name and name not in seen and len(name) > 1:
            seen.add(name)
            names.append(name)

    return names


# ========== 使用示例 ==========
if __name__ == '__main__':
    # 列出 115 网盘根目录
    print("📂 115 网盘:")
    for name in list_dir("/115open"):
        print(f"   • {name}")

    print()
    print("📂 OneDrive:")
    for name in list_dir("/OneDrive"):
        print(f"   • {name}")

    print()
    print("📂 阿里云盘:")
    for name in list_dir("/阿里云盘Open"):
        print(f"   • {name}")

运行

python3 clouddrive_client.py

不需要安装任何第三方依赖(纯标准库 + curl)。


✅ 已验证的 gRPC 方法

GetSubFiles — 列出目录

请求{path: "/115open"} (field 1, string)

body = call("GetSubFiles", encode_path("/115open"))

实测路径

路径 内容
/ 列出所有挂载的云盘
/115open 115 网盘根目录(含 Atomic Heart、PRAGMATA 等)
/OneDrive OneDrive 目录(DOS、KaliLinux、PythonStudy、SQL 等)
/阿里云盘Open 阿里云盘根目录

GetApiTokenInfo — 获取 token 信息

请求{token: "9b194899-..."} (field 1, string)

req = bytes([0x0a, 36]) + TOKEN.encode()
body = call("GetApiTokenInfo", req)
# 返回 token 元数据(根目录、权限等)

GetMountPoints — 获取挂载点列表

body = call("GetMountPoints")
# 返回本地挂载路径信息

GetSpaceInfo — 空间使用情况

请求{path: "/115open"}

body = call("GetSpaceInfo", encode_path("/115open"))

常用 gRPC 方法速查

⚠️ 方法名从 CloudDrive2 二进制中提取,注意不要用 GetSubFilesGetSubFiles 这种重复名(grep 拼接产物),实际方法名是 GetSubFiles

文件操作

方法名 功能 参数
GetSubFiles 列出子目录 path (field 1, string)
CreateFolder 创建文件夹 path
CreateFile 创建文件 path, content
RenameFile 重命名 old_path, new_path
DeleteFile 删除(回收站) path
DeleteFilePermanently 永久删除 path
CopyFile 复制 src, dst
MoveFile 移动 src, dst
GetSearchResults 搜索 query, path
GetFileDetailProperties 文件详情 path

下载/上传

方法名 功能
GetDownloadUrlPath 获取下载直链
WriteToFile 写入文件
GetDownloadFileList 下载任务列表
GetUploadFileList 上传任务列表

空间/系统

方法名 功能
GetSpaceInfo 空间使用情况
GetRuntimeInfo 运行时信息
GetSystemInfo 系统信息
GetSystemSettings 系统设置
SetSystemSettings 修改设置

Token 管理

方法名 功能
GetApiTokenInfo 查询 token 信息
ListTokens 列出所有 token
CreateToken 创建 token
RemoveToken 删除 token
ModifyToken 修改 token 权限

挂载点

方法名 功能
GetMountPoints 列出挂载点
AddMountPoint 添加挂载点
RemoveMountPoint 移除挂载点
Mount / Unmount 挂载/卸载

离线下载

方法名 功能
AddOfflineFiles 添加离线下载
ListOfflineFilesByPath 按路径列离线任务
ListAllOfflineFiles 所有离线任务
RemoveOfflineFiles 删除离线任务

Protobuf 编码速查

CloudDrive2 的请求参数简单,主要是 string 字段:

# field N, string value
def encode_string(field_num: int, value: str) -> bytes:
    tag = (field_num << 3) | 2   # wire type 2 = LEN
    data = value.encode('utf-8')
    return bytes([tag, len(data)]) + data

# 常用:
# path -> field 1:  0x0a + len + path_bytes
# token -> field 1: 0x0a + len + token_bytes

实测参数映射(通过交叉验证推测):

方法 参数 编码
GetSubFiles path (field 1) 0x0a + len + path
GetApiTokenInfo token (field 1) 0x0a + len + token
GetSpaceInfo path (field 1) 0x0a + len + path
CreateFolder path (field 1) 0x0a + len + path

踩坑记录

1. 方法名不要用 grep 拼接结果

从二进制 strings 提取时,GetSubFilesGetSubFiles 是 grep 把相邻同名方法串在一起。实际方法名是 GetSubFiles

2. gRPC-web response trailer 解析

CloudDrive2 的 gRPC 响应可能把 grpc-status 放在最后的 trailer 帧中(flags=0x80),而不是 HTTP header。需要解析 trailer 帧才能拿到正确的状态码。

if flags & 0x80:
    # trailer 帧内容: "grpc-status:0\r\ngrpc-message:...\r\n"
    for line in chunk.decode('utf-8').split('\r\n'):
        if line.startswith('grpc-status:'):
            status = int(line.split(':')[1])

3. Python requests 库版本问题

Hermes venv 中的 Python 3.11 + requests 的 urllib3 有类型注解兼容性问题。直接用 subprocess + curl 绕过,零依赖。

4. 路径编码

中文路径(如 /阿里云盘Open)直接用 UTF-8 编码即可,CloudDrive2 内部处理正确。

5. 空目录返回

GetSubFiles 对空目录返回空 body(0 字节 protobuf),这是正常的。


WebDAV(替代方案)

WebDAV 已启用但没有配置用户(dav_users: [])。

要在 Web UI 中添加 DAV 用户后使用:

# macOS 挂载
mkdir -p /tmp/clouddrive
mount_webdav http://192.168.6.23:19798/dav/ /tmp/clouddrive

相关笔记

  • [[家庭实验室/Docker服务/Docker-compose合集/Clouddrive2|CloudDrive2 Docker 部署]]
  • [[VPS/服务器清单|服务器清单]]
  • [[家庭实验室/群晖NAS/群晖密钥登录|群晖 SSH 密钥登录]]

更新日志

  • 2026-08-01 v2:添加完整 Python 客户端(已验证)、修复方法名、添加 trailer 解析、添加踩坑记录、添加全部 5 个云盘挂载信息
  • 2026-08-01 v1:初始版本