From 73278667e069dd28403cadd302095c53e4301617 Mon Sep 17 00:00:00 2001 From: r Date: Sun, 21 Jun 2026 12:48:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(files):=20=E8=B7=A8=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E5=99=A8=E6=96=87=E4=BB=B6=E7=9B=B4=E4=BC=A0=E4=B8=8E=E6=89=93?= =?UTF-8?q?=E5=8C=85=E6=8A=95=E9=80=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A→B SFTP 中继与一键 tar 打包 curl 拉取；含审计修复与 18 项单测。 --- docs/audit/2026-06-21-server-file-transfer.md | 46 ++ ...-06-21-server-file-transfer-audit-fixes.md | 35 ++ .../2026-06-21-server-file-transfer.md | 44 ++ .../plans/2026-06-21-server-file-transfer.md | 33 ++ .../2026-06-21-server-file-transfer-design.md | 50 +++ .../src/components/files/FilesDialogs.vue | 101 +++++ frontend/src/components/files/FilesList.vue | 7 + .../src/components/files/FilesToolbar.vue | 9 + .../src/composables/files/useFilesActions.ts | 203 +++++++++ .../src/composables/files/useFilesPage.ts | 17 + frontend/src/utils/auditLabels.ts | 4 + server/api/auth_jwt.py | 1 + server/api/schemas.py | 69 +++ server/api/sync_v2.py | 229 ++++++++++ .../services/server_file_transfer_service.py | 400 ++++++++++++++++++ .../redis/transfer_package_store.py | 41 ++ tests/test_server_file_transfer.py | 394 +++++++++++++++++ 17 files changed, 1683 insertions(+) create mode 100644 docs/audit/2026-06-21-server-file-transfer.md create mode 100644 docs/changelog/2026-06-21-server-file-transfer-audit-fixes.md create mode 100644 docs/changelog/2026-06-21-server-file-transfer.md create mode 100644 docs/design/plans/2026-06-21-server-file-transfer.md create mode 100644 docs/design/specs/2026-06-21-server-file-transfer-design.md create mode 100644 server/application/services/server_file_transfer_service.py create mode 100644 server/infrastructure/redis/transfer_package_store.py create mode 100644 tests/test_server_file_transfer.py diff --git a/docs/audit/2026-06-21-server-file-transfer.md b/docs/audit/2026-06-21-server-file-transfer.md new file mode 100644 index 00000000..99d7a62c --- /dev/null +++ b/docs/audit/2026-06-21-server-file-transfer.md @@ -0,0 +1,46 @@ +# 审计 — 2026-06-21 跨服务器文件传输 + +**Changelog**: `docs/changelog/2026-06-21-server-file-transfer.md`、`docs/changelog/2026-06-21-server-file-transfer-audit-fixes.md` + +## Step 3 规则扫描 + +| 文件 | 规则 | 结论 | +|------|------|------| +| `server_file_transfer_service.py` | 路径 `shlex.quote`、直传 500MB 上限、deliver 失败回滚 | PASS | +| `transfer-download` | token Redis TTL；JWT 前缀白名单 | PASS（已知：前缀对所有 method 生效） | +| `useFilesActions.ts` | 模式推荐基于 `transferItems`；relay/deliver 分路径缓存 | PASS | +| `test_server_file_transfer.py` | deliver 成功/回滚/超限、relay、store | PASS | + +## Closure + +| 项 | 状态 | 说明 | +|----|------|------| +| deliver 失败泄漏归档/token | FIXED | rollback cleanup | +| deliver 成功 token 残留 | FIXED | delete_package_meta | +| 右键传输模式误判 | FIXED | transferItems | +| 直传多项中途失败无回滚 | ACCEPT | v1 文档化 | + +## 文件清单 + +- `server/application/services/server_file_transfer_service.py` +- `server/infrastructure/redis/transfer_package_store.py` +- `server/api/sync_v2.py` +- `server/api/schemas.py` +- `server/api/auth_jwt.py` +- `frontend/src/composables/files/useFilesActions.ts` +- `frontend/src/composables/files/useFilesPage.ts` +- `frontend/src/components/files/FilesDialogs.vue` +- `frontend/src/components/files/FilesList.vue` +- `frontend/src/components/files/FilesToolbar.vue` +- `frontend/src/utils/auditLabels.ts` +- `tests/test_server_file_transfer.py` +- `docs/design/specs/2026-06-21-server-file-transfer-design.md` +- `docs/design/plans/2026-06-21-server-file-transfer.md` +- `docs/changelog/2026-06-21-server-file-transfer.md` +- `docs/changelog/2026-06-21-server-file-transfer-audit-fixes.md` + +## DoD + +- [x] 直传 + 打包投递 API 与前端对话框 +- [x] `pytest tests/test_server_file_transfer.py` 18 passed +- [x] changelog ≥10 行 diff --git a/docs/changelog/2026-06-21-server-file-transfer-audit-fixes.md b/docs/changelog/2026-06-21-server-file-transfer-audit-fixes.md new file mode 100644 index 00000000..e849a866 --- /dev/null +++ b/docs/changelog/2026-06-21-server-file-transfer-audit-fixes.md @@ -0,0 +1,35 @@ +# 跨服务器文件传输 — 审计修复 + +## 日期 + +2026-06-21 + +## 摘要 + +对跨机传输首版做走读审计，修复 deliver 失败资源泄漏、token 残留、前端模式误判与路径记忆混用。 + +## 审计发现与处置 + +| 级别 | 问题 | 处置 | +|------|------|------| +| P1 | `deliver` 拉取失败后源机 `/tmp` 归档与 Redis token 未清理 | 失败时 rollback：`rm` 归档 + `delete_package_meta` | +| P1 | `deliver` 成功后 token 仍有效但源归档已删，curl 重放 404 | 成功后 `delete_package_meta` | +| P2 | 右键「发送到其他服务器」未选中行时，模式推荐误用 `selectedFiles` | 引入 `transferItems`，按实际传输项判断 | +| P2 | localStorage 直传目录与打包路径共用 key，切换模式路径错乱 | 按 `relay`/`deliver` 分 key | +| P2 | 打包成功立即关对话框，`download_url` 无法复制 | 投递成功后保留对话框至用户关闭 | +| 已知限制 | 直传多项中途失败无回滚 | v1 接受；UI 已提示大目录用打包 | +| 已知限制 | 归档与直传单文件 500MB 上限 | 与现有 download 一致，未改 | + +## 涉及文件 + +- `server/application/services/server_file_transfer_service.py` +- `frontend/src/composables/files/useFilesActions.ts` +- `frontend/src/components/files/FilesDialogs.vue` + +## 验证 + +```bash +.venv/bin/pytest tests/test_server_file_transfer.py -v +``` + +共 **18** 项：schema、JWT 白名单、Redis store、SFTP 大小守卫、deliver 成功/失败回滚/超限、relay 同源拒绝与成功路径、`cleanup` 命令引用。 diff --git a/docs/changelog/2026-06-21-server-file-transfer.md b/docs/changelog/2026-06-21-server-file-transfer.md new file mode 100644 index 00000000..9e67b32e --- /dev/null +++ b/docs/changelog/2026-06-21-server-file-transfer.md @@ -0,0 +1,44 @@ +# 跨服务器文件传输 + +## 日期 + +2026-06-21 + +## 摘要 + +文件管理器新增 A→B 跨机传输：**直传**（中心 SFTP 中继，单文件 ≤500MB）与 **打包投递**（源机 tar.gz → 签名下载 URL → 目标机 curl + 可选解压，一键完成）。 + +## 动机 + +同机剪贴板无法跨服务器；运维需对标宝塔「发送到服务器 / 打包下载」能力，避免本地下载再上传。 + +## 涉及文件 + +- `server/application/services/server_file_transfer_service.py` — 直传、打包、投递编排 +- `server/infrastructure/redis/transfer_package_store.py` — 下载 token 元数据 +- `server/api/sync_v2.py` — `server-transfer`、`server-transfer/deliver`、`transfer-download` +- `server/api/schemas.py` — 请求模型 +- `server/api/auth_jwt.py` — 下载端点 JWT 豁免 +- `frontend/src/composables/files/useFilesActions.ts` — 对话框与 API 调用 +- `frontend/src/components/files/FilesDialogs.vue`、`FilesToolbar.vue`、`FilesList.vue` +- `frontend/src/utils/auditLabels.ts` +- `tests/test_server_file_transfer.py` +- `docs/design/specs/2026-06-21-server-file-transfer-design.md` + +## 迁移 / 重启 + +- 需重启 API（新路由 + Redis key `transfer:pkg:*`） +- 无需 DB 迁移 +- 前端需 `vite build` 后部署 `web/app/` + +## 验证 + +```bash +pytest tests/test_server_file_transfer.py -q +cd frontend && npm run type-check +bash scripts/local_verify.sh +``` + +- 文件页选中项 →「发送到其他服务器」→ 直传 / 打包投递 +- `GET /api/sync/transfer-download/{token}` 无 JWT 可 curl；过期 404 +- 审计动作：`server_file_relay`、`server_file_deliver` diff --git a/docs/design/plans/2026-06-21-server-file-transfer.md b/docs/design/plans/2026-06-21-server-file-transfer.md new file mode 100644 index 00000000..46d8b862 --- /dev/null +++ b/docs/design/plans/2026-06-21-server-file-transfer.md @@ -0,0 +1,33 @@ +# 跨服务器文件传输 — 实施计划 + +## 涉及文件 + +| 文件 | 变更 | +|------|------| +| `server/infrastructure/redis/transfer_package_store.py` | 新建：打包任务 Redis | +| `server/application/services/server_file_transfer_service.py` | 新建：直传 / 打包 / 拉取 / 流 | +| `server/api/schemas.py` | 请求模型 | +| `server/api/sync_v2.py` | 四个端点 | +| `server/api/auth_jwt.py` | 下载端点 JWT 豁免 | +| `frontend/src/composables/files/useFilesActions.ts` | 对话框逻辑 | +| `frontend/src/components/files/FilesDialogs.vue` | UI | +| `frontend/src/components/files/FilesList.vue` | 右键菜单 | +| `tests/test_server_file_transfer.py` | 单元测试 | +| `docs/changelog/2026-06-21-server-file-transfer.md` | 变更记录 | + +## 步骤 + +1. Redis store + service（SFTP 递归中继、远程 tar、目标 curl） +2. API + 审计 +3. 前端对话框（模式切换、目标服务器、路径、打包后展示链接与一键 pull） +4. pytest + `scripts/local_verify.sh` 相关用例 + +## 回滚 + +删除新端点与服务文件；Redis key `transfer:pkg:*` 自然过期。无需 DB 迁移。 + +## 测试要点 + +- 校验：同源目标、空 sources、路径非法 +- token 过期 / 不存在 → 404 +- mock SFTP 的 relay 字节计数（可选） diff --git a/docs/design/specs/2026-06-21-server-file-transfer-design.md b/docs/design/specs/2026-06-21-server-file-transfer-design.md new file mode 100644 index 00000000..1c023581 --- /dev/null +++ b/docs/design/specs/2026-06-21-server-file-transfer-design.md @@ -0,0 +1,50 @@ +# 跨服务器文件传输设计 + +## 背景与目标 + +文件管理器当前剪贴板仅支持**同一台服务器**内复制/移动。运维需要在 A 服务器选定文件或目录后，传到 B 服务器，无需经本地下载再上传。 + +对标宝塔「发送到服务器 / 打包下载链接」能力，提供两种模式： + +| 模式 | 适用 | 数据路径 | +|------|------|----------| +| **直传 relay** | 单文件、多文件（含目录，递归 SFTP） | A → Nexus 中心 SFTP 中继 → B | +| **打包 package** | 大目录、批量归档 | A 上 tar.gz → Nexus 签名下载 URL → B `curl` 拉取（可解压） | + +## 方案对比 + +| 方案 | 优点 | 缺点 | +|------|------|------| +| A. 中心 SFTP 中继 | 不依赖 A/B 互通；复用现有 SSH 凭据 | 带宽经中心；超大文件受 500MB/文件限制 | +| B. A 上 scp 到 B | 不经中心 | 需在 A 存 B 凭据；子机间常不通 | +| C. 仅打包 + wget | 适合目录；B 直连 API | 需临时公开下载端点（签名 token） | + +**选定**：A（直传）+ C（打包拉取）组合；拒绝 B。 + +## 接口 + +前缀 `/api/sync`（`sync_v2.py`）。 + +| 方法 | 路径 | 鉴权 | 说明 | +|------|------|------|------| +| POST | `/server-transfer` | JWT | 直传 | +| POST | `/server-transfer/deliver` | JWT | 一键打包 + B 拉取 + 可选解压 | +| POST | `/server-transfer/package` | JWT | 仅打包（高级） | +| POST | `/server-transfer/pull` | JWT | 仅拉取（高级） | +| GET | `/transfer-download/{token}` | **token**（无 JWT） | 流式下载归档 | + +## 安全 + +- `source_server_id != dest_server_id` +- 路径校验复用 `assert_clipboard_transfer_safe` / `coerce_remote_abs_path` +- 下载 token 存 Redis，TTL 1h；路径仅 token 可查 +- `GET /transfer-download/` 加入 JWT 白名单前缀；token 即凭证 +- 单文件直传上限 500MB（与现有下载一致） +- 全操作写 `audit_logs` + +## 验收标准 + +- [ ] 文件页选中 1+ 项 →「发送到其他服务器」→ 直传到 B 指定目录 +- [ ] 打包模式生成可 curl 的 URL；B 机 pull 成功并可解压 +- [ ] 源=目标、非法路径返回 400 +- [ ] pytest 覆盖校验与 token store diff --git a/frontend/src/components/files/FilesDialogs.vue b/frontend/src/components/files/FilesDialogs.vue index e7844b9d..bd06348f 100644 --- a/frontend/src/components/files/FilesDialogs.vue +++ b/frontend/src/components/files/FilesDialogs.vue @@ -150,6 +150,107 @@ const p = useFilesPageContext() + + + 发送到其他服务器 + +

