feat(backups): validate VHD streams (#6770)

@vates/stream-reader/.USAGE.md

@@ -0,0 +1,42 @@
+```js
+import StreamReader from '@vates/stream-reader'
+
+const reader = new StreamReader(stream)
+```
+
+### `.read([size])`
+
+- returns the next available chunk of data
+- like `stream.read()`, a number of bytes can be specified
+- returns with less data than expected if stream has ended
+- returns `null` if the stream has ended and no data has been read
+
+```js
+const chunk = await reader.read(512)
+```
+
+### `.readStrict([size])`
+
+Similar behavior to `readChunk` but throws if the stream ended before the requested data could be read.
+
+```js
+const chunk = await reader.readStrict(512)
+```
+
+### `.skip(size)`
+
+Skips a given number of bytes from a stream.
+
+Returns the number of bytes actually skipped, which may be less than the requested size if the stream has ended.
+
+```js
+const bytesSkipped = await reader.skip(2 * 1024 * 1024 * 1024)
+```
+
+### `.skipStrict(size)`
+
+Skips a given number of bytes from a stream and throws if the stream ended before enough stream has been skipped.
+
+```js
+await reader.skipStrict(2 * 1024 * 1024 * 1024)
+```

@vates/stream-reader/.npmignore

@@ -0,0 +1 @@
+../../scripts/npmignore

@vates/stream-reader/README.md

@@ -0,0 +1,75 @@
+<!-- DO NOT EDIT MANUALLY, THIS FILE HAS BEEN GENERATED -->
+
+# @vates/stream-reader
+
+[![Package Version](https://badgen.net/npm/v/@vates/stream-reader)](https://npmjs.org/package/@vates/stream-reader) ![License](https://badgen.net/npm/license/@vates/stream-reader) [![PackagePhobia](https://badgen.net/bundlephobia/minzip/@vates/stream-reader)](https://bundlephobia.com/result?p=@vates/stream-reader) [![Node compatibility](https://badgen.net/npm/node/@vates/stream-reader)](https://npmjs.org/package/@vates/stream-reader)
+
+> Efficiently reads and skips chunks of a given size in a stream
+
+## Install
+
+Installation of the [npm package](https://npmjs.org/package/@vates/stream-reader):
+
+```sh
+npm install --save @vates/stream-reader
+```
+
+## Usage
+
+```js
+import StreamReader from '@vates/stream-reader'
+
+const reader = new StreamReader(stream)
+```
+
+### `.read([size])`
+
+- returns the next available chunk of data
+- like `stream.read()`, a number of bytes can be specified
+- returns with less data than expected if stream has ended
+- returns `null` if the stream has ended and no data has been read
+
+```js
+const chunk = await reader.read(512)
+```
+
+### `.readStrict([size])`
+
+Similar behavior to `readChunk` but throws if the stream ended before the requested data could be read.
+
+```js
+const chunk = await reader.readStrict(512)
+```
+
+### `.skip(size)`
+
+Skips a given number of bytes from a stream.
+
+Returns the number of bytes actually skipped, which may be less than the requested size if the stream has ended.
+
+```js
+const bytesSkipped = await reader.skip(2 * 1024 * 1024 * 1024)
+```
+
+### `.skipStrict(size)`
+
+Skips a given number of bytes from a stream and throws if the stream ended before enough stream has been skipped.
+
+```js
+await reader.skipStrict(2 * 1024 * 1024 * 1024)
+```
+
+## Contributions
+
+Contributions are _very_ welcomed, either on the documentation or on
+the code.
+
+You may:
+
+- report any [issue](https://github.com/vatesfr/xen-orchestra/issues)
+  you've encountered;
+- fork and create a pull request.
+
+## License
+
+[ISC](https://spdx.org/licenses/ISC) © [Vates SAS](https://vates.fr)

@vates/stream-reader/index.js

@@ -0,0 +1,123 @@
+'use strict'
+
+const assert = require('node:assert')
+const { finished, Readable } = require('node:stream')
+
+const noop = Function.prototype
+
+// Inspired by https://github.com/nodejs/node/blob/85705a47958c9ae5dbaa1f57456db19bdefdc494/lib/internal/streams/readable.js#L1107
+class StreamReader {
+  #ended = false
+  #error
+  #executor = resolve => {
+    this.#resolve = resolve
+  }
+  #stream
+  #resolve = noop
+
+  constructor(stream) {
+    stream = typeof stream.pipe === 'function' ? stream : Readable.from(stream)
+
+    this.#stream = stream
+
+    stream.on('readable', () => this.#resolve())
+
+    finished(stream, { writable: false }, error => {
+      this.#error = error
+      this.#ended = true
+      this.#resolve()
+    })
+  }
+
+  async read(size) {
+    if (size !== undefined) {
+      assert(size > 0)
+    }
+
+    do {
+      if (this.#ended) {
+        if (this.#error) {
+          throw this.#error
+        }
+        return null
+      }
+
+      const value = this.#stream.read(size)
+      if (value !== null) {
+        return value
+      }
+
+      await new Promise(this.#executor)
+    } while (true)
+  }
+
+  async readStrict(size) {
+    const chunk = await this.read(size)
+    if (chunk === null) {
+      throw new Error('stream has ended without data')
+    }
+
+    if (size !== undefined && chunk.length !== size) {
+      const error = new Error(`stream has ended with not enough data (actual: ${chunk.length}, expected: ${size})`)
+      Object.defineProperties(error, {
+        chunk: {
+          value: chunk,
+        },
+      })
+      throw error
+    }
+
+    return chunk
+  }
+
+  async skip(size) {
+    if (size === 0) {
+      return size
+    }
+
+    let toSkip = size
+    do {
+      if (this.#ended) {
+        if (this.#error) {
+          throw this.#error
+        }
+        return size - toSkip
+      }
+
+      const data = this.#stream.read()
+      if (data !== null) {
+        toSkip -= data === null ? 0 : data.length
+        if (toSkip > 0) {
+          // continue to read
+        } else {
+          // if more than wanted has been read, push back the rest
+          if (toSkip < 0) {
+            this.#stream.unshift(data.slice(toSkip))
+          }
+
+          return size
+        }
+      }
+
+      await new Promise(this.#executor)
+    } while (true)
+  }
+
+  async skipStrict(size) {
+    const bytesSkipped = await this.skip(size)
+    if (bytesSkipped !== size) {
+      const error = new Error(`stream has ended with not enough data (actual: ${bytesSkipped}, expected: ${size})`)
+      error.bytesSkipped = bytesSkipped
+      throw error
+    }
+  }
+}
+
+StreamReader.prototype[Symbol.asyncIterator] = async function* asyncIterator() {
+  let chunk
+  while ((chunk = await this.read()) !== null) {
+    yield chunk
+  }
+}
+
+module.exports = StreamReader

@vates/stream-reader/index.test.js

@@ -0,0 +1,141 @@
+'use strict'
+
+const { describe, it } = require('test')
+const assert = require('node:assert').strict
+
+const { Readable } = require('stream')
+
+const StreamReader = require('./index.js')
+
+const makeStream = it => Readable.from(it, { objectMode: false })
+makeStream.obj = Readable.from
+
+const rejectionOf = promise =>
+  promise.then(
+    value => {
+      throw value
+    },
+    error => error
+  )
+
+const makeErrorTests = method => {
+  it('rejects if the stream errors', async () => {
+    const error = new Error()
+    const stream = makeStream([])
+
+    const pError = rejectionOf(new StreamReader(stream)[method](10))
+    stream.destroy(error)
+
+    assert.strict(await pError, error)
+  })
+
+  it('rejects if the stream has already errored', async () => {
+    const error = new Error()
+    const stream = makeStream([])
+
+    await new Promise(resolve => {
+      stream.once('error', resolve).destroy(error)
+    })
+
+    assert.strict(await rejectionOf(new StreamReader(stream)[method](10)), error)
+  })
+}
+
+describe('read()', () => {
+  it('rejects if size is less than or equal to 0', async () => {
+    const error = await rejectionOf(new StreamReader(makeStream([])).read(0))
+    assert.strictEqual(error.code, 'ERR_ASSERTION')
+  })
+
+  it('returns null if stream is empty', async () => {
+    assert.strictEqual(await new StreamReader(makeStream([])).read(), null)
+  })
+
+  makeErrorTests('read')
+
+  it('returns null if the stream is already ended', async () => {
+    const reader = new StreamReader(makeStream([]))
+
+    await reader.read()
+
+    assert.strictEqual(await reader.read(), null)
+  })
+
+  describe('with binary stream', () => {
+    it('returns the first chunk of data', async () => {
+      assert.deepEqual(await new StreamReader(makeStream(['foo', 'bar'])).read(), Buffer.from('foo'))
+    })
+
+    it('returns a chunk of the specified size (smaller than first)', async () => {
+      assert.deepEqual(await new StreamReader(makeStream(['foo', 'bar'])).read(2), Buffer.from('fo'))
+    })
+
+    it('returns a chunk of the specified size (larger than first)', async () => {
+      assert.deepEqual(await new StreamReader(makeStream(['foo', 'bar'])).read(4), Buffer.from('foob'))
+    })
+
+    it('returns less data if stream ends', async () => {
+      assert.deepEqual(await new StreamReader(makeStream(['foo', 'bar'])).read(10), Buffer.from('foobar'))
+    })
+  })
+
+  describe('with object stream', () => {
+    it('returns the first chunk of data verbatim', async () => {
+      const chunks = [{}, {}]
+      assert.strictEqual(await new StreamReader(makeStream.obj(chunks)).read(), chunks[0])
+    })
+  })
+})
+
+describe('readStrict()', function () {
+  it('throws if stream is empty', async () => {
+    const error = await rejectionOf(new StreamReader(makeStream([])).readStrict())
+    assert(error instanceof Error)
+    assert.strictEqual(error.message, 'stream has ended without data')
+    assert.strictEqual(error.chunk, undefined)
+  })
+
+  it('throws if stream ends with not enough data', async () => {
+    const error = await rejectionOf(new StreamReader(makeStream(['foo', 'bar'])).readStrict(10))
+    assert(error instanceof Error)
+    assert.strictEqual(error.message, 'stream has ended with not enough data (actual: 6, expected: 10)')
+    assert.deepEqual(error.chunk, Buffer.from('foobar'))
+  })
+})
+
+describe('skip()', function () {
+  makeErrorTests('skip')
+
+  it('returns 0 if size is 0', async () => {
+    assert.strictEqual(await new StreamReader(makeStream(['foo'])).skip(0), 0)
+  })
+
+  it('returns 0 if the stream is already ended', async () => {
+    const reader = new StreamReader(makeStream([]))
+
+    await reader.read()
+
+    assert.strictEqual(await reader.skip(10), 0)
+  })
+
+  it('skips a number of bytes', async () => {
+    const reader = new StreamReader(makeStream('foo bar'))
+
+    assert.strictEqual(await reader.skip(4), 4)
+    assert.deepEqual(await reader.read(4), Buffer.from('bar'))
+  })
+
+  it('returns less size if stream ends', async () => {
+    assert.deepEqual(await new StreamReader(makeStream('foo bar')).skip(10), 7)
+  })
+})
+
+describe('skipStrict()', function () {
+  it('throws if stream ends with not enough data', async () => {
+    const error = await rejectionOf(new StreamReader(makeStream('foo bar')).skipStrict(10))
+
+    assert(error instanceof Error)
+    assert.strictEqual(error.message, 'stream has ended with not enough data (actual: 7, expected: 10)')
+    assert.deepEqual(error.bytesSkipped, 7)
+  })
+})

@vates/stream-reader/package.json

@@ -0,0 +1,39 @@
+{
+  "private": false,
+  "name": "@vates/stream-reader",
+  "description": "Efficiently reads and skips chunks of a given size in a stream",
+  "keywords": [
+    "async",
+    "chunk",
+    "data",
+    "node",
+    "promise",
+    "read",
+    "reader",
+    "skip",
+    "stream"
+  ],
+  "homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@vates/stream-reader",
+  "bugs": "https://github.com/vatesfr/xen-orchestra/issues",
+  "repository": {
+    "directory": "@vates/stream-reader",
+    "type": "git",
+    "url": "https://github.com/vatesfr/xen-orchestra.git"
+  },
+  "author": {
+    "name": "Vates SAS",
+    "url": "https://vates.fr"
+  },
+  "license": "ISC",
+  "version": "0.0.0",
+  "engines": {
+    "node": ">=10"
+  },
+  "scripts": {
+    "postversion": "npm publish --access public",
+    "test": "node--test"
+  },
+  "devDependencies": {
+    "test": "^3.3.0"
+  }
+}