From ca2268ac3a1e46d50657454a40c96eacea54b8f4 Mon Sep 17 00:00:00 2001 From: freeof123 Date: Tue, 23 Jun 2026 16:51:53 +0800 Subject: [PATCH 1/4] feat: add onDownloadReady callback and fix signMessage Ethereum prefix Expose onDownloadReady so UI can dismiss loading overlays after HEAD+tail validation; fix eth_hash_prefix to restore v4.x-compatible signatures. Co-authored-by: Cursor --- src/browser/HttpSealedFileStream.js | 12 +++++++++- src/browser/README.md | 1 + src/browser/blob_download.js | 6 ++--- src/browser/downloadUnsealed.js | 11 +++++++-- src/browser/stream_download.js | 4 ++-- src/common/ypccrypto.common.js | 9 +++++--- test/BrowserCrypto.spec.mjs | 30 +++++++++++++++++++++++++ test/HttpSealedFileStream.spec.mjs | 35 +++++++++++++++++++++++++++++ 8 files changed, 97 insertions(+), 11 deletions(-) diff --git a/src/browser/HttpSealedFileStream.js b/src/browser/HttpSealedFileStream.js index d7a6b91..a0baf47 100644 --- a/src/browser/HttpSealedFileStream.js +++ b/src/browser/HttpSealedFileStream.js @@ -23,8 +23,9 @@ export class HttpSealedFileStream extends ReadableStream { * @param {object} [options] * @param {number} [options.chunkSize] - bytes per Range request (default 1 MB) * @param {Function} [options.fetch] - fetch impl (defaults to globalThis.fetch) + * @param {Function} [options.onReady] - called after HEAD+tail validated, before body Range fetches */ - constructor(url, { chunkSize = DEFAULT_CHUNK, fetch: fetchFn } = {}) { + constructor(url, { chunkSize = DEFAULT_CHUNK, fetch: fetchFn, onReady } = {}) { const _fetch = fetchFn || fetch.bind(globalThis); const state = { url, @@ -84,6 +85,15 @@ export class HttpSealedFileStream extends ReadableStream { return; } + if (onReady) { + try { + onReady(); + } catch (e) { + controller.error(e); + return; + } + } + let pos = 0; while (pos < state.contentSize) { const chunkEnd = Math.min(pos + CHUNK, state.contentSize); diff --git a/src/browser/README.md b/src/browser/README.md index 787c931..e7e17fa 100644 --- a/src/browser/README.md +++ b/src/browser/README.md @@ -47,6 +47,7 @@ npm install @yeez-tech/meta-encryptor | `filename` | string | ✅ | 下载文件名（必需，无默认值） | | `onLog` | function | ❌ | 日志回调 `(message: string) => void` | | `onProgress` | function | ❌ | 进度回调 `(total, processed, readBytes, writeBytes) => void` | +| `onDownloadReady` | function | ❌ | 下载就绪回调：HEAD+tail 校验完成、正文 Range 开始前触发，可用于关闭准备蒙层 | | `onSuccess` | function | ❌ | 成功回调 `(data: { filename }) => void` | | `onError` | function | ❌ | 错误回调 `(error: Error) => void` | diff --git a/src/browser/blob_download.js b/src/browser/blob_download.js index 1cee387..72b2b40 100644 --- a/src/browser/blob_download.js +++ b/src/browser/blob_download.js @@ -6,14 +6,14 @@ import { createProgressTransformer } from '../common/progress.js'; * @param {string} url * @param {string} privateKeyHex * @param {string} filename - * @param {{ log?: Function, onProgress?: Function, fetch?: Function, size?: number }} [opts] + * @param {{ log?: Function, onProgress?: Function, fetch?: Function, size?: number, onDownloadReady?: Function }} [opts] */ -export async function blobDownloadAndDecrypt(url, privateKeyHex, filename, { log, onProgress, fetch: _fetch, size } = {}) { +export async function blobDownloadAndDecrypt(url, privateKeyHex, filename, { log, onProgress, fetch: _fetch, size, onDownloadReady } = {}) { log = log || (() => {}) try { const chunks = [] - const stream = new HttpSealedFileStream(url, { fetch: _fetch }) + const stream = new HttpSealedFileStream(url, { fetch: _fetch, onReady: onDownloadReady }) const unsealer = new Unsealer({ privateKeyHex: privateKeyHex.trim(), progressHandler: (total, processed, readBytes, writeBytes) => { diff --git a/src/browser/downloadUnsealed.js b/src/browser/downloadUnsealed.js index 910506c..ab099f8 100644 --- a/src/browser/downloadUnsealed.js +++ b/src/browser/downloadUnsealed.js @@ -85,6 +85,7 @@ async function inspectSealed(url, log) { * @param {string} options.filename * @param {Function} [options.onLog] * @param {Function} [options.onProgress] - (total, processed, readBytes, writeBytes) => {} + * @param {Function} [options.onDownloadReady] - HEAD+tail 完成、正文 Range 开始前触发（可关蒙层） * @param {Function} [options.onSuccess] * @param {Function} [options.onError] * @returns {Promise} @@ -95,6 +96,7 @@ export async function downloadUnsealed({ filename, onLog, onProgress, + onDownloadReady, onSuccess, onError }) { @@ -128,7 +130,12 @@ export async function downloadUnsealed({ if (!mobile) { try { log('Trying stream download...') - await streamDownloadAndDecrypt(url, key, filename, { log, onProgress, size: meta.plaintextSize }) + await streamDownloadAndDecrypt(url, key, filename, { + log, + onProgress, + size: meta.plaintextSize, + onDownloadReady, + }) if (onSuccess) onSuccess({ filename }) return } catch (e) { @@ -137,7 +144,7 @@ export async function downloadUnsealed({ } log('Using Blob download...') - await blobDownloadAndDecrypt(url, key, filename, { log, onProgress }) + await blobDownloadAndDecrypt(url, key, filename, { log, onProgress, onDownloadReady }) if (onSuccess) onSuccess({ filename }) } catch (error) { log('Download failed: ' + error.message) diff --git a/src/browser/stream_download.js b/src/browser/stream_download.js index aac78e9..3b83ffa 100644 --- a/src/browser/stream_download.js +++ b/src/browser/stream_download.js @@ -50,14 +50,14 @@ export async function getBestWritable(filename, { log, size } = {}) { return null; } -export async function streamDownloadAndDecrypt(url, privateKeyHex, filename, { log, onProgress, writable, size, fetch: _fetch } = {}) { +export async function streamDownloadAndDecrypt(url, privateKeyHex, filename, { log, onProgress, writable, size, fetch: _fetch, onDownloadReady } = {}) { log = log || (() => {}) try { const out = writable || await getBestWritable(filename, { log, size }) if (!out) throw new MetaEncryptorError('ERR_NO_STREAM_WRITABLE'); log('Stream download starting...'); - const stream = new HttpSealedFileStream(url, { fetch: _fetch }) + const stream = new HttpSealedFileStream(url, { fetch: _fetch, onReady: onDownloadReady }) const unsealer = new Unsealer({ privateKeyHex: privateKeyHex.trim(), progressHandler: (total, processed, readBytes, writeBytes) => { diff --git a/src/common/ypccrypto.common.js b/src/common/ypccrypto.common.js index 2f6ddd1..17282d1 100644 --- a/src/common/ypccrypto.common.js +++ b/src/common/ypccrypto.common.js @@ -132,14 +132,17 @@ function generateAESKeyFrom(pkey, skey){ return derived_key; } -const eth_hash_prefix = Buffer.from("\\x19Ethereum Signed Message:\\n32"); +const eth_hash_prefix = Buffer.concat([ + Buffer.from([0x19]), + Buffer.from("Ethereum Signed Message:\n32"), +]); function signMessage(skey, raw) { - let raw_hash = Buffer.from(keccak256(Buffer.from(raw))); + let raw_hash = Buffer.from(keccak256(toUint8Array(raw))); let msg = new Uint8Array(eth_hash_prefix.length + raw_hash.length) msg.set(eth_hash_prefix) msg.set(raw_hash, eth_hash_prefix.length) - msg = Buffer.from(keccak256(Buffer.from(msg))) + msg = Buffer.from(keccak256(toUint8Array(msg))) const msgBytes = toUint8Array(msg); const skeyBytes = toUint8Array(skey); diff --git a/test/BrowserCrypto.spec.mjs b/test/BrowserCrypto.spec.mjs index eadf889..6cbbddd 100644 --- a/test/BrowserCrypto.spec.mjs +++ b/test/BrowserCrypto.spec.mjs @@ -1,4 +1,5 @@ import { BrowserCrypto } from '../src/browser/ypccrypto.browser.js'; +import { eth_hash_prefix } from '../src/common/ypccrypto.common.js'; import fs from 'fs/promises'; import path from 'path'; @@ -256,6 +257,35 @@ describe('BrowserCrypto compatibility with Node YPCCrypto', () => { }); }); +describe('signMessage Ethereum prefix regression', () => { + test('eth_hash_prefix uses correct Ethereum Signed Message format', () => { + expect(eth_hash_prefix.length).toBe(28); + expect(eth_hash_prefix[0]).toBe(0x19); + expect(Buffer.from(eth_hash_prefix.subarray(1)).toString()).toBe( + 'Ethereum Signed Message:\n32' + ); + }); + + test('generateSignature matches v4.x compatible vector', () => { + const skey = Buffer.from( + '0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef', + 'hex' + ); + const epkey = Buffer.from( + '2acc91b839b115519fa43b7a0bcf20f7e83ef983c00e854343025a5a26a08a3e2b1bbb2c178989126d09d7a7def36fd029eafd60bdfe046b12b226877e018d92', + 'hex' + ); + const ehash = Buffer.from( + '30890aacb90c4a6440023df1a51a74050756d04811ce0a47eea86dd47dec320a', + 'hex' + ); + const sig = BrowserCrypto.generateSignature(skey, epkey, ehash); + expect(Buffer.from(sig).toString('hex')).toBe( + '568643ef491fb8e32d061783a6ac94aacdf7967b39ef53490121c39d7997e043630af675554b5538db2b2bab96068351842490a1ee7551d13417fd1052e41cf61c' + ); + }); +}); + // --- Fixtures-based consistency checks --- function hexToBytes(hex){ if (hex.startsWith('0x')) hex = hex.slice(2); diff --git a/test/HttpSealedFileStream.spec.mjs b/test/HttpSealedFileStream.spec.mjs index d656bb0..f415ade 100644 --- a/test/HttpSealedFileStream.spec.mjs +++ b/test/HttpSealedFileStream.spec.mjs @@ -117,6 +117,41 @@ describe('HttpSealedFileStream', () => { expect(chunks[0].length).toBe(64); }); + test('calls onReady after HEAD+tail, before body Range fetches', async () => { + const original = Buffer.alloc(100, 'A'); + const { diskBuf } = sealBuffer(original); + let ready = false; + let rangeFetchCount = 0; + + const fetch = async (_url, init = {}) => { + if (init.method === 'HEAD') { + return { + ok: true, status: 200, + headers: { get: (n) => n.toLowerCase() === 'content-length' ? String(diskBuf.length) : null } + }; + } + if (init.headers?.Range) { + rangeFetchCount++; + } + return createMockFetch(diskBuf)(_url, init); + }; + + const hsfs = new HttpSealedFileStream('http://x', { + chunkSize: 4096, + fetch, + onReady: () => { ready = true; }, + }); + + const reader = hsfs.getReader(); + while (true) { + const { done } = await reader.read(); + if (done) break; + } + + expect(ready).toBe(true); + expect(rangeFetchCount).toBeGreaterThan(0); + }); + test('streams content in multiple chunks with small chunkSize', async () => { const original = Buffer.alloc(10000, 'Z'); const { diskBuf } = sealBuffer(original); From 5be506d960f56e783de03f2ee30a020a0576a575 Mon Sep 17 00:00:00 2001 From: freeof123 Date: Tue, 23 Jun 2026 17:03:57 +0800 Subject: [PATCH 2/4] chore: bump version to 5.0.4 Co-authored-by: Cursor --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index 57d8a39..54422d9 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@yeez-tech/meta-encryptor", - "version": "5.0.3", + "version": "5.0.4", "description": "Data Seal/Unseal for Fidelius", "main": "build/commonjs/index.node.cjs", "module": "build/es/index.node.js", From 217bd71868d634642997fc0bf071f03f91395cec Mon Sep 17 00:00:00 2001 From: freeof123 Date: Tue, 23 Jun 2026 17:32:59 +0800 Subject: [PATCH 3/4] refactor: implement onDownloadReady via createDownloadReadyTransformer Move download-ready signaling out of HttpSealedFileStream into a TransformStream (same pattern as createProgressTransformer) and pipe it in stream/blob download paths without modifying the HTTP stream class. Co-authored-by: Cursor --- jest.config.browser.mjs | 3 +- src/browser/HttpSealedFileStream.js | 12 +------ src/browser/README.md | 2 +- src/browser/blob_download.js | 5 +-- src/browser/downloadUnsealed.js | 2 +- src/browser/stream_download.js | 5 +-- src/common/progress.js | 22 ++++++++++++ test/HttpSealedFileStream.spec.mjs | 22 +++--------- test/progress.spec.mjs | 55 +++++++++++++++++++++++++++++ 9 files changed, 92 insertions(+), 36 deletions(-) create mode 100644 test/progress.spec.mjs diff --git a/jest.config.browser.mjs b/jest.config.browser.mjs index bbf1b3b..a02a9f2 100644 --- a/jest.config.browser.mjs +++ b/jest.config.browser.mjs @@ -4,7 +4,8 @@ export default { testMatch: [ "/test/Browser*.spec.mjs", "/test/Http*.spec.mjs", - "/test/downloadFunctions.spec.mjs" + "/test/downloadFunctions.spec.mjs", + "/test/progress.spec.mjs" ], roots: ['/test', '/src'], diff --git a/src/browser/HttpSealedFileStream.js b/src/browser/HttpSealedFileStream.js index a0baf47..d7a6b91 100644 --- a/src/browser/HttpSealedFileStream.js +++ b/src/browser/HttpSealedFileStream.js @@ -23,9 +23,8 @@ export class HttpSealedFileStream extends ReadableStream { * @param {object} [options] * @param {number} [options.chunkSize] - bytes per Range request (default 1 MB) * @param {Function} [options.fetch] - fetch impl (defaults to globalThis.fetch) - * @param {Function} [options.onReady] - called after HEAD+tail validated, before body Range fetches */ - constructor(url, { chunkSize = DEFAULT_CHUNK, fetch: fetchFn, onReady } = {}) { + constructor(url, { chunkSize = DEFAULT_CHUNK, fetch: fetchFn } = {}) { const _fetch = fetchFn || fetch.bind(globalThis); const state = { url, @@ -85,15 +84,6 @@ export class HttpSealedFileStream extends ReadableStream { return; } - if (onReady) { - try { - onReady(); - } catch (e) { - controller.error(e); - return; - } - } - let pos = 0; while (pos < state.contentSize) { const chunkEnd = Math.min(pos + CHUNK, state.contentSize); diff --git a/src/browser/README.md b/src/browser/README.md index e7e17fa..f30d979 100644 --- a/src/browser/README.md +++ b/src/browser/README.md @@ -47,7 +47,7 @@ npm install @yeez-tech/meta-encryptor | `filename` | string | ✅ | 下载文件名（必需，无默认值） | | `onLog` | function | ❌ | 日志回调 `(message: string) => void` | | `onProgress` | function | ❌ | 进度回调 `(total, processed, readBytes, writeBytes) => void` | -| `onDownloadReady` | function | ❌ | 下载就绪回调：HEAD+tail 校验完成、正文 Range 开始前触发，可用于关闭准备蒙层 | +| `onDownloadReady` | function | ❌ | 下载就绪回调：HTTP 流首个数据块进入管道时触发（HEAD+tail 完成后），可用于关闭准备蒙层 | | `onSuccess` | function | ❌ | 成功回调 `(data: { filename }) => void` | | `onError` | function | ❌ | 错误回调 `(error: Error) => void` | diff --git a/src/browser/blob_download.js b/src/browser/blob_download.js index 72b2b40..a6b1d6d 100644 --- a/src/browser/blob_download.js +++ b/src/browser/blob_download.js @@ -1,6 +1,6 @@ import { Unsealer } from './Unsealer.js' import { HttpSealedFileStream } from './HttpSealedFileStream.js' -import { createProgressTransformer } from '../common/progress.js'; +import { createProgressTransformer, createDownloadReadyTransformer } from '../common/progress.js'; /** * @param {string} url @@ -13,7 +13,7 @@ export async function blobDownloadAndDecrypt(url, privateKeyHex, filename, { log try { const chunks = [] - const stream = new HttpSealedFileStream(url, { fetch: _fetch, onReady: onDownloadReady }) + const stream = new HttpSealedFileStream(url, { fetch: _fetch }) const unsealer = new Unsealer({ privateKeyHex: privateKeyHex.trim(), progressHandler: (total, processed, readBytes, writeBytes) => { @@ -22,6 +22,7 @@ export async function blobDownloadAndDecrypt(url, privateKeyHex, filename, { log }) await stream + .pipeThrough(createDownloadReadyTransformer(onDownloadReady)) .pipeThrough(unsealer) .pipeThrough(createProgressTransformer(size, onProgress)) .pipeTo(new WritableStream({ diff --git a/src/browser/downloadUnsealed.js b/src/browser/downloadUnsealed.js index ab099f8..9352a89 100644 --- a/src/browser/downloadUnsealed.js +++ b/src/browser/downloadUnsealed.js @@ -85,7 +85,7 @@ async function inspectSealed(url, log) { * @param {string} options.filename * @param {Function} [options.onLog] * @param {Function} [options.onProgress] - (total, processed, readBytes, writeBytes) => {} - * @param {Function} [options.onDownloadReady] - HEAD+tail 完成、正文 Range 开始前触发（可关蒙层） + * @param {Function} [options.onDownloadReady] - HTTP 流首个数据块进入管道时触发（可关蒙层） * @param {Function} [options.onSuccess] * @param {Function} [options.onError] * @returns {Promise} diff --git a/src/browser/stream_download.js b/src/browser/stream_download.js index 3b83ffa..94eb67c 100644 --- a/src/browser/stream_download.js +++ b/src/browser/stream_download.js @@ -1,7 +1,7 @@ import { Unsealer } from './Unsealer.js' import { HttpSealedFileStream } from './HttpSealedFileStream.js' import { MetaEncryptorError } from '../common/errors.js'; -import { createProgressTransformer } from '../common/progress.js'; +import { createProgressTransformer, createDownloadReadyTransformer } from '../common/progress.js'; async function ensureStreamSaver(log) { if (typeof window === 'undefined') return null; @@ -57,7 +57,7 @@ export async function streamDownloadAndDecrypt(url, privateKeyHex, filename, { l if (!out) throw new MetaEncryptorError('ERR_NO_STREAM_WRITABLE'); log('Stream download starting...'); - const stream = new HttpSealedFileStream(url, { fetch: _fetch, onReady: onDownloadReady }) + const stream = new HttpSealedFileStream(url, { fetch: _fetch }) const unsealer = new Unsealer({ privateKeyHex: privateKeyHex.trim(), progressHandler: (total, processed, readBytes, writeBytes) => { @@ -66,6 +66,7 @@ export async function streamDownloadAndDecrypt(url, privateKeyHex, filename, { l }) await stream + .pipeThrough(createDownloadReadyTransformer(onDownloadReady)) .pipeThrough(unsealer) .pipeThrough(createProgressTransformer(size, onProgress)) .pipeTo(out) diff --git a/src/common/progress.js b/src/common/progress.js index e9a3b3b..f501738 100644 --- a/src/common/progress.js +++ b/src/common/progress.js @@ -18,3 +18,25 @@ export function createProgressTransformer(plaintextSize, onProgress) { } }) } + +/** + * Create a TransformStream that fires once when the first chunk flows through. + * Place after HttpSealedFileStream to signal download stream is ready (e.g. dismiss overlay). + * + * @param {Function} [onDownloadReady] + * @returns {TransformStream} + */ +export function createDownloadReadyTransformer(onDownloadReady) { + let called = false + return new TransformStream({ + transform(chunk, controller) { + if (!called) { + called = true + if (onDownloadReady) { + onDownloadReady() + } + } + controller.enqueue(chunk) + } + }) +} diff --git a/test/HttpSealedFileStream.spec.mjs b/test/HttpSealedFileStream.spec.mjs index f415ade..4b1042e 100644 --- a/test/HttpSealedFileStream.spec.mjs +++ b/test/HttpSealedFileStream.spec.mjs @@ -13,6 +13,7 @@ import { HeaderSize, BlockInfoSize } from '../src/common/limits.js'; import { BrowserCrypto } from '../src/browser/ypccrypto.browser.js'; import { Unsealer } from '../src/browser/Unsealer.js'; import { HttpSealedFileStream } from '../src/browser/HttpSealedFileStream.js'; +import { createDownloadReadyTransformer } from '../src/common/progress.js'; import fs from 'fs'; import path from 'path'; import crypto from 'crypto'; @@ -117,30 +118,16 @@ describe('HttpSealedFileStream', () => { expect(chunks[0].length).toBe(64); }); - test('calls onReady after HEAD+tail, before body Range fetches', async () => { + test('createDownloadReadyTransformer fires on first chunk from HttpSealedFileStream', async () => { const original = Buffer.alloc(100, 'A'); const { diskBuf } = sealBuffer(original); let ready = false; - let rangeFetchCount = 0; - - const fetch = async (_url, init = {}) => { - if (init.method === 'HEAD') { - return { - ok: true, status: 200, - headers: { get: (n) => n.toLowerCase() === 'content-length' ? String(diskBuf.length) : null } - }; - } - if (init.headers?.Range) { - rangeFetchCount++; - } - return createMockFetch(diskBuf)(_url, init); - }; + const fetch = createMockFetch(diskBuf); const hsfs = new HttpSealedFileStream('http://x', { chunkSize: 4096, fetch, - onReady: () => { ready = true; }, - }); + }).pipeThrough(createDownloadReadyTransformer(() => { ready = true; })); const reader = hsfs.getReader(); while (true) { @@ -149,7 +136,6 @@ describe('HttpSealedFileStream', () => { } expect(ready).toBe(true); - expect(rangeFetchCount).toBeGreaterThan(0); }); test('streams content in multiple chunks with small chunkSize', async () => { diff --git a/test/progress.spec.mjs b/test/progress.spec.mjs new file mode 100644 index 0000000..0eaf187 --- /dev/null +++ b/test/progress.spec.mjs @@ -0,0 +1,55 @@ +import { createProgressTransformer, createDownloadReadyTransformer } from '../src/common/progress.js'; + +describe('progress transformers', () => { + test('createProgressTransformer reports cumulative bytes', async () => { + const calls = []; + const { readable, writable } = new TransformStream(); + const writer = writable.getWriter(); + writer.write(new Uint8Array(3)); + writer.write(new Uint8Array(2)); + writer.close(); + + const reader = readable + .pipeThrough(createProgressTransformer(10, (total, received) => { + calls.push([total, received]); + })) + .getReader(); + + while (!(await reader.read()).done) {} + + expect(calls).toEqual([[10, 3], [10, 5]]); + }); + + test('createDownloadReadyTransformer fires once on first chunk', async () => { + let readyCount = 0; + const { readable, writable } = new TransformStream(); + const writer = writable.getWriter(); + writer.write(new Uint8Array(64)); + writer.write(new Uint8Array(8)); + writer.close(); + + const reader = readable + .pipeThrough(createDownloadReadyTransformer(() => { readyCount++; })) + .getReader(); + + await reader.read(); + expect(readyCount).toBe(1); + await reader.read(); + expect(readyCount).toBe(1); + }); + + test('createDownloadReadyTransformer no-ops when callback omitted', async () => { + const { readable, writable } = new TransformStream(); + const writer = writable.getWriter(); + writer.write(new Uint8Array(1)); + writer.close(); + + const reader = readable + .pipeThrough(createDownloadReadyTransformer()) + .getReader(); + + const { value, done } = await reader.read(); + expect(done).toBe(false); + expect(value.length).toBe(1); + }); +}); From 6479b1bdfe93ac41dd243ff020d199b74e6557ec Mon Sep 17 00:00:00 2001 From: freeof123 Date: Thu, 9 Jul 2026 13:50:52 +0800 Subject: [PATCH 4/4] =?UTF-8?q?=E5=A4=9A=E6=AC=A1=E6=9A=82=E5=81=9C?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E9=80=9A=E8=BF=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/node/PipelineConext.js | 94 ++++++++++++++++++++++++++++---------- src/node/Recoverable.js | 44 ++++++++++-------- src/node/Unsealer.js | 12 +++-- 3 files changed, 103 insertions(+), 47 deletions(-) diff --git a/src/node/PipelineConext.js b/src/node/PipelineConext.js index f1c27fd..10aa012 100644 --- a/src/node/PipelineConext.js +++ b/src/node/PipelineConext.js @@ -5,10 +5,22 @@ import { promisify } from 'util'; import { MetaEncryptorError } from '../common/errors.js'; -const readFile = promisify(fs.readFile); -const writeFile = promisify(fs.writeFile); +const open = promisify(fs.open); +const close = promisify(fs.close); +const fsync = promisify(fs.fsync); +const rename = promisify(fs.rename); const logger = log.getLogger("meta-encryptor/PipelineContext"); +function toBinaryChunk(value) { + if (Buffer.isBuffer(value)) { + return value; + } + if (value instanceof Uint8Array) { + return Buffer.from(value.buffer, value.byteOffset, value.byteLength); + } + return null; +} + export class PipelineContext { constructor(options) { this.context = {}; @@ -37,25 +49,25 @@ export class PipelineContextInFile extends PipelineContext { constructor(filePath, options) { super(options); this.filePath = filePath; + this._saveTail = Promise.resolve(); + this._saveDirty = false; } - async saveContext() { + _buildPayload() { const binaryChunks = []; const meta = {}; let offset = 0; - logger.debug("Context before save:", this.context); - logger.debug("PipelineContextInFile::saveContext saving to ", this.filePath); - for (const [key, value] of Object.entries(this.context)) { - if (Buffer.isBuffer(value)) { - binaryChunks.push(value); + const binary = toBinaryChunk(value); + if (binary) { + binaryChunks.push(binary); meta[key] = { type: 'binary', offset, - length: value.length + length: binary.length }; - offset += value.length; + offset += binary.length; } else { meta[key] = { type: 'json', @@ -68,26 +80,58 @@ export class PipelineContextInFile extends PipelineContext { const metaBuffer = Buffer.from(metaStr); const metaLength = metaBuffer.length; + const totalSize = 4 + metaLength + offset; + const fileBuffer = Buffer.alloc(totalSize); + fileBuffer.writeUInt32BE(metaLength, 0); + metaBuffer.copy(fileBuffer, 4); + let currentOffset = 4 + metaLength; + for (const chunk of binaryChunks) { + chunk.copy(fileBuffer, currentOffset); + currentOffset += chunk.length; + } + + return fileBuffer; + } + + async _writeContextAtomic() { + const fileBuffer = this._buildPayload(); + const tmpPath = `${this.filePath}.tmp`; + + logger.debug("PipelineContextInFile::saveContext saving to ", this.filePath); + + const fd = await open(tmpPath, 'w'); try { - const fd = await promisify(fs.open)(this.filePath, 'w'); - const lengthBuffer = Buffer.alloc(4); - lengthBuffer.writeUInt32BE(metaLength); - await promisify(fs.write)(fd, lengthBuffer, 0, 4, 0); - await promisify(fs.write)(fd, metaBuffer, 0, metaLength, 4); - let currentOffset = 4 + metaLength; - for (const chunk of binaryChunks) { - await promisify(fs.write)(fd, chunk, 0, chunk.length, currentOffset); - currentOffset += chunk.length; + await promisify(fs.write)(fd, fileBuffer, 0, fileBuffer.length, 0); + await fsync(fd); + } finally { + await close(fd); + } + + await rename(tmpPath, this.filePath); + } + + async _flushAll() { + while (this._saveDirty) { + this._saveDirty = false; + try { + await this._writeContextAtomic(); + } catch (error) { + logger.error('PipelineContextInFile::saveContext error:', error.message); + throw error; } - await promisify(fs.close)(fd); - } catch (error) { - logger.error('PipelineContextInFile::saveContext error:', error.message); - throw error; } } + saveContext() { + this._saveDirty = true; + this._saveTail = this._saveTail.then(() => this._flushAll()); + return this._saveTail; + } + async loadContext() { try { + await this._saveTail; + if (!fs.existsSync(this.filePath)) { this.context = {}; this.runtime.rawCommitted = 0; @@ -96,7 +140,7 @@ export class PipelineContextInFile extends PipelineContext { return; } - const fd = await promisify(fs.open)(this.filePath, 'r'); + const fd = await open(this.filePath, 'r'); const metaLengthBuffer = Buffer.alloc(4); await promisify(fs.read)(fd, metaLengthBuffer, 0, 4, 0); const metaLength = metaLengthBuffer.readUInt32BE(); @@ -118,7 +162,7 @@ export class PipelineContextInFile extends PipelineContext { } } - await promisify(fs.close)(fd); + await close(fd); const readStart = this.context.readStart || 0; const writeStart = this.context.writeStart || 0; diff --git a/src/node/Recoverable.js b/src/node/Recoverable.js index 9eea928..7ad60df 100644 --- a/src/node/Recoverable.js +++ b/src/node/Recoverable.js @@ -219,7 +219,7 @@ export class RecoverableWriteStream extends Writable { const blocks = runtime.pendingBlocks || []; let hasCommittedBlock = false; - let committedRawBytes = 0; + let committedItems = 0; logger.debug("On plaintext written:", writtenBytes, " bytes. Current runtime:", runtime); while(remain > 0 && blocks.length > 0){ @@ -233,9 +233,9 @@ export class RecoverableWriteStream extends Writable { // Block fully committed runtime.rawCommitted += block.rawSize; runtime.plainCommitted += block.plainSize; - committedRawBytes += block.rawSize; blocks.shift(); hasCommittedBlock = true; + committedItems += 1; } } logger.debug("After committing, remaining to commit:", remain, " bytes. Updated runtime:", runtime); @@ -244,31 +244,33 @@ export class RecoverableWriteStream extends Writable { return Promise.resolve(); } if(this.context.context){ - const buf = this.context.context['data']; - if(Buffer.isBuffer(buf) && buf.length > 0 && - committedRawBytes > 0){ - if(committedRawBytes >= buf.length){ - // All data committed - this.context.context['data'] = Buffer.alloc(0); - }else{ - // Partial data committed - this.context.context['data'] = buf.subarray(committedRawBytes); - } - } this.context.context['readStart'] = runtime.rawCommitted; this.context.context['writeStart'] = runtime.plainCommitted; + if (committedItems > 0) { + this.context.context['readItemCount'] = + (this.context.context['readItemCount'] || 0) + committedItems; + } + // Checkpoint only fully committed progress; discard in-flight cipher tail. + this.context.context['data'] = Buffer.alloc(0); logger.debug("After writing, updated readStart to:", this.context.context['readStart'], " writeStart to:", this.context.context['writeStart'], - " data length to:", this.context.context['data'] ? this.context.context['data'].length : 0); - // - this.context.saveContext(); + " readItemCount to:", this.context.context['readItemCount']); + return this.context.saveContext(); } return Promise.resolve(); } _final(callback) { this.writeStream.on('finish', () => { - const writeStart = this.context.context['writeStart'] || 0; + const ctx = this.context && this.context.context; + const runtime = this.context && this.context.runtime; + const writeStart = (ctx && ctx['writeStart']) || (runtime && runtime.plainCommitted) || 0; + + if (ctx && runtime) { + ctx['readStart'] = runtime.rawCommitted; + ctx['writeStart'] = runtime.plainCommitted; + ctx['data'] = Buffer.alloc(0); + } // Always truncate to writeStart to remove any residual // data from previous incomplete write attempts. @@ -276,8 +278,14 @@ export class RecoverableWriteStream extends Writable { if (truncateErr) { logger.warn("Error truncating file:", truncateErr); callback(truncateErr); + return; + } + logger.debug("File truncated successfully to length:", writeStart); + if (this.context && typeof this.context.saveContext === 'function') { + Promise.resolve(this.context.saveContext()) + .then(() => callback()) + .catch((err) => callback(err)); } else { - logger.debug("File truncated successfully to length:", writeStart); callback(); } }); diff --git a/src/node/Unsealer.js b/src/node/Unsealer.js index b26b216..f31d503 100644 --- a/src/node/Unsealer.js +++ b/src/node/Unsealer.js @@ -18,6 +18,7 @@ export class Unsealer extends Transform { const keyPair = options.keyPair; const progressHandler = options.progressHandler; const context = options ? options.context : undefined; + const ctx = context && context.context ? context.context : {}; // Node-specific rolling keccak256 hash let dataHash = Buffer.from(keccak256(Buffer.from("Fidelius", "utf-8"))); @@ -55,9 +56,9 @@ export class Unsealer extends Transform { } }, initialState: { - readItemCount: options ? (options.processedItemCount || 0) : 0, - processedBytes: options ? (options.processedBytes || 0) : 0, - writeBytes: options ? (options.writeBytes || 0) : 0, + readItemCount: options?.processedItemCount ?? ctx.readItemCount ?? 0, + processedBytes: options?.processedBytes ?? ctx.readStart ?? 0, + writeBytes: options?.writeBytes ?? ctx.writeStart ?? 0, } }); @@ -81,7 +82,10 @@ export class Unsealer extends Transform { // persist trailing unconsumed bytes for recoverable stream context if (this._context && this._context.context && this._context.context["status"] === "file") { - this._context.context["data"] = this.#core.remaining; + const remaining = this.#core.remaining; + this._context.context["data"] = remaining.length > 0 + ? Buffer.from(remaining.buffer, remaining.byteOffset, remaining.byteLength) + : Buffer.alloc(0); } callback();