diff --git a/plugins/file-manager/src/backend/download-routes.ts b/plugins/file-manager/src/backend/download-routes.ts index 21ed7c9cf..9de1faf97 100644 --- a/plugins/file-manager/src/backend/download-routes.ts +++ b/plugins/file-manager/src/backend/download-routes.ts @@ -1,6 +1,7 @@ import type { Express } from "express"; import type { PluginContext } from "@termix/plugin-sdk/backend"; import { getMimeType } from "./utils.js"; +import { createDownloadStream } from "./download-stream.js"; import { execChannel, getSessionSftp, type SSHSession } from "./session.js"; type FileDownloadRoutesDeps = { @@ -296,9 +297,12 @@ export function registerFileDownloadRoutes( res.setHeader("Content-Length", String(stats.size)); - const readStream = sftp.createReadStream(filePath); + if (res.destroyed) return; + const readStream = createDownloadStream(sftp, filePath, stats.size); + res.on("close", () => readStream.destroy()); readStream.on("error", (err) => { if (!res.headersSent) { + res.removeHeader("Content-Length"); res.status(500).json({ error: `Download failed: ${err.message}` }); } else { res.destroy(); diff --git a/plugins/file-manager/src/backend/download-stream.ts b/plugins/file-manager/src/backend/download-stream.ts new file mode 100644 index 000000000..27f939e4a --- /dev/null +++ b/plugins/file-manager/src/backend/download-stream.ts @@ -0,0 +1,68 @@ +import { Readable } from "node:stream"; +import type { SFTPWrapper } from "ssh2"; +import { + promisifySftpClose, + promisifySftpOpen, + SFTP_OPEN_READ, +} from "./sftp-promisify.js"; + +const CHUNK_SIZE = 32 * 1024; +const CONCURRENCY = 8; + +/** Bounded read-ahead; batches retain file order even if replies arrive out of order. */ +export function createDownloadStream( + sftp: SFTPWrapper, + path: string, + size: number, +): Readable { + const stream = Readable.from(readFile(), { + objectMode: false, + highWaterMark: CHUNK_SIZE, + }); + return stream; + + async function readChunk(handle: Buffer, position: number, length: number) { + const buffer = Buffer.allocUnsafe(length); + let offset = 0; + while (offset < length && !stream.destroyed) { + const bytesRead = await new Promise((resolve, reject) => { + sftp.read( + handle, + buffer, + offset, + length - offset, + position + offset, + (err, bytes) => (err ? reject(err) : resolve(bytes)), + ); + }); + if (bytesRead === 0) + throw new Error("File ended before the advertised size"); + offset += bytesRead; + } + return buffer; + } + + async function* readFile() { + if (stream.destroyed) return; + const handle = await promisifySftpOpen(sftp, path, SFTP_OPEN_READ, 0o666); + try { + for (let position = 0; position < size && !stream.destroyed;) { + const reads: Promise[] = []; + for (let i = 0; i < CONCURRENCY && position < size; i++) { + const length = Math.min(CHUNK_SIZE, size - position); + reads.push(readChunk(handle, position, length)); + position += length; + } + // Settle all reads before closing the shared handle after an error. + const results = await Promise.allSettled(reads); + if (stream.destroyed) return; + for (const result of results) { + if (result.status === "rejected") throw result.reason; + yield result.value; + } + } + } finally { + await promisifySftpClose(sftp, handle); + } + } +} diff --git a/plugins/file-manager/src/frontend/FileManagerGrid.tsx b/plugins/file-manager/src/frontend/FileManagerGrid.tsx index e1202aa4a..c1f67bb61 100644 --- a/plugins/file-manager/src/frontend/FileManagerGrid.tsx +++ b/plugins/file-manager/src/frontend/FileManagerGrid.tsx @@ -458,6 +458,9 @@ export function FileManagerGrid({ }; const handleFileDrop = (e: React.DragEvent, targetFile: FileItem) => { + // OS drops belong to the pane's upload handler, even over an existing row. + if (dragState.type !== "internal" && !isLocalFilesDrag(e.dataTransfer)) + return; e.preventDefault(); e.stopPropagation(); diff --git a/plugins/file-manager/tests/backend/download-routes.test.ts b/plugins/file-manager/tests/backend/download-routes.test.ts new file mode 100644 index 000000000..b934cc733 --- /dev/null +++ b/plugins/file-manager/tests/backend/download-routes.test.ts @@ -0,0 +1,124 @@ +import { once } from "node:events"; +import { PassThrough } from "node:stream"; +import type { Express, Request, Response } from "express"; +import type { PluginContext } from "@termix/plugin-sdk/backend"; +import { describe, expect, it, vi } from "vitest"; +import { registerFileDownloadRoutes } from "../../src/backend/download-routes.js"; +import type { SSHSession } from "../../src/backend/session.js"; + +function setup(remote: object, owned = true) { + const routes = new Map< + string, + (req: Request, res: Response) => Promise + >(); + registerFileDownloadRoutes( + { post: (path, handler) => routes.set(path, handler) } as Express, + { + ctx: { + currentActor: () => "user", + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + } as unknown as PluginContext, + sshSessions: { s: { isConnected: true, sftp: remote } as SSHSession }, + scheduleSessionCleanup: vi.fn(), + verifySessionOwnership: () => owned, + }, + ); + const headers = new Map(); + const res = Object.assign(new PassThrough(), { + headersSent: false, + setHeader: (name: string, value: string) => headers.set(name, value), + removeHeader: (name: string) => headers.delete(name), + status: vi.fn().mockReturnThis(), + json: vi.fn(), + }); + const start = () => + routes.get("/downloadFileStream")!( + { body: { sessionId: "s", path: "/file" } } as Request, + res as unknown as Response, + ); + return { res, start, headers }; +} + +function source() { + return { + stat: vi.fn((_path, cb) => cb(undefined, { size: 5, isFile: () => true })), + open: vi.fn((_path, _flags, _mode, cb) => + cb(undefined, Buffer.from("handle")), + ), + read: vi.fn((_handle, buffer, offset, _length, _position, cb) => { + Buffer.from("hello").copy(buffer, offset); + cb(undefined, 5); + }), + close: vi.fn((_handle, cb) => cb()), + }; +} + +describe("stream download route", () => { + it("streams the file content with its advertised length", async () => { + const remote = source(); + const { res, start, headers } = setup(remote); + await start(); + const chunks: Buffer[] = []; + for await (const chunk of res) chunks.push(chunk); + expect(Buffer.concat(chunks).toString()).toBe("hello"); + expect(headers.get("Content-Length")).toBe("5"); + expect(remote.close).toHaveBeenCalledOnce(); + }); + + it("does not open a file after the HTTP client disconnected during stat", async () => { + const remote = source(); + let finishStat!: () => void; + remote.stat.mockImplementation((_path, cb) => { + finishStat = () => cb(undefined, { size: 5, isFile: () => true }); + }); + const { res, start } = setup(remote); + const started = start(); + await Promise.resolve(); + res.destroy(); + finishStat(); + await started; + expect(remote.open).not.toHaveBeenCalled(); + }); + + it("closes the SFTP handle after HTTP disconnect during a read", async () => { + const remote = source(); + let finishRead!: () => void; + remote.read.mockImplementation( + (_h, buffer, offset, length, position, cb) => { + finishRead = () => cb(undefined, 0); + }, + ); + const { res, start } = setup(remote); + await start(); + await new Promise((resolve) => setImmediate(resolve)); + res.destroy(); + await once(res, "close"); + finishRead(); + await vi.waitFor(() => expect(remote.close).toHaveBeenCalledOnce()); + expect(remote.read).toHaveBeenCalledOnce(); + }); + + it("removes the file length before returning an early stream error as JSON", async () => { + const remote = source(); + remote.open.mockImplementation((_p, _f, _m, cb) => + cb(new Error("permission denied"), undefined), + ); + const { res, start, headers } = setup(remote); + await start(); + await vi.waitFor(() => expect(res.status).toHaveBeenCalledWith(500)); + expect(headers.has("Content-Length")).toBe(false); + expect(res.json).toHaveBeenCalledWith({ + error: "Download failed: permission denied", + }); + res.destroy(); + }); + + it("refuses another user's SSH session before reading a file", async () => { + const remote = source(); + const { res, start } = setup(remote, false); + await start(); + expect(res.status).toHaveBeenCalledWith(403); + expect(remote.stat).not.toHaveBeenCalled(); + res.destroy(); + }); +}); diff --git a/plugins/file-manager/tests/backend/download-stream-sftp.test.ts b/plugins/file-manager/tests/backend/download-stream-sftp.test.ts new file mode 100644 index 000000000..ea3b350ae --- /dev/null +++ b/plugins/file-manager/tests/backend/download-stream-sftp.test.ts @@ -0,0 +1,74 @@ +import { once } from "node:events"; +import type { AddressInfo } from "node:net"; +import { Client, Server, utils, type SFTPWrapper } from "ssh2"; +import { expect, it } from "vitest"; +import { createDownloadStream } from "../../src/backend/download-stream.js"; + +it("overlaps real SFTP reads while retaining the serial stream's byte content", async () => { + const data = Buffer.alloc(1024 * 1024 + 71); + for (let i = 0; i < data.length; i++) data[i] = (i * 7 + (i >>> 10)) % 256; + let pending = 0; + let peak = 0; + let closed = 0; + const server = new Server( + { + hostKeys: [utils.generateKeyPairSync("rsa", { bits: 2048 }).private], + }, + (client) => { + client.on("authentication", (ctx) => ctx.accept()); + client.on("ready", () => + client.on("session", (accept) => { + accept().on("sftp", (acceptSftp) => { + const sftp = acceptSftp(); + sftp.on("OPEN", (id) => sftp.handle(id, Buffer.from("file"))); + sftp.on("READ", (id, _handle, offset, length) => { + peak = Math.max(peak, ++pending); + setTimeout(() => { + pending--; + if (offset >= data.length) sftp.status(id, 1); + else sftp.data(id, data.subarray(offset, offset + length)); + }, 10); + }); + sftp.on("CLOSE", (id) => { + closed++; + sftp.status(id, 0); + }); + }); + }), + ); + }, + ); + const client = new Client(); + try { + await new Promise((resolve) => + server.listen(0, "127.0.0.1", resolve), + ); + client.connect({ + host: "127.0.0.1", + port: (server.address() as AddressInfo).port, + username: "test", + }); + await once(client, "ready"); + const sftp = await new Promise((resolve, reject) => + client.sftp((err, channel) => (err ? reject(err) : resolve(channel))), + ); + const serial: Buffer[] = []; + for await (const chunk of sftp.createReadStream("/file")) + serial.push(chunk); + expect(Buffer.concat(serial)).toEqual(data); + expect(peak).toBe(1); + peak = 0; + const pipelined: Buffer[] = []; + for await (const chunk of createDownloadStream(sftp, "/file", data.length)) + pipelined.push(chunk); + expect(Buffer.concat(pipelined)).toEqual(data); + expect(peak).toBeGreaterThan(1); + expect(peak).toBeLessThanOrEqual(8); + expect(closed).toBe(2); + } finally { + const closedClient = once(client, "close"); + client.destroy(); + await closedClient; + await new Promise((resolve) => server.close(() => resolve())); + } +}, 15000); diff --git a/plugins/file-manager/tests/backend/download-stream.test.ts b/plugins/file-manager/tests/backend/download-stream.test.ts new file mode 100644 index 000000000..6aa694431 --- /dev/null +++ b/plugins/file-manager/tests/backend/download-stream.test.ts @@ -0,0 +1,133 @@ +import { once } from "node:events"; +import { setImmediate as nextTurn } from "node:timers/promises"; +import type { SFTPWrapper } from "ssh2"; +import { describe, expect, it, vi } from "vitest"; +import { createDownloadStream } from "../../src/backend/download-stream.js"; + +function source( + data: Buffer, + options: { shortRead?: number; failAt?: number } = {}, +) { + let pending = 0; + let peak = 0; + const closedWithPending: number[] = []; + const sftp = { + open: vi.fn((_path, _flags, _mode, cb) => + cb(undefined, Buffer.from("handle")), + ), + close: vi.fn((_handle, cb) => { + closedWithPending.push(pending); + cb(); + }), + read: vi.fn((_handle, buffer, offset, length, position, cb) => { + peak = Math.max(peak, ++pending); + // Later file offsets can finish first. + setTimeout( + () => { + pending--; + if (position === options.failAt) return cb(new Error("read failed")); + const bytes = Math.min( + length, + options.shortRead ?? length, + Math.max(0, data.length - position), + ); + if (bytes > 0) data.copy(buffer, offset, position, position + bytes); + cb(undefined, bytes); + }, + position === 0 ? 10 : 0, + ); + }), + }; + return { + sftp: sftp as unknown as SFTPWrapper, + close: sftp.close, + read: sftp.read, + peak: () => peak, + closedWithPending, + }; +} + +async function contents(stream: ReturnType) { + const buffers: Buffer[] = []; + for await (const buffer of stream) buffers.push(buffer); + return Buffer.concat(buffers); +} + +describe("SFTP download stream", () => { + it("pipelines bounded reads and preserves bytes despite out-of-order replies", async () => { + const data = Buffer.alloc(1024 * 1024 + 123); + for (let i = 0; i < data.length; i++) data[i] = (i * 13 + (i >>> 8)) % 256; + const remote = source(data); + expect( + await contents(createDownloadStream(remote.sftp, "/file", data.length)), + ).toEqual(data); + expect(remote.peak()).toBe(8); + expect(remote.closedWithPending).toEqual([0]); + }); + + it("fills short server reads without losing or repeating bytes", async () => { + const data = Buffer.from("short reads ".repeat(10000)); + const remote = source(data, { shortRead: 4096 }); + expect( + await contents(createDownloadStream(remote.sftp, "/file", data.length)), + ).toEqual(data); + expect(remote.closedWithPending).toEqual([0]); + }); + + it("handles empty files without issuing reads", async () => { + const remote = source(Buffer.alloc(0)); + expect( + await contents(createDownloadStream(remote.sftp, "/empty", 0)), + ).toEqual(Buffer.alloc(0)); + expect(remote.read).not.toHaveBeenCalled(); + expect(remote.close).toHaveBeenCalledOnce(); + }); + + it("fails on premature EOF instead of returning uninitialized bytes", async () => { + const remote = source(Buffer.from("truncated")); + await expect( + contents(createDownloadStream(remote.sftp, "/file", 100000)), + ).rejects.toThrow("File ended"); + expect(remote.closedWithPending).toEqual([0]); + }); + + it("settles outstanding reads before closing on a read failure", async () => { + const remote = source(Buffer.alloc(1024 * 1024), { failAt: 32768 }); + await expect( + contents(createDownloadStream(remote.sftp, "/file", 1024 * 1024)), + ).rejects.toThrow("read failed"); + expect(remote.read).toHaveBeenCalledTimes(8); + expect(remote.closedWithPending).toEqual([0]); + }); + + it("stops scheduling reads and closes the handle when destroyed mid-batch", async () => { + const remote = source(Buffer.alloc(1024 * 1024), { shortRead: 4096 }); + const replies: Array<() => void> = []; + remote.read.mockImplementation( + (_handle, _buffer, _offset, _length, _position, cb) => { + replies.push(() => cb(undefined, 4096)); + }, + ); + const stream = createDownloadStream(remote.sftp, "/file", 1024 * 1024); + stream.resume(); + await nextTurn(); + const closed = once(stream, "close"); + stream.destroy(); + for (const reply of replies) reply(); + await closed; + expect(remote.read).toHaveBeenCalledTimes(8); + expect(remote.closedWithPending).toEqual([0]); + }); + + it("bounds read-ahead while the consumer is paused", async () => { + const remote = source(Buffer.alloc(8 * 1024 * 1024)); + const stream = createDownloadStream(remote.sftp, "/file", 8 * 1024 * 1024); + stream.read(0); + await once(stream, "readable"); + expect(remote.read).toHaveBeenCalledTimes(8); + const closed = once(stream, "close"); + stream.destroy(); + await closed; + expect(remote.closedWithPending).toEqual([0]); + }); +}); diff --git a/plugins/file-manager/tests/frontend/FileManagerGrid.test.tsx b/plugins/file-manager/tests/frontend/FileManagerGrid.test.tsx index fafd1636b..198a04da8 100644 --- a/plugins/file-manager/tests/frontend/FileManagerGrid.test.tsx +++ b/plugins/file-manager/tests/frontend/FileManagerGrid.test.tsx @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { cleanup, fireEvent, render, screen } from "@testing-library/react"; import "@testing-library/jest-dom/vitest"; import { FileManagerGrid } from "../../src/frontend/FileManagerGrid"; @@ -44,6 +44,7 @@ const props = { afterEach(() => { cleanup(); vi.restoreAllMocks(); + vi.unstubAllGlobals(); }); describe("remote file rename", () => { @@ -89,3 +90,52 @@ describe("remote file rename", () => { expect(screen.getByText(file.modified!)).toBeInTheDocument(); }); }); + +describe("external file drops", () => { + beforeEach(() => { + vi.stubGlobal( + "ResizeObserver", + class { + observe() {} + disconnect() {} + }, + ); + }); + it.each(["list", "grid"] as const)( + "uploads dropped folders over a file in %s view", + (viewMode) => { + const onUploadItems = vi.fn(); + render( + , + ); + const folder = { + name: "logs", + isDirectory: true, + isFile: false, + } as FileSystemEntry; + fireEvent.drop(screen.getByText(file.name), { + dataTransfer: { + types: ["Files"], + files: [], + getData: () => "", + items: [{ webkitGetAsEntry: () => folder }], + }, + }); + expect(onUploadItems).toHaveBeenCalledExactlyOnceWith([folder]); + }, + ); + + it("uploads a plain OS file dropped over a row", () => { + const onUpload = vi.fn(); + render(); + const files = [new File(["log"], "app.log")]; + fireEvent.drop(screen.getByText(file.name), { + dataTransfer: { types: ["Files"], files, items: [], getData: () => "" }, + }); + expect(onUpload).toHaveBeenCalledExactlyOnceWith(files); + }); +});