Merge pull request #11 from openclaw/fix/manifest-safe-exdev-move
fix: preserve concurrent move fallback writes
This commit is contained in:
Peter Steinberger
GitHub
commit
7ca0af4bac
@ -6,6 +6,7 @@
|
||||
|
||||
- Harden temp filename prefixes, local-root reads, private store imports, durable queue reads, and regular-file byte caps against Deepsec-reported path traversal, symlink, and oversized-read races.
|
||||
- Harden sidecar lock cleanup and stale-lock handling so stale third-party locks fail closed instead of being deleted by path.
|
||||
- Make cross-device move fallbacks reject source changes during staged copies and clean up only the source entries copied into the staged destination, preserving concurrent source additions or replacements instead of recursively deleting them.
|
||||
|
||||
### Tests
|
||||
|
||||
|
||||
@ -205,7 +205,7 @@ JSON5-backed plugin manifests.
|
||||
|
||||
## Atomic writes
|
||||
|
||||
`replaceFileAtomic()` writes a sibling temp file, optionally fsyncs it, and renames it over the destination. Mode preservation, rename retry / copy fallback on `EPERM`, parent-directory fsync, and a `beforeRename` hook for backup or observer flows are all opt-in.
|
||||
`replaceFileAtomic()` writes a sibling temp file, optionally fsyncs it, and renames it over the destination. Mode preservation, rename retry / copy fallback on `EPERM`, parent-directory fsync, and a `beforeRename` hook for backup or observer flows are all opt-in. `movePathWithCopyFallback()` stages cross-device moves before commit and removes only the copied source entries, so concurrent source additions or replacements are preserved.
|
||||
|
||||
```ts
|
||||
import { replaceFileAtomic } from "@openclaw/fs-safe/atomic";
|
||||
|
||||
@ -118,19 +118,25 @@ await writeTextAtomic("/srv/workspace/rendered.md", rendered, {
|
||||
|
||||
Rename a path. If the rename fails with `EXDEV` (cross-device), fall back to
|
||||
copying into a staged sibling path, renaming that staged path into place, and
|
||||
then removing the source. The fallback avoids buffering regular files into
|
||||
memory and does not tighten the destination parent directory mode.
|
||||
then removing only the source entries that were copied. The fallback avoids
|
||||
buffering regular files into memory and does not tighten the destination parent
|
||||
directory mode.
|
||||
|
||||
```ts
|
||||
import { movePathWithCopyFallback } from "@openclaw/fs-safe/atomic";
|
||||
|
||||
await movePathWithCopyFallback({
|
||||
from: "/srv/cache/blob.bin",
|
||||
sourceHardlinks: "reject",
|
||||
to: "/srv/persistent/blob.bin",
|
||||
});
|
||||
```
|
||||
|
||||
Use it when source and destination might live on different filesystems (containers, tmpfs, separate volumes).
|
||||
If another writer changes source entries during the fallback, the staged copy
|
||||
throws `ESTALE` before commit when possible. If the destination has already
|
||||
been committed, cleanup still preserves the changed source entries and throws
|
||||
`ESTALE`.
|
||||
|
||||
## Difference from `root()`
|
||||
|
||||
|
||||
260
src/move-path.ts
260
src/move-path.ts
@ -1,13 +1,259 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { constants as fsConstants } from "node:fs";
|
||||
import type { FileHandle } from "node:fs/promises";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { guardedRename, guardedRm } from "./guarded-mutation.js";
|
||||
import { guardedRename } from "./guarded-mutation.js";
|
||||
|
||||
export type MovePathWithCopyFallbackOptions = {
|
||||
from: string;
|
||||
sourceHardlinks?: "allow" | "reject";
|
||||
to: string;
|
||||
};
|
||||
|
||||
type EntryIdentity = {
|
||||
ctimeMs: number;
|
||||
dev: number;
|
||||
ino: number;
|
||||
mode: number;
|
||||
mtimeMs: number;
|
||||
size: number;
|
||||
};
|
||||
|
||||
type CopiedEntryManifest =
|
||||
| (EntryIdentity & {
|
||||
children: Array<{ name: string; manifest: CopiedEntryManifest }>;
|
||||
kind: "directory";
|
||||
})
|
||||
| (EntryIdentity & { kind: "leaf" });
|
||||
|
||||
type CleanupCopiedEntryResult = "removed" | "stale";
|
||||
|
||||
function entryIdentity(stat: {
|
||||
ctimeMs: number;
|
||||
dev: number;
|
||||
ino: number;
|
||||
mode: number;
|
||||
mtimeMs: number;
|
||||
size: number;
|
||||
}): EntryIdentity {
|
||||
return {
|
||||
ctimeMs: stat.ctimeMs,
|
||||
dev: stat.dev,
|
||||
ino: stat.ino,
|
||||
mode: stat.mode,
|
||||
mtimeMs: stat.mtimeMs,
|
||||
size: stat.size,
|
||||
};
|
||||
}
|
||||
|
||||
function sameIdentity(a: EntryIdentity, b: EntryIdentity): boolean {
|
||||
return (
|
||||
a.dev === b.dev &&
|
||||
a.ino === b.ino &&
|
||||
a.mode === b.mode &&
|
||||
a.size === b.size &&
|
||||
a.mtimeMs === b.mtimeMs &&
|
||||
a.ctimeMs === b.ctimeMs
|
||||
);
|
||||
}
|
||||
|
||||
function sameDirectoryNode(a: EntryIdentity, b: EntryIdentity): boolean {
|
||||
return a.dev === b.dev && a.ino === b.ino;
|
||||
}
|
||||
|
||||
function modeBits(mode: number): number {
|
||||
return mode & 0o777;
|
||||
}
|
||||
|
||||
function sourceChangedError(sourcePath: string): Error {
|
||||
return Object.assign(new Error(`Source changed during move fallback: ${sourcePath}`), {
|
||||
code: "ESTALE",
|
||||
});
|
||||
}
|
||||
|
||||
async function assertSourceStillMatches(
|
||||
sourcePath: string,
|
||||
identity: EntryIdentity,
|
||||
): Promise<void> {
|
||||
if (!sameIdentity(identity, entryIdentity(await fs.lstat(sourcePath)))) {
|
||||
throw sourceChangedError(sourcePath);
|
||||
}
|
||||
}
|
||||
|
||||
function regularReadFlags(): number {
|
||||
return (
|
||||
fsConstants.O_RDONLY |
|
||||
(typeof fsConstants.O_NOFOLLOW === "number" && process.platform !== "win32"
|
||||
? fsConstants.O_NOFOLLOW
|
||||
: 0)
|
||||
);
|
||||
}
|
||||
|
||||
async function writeAll(handle: FileHandle, buffer: Buffer, bytesRead: number): Promise<void> {
|
||||
let offset = 0;
|
||||
while (offset < bytesRead) {
|
||||
const { bytesWritten } = await handle.write(buffer, offset, bytesRead - offset);
|
||||
offset += bytesWritten;
|
||||
}
|
||||
}
|
||||
|
||||
async function copyRegularFilePinned(params: {
|
||||
from: string;
|
||||
identity: EntryIdentity;
|
||||
mode: number;
|
||||
to: string;
|
||||
}): Promise<void> {
|
||||
let destinationCreated = false;
|
||||
let sourceHandle: FileHandle;
|
||||
try {
|
||||
sourceHandle = await fs.open(params.from, regularReadFlags());
|
||||
} catch (error) {
|
||||
const code = (error as NodeJS.ErrnoException | null)?.code;
|
||||
if (code === "ELOOP" || code === "ENOENT" || code === "ENOTDIR") {
|
||||
throw sourceChangedError(params.from);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
try {
|
||||
const openedStat = await sourceHandle.stat();
|
||||
if (!openedStat.isFile() || !sameIdentity(params.identity, entryIdentity(openedStat))) {
|
||||
throw sourceChangedError(params.from);
|
||||
}
|
||||
|
||||
const destinationHandle = await fs.open(
|
||||
params.to,
|
||||
fsConstants.O_WRONLY | fsConstants.O_CREAT | fsConstants.O_EXCL,
|
||||
modeBits(params.mode) || 0o666,
|
||||
);
|
||||
destinationCreated = true;
|
||||
try {
|
||||
const scratch = Buffer.allocUnsafe(64 * 1024);
|
||||
while (true) {
|
||||
const { bytesRead } = await sourceHandle.read(scratch, 0, scratch.length, null);
|
||||
if (bytesRead === 0) {
|
||||
break;
|
||||
}
|
||||
await writeAll(destinationHandle, scratch, bytesRead);
|
||||
}
|
||||
} finally {
|
||||
await destinationHandle.close();
|
||||
}
|
||||
|
||||
// Re-check the opened handle before the staged tree can be committed. If
|
||||
// the source changed while we copied, the caller should retry the move.
|
||||
if (!sameIdentity(params.identity, entryIdentity(await sourceHandle.stat()))) {
|
||||
throw sourceChangedError(params.from);
|
||||
}
|
||||
await fs.chmod(params.to, modeBits(params.mode)).catch(() => undefined);
|
||||
} catch (error) {
|
||||
if (destinationCreated) {
|
||||
await fs.rm(params.to, { force: true }).catch(() => undefined);
|
||||
}
|
||||
throw error;
|
||||
} finally {
|
||||
await sourceHandle.close();
|
||||
}
|
||||
}
|
||||
|
||||
async function copyEntryWithManifest(
|
||||
from: string,
|
||||
to: string,
|
||||
options: { sourceHardlinks: "allow" | "reject" },
|
||||
): Promise<CopiedEntryManifest> {
|
||||
const sourceStat = await fs.lstat(from);
|
||||
const identity = entryIdentity(sourceStat);
|
||||
|
||||
if (sourceStat.isSymbolicLink()) {
|
||||
await fs.symlink(await fs.readlink(from), to);
|
||||
// readlink() is path-based; verify the symlink we copied is still the one
|
||||
// we inspected before letting the staged destination become visible.
|
||||
await assertSourceStillMatches(from, identity);
|
||||
return { ...identity, kind: "leaf" };
|
||||
}
|
||||
|
||||
if (sourceStat.isDirectory()) {
|
||||
await fs.mkdir(to, { mode: modeBits(sourceStat.mode) || 0o755 });
|
||||
const children: Array<{ name: string; manifest: CopiedEntryManifest }> = [];
|
||||
for (const child of await fs.readdir(from)) {
|
||||
children.push({
|
||||
name: child,
|
||||
manifest: await copyEntryWithManifest(path.join(from, child), path.join(to, child), options),
|
||||
});
|
||||
}
|
||||
// Directory traversal is path-based in Node. Treat a changed parent as a
|
||||
// stale move before committing so swapped-in outside trees are not imported.
|
||||
await assertSourceStillMatches(from, identity);
|
||||
// mkdir() honors process umask. Restore the source mode before commit so
|
||||
// EXDEV fallback preserves directory permissions like fs.cp did.
|
||||
await fs.chmod(to, modeBits(sourceStat.mode));
|
||||
return { ...identity, children, kind: "directory" };
|
||||
}
|
||||
|
||||
if (!sourceStat.isFile()) {
|
||||
throw new Error(`Refusing to move non-file path with copy fallback: ${from}`);
|
||||
}
|
||||
if (options.sourceHardlinks === "reject" && sourceStat.nlink > 1) {
|
||||
throw new Error(`Refusing to move hardlinked file with copy fallback: ${from}`);
|
||||
}
|
||||
|
||||
await copyRegularFilePinned({ from, identity, mode: sourceStat.mode, to });
|
||||
return { ...identity, kind: "leaf" };
|
||||
}
|
||||
|
||||
function mergeCleanupResults(
|
||||
a: CleanupCopiedEntryResult,
|
||||
b: CleanupCopiedEntryResult,
|
||||
): CleanupCopiedEntryResult {
|
||||
return a === "stale" || b === "stale" ? "stale" : "removed";
|
||||
}
|
||||
|
||||
async function cleanupCopiedEntry(
|
||||
sourcePath: string,
|
||||
manifest: CopiedEntryManifest,
|
||||
): Promise<CleanupCopiedEntryResult> {
|
||||
let currentStat: Awaited<ReturnType<typeof fs.lstat>>;
|
||||
try {
|
||||
currentStat = await fs.lstat(sourcePath);
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException | null)?.code === "ENOENT") {
|
||||
return "removed";
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
||||
if (manifest.kind === "directory") {
|
||||
if (!currentStat.isDirectory() || !sameDirectoryNode(manifest, entryIdentity(currentStat))) {
|
||||
return "stale";
|
||||
}
|
||||
// A same-inode directory can gain unrelated children after commit. Still
|
||||
// clean manifest children so the fallback does not duplicate copied files.
|
||||
let result: CleanupCopiedEntryResult = "removed";
|
||||
for (const child of manifest.children) {
|
||||
result = mergeCleanupResults(
|
||||
result,
|
||||
await cleanupCopiedEntry(path.join(sourcePath, child.name), child.manifest),
|
||||
);
|
||||
}
|
||||
try {
|
||||
await fs.rmdir(sourcePath);
|
||||
} catch (error) {
|
||||
const code = (error as NodeJS.ErrnoException | null)?.code;
|
||||
if (code === "ENOTEMPTY" || code === "EEXIST") {
|
||||
return "stale";
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
if (!sameIdentity(manifest, entryIdentity(currentStat))) {
|
||||
return "stale";
|
||||
}
|
||||
await fs.unlink(sourcePath);
|
||||
return "removed";
|
||||
}
|
||||
|
||||
export async function movePathWithCopyFallback(
|
||||
options: MovePathWithCopyFallbackOptions,
|
||||
): Promise<void> {
|
||||
@ -22,14 +268,14 @@ export async function movePathWithCopyFallback(
|
||||
const targetDir = path.dirname(path.resolve(options.to));
|
||||
const staged = path.join(targetDir, `.fs-safe-move-${process.pid}-${randomUUID()}.tmp`);
|
||||
try {
|
||||
await fs.cp(options.from, staged, {
|
||||
recursive: true,
|
||||
force: false,
|
||||
errorOnExist: true,
|
||||
dereference: false,
|
||||
const manifest = await copyEntryWithManifest(options.from, staged, {
|
||||
sourceHardlinks: options.sourceHardlinks ?? "allow",
|
||||
});
|
||||
await guardedRename({ from: staged, to: options.to });
|
||||
await guardedRm({ target: options.from, recursive: true, force: true, verifyAfter: false });
|
||||
const cleanupResult = await cleanupCopiedEntry(options.from, manifest);
|
||||
if (cleanupResult === "stale") {
|
||||
throw sourceChangedError(options.from);
|
||||
}
|
||||
} finally {
|
||||
await fs.rm(staged, { recursive: true, force: true }).catch(() => undefined);
|
||||
}
|
||||
|
||||
179
test/move-path-regression.test.ts
Normal file
179
test/move-path-regression.test.ts
Normal file
@ -0,0 +1,179 @@
|
||||
import fsp from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { movePathWithCopyFallback } from "../src/move-path.js";
|
||||
|
||||
const tempDirs: string[] = [];
|
||||
|
||||
async function tempRoot(prefix: string): Promise<string> {
|
||||
const dir = await fsp.mkdtemp(path.join(os.tmpdir(), prefix));
|
||||
tempDirs.push(dir);
|
||||
return dir;
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks();
|
||||
await Promise.all(tempDirs.splice(0).map((dir) => fsp.rm(dir, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
describe("movePathWithCopyFallback regressions", () => {
|
||||
it.runIf(process.platform !== "win32")(
|
||||
"does not delete source entries replaced after an EXDEV copy",
|
||||
async () => {
|
||||
const base = await tempRoot("fs-safe-move-exdev-replaced-source-");
|
||||
const source = path.join(base, "source-dir");
|
||||
const dest = path.join(base, "dest-dir");
|
||||
await fsp.mkdir(source);
|
||||
await fsp.writeFile(path.join(source, "copied.txt"), "copied");
|
||||
const realRename = fsp.rename;
|
||||
vi.spyOn(fsp, "rename").mockImplementation(async (from, to) => {
|
||||
if (from === source && to === dest) {
|
||||
throw Object.assign(new Error("cross-device"), { code: "EXDEV" });
|
||||
}
|
||||
await realRename(from, to);
|
||||
if (to === dest && String(from).includes(".fs-safe-move-")) {
|
||||
await fsp.rm(path.join(source, "copied.txt"));
|
||||
await fsp.writeFile(path.join(source, "copied.txt"), "replacement");
|
||||
await fsp.writeFile(path.join(source, "late.txt"), "late");
|
||||
}
|
||||
});
|
||||
|
||||
await expect(movePathWithCopyFallback({ from: source, to: dest })).rejects.toMatchObject({
|
||||
code: "ESTALE",
|
||||
});
|
||||
|
||||
await expect(fsp.readFile(path.join(dest, "copied.txt"), "utf8")).resolves.toBe("copied");
|
||||
await expect(fsp.readFile(path.join(source, "copied.txt"), "utf8")).resolves.toBe(
|
||||
"replacement",
|
||||
);
|
||||
await expect(fsp.readFile(path.join(source, "late.txt"), "utf8")).resolves.toBe("late");
|
||||
},
|
||||
);
|
||||
|
||||
it.runIf(process.platform !== "win32")(
|
||||
"can reject hardlinked files during EXDEV move fallback",
|
||||
async () => {
|
||||
const base = await tempRoot("fs-safe-move-exdev-hardlink-");
|
||||
const source = path.join(base, "source.txt");
|
||||
const hardlink = path.join(base, "hardlink.txt");
|
||||
const dest = path.join(base, "dest.txt");
|
||||
await fsp.writeFile(source, "source");
|
||||
await fsp.link(source, hardlink);
|
||||
const realRename = fsp.rename;
|
||||
vi.spyOn(fsp, "rename").mockImplementation(async (from, to) => {
|
||||
if (from === source && to === dest) {
|
||||
throw Object.assign(new Error("cross-device"), { code: "EXDEV" });
|
||||
}
|
||||
return await realRename(from, to);
|
||||
});
|
||||
|
||||
await expect(
|
||||
movePathWithCopyFallback({ from: source, sourceHardlinks: "reject", to: dest }),
|
||||
).rejects.toThrow("Refusing to move hardlinked file");
|
||||
|
||||
await expect(fsp.readFile(source, "utf8")).resolves.toBe("source");
|
||||
await expect(fsp.stat(dest)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
},
|
||||
);
|
||||
|
||||
it.runIf(process.platform !== "win32")(
|
||||
"preserves directory modes during EXDEV move fallback",
|
||||
async () => {
|
||||
const base = await tempRoot("fs-safe-move-exdev-dir-mode-");
|
||||
const source = path.join(base, "source-dir");
|
||||
const dest = path.join(base, "dest-dir");
|
||||
await fsp.mkdir(source);
|
||||
await fsp.writeFile(path.join(source, "copied.txt"), "copied");
|
||||
await fsp.chmod(source, 0o777);
|
||||
const realRename = fsp.rename;
|
||||
vi.spyOn(fsp, "rename").mockImplementation(async (from, to) => {
|
||||
if (from === source && to === dest) {
|
||||
throw Object.assign(new Error("cross-device"), { code: "EXDEV" });
|
||||
}
|
||||
return await realRename(from, to);
|
||||
});
|
||||
const realMkdir = fsp.mkdir;
|
||||
vi.spyOn(fsp, "mkdir").mockImplementation(async (target, options) => {
|
||||
const result = await realMkdir(target, options as never);
|
||||
if (String(target).includes(".fs-safe-move-")) {
|
||||
await fsp.chmod(target, 0o700);
|
||||
}
|
||||
return result;
|
||||
});
|
||||
|
||||
await movePathWithCopyFallback({ from: source, to: dest });
|
||||
|
||||
expect((await fsp.stat(dest)).mode & 0o777).toBe(0o777);
|
||||
await expect(fsp.readFile(path.join(dest, "copied.txt"), "utf8")).resolves.toBe("copied");
|
||||
},
|
||||
);
|
||||
|
||||
it.runIf(process.platform !== "win32")(
|
||||
"removes unchanged copied children when source directory gains a late child",
|
||||
async () => {
|
||||
const base = await tempRoot("fs-safe-move-exdev-added-source-");
|
||||
const source = path.join(base, "source-dir");
|
||||
const dest = path.join(base, "dest-dir");
|
||||
await fsp.mkdir(source);
|
||||
await fsp.writeFile(path.join(source, "copied.txt"), "copied");
|
||||
const realRename = fsp.rename;
|
||||
vi.spyOn(fsp, "rename").mockImplementation(async (from, to) => {
|
||||
if (from === source && to === dest) {
|
||||
throw Object.assign(new Error("cross-device"), { code: "EXDEV" });
|
||||
}
|
||||
await realRename(from, to);
|
||||
if (to === dest && String(from).includes(".fs-safe-move-")) {
|
||||
await fsp.writeFile(path.join(source, "late.txt"), "late");
|
||||
}
|
||||
});
|
||||
|
||||
await expect(movePathWithCopyFallback({ from: source, to: dest })).rejects.toMatchObject({
|
||||
code: "ESTALE",
|
||||
});
|
||||
|
||||
await expect(fsp.readFile(path.join(dest, "copied.txt"), "utf8")).resolves.toBe("copied");
|
||||
await expect(fsp.stat(path.join(source, "copied.txt"))).rejects.toMatchObject({
|
||||
code: "ENOENT",
|
||||
});
|
||||
await expect(fsp.readFile(path.join(source, "late.txt"), "utf8")).resolves.toBe("late");
|
||||
},
|
||||
);
|
||||
|
||||
it.runIf(process.platform !== "win32")(
|
||||
"does not commit bytes from a source swapped after validation",
|
||||
async () => {
|
||||
const base = await tempRoot("fs-safe-move-exdev-source-swap-");
|
||||
const outside = await tempRoot("fs-safe-move-exdev-source-swap-outside-");
|
||||
const source = path.join(base, "source.txt");
|
||||
const dest = path.join(base, "dest.txt");
|
||||
const outsideFile = path.join(outside, "secret.txt");
|
||||
await fsp.writeFile(source, "inside");
|
||||
await fsp.writeFile(outsideFile, "secret");
|
||||
|
||||
const realRename = fsp.rename;
|
||||
vi.spyOn(fsp, "rename").mockImplementation(async (from, to) => {
|
||||
if (from === source && to === dest) {
|
||||
throw Object.assign(new Error("cross-device"), { code: "EXDEV" });
|
||||
}
|
||||
return await realRename(from, to);
|
||||
});
|
||||
const realLstat = fsp.lstat;
|
||||
let swapped = false;
|
||||
vi.spyOn(fsp, "lstat").mockImplementation(async (candidate, options) => {
|
||||
const stat = await realLstat(candidate, options as never);
|
||||
if (!swapped && candidate === source) {
|
||||
swapped = true;
|
||||
await fsp.rm(source);
|
||||
await fsp.symlink(outsideFile, source, "file");
|
||||
}
|
||||
return stat;
|
||||
});
|
||||
|
||||
await expect(movePathWithCopyFallback({ from: source, to: dest })).rejects.toMatchObject({
|
||||
code: "ESTALE",
|
||||
});
|
||||
await expect(fsp.stat(dest)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
},
|
||||
);
|
||||
});
|
||||
Loading…
Reference in New Issue
Block a user