CloudDrive2 API 调用指南.md 12 KB


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

CloudDrive2 API 调用指南

基本信息

  • 服务器:[[VPS/服务器清单|ZLSH_NAS]](192.168.6.23,公司白群晖 DS923+)
  • CloudDrive2 版本:v1.0.13(Synology 套件安装)
  • 安装路径:/volume1/@appstore/CloudDrive2/
  • 数据路径:/volume1/@appdata/CloudDrive2/Config/
  • Web UI:http://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:初始版本