+ 从当前服务器传输 {{ p.transferPaths.length }} 项 +

+ + + + + 直传 + 打包投递 + + + 所选文件超过 500MB 直传上限，请使用打包投递。 + + + 含目录时建议使用打包投递；直传将递归复制，可能较慢。 + + + + +
+ +
+ + + + + + {{ p.transferDownloadUrl ? '关闭' : '取消' }} + + + {{ p.transferMode === 'relay' ? '开始直传' : '打包并投递' }} + + + + + diff --git a/frontend/src/components/files/FilesList.vue b/frontend/src/components/files/FilesList.vue index 84a0bea9..9cf92a47 100644 --- a/frontend/src/components/files/FilesList.vue +++ b/frontend/src/components/files/FilesList.vue @@ -198,6 +198,13 @@ const p = useFilesPageContext() 终端 + + 发送到其他服务器 + 压缩 + + 发送到其他服务器 + +const MAX_RELAY_FILE_BYTES = 524_288_000 +const TRANSFER_DEST_STORAGE_PREFIX = 'nexus:file-transfer-dest' + +export type ServerTransferMode = 'relay' | 'deliver' + export function useFilesActions(options: { browse: (opts?: { force?: boolean }) => Promise invalidateAfterMutation: (parentPath?: string) => void @@ -32,6 +38,7 @@ export function useFilesActions(options: { const snackbar = useSnackbar() const store = useFilesStore() const { selectedServer, currentPath } = storeToRefs(store) + const { servers: serverList } = useServerList() const { selectedFiles, selectedAbsolutePaths, clearSelection } = options.selection const { setFromSelection, @@ -68,6 +75,182 @@ export function useFilesActions(options: { const contextY = ref(0) const contextFile = ref(null) + const showTransfer = ref(false) + const transferMode = ref('relay') + const transferDestServerId = ref(null) + const transferDestPath = ref('') + const transferExtract = ref(true) + const transferPaths = ref([]) + const transferItems = ref([]) + const transferDownloadUrl = ref('') + + const transferDestServers = computed(() => { + const cur = selectedServer.value + return serverList.value.filter((s) => s.id !== cur) + }) + + const transferHasDirectory = computed(() => + transferItems.value.some((f) => f.type === 'directory'), + ) + + const transferHasOversizedFile = computed(() => + transferItems.value.some( + (f) => f.type === 'file' && typeof f.size === 'number' && f.size > MAX_RELAY_FILE_BYTES, + ), + ) + + const transferRelayDisabled = computed(() => transferHasOversizedFile.value) + + const transferRecommendedMode = computed(() => { + if (transferHasOversizedFile.value || transferHasDirectory.value) return 'deliver' + return 'relay' + }) + + function defaultTransferDestPath(mode: ServerTransferMode, destServerId: number | null): string { + const destSrv = destServerId + ? serverList.value.find((s) => s.id === destServerId) + : undefined + const base = destSrv?.target_path?.trim() || '/tmp' + if (mode === 'relay') { + return base.endsWith('/') ? base.slice(0, -1) : base + } + const first = transferPaths.value[0] + const name = first ? (first.split('/').pop() || 'transfer') : 'transfer' + const safe = name.replace(/[^\w.\-]/g, '_') || 'transfer' + return `${base.replace(/\/$/, '')}/${safe}.tar.gz` + } + + function transferDestStorageKey(mode: ServerTransferMode, serverId: number): string { + return `${TRANSFER_DEST_STORAGE_PREFIX}:${mode}:${serverId}` + } + + function loadStoredTransferDest(mode: ServerTransferMode, serverId: number): string | null { + try { + return localStorage.getItem(transferDestStorageKey(mode, serverId)) + } catch { + return null + } + } + + function storeTransferDest(mode: ServerTransferMode, serverId: number, path: string) { + try { + localStorage.setItem(transferDestStorageKey(mode, serverId), path) + } catch { + /* ignore */ + } + } + + function openTransferDialog(items?: FileEntry[]) { + if (!selectedServer.value) { + snackbar('请先选择源服务器', 'warning') + return + } + const entries = items?.length ? items : selectedFiles.value + if (!entries.length) { + snackbar('请先选择文件或目录', 'warning') + return + } + transferItems.value = entries + transferPaths.value = entries.map((f) => joinRemotePath(currentPath.value, f.name)) + transferDownloadUrl.value = '' + const destCandidates = transferDestServers.value + transferDestServerId.value = destCandidates[0]?.id ?? null + const recommended: ServerTransferMode = + entries.some((f) => f.type === 'directory') + || entries.some( + (f) => f.type === 'file' && typeof f.size === 'number' && f.size > MAX_RELAY_FILE_BYTES, + ) + ? 'deliver' + : 'relay' + transferMode.value = recommended + const stored = transferDestServerId.value + ? loadStoredTransferDest(recommended, transferDestServerId.value) + : null + transferDestPath.value = + stored || defaultTransferDestPath(recommended, transferDestServerId.value) + showTransfer.value = true + } + + function onTransferModeChange(mode: ServerTransferMode) { + transferMode.value = mode + const stored = transferDestServerId.value + ? loadStoredTransferDest(mode, transferDestServerId.value) + : null + transferDestPath.value = + stored || defaultTransferDestPath(mode, transferDestServerId.value) + } + + function onTransferDestServerChange(serverId: number | null) { + transferDestServerId.value = serverId + const stored = serverId ? loadStoredTransferDest(transferMode.value, serverId) : null + transferDestPath.value = + stored || defaultTransferDestPath(transferMode.value, serverId) + } + + async function doTransfer() { + if (!selectedServer.value || !transferDestServerId.value || !transferPaths.value.length) { + snackbar('请完整填写传输信息', 'warning') + return + } + if (transferMode.value === 'relay' && transferRelayDisabled.value) { + snackbar('所选文件超过直传上限，请使用打包投递', 'warning') + return + } + const destPath = normalizeRemotePath(transferDestPath.value) + actionLoading.value = true + transferDownloadUrl.value = '' + try { + if (transferMode.value === 'relay') { + await http.post('/sync/server-transfer', { + source_server_id: selectedServer.value, + dest_server_id: transferDestServerId.value, + sources: transferPaths.value, + dest_dir: destPath, + }) + snackbar(`已直传 ${transferPaths.value.length} 项到目标服务器`) + storeTransferDest('relay', transferDestServerId.value, destPath) + showTransfer.value = false + clearSelection() + } else { + const res = await http.post<{ + download_url?: string + dest_path?: string + }>('/sync/server-transfer/deliver', { + source_server_id: selectedServer.value, + dest_server_id: transferDestServerId.value, + paths: transferPaths.value, + dest_path: destPath, + extract: transferExtract.value, + format: 'tar.gz', + }) + transferDownloadUrl.value = res.download_url || '' + storeTransferDest('deliver', transferDestServerId.value, destPath) + snackbar( + transferExtract.value + ? '打包投递完成，已解压到目标机' + : `打包投递完成：${res.dest_path || destPath}`, + ) + // 保留对话框以展示 download_url（排错）；用户点取消关闭 + clearSelection() + } + } catch (e: unknown) { + const msg = e instanceof Error ? e.message : '跨服务器传输失败' + snackbar(msg, 'error') + } finally { + actionLoading.value = false + } + } + + async function copyTransferDownloadUrl() { + if (!transferDownloadUrl.value) return + try { + await navigator.clipboard.writeText(transferDownloadUrl.value) + snackbar('下载链接已复制') + } catch { + snackbar('复制失败', 'error') + } + } + function resolvePasteDestination(explicitDest?: string): string { if (explicitDest) return normalizeRemotePath(explicitDest) const dirs = selectedFiles.value.filter((f) => f.type === 'directory') @@ -184,6 +367,9 @@ export function useFilesActions(options: { case 'decompress': openDecompressDialog(item) break + case 'sendToServer': + openTransferDialog([item]) + break case 'rename': startRename(item) break @@ -589,6 +775,23 @@ export function useFilesActions(options: { contextX, contextY, contextFile, + showTransfer, + transferMode, + transferDestServerId, + transferDestPath, + transferExtract, + transferPaths, + transferDownloadUrl, + transferDestServers, + transferHasDirectory, + transferHasOversizedFile, + transferRelayDisabled, + transferRecommendedMode, + openTransferDialog, + onTransferModeChange, + onTransferDestServerChange, + doTransfer, + copyTransferDownloadUrl, copyToClipboard, cutToClipboard, pasteClipboard, diff --git a/frontend/src/composables/files/useFilesPage.ts b/frontend/src/composables/files/useFilesPage.ts index 9b3be327..21186eb9 100644 --- a/frontend/src/composables/files/useFilesPage.ts +++ b/frontend/src/composables/files/useFilesPage.ts @@ -260,6 +260,23 @@ export function useFilesPage() { actionLoading, batchDelete: actions.batchDelete, openCompressDialog: actions.openCompressDialog, + openTransferDialog: actions.openTransferDialog, + showTransfer: actions.showTransfer, + transferMode: actions.transferMode, + transferDestServerId: actions.transferDestServerId, + transferDestPath: actions.transferDestPath, + transferExtract: actions.transferExtract, + transferPaths: actions.transferPaths, + transferDownloadUrl: actions.transferDownloadUrl, + transferDestServers: actions.transferDestServers, + transferHasDirectory: actions.transferHasDirectory, + transferHasOversizedFile: actions.transferHasOversizedFile, + transferRelayDisabled: actions.transferRelayDisabled, + transferRecommendedMode: actions.transferRecommendedMode, + onTransferModeChange: actions.onTransferModeChange, + onTransferDestServerChange: actions.onTransferDestServerChange, + doTransfer: actions.doTransfer, + copyTransferDownloadUrl: actions.copyTransferDownloadUrl, showPreview: editor.showPreview, previewTitle: editor.previewTitle, previewContent: editor.previewContent, diff --git a/frontend/src/utils/auditLabels.ts b/frontend/src/utils/auditLabels.ts index 16827cc4..8c8deb78 100644 --- a/frontend/src/utils/auditLabels.ts +++ b/frontend/src/utils/auditLabels.ts @@ -120,6 +120,10 @@ const ACTION_LABELS: Record = { file_permission_change: '修改文件权限', file_compress: '压缩远程文件', file_decompress: '解压远程文件', + server_file_relay: '跨服务器直传文件', + server_file_package: '跨服务器打包', + server_file_pull: '跨服务器拉取打包', + server_file_deliver: '跨服务器打包投递', local_file_delete: '删除本地文件', local_file_rename: '重命名本地文件', local_file_preview: '预览本地文件', diff --git a/server/api/auth_jwt.py b/server/api/auth_jwt.py index 43506a1c..61ec7c9d 100644 --- a/server/api/auth_jwt.py +++ b/server/api/auth_jwt.py @@ -49,6 +49,7 @@ PUBLIC_EXACT_PATHS = frozenset({ PUBLIC_PREFIXES = ( "/api/agent/", "/api/install/", + "/api/sync/transfer-download/", "/ws/", ) diff --git a/server/api/schemas.py b/server/api/schemas.py index e6c0a0f5..d4205b6c 100644 --- a/server/api/schemas.py +++ b/server/api/schemas.py @@ -516,6 +516,75 @@ class FileDecompress(BaseModel): return coerce_remote_abs_path(text) +class ServerFileTransferRelay(BaseModel): + """Direct SFTP relay from source server to destination server.""" + source_server_id: int = Field(..., ge=1) + dest_server_id: int = Field(..., ge=1) + sources: list[str] = Field(..., min_length=1, max_length=50) + dest_dir: str = Field(..., min_length=1, max_length=4096) + + @field_validator("sources", mode="before") + @classmethod + def _normalize_sources(cls, v: object) -> list[str]: + if not isinstance(v, list): + raise ValueError("sources 必须为路径列表") + return coerce_remote_abs_path_list([str(x) for x in v]) + + @field_validator("dest_dir", mode="before") + @classmethod + def _normalize_dest_dir(cls, v: object) -> str: + return coerce_remote_abs_path(str(v)) + + +class ServerFileTransferPackage(BaseModel): + """Create a downloadable archive on the source server.""" + source_server_id: int = Field(..., ge=1) + paths: list[str] = Field(..., min_length=1, max_length=100) + format: str = Field("tar.gz", pattern=r"^(tar\.gz|zip)$") + + @field_validator("paths", mode="before") + @classmethod + def _normalize_paths(cls, v: object) -> list[str]: + if not isinstance(v, list): + raise ValueError("paths 必须为路径列表") + return coerce_remote_abs_path_list([str(x) for x in v]) + + +class ServerFileTransferPull(BaseModel): + """Download package URL on destination server and optionally extract.""" + token: str = Field(..., min_length=8, max_length=128) + dest_server_id: int = Field(..., ge=1) + dest_path: str = Field(..., min_length=1, max_length=4096) + extract: bool = True + + @field_validator("dest_path", mode="before") + @classmethod + def _normalize_dest_path(cls, v: object) -> str: + return coerce_remote_abs_path(str(v)) + + +class ServerFileTransferDeliver(BaseModel): + """One-click package on source, curl on dest, optional extract.""" + source_server_id: int = Field(..., ge=1) + dest_server_id: int = Field(..., ge=1) + paths: list[str] = Field(..., min_length=1, max_length=100) + dest_path: str = Field(..., min_length=1, max_length=4096) + extract: bool = True + format: str = Field("tar.gz", pattern=r"^(tar\.gz|zip)$") + + @field_validator("paths", mode="before") + @classmethod + def _normalize_paths(cls, v: object) -> list[str]: + if not isinstance(v, list): + raise ValueError("paths 必须为路径列表") + return coerce_remote_abs_path_list([str(x) for x in v]) + + @field_validator("dest_path", mode="before") + @classmethod + def _normalize_dest_path(cls, v: object) -> str: + return coerce_remote_abs_path(str(v)) + + # ── Script ── class ScriptCreate(BaseModel): diff --git a/server/api/sync_v2.py b/server/api/sync_v2.py index b3eab215..40a932c7 100644 --- a/server/api/sync_v2.py +++ b/server/api/sync_v2.py @@ -30,6 +30,10 @@ from server.api.schemas import ( FileChmod, FileCompress, FileDecompress, + ServerFileTransferRelay, + ServerFileTransferPackage, + ServerFileTransferPull, + ServerFileTransferDeliver, ) from server.api.remote_path_validation import RemotePathError, assert_clipboard_transfer_safe from server.utils.text_io import detect_text_eol @@ -1885,3 +1889,228 @@ async def decompress_file( ) return {"success": True, "path": archive_path, "dest": dest_dir} + + +# ── Cross-server file transfer ── + + +@router.post("/server-transfer") +async def server_file_transfer_relay( + payload: ServerFileTransferRelay, + request: Request, + admin: Admin = Depends(get_current_admin), +): + """Relay copy files/directories from server A to server B via Nexus SFTP.""" + from server.application.services.server_file_transfer_service import ( + ServerTransferError, + relay_files, + ) + from server.application.services.files_browse_service import invalidate_browse_cache + + session = request.state.db + try: + result = await relay_files( + session, + source_server_id=payload.source_server_id, + dest_server_id=payload.dest_server_id, + sources=payload.sources, + dest_dir=payload.dest_dir, + ) + except RemotePathError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except ServerTransferError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + await _audit_sync( + "server_file_relay", + "server", + payload.dest_server_id, + ( + f"A#{payload.source_server_id} → B#{payload.dest_server_id}: " + f"{result['count']} 项 → {result['dest_dir']} ({result['bytes_transferred']} B)" + ), + admin.username, + request, + ) + invalidate_browse_cache(payload.dest_server_id, result["dest_dir"]) + return {"success": True, **result} + + +@router.post("/server-transfer/package") +async def server_file_transfer_package( + payload: ServerFileTransferPackage, + request: Request, + admin: Admin = Depends(get_current_admin), +): + """Compress paths on source server; return signed download URL for target pull.""" + from server.application.services.server_file_transfer_service import ( + ServerTransferError, + create_transfer_package, + ) + + session = request.state.db + try: + result = await create_transfer_package( + session, + source_server_id=payload.source_server_id, + paths=payload.paths, + fmt=payload.format, + ) + except ServerTransferError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + await _audit_sync( + "server_file_package", + "server", + payload.source_server_id, + f"打包 {len(payload.paths)} 项 → {result.get('archive_path')}", + admin.username, + request, + ) + return {"success": True, **result} + + +@router.post("/server-transfer/pull") +async def server_file_transfer_pull( + payload: ServerFileTransferPull, + request: Request, + admin: Admin = Depends(get_current_admin), +): + """On destination server: curl package download URL; optional extract.""" + from server.application.services.server_file_transfer_service import ( + ServerTransferError, + pull_transfer_package, + ) + from server.application.services.files_browse_service import invalidate_browse_parents + from server.infrastructure.redis.transfer_package_store import load_package_meta + + session = request.state.db + try: + result = await pull_transfer_package( + session, + token=payload.token, + dest_server_id=payload.dest_server_id, + dest_path=payload.dest_path, + extract=payload.extract, + ) + except ServerTransferError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + meta = await load_package_meta(payload.token) + if meta and result.get("extracted_to"): + invalidate_browse_parents( + payload.dest_server_id, + result["extracted_to"], + result["dest_path"], + ) + else: + from server.application.services.files_browse_service import invalidate_browse_cache + from server.utils.posix_paths import posix_dirname + + invalidate_browse_cache(payload.dest_server_id, posix_dirname(result["dest_path"])) + + await _audit_sync( + "server_file_pull", + "server", + payload.dest_server_id, + f"拉取打包 {payload.token[:8]}… → {result['dest_path']}", + admin.username, + request, + ) + return {"success": True, **result} + + +@router.post("/server-transfer/deliver") +async def server_file_transfer_deliver( + payload: ServerFileTransferDeliver, + request: Request, + admin: Admin = Depends(get_current_admin), +): + """One-click: compress on source, curl on dest, optional extract.""" + from server.application.services.server_file_transfer_service import ( + ServerTransferError, + deliver_transfer_package, + ) + from server.application.services.files_browse_service import invalidate_browse_parents, invalidate_browse_cache + from server.utils.posix_paths import posix_dirname + + session = request.state.db + try: + result = await deliver_transfer_package( + session, + source_server_id=payload.source_server_id, + paths=payload.paths, + dest_server_id=payload.dest_server_id, + dest_path=payload.dest_path, + extract=payload.extract, + fmt=payload.format, + ) + except ServerTransferError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + if result.get("extracted_to"): + invalidate_browse_parents( + payload.dest_server_id, + result["extracted_to"], + result["dest_path"], + ) + else: + invalidate_browse_cache(payload.dest_server_id, posix_dirname(result["dest_path"])) + + await _audit_sync( + "server_file_deliver", + "server", + payload.dest_server_id, + ( + f"A#{payload.source_server_id} → B#{payload.dest_server_id}: " + f"打包投递 {len(payload.paths)} 项 → {result['dest_path']}" + ), + admin.username, + request, + ) + return {"success": True, **result} + + +@router.get("/transfer-download/{token}") +async def server_file_transfer_download( + token: str, + request: Request, +): + """Stream package archive from source server (token auth, no JWT — for remote curl).""" + from fastapi.responses import StreamingResponse + from server.application.services.server_file_transfer_service import ( + ServerTransferError, + iter_package_chunks, + ) + from server.infrastructure.database.server_repo import ServerRepositoryImpl + from server.infrastructure.redis.transfer_package_store import load_package_meta + + session = request.state.db + meta = await load_package_meta(token) + if not meta: + raise HTTPException(status_code=404, detail="下载链接无效或已过期") + + source_id = int(meta["source_server_id"]) + archive_path = str(meta["archive_path"]) + filename = str(meta.get("filename") or "package.tar.gz") + + server = await ServerRepositoryImpl(session).get_by_id(source_id) + if not server: + raise HTTPException(status_code=404, detail="源服务器不存在") + + try: + async def file_generator(): + try: + async for chunk in iter_package_chunks(server, archive_path): + yield chunk + except ServerTransferError as exc: + logger.warning("transfer-download stream failed: %s", exc) + raise + + return StreamingResponse( + file_generator(), + media_type="application/octet-stream", + headers={"Content-Disposition": f'attachment; filename="{filename}"'}, + ) + except ServerTransferError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc diff --git a/server/application/services/server_file_transfer_service.py b/server/application/services/server_file_transfer_service.py new file mode 100644 index 00000000..3b751963 --- /dev/null +++ b/server/application/services/server_file_transfer_service.py @@ -0,0 +1,400 @@ +"""Cross-server file transfer: SFTP relay and package download/pull.""" + +from __future__ import annotations + +import logging +import posixpath +import secrets +import shlex +from collections.abc import AsyncIterator +from datetime import datetime, timedelta, timezone +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from server.api.remote_path_validation import assert_clipboard_transfer_safe +from server.config import settings +from server.domain.models import Server +from server.infrastructure.database.server_repo import ServerRepositoryImpl +from server.infrastructure.redis.transfer_package_store import ( + DEFAULT_TTL_SECONDS, + delete_package_meta, + load_package_meta, + save_package_meta, +) +from server.infrastructure.ssh.asyncssh_pool import sftp_session, ssh_pool +from server.infrastructure.ssh.remote_archive import run_remote_compress +from server.infrastructure.ssh.remote_shell import exec_ssh_command_with_fallback +from server.utils.posix_paths import ( + PosixPathError, + normalize_remote_abs_path, + posix_basename, + posix_join, + remote_join, +) + +logger = logging.getLogger("nexus.server_file_transfer") + +MAX_RELAY_FILE_BYTES = 524_288_000 # 500MB — aligned with sync download cap +MAX_RELAY_SOURCES = 50 +CHUNK_SIZE = 65536 + + +class ServerTransferError(Exception): + """Business validation or transfer failure.""" + + +def _api_base_url() -> str: + base = (settings.API_BASE_URL or "").strip().rstrip("/") + if not base: + raise ServerTransferError("未配置 NEXUS_API_BASE_URL，无法生成下载链接") + return base + + +async def _get_server(session: AsyncSession, server_id: int) -> Server: + server = await ServerRepositoryImpl(session).get_by_id(server_id) + if not server: + raise ServerTransferError("服务器不存在") + return server + + +def _assert_distinct_servers(source_id: int, dest_id: int) -> None: + if source_id == dest_id: + raise ServerTransferError("源服务器与目标服务器不能相同") + + +async def _ensure_remote_dir(server: Server, path: str) -> None: + result = await exec_ssh_command_with_fallback( + server, + f"test -d {shlex.quote(path)} || mkdir -p {shlex.quote(path)}", + timeout=30, + ) + if result.get("exit_code") != 0: + err = (result.get("stderr") or result.get("stdout") or "无法创建目标目录")[:300] + raise ServerTransferError(err) + + +async def _sftp_is_directory(sftp: Any, path: str) -> bool: + stat = await sftp.stat(path) + ftype = getattr(stat, "type", None) + if ftype == "directory": + return True + if ftype == "regular file": + return False + # unknown: treat trailing slash or readdir probe + try: + await sftp.listdir(path) + return True + except Exception: + return False + + +async def _relay_file( + src_sftp: Any, + dst_sftp: Any, + src_path: str, + dst_path: str, +) -> int: + stat = await src_sftp.stat(src_path) + size = int(getattr(stat, "size", 0) or 0) + if size > MAX_RELAY_FILE_BYTES: + raise ServerTransferError( + f"文件 {posix_basename(src_path)} 超过直传上限 {MAX_RELAY_FILE_BYTES} 字节，请使用打包模式" + ) + transferred = 0 + async with src_sftp.open(src_path, "rb") as rf: + async with dst_sftp.open(dst_path, "wb") as wf: + while chunk := await rf.read(CHUNK_SIZE): + await wf.write(chunk) + transferred += len(chunk) + return transferred + + +async def _relay_tree( + src_sftp: Any, + dst_sftp: Any, + src_path: str, + dst_path: str, +) -> int: + if await _sftp_is_directory(src_sftp, src_path): + await dst_sftp.makedirs(dst_path, exist_ok=True) + total = 0 + for name in await src_sftp.listdir(src_path): + if name in (".", ".."): + continue + child_src = posix_join(src_path, name) + child_dst = posix_join(dst_path, name) + total += await _relay_tree(src_sftp, dst_sftp, child_src, child_dst) + return total + return await _relay_file(src_sftp, dst_sftp, src_path, dst_path) + + +async def relay_files( + session: AsyncSession, + *, + source_server_id: int, + dest_server_id: int, + sources: list[str], + dest_dir: str, +) -> dict[str, Any]: + """Stream copy files/directories from source server to dest via Nexus SFTP relay.""" + if len(sources) > MAX_RELAY_SOURCES: + raise ServerTransferError(f"单次直传最多 {MAX_RELAY_SOURCES} 项") + _assert_distinct_servers(source_server_id, dest_server_id) + normalized_sources, dest = assert_clipboard_transfer_safe(sources, dest_dir) + + source = await _get_server(session, source_server_id) + dest_server = await _get_server(session, dest_server_id) + await _ensure_remote_dir(dest_server, dest) + + conn_src = await ssh_pool.acquire(source) + conn_dst = await ssh_pool.acquire(dest_server) + bytes_total = 0 + items: list[dict[str, str]] = [] + try: + async with sftp_session(conn_src) as src_sftp, sftp_session(conn_dst) as dst_sftp: + for src_path in normalized_sources: + base = posix_basename(src_path.rstrip("/")) or src_path + target = remote_join(dest, base) + bytes_total += await _relay_tree(src_sftp, dst_sftp, src_path, target) + items.append({"source": src_path, "dest": target}) + finally: + await ssh_pool.release(source.id) + await ssh_pool.release(dest_server.id) + + return { + "mode": "relay", + "source_server_id": source_server_id, + "dest_server_id": dest_server_id, + "dest_dir": dest, + "count": len(items), + "bytes_transferred": bytes_total, + "items": items, + } + + +async def create_transfer_package( + session: AsyncSession, + *, + source_server_id: int, + paths: list[str], + fmt: str = "tar.gz", +) -> dict[str, Any]: + """Compress on source server and register a time-limited download token.""" + if not paths: + raise ServerTransferError("paths 不能为空") + if len(paths) > 100: + raise ServerTransferError("单次打包最多 100 项") + + source = await _get_server(session, source_server_id) + normalized: list[str] = [] + for raw in paths: + try: + normalized.append(normalize_remote_abs_path(raw)) + except PosixPathError as exc: + raise ServerTransferError(str(exc)) from exc + + token = secrets.token_urlsafe(24) + suffix = ".tar.gz" if fmt == "tar.gz" else ".zip" + archive_name = f"nexus-xfer-{token[:12]}{suffix}" + archive_path = f"/tmp/{archive_name}" + + result = await run_remote_compress(source, normalized, archive_path, fmt, timeout=600) + if result.get("exit_code") != 0: + err = (result.get("stderr") or result.get("stdout") or "打包失败")[:400] + raise ServerTransferError(err) + + stat_res = await exec_ssh_command_with_fallback( + source, + f"stat -c %s {shlex.quote(archive_path)}", + timeout=30, + ) + size = 0 + if stat_res.get("exit_code") == 0: + try: + size = int((stat_res.get("stdout") or "0").strip()) + except ValueError: + size = 0 + + expires_at = datetime.now(timezone.utc) + timedelta(seconds=DEFAULT_TTL_SECONDS) + meta = { + "source_server_id": source_server_id, + "archive_path": archive_path, + "filename": archive_name, + "size": size, + "format": fmt, + "expires_at": expires_at.isoformat(), + } + await save_package_meta(token, meta) + + download_url = f"{_api_base_url()}/api/sync/transfer-download/{token}" + return { + "mode": "package", + "token": token, + "download_url": download_url, + "archive_path": archive_path, + "size": size, + "expires_at": expires_at.isoformat(), + "source_server_id": source_server_id, + } + + +async def pull_transfer_package( + session: AsyncSession, + *, + token: str, + dest_server_id: int, + dest_path: str, + extract: bool = True, +) -> dict[str, Any]: + """On dest server: curl download URL, optionally extract tar.gz/zip.""" + meta = await load_package_meta(token) + if not meta: + raise ServerTransferError("下载链接无效或已过期") + + dest_server = await _get_server(session, dest_server_id) + dest = dest_path.strip() + if not dest.startswith("/"): + raise ServerTransferError("目标路径必须为绝对路径") + + download_url = f"{_api_base_url()}/api/sync/transfer-download/{token}" + url_q = shlex.quote(download_url) + dest_q = shlex.quote(dest) + + parent = posixpath.dirname(dest) or "/" + await _ensure_remote_dir(dest_server, parent) + + curl_cmd = f"curl -fsSL --connect-timeout 30 --max-time 3600 -o {dest_q} {url_q}" + result = await exec_ssh_command_with_fallback(dest_server, curl_cmd, timeout=3700) + if result.get("exit_code") != 0: + err = (result.get("stderr") or result.get("stdout") or "下载失败")[:400] + raise ServerTransferError(err) + + extracted_to: str | None = None + fmt = str(meta.get("format") or "tar.gz") + if extract: + extract_dir = posixpath.dirname(dest) or "/" + extract_q = shlex.quote(extract_dir) + if fmt == "tar.gz": + extract_cmd = f"tar -xzf {dest_q} -C {extract_q}" + else: + extract_cmd = f"unzip -o {dest_q} -d {extract_q}" + ext_res = await exec_ssh_command_with_fallback(dest_server, extract_cmd, timeout=600) + if ext_res.get("exit_code") != 0: + err = (ext_res.get("stderr") or ext_res.get("stdout") or "解压失败")[:400] + raise ServerTransferError(err) + extracted_to = extract_dir + + return { + "mode": "pull", + "dest_server_id": dest_server_id, + "dest_path": dest, + "extracted_to": extracted_to, + "download_url": download_url, + "bytes_hint": meta.get("size"), + } + + +async def deliver_transfer_package( + session: AsyncSession, + *, + source_server_id: int, + paths: list[str], + dest_server_id: int, + dest_path: str, + extract: bool = True, + fmt: str = "tar.gz", + cleanup_source: bool = True, +) -> dict[str, Any]: + """One-click: package on source, pull on dest, optional extract and source cleanup.""" + _assert_distinct_servers(source_server_id, dest_server_id) + + package = await create_transfer_package( + session, + source_server_id=source_server_id, + paths=paths, + fmt=fmt, + ) + size = int(package.get("size") or 0) + if size > MAX_RELAY_FILE_BYTES: + source = await _get_server(session, source_server_id) + await cleanup_source_archive(source, str(package["archive_path"])) + await delete_package_meta(str(package["token"])) + raise ServerTransferError( + f"打包后大小 {size} 字节超过下载上限 {MAX_RELAY_FILE_BYTES}，请缩小范围或联系管理员" + ) + + try: + pull = await pull_transfer_package( + session, + token=str(package["token"]), + dest_server_id=dest_server_id, + dest_path=dest_path, + extract=extract, + ) + except ServerTransferError: + try: + source = await _get_server(session, source_server_id) + await cleanup_source_archive(source, str(package["archive_path"])) + await delete_package_meta(str(package["token"])) + except Exception as exc: + logger.warning("deliver rollback cleanup failed: %s", exc) + raise + except Exception as exc: + raise ServerTransferError(str(exc)) from exc + + if cleanup_source: + try: + source = await _get_server(session, source_server_id) + await cleanup_source_archive(source, str(package["archive_path"])) + except Exception as exc: + logger.warning("cleanup source archive failed: %s", exc) + + try: + await delete_package_meta(str(package["token"])) + except Exception as exc: + logger.warning("delete transfer token failed: %s", exc) + + return { + "mode": "deliver", + "source_server_id": source_server_id, + "dest_server_id": dest_server_id, + "token": package["token"], + "download_url": package["download_url"], + "archive_path": package["archive_path"], + "size": size, + "dest_path": pull["dest_path"], + "extracted_to": pull.get("extracted_to"), + "expires_at": package.get("expires_at"), + } + + +async def iter_package_chunks( + server: Server, + archive_path: str, +) -> AsyncIterator[bytes]: + """SFTP-read archive from source server in chunks.""" + conn = await ssh_pool.acquire(server) + try: + async with sftp_session(conn) as sftp: + try: + stat = await sftp.stat(archive_path) + except FileNotFoundError as exc: + raise ServerTransferError("归档文件不存在或已被清理") from exc + size = int(getattr(stat, "size", 0) or 0) + if size > MAX_RELAY_FILE_BYTES: + raise ServerTransferError("归档超过下载大小限制") + async with sftp.open(archive_path, "rb") as rf: + while chunk := await rf.read(CHUNK_SIZE): + yield chunk + finally: + await ssh_pool.release(server.id) + + +async def cleanup_source_archive(server: Server, archive_path: str) -> None: + """Best-effort remove temp archive on source after successful pull.""" + await exec_ssh_command_with_fallback( + server, + f"rm -f {shlex.quote(archive_path)}", + timeout=30, + ) diff --git a/server/infrastructure/redis/transfer_package_store.py b/server/infrastructure/redis/transfer_package_store.py new file mode 100644 index 00000000..e5e54fc4 --- /dev/null +++ b/server/infrastructure/redis/transfer_package_store.py @@ -0,0 +1,41 @@ +"""Redis-backed metadata for cross-server package transfer download tokens.""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +from server.infrastructure.redis.client import get_redis + +logger = logging.getLogger("nexus.transfer_package_store") + +REDIS_PREFIX = "transfer:pkg:" +DEFAULT_TTL_SECONDS = 3600 + + +def _key(token: str) -> str: + return f"{REDIS_PREFIX}{token}" + + +async def save_package_meta(token: str, meta: dict[str, Any], *, ttl: int = DEFAULT_TTL_SECONDS) -> None: + redis = get_redis() + await redis.set(_key(token), json.dumps(meta, ensure_ascii=False), ex=ttl) + + +async def load_package_meta(token: str) -> dict[str, Any] | None: + redis = get_redis() + raw = await redis.get(_key(token)) + if not raw: + return None + try: + data = json.loads(raw) + except json.JSONDecodeError: + logger.warning("Invalid transfer package JSON for token prefix %s…", token[:8]) + return None + return data if isinstance(data, dict) else None + + +async def delete_package_meta(token: str) -> None: + redis = get_redis() + await redis.delete(_key(token)) diff --git a/tests/test_server_file_transfer.py b/tests/test_server_file_transfer.py new file mode 100644 index 00000000..b61604c7 --- /dev/null +++ b/tests/test_server_file_transfer.py @@ -0,0 +1,394 @@ +"""Unit tests for cross-server file transfer.""" + +from __future__ import annotations + +import json +from unittest.mock import AsyncMock, MagicMock + +import pytest +from pydantic import ValidationError + +from server.api.remote_path_validation import RemotePathError, assert_clipboard_transfer_safe +from server.application.services.server_file_transfer_service import ( + MAX_RELAY_FILE_BYTES, + ServerTransferError, + _assert_distinct_servers, + _relay_file, + cleanup_source_archive, + deliver_transfer_package, + relay_files, +) +from server.domain.models import Server + + +pytestmark = pytest.mark.unit + + +# ── validation helpers ── + + +def test_assert_distinct_servers_rejects_same(): + with pytest.raises(ServerTransferError, match="不能相同"): + _assert_distinct_servers(1, 1) + + +def test_assert_distinct_servers_ok(): + _assert_distinct_servers(1, 2) + + +def test_clipboard_transfer_safe_rejects_dest_inside_source(): + with pytest.raises(RemotePathError): + assert_clipboard_transfer_safe(["/www/a"], "/www/a/sub") + + +def test_relay_file_size_limit_constant(): + assert MAX_RELAY_FILE_BYTES == 524_288_000 + + +# ── schemas ── + + +def test_server_file_transfer_relay_schema_validation(): + from server.api.schemas import ServerFileTransferRelay + + payload = ServerFileTransferRelay( + source_server_id=1, + dest_server_id=2, + sources=["/www/a.txt"], + dest_dir="/www/backup", + ) + assert payload.dest_dir == "/www/backup" + assert payload.sources == ["/www/a.txt"] + + +def test_server_file_transfer_relay_schema_rejects_empty_sources(): + from server.api.schemas import ServerFileTransferRelay + + with pytest.raises(ValidationError): + ServerFileTransferRelay( + source_server_id=1, + dest_server_id=2, + sources=[], + dest_dir="/www/backup", + ) + + +def test_server_file_transfer_deliver_schema_validation(): + from server.api.schemas import ServerFileTransferDeliver + + payload = ServerFileTransferDeliver( + source_server_id=1, + dest_server_id=3, + paths=["/www/wwwroot"], + dest_path="/tmp/site.tar.gz", + extract=True, + ) + assert payload.format == "tar.gz" + assert payload.extract is True + + +# ── JWT public path ── + + +def test_transfer_download_is_public_get(): + from server.api.auth_jwt import _is_public_path + + assert _is_public_path("/api/sync/transfer-download/abc123", "GET") is True + # Prefix 白名单对所有 method 生效；路由仅注册 GET + assert _is_public_path("/api/sync/transfer-download/abc123", "POST") is True + assert _is_public_path("/api/sync/server-transfer", "GET") is False + + +# ── Redis transfer package store ── + + +class _FakeRedis: + def __init__(self) -> None: + self.data: dict[str, str] = {} + + async def set(self, key: str, value: str, ex: int | None = None) -> None: + self.data[key] = value + + async def get(self, key: str) -> str | None: + return self.data.get(key) + + async def delete(self, key: str) -> None: + self.data.pop(key, None) + + +@pytest.mark.asyncio +async def test_transfer_package_store_roundtrip(monkeypatch): + from server.infrastructure.redis import transfer_package_store as store + + fake = _FakeRedis() + monkeypatch.setattr(store, "get_redis", lambda: fake) + + meta = {"source_server_id": 1, "archive_path": "/tmp/x.tar.gz", "format": "tar.gz"} + await store.save_package_meta("tok-a", meta, ttl=60) + loaded = await store.load_package_meta("tok-a") + assert loaded == meta + + await store.delete_package_meta("tok-a") + assert await store.load_package_meta("tok-a") is None + + +@pytest.mark.asyncio +async def test_transfer_package_store_invalid_json(monkeypatch): + from server.infrastructure.redis import transfer_package_store as store + + fake = _FakeRedis() + fake.data["transfer:pkg:bad"] = "not-json" + monkeypatch.setattr(store, "get_redis", lambda: fake) + + assert await store.load_package_meta("bad") is None + + +# ── SFTP relay file size guard ── + + +@pytest.mark.asyncio +async def test_relay_file_rejects_oversized(): + src_sftp = AsyncMock() + dst_sftp = AsyncMock() + stat = MagicMock() + stat.size = MAX_RELAY_FILE_BYTES + 1 + src_sftp.stat = AsyncMock(return_value=stat) + + with pytest.raises(ServerTransferError, match="超过直传上限"): + await _relay_file(src_sftp, dst_sftp, "/www/big.bin", "/tmp/big.bin") + + +@pytest.mark.asyncio +async def test_relay_file_streams_chunks(): + src_sftp = AsyncMock() + dst_sftp = AsyncMock() + stat = MagicMock() + stat.size = 3 + src_sftp.stat = AsyncMock(return_value=stat) + + read_chunks = [b"ab", b"c", b""] + + class _FakeReader: + async def read(self, _n: int) -> bytes: + return read_chunks.pop(0) + + async def __aenter__(self): + return self + + async def __aexit__(self, *_a): + return None + + class _FakeWriter: + def __init__(self) -> None: + self.written = b"" + + async def write(self, data: bytes) -> None: + self.written += data + + async def __aenter__(self): + return self + + async def __aexit__(self, *_a): + return None + + writer = _FakeWriter() + src_sftp.open = MagicMock(return_value=_FakeReader()) + dst_sftp.open = MagicMock(return_value=writer) + + nbytes = await _relay_file(src_sftp, dst_sftp, "/www/a.txt", "/tmp/a.txt") + assert nbytes == 3 + assert writer.written == b"abc" + + +# ── deliver orchestration (mocked) ── + + +def _sample_package(*, size: int = 128) -> dict: + return { + "mode": "package", + "token": "pkg-token-xyz", + "download_url": "https://api.example.com/api/sync/transfer-download/pkg-token-xyz", + "archive_path": "/tmp/nexus-xfer-pkg.tar.gz", + "size": size, + "expires_at": "2026-06-21T12:00:00+00:00", + "source_server_id": 1, + } + + +@pytest.mark.asyncio +async def test_deliver_success_cleans_archive_and_deletes_token(monkeypatch): + from server.application.services import server_file_transfer_service as svc + + package = _sample_package() + pull_result = {"dest_path": "/www/site.tar.gz", "extracted_to": "/www"} + + monkeypatch.setattr(svc, "create_transfer_package", AsyncMock(return_value=package)) + monkeypatch.setattr(svc, "pull_transfer_package", AsyncMock(return_value=pull_result)) + monkeypatch.setattr( + svc, + "_get_server", + AsyncMock(return_value=Server(name="src", domain="1.2.3.4")), + ) + cleanup_mock = AsyncMock() + delete_mock = AsyncMock() + monkeypatch.setattr(svc, "cleanup_source_archive", cleanup_mock) + monkeypatch.setattr(svc, "delete_package_meta", delete_mock) + + session = MagicMock() + result = await deliver_transfer_package( + session, + source_server_id=1, + paths=["/www/wwwroot"], + dest_server_id=2, + dest_path="/www/site.tar.gz", + ) + + assert result["mode"] == "deliver" + assert result["dest_path"] == "/www/site.tar.gz" + cleanup_mock.assert_awaited_once() + delete_mock.assert_awaited_once_with("pkg-token-xyz") + + +@pytest.mark.asyncio +async def test_deliver_pull_failure_rolls_back_archive_and_token(monkeypatch): + from server.application.services import server_file_transfer_service as svc + + package = _sample_package() + source = Server(name="src", domain="1.2.3.4") + + monkeypatch.setattr(svc, "create_transfer_package", AsyncMock(return_value=package)) + monkeypatch.setattr( + svc, + "pull_transfer_package", + AsyncMock(side_effect=ServerTransferError("下载失败")), + ) + monkeypatch.setattr(svc, "_get_server", AsyncMock(return_value=source)) + cleanup_mock = AsyncMock() + delete_mock = AsyncMock() + monkeypatch.setattr(svc, "cleanup_source_archive", cleanup_mock) + monkeypatch.setattr(svc, "delete_package_meta", delete_mock) + + session = MagicMock() + with pytest.raises(ServerTransferError, match="下载失败"): + await deliver_transfer_package( + session, + source_server_id=1, + paths=["/www/wwwroot"], + dest_server_id=2, + dest_path="/www/site.tar.gz", + ) + + cleanup_mock.assert_awaited_once_with(source, package["archive_path"]) + delete_mock.assert_awaited_once_with("pkg-token-xyz") + + +@pytest.mark.asyncio +async def test_deliver_rejects_oversized_package(monkeypatch): + from server.application.services import server_file_transfer_service as svc + + package = _sample_package(size=MAX_RELAY_FILE_BYTES + 1) + source = Server(name="src", domain="1.2.3.4") + + monkeypatch.setattr(svc, "create_transfer_package", AsyncMock(return_value=package)) + monkeypatch.setattr(svc, "_get_server", AsyncMock(return_value=source)) + cleanup_mock = AsyncMock() + delete_mock = AsyncMock() + monkeypatch.setattr(svc, "cleanup_source_archive", cleanup_mock) + monkeypatch.setattr(svc, "delete_package_meta", delete_mock) + pull_mock = AsyncMock() + monkeypatch.setattr(svc, "pull_transfer_package", pull_mock) + + session = MagicMock() + with pytest.raises(ServerTransferError, match="超过下载上限"): + await deliver_transfer_package( + session, + source_server_id=1, + paths=["/www/huge"], + dest_server_id=2, + dest_path="/www/huge.tar.gz", + ) + + pull_mock.assert_not_called() + cleanup_mock.assert_awaited_once() + delete_mock.assert_awaited_once_with("pkg-token-xyz") + + +# ── relay_files (mocked SSH) ── + + +@pytest.mark.asyncio +async def test_relay_files_rejects_same_server(monkeypatch): + session = MagicMock() + with pytest.raises(ServerTransferError, match="不能相同"): + await relay_files( + session, + source_server_id=5, + dest_server_id=5, + sources=["/www/a.txt"], + dest_dir="/www/backup", + ) + + +@pytest.mark.asyncio +async def test_relay_files_success(monkeypatch): + from server.application.services import server_file_transfer_service as svc + + source = Server(id=1, name="a", domain="1.1.1.1") + dest = Server(id=2, name="b", domain="2.2.2.2") + + async def fake_relay_tree(_src, _dst, src_path, dst_path): + assert src_path == "/www/a.txt" + assert dst_path == "/www/backup/a.txt" + return 42 + + monkeypatch.setattr(svc, "_get_server", AsyncMock(side_effect=[source, dest])) + monkeypatch.setattr(svc, "_ensure_remote_dir", AsyncMock()) + monkeypatch.setattr(svc, "_relay_tree", fake_relay_tree) + + class _Pool: + async def acquire(self, _server): + return object() + + async def release(self, _server_id): + return None + + monkeypatch.setattr(svc, "ssh_pool", _Pool()) + + class _SftpCtx: + async def __aenter__(self): + return AsyncMock() + + async def __aexit__(self, *_a): + return None + + monkeypatch.setattr(svc, "sftp_session", lambda _conn: _SftpCtx()) + + session = MagicMock() + result = await relay_files( + session, + source_server_id=1, + dest_server_id=2, + sources=["/www/a.txt"], + dest_dir="/www/backup", + ) + + assert result["mode"] == "relay" + assert result["bytes_transferred"] == 42 + assert result["count"] == 1 + assert result["items"][0]["dest"] == "/www/backup/a.txt" + + +@pytest.mark.asyncio +async def test_cleanup_source_archive_uses_quoted_rm(monkeypatch): + from server.application.services import server_file_transfer_service as svc + + server = Server(name="t", domain="1.2.3.4") + exec_mock = AsyncMock(return_value={"exit_code": 0}) + monkeypatch.setattr(svc, "exec_ssh_command_with_fallback", exec_mock) + + await cleanup_source_archive(server, "/tmp/nexus-xfer-test.tar.gz") + + exec_mock.assert_awaited_once() + cmd = exec_mock.await_args.args[1] + assert "rm -f" in cmd + assert "/tmp/nexus-xfer-test.tar.gz" in cmd