adds incremental verification to CAR files.

.gitignore

@@ -1 +1 @@
-node_modules/
+node_modules/

.npmignore

@@ -1,4 +1,4 @@
 node_modules/
 test/
 .*/
-*.config.*js
+*.config.*js

package.json

@@ -26,6 +26,8 @@
     "@ipld/dag-json": "^8.0.11",
     "@ipld/dag-pb": "^2.1.18",
     "@multiformats/blake2": "^1.0.11",
+    "browser-readablestream-to-it": "^2.0.4",
+    "ipfs-unixfs-exporter": "^13.1.7",
     "multiformats": "^9.9.0"
   },
   "devDependencies": {
@@ -42,5 +44,8 @@
       "ecmaVersion": "latest",
       "sourceType": "module"
     }
+  },
+  "imports": {
+    "#src/*": "./src/*"
   }
 }

src/utils/car-block-getter.js

@@ -0,0 +1,51 @@
+import { CarBlockIterator } from '@ipld/car/iterator'
+import toIterable from 'browser-readablestream-to-it'
+
+import { verifyBlock } from './car.js'
+import { promiseTimeout } from './timers.js'
+import { TimeoutError, VerificationError } from './errors.js'
+
+// Assumptions
+// * client and server are both using DFS traversal.
+// * Server is sending CARs with duplicate blocks.
+export class CarBlockGetter {
+  constructor (carItr, opts = {}) {
+    this.carItr = carItr
+    this.getBlockTimeout = opts.getBlockTimeout ?? 1_000 * 10
+  }
+
+  static async fromStream (carStream) {
+    const iterable = await CarBlockIterator.fromIterable(
+      asAsyncIterable(carStream)
+    )
+    const carItr = iterable[Symbol.asyncIterator]()
+    return new CarBlockGetter(carItr)
+  }
+
+  async get (cid, options) {
+    const { value, done } = await promiseTimeout(
+      this.carItr.next(),
+      this.getBlockTimeout,
+      new TimeoutError(`get block ${cid} timed out`)
+    )
+
+    if (!value && done) {
+      throw new VerificationError('CAR file has no more blocks.')
+    }
+
+    const { cid: blockCid, bytes } = value
+    await verifyBlock(blockCid, bytes)
+
+    if (!cid.equals(blockCid)) {
+      throw new VerificationError(
+        `received block with cid ${blockCid}, expected ${cid}`
+      )
+    }
+
+    return bytes
+  }
+}
+
+function asAsyncIterable (readable) {
+  return Symbol.asyncIterator in readable ? readable : toIterable(readable)
+}

src/utils/car.js

@@ -8,6 +8,10 @@ import * as json from 'multiformats/codecs/json'
 import { sha256 } from 'multiformats/hashes/sha2'
 import { from as hasher } from 'multiformats/hashes/hasher'
 import { blake2b256 } from '@multiformats/blake2/blake2b'
+import { recursive } from 'ipfs-unixfs-exporter'
+
+import { CarBlockGetter } from './car-block-getter.js'
+import { VerificationError } from './errors.js'
 
 const { toHex } = bytes
 
@@ -33,20 +37,49 @@ const hashes = {
 export async function validateBody (body) {
   const carBlockIterator = await CarBlockIterator.fromIterable(body)
   for await (const { cid, bytes } of carBlockIterator) {
-    if (!codecs[cid.code]) {
-      throw new Error(`Unexpected codec: 0x${cid.code.toString(16)}`)
-    }
-    if (!hashes[cid.multihash.code]) {
-      throw new Error(`Unexpected multihash code: 0x${cid.multihash.code.toString(16)}`)
-    }
-
-    // Verify step 2: if we hash the bytes, do we get the same digest as reported by the CID?
-    // Note that this step is sufficient if you just want to safely verify the CAR's reported CIDs
-    const hash = await hashes[cid.multihash.code].digest(bytes)
-    if (toHex(hash.digest) !== toHex(cid.multihash.digest)) {
-      throw new Error('Hash mismatch')
-    }
+    await verifyBlock(cid, bytes)
   }
 
   return true
 }
+
+/**
+ * Verifies a block
+ *
+ * @param {CID} cid
+ * @param {Uint8Array} bytes
+ */
+export async function verifyBlock (cid, bytes) {
+  // Verify step 1: is this a CID we know how to deal with?
+  if (!codecs[cid.code]) {
+    throw new VerificationError(`Unexpected codec: 0x${cid.code.toString(16)}`)
+  }
+  if (!hashes[cid.multihash.code]) {
+    throw new VerificationError(`Unexpected multihash code: 0x${cid.multihash.code.toString(16)}`)
+  }
+
+  // Verify step 2: if we hash the bytes, do we get the same digest as
+  // reported by the CID? Note that this step is sufficient if you just
+  // want to safely verify the CAR's reported CIDs
+  const hash = await hashes[cid.multihash.code].digest(bytes)
+  if (toHex(hash.digest) !== toHex(cid.multihash.digest)) {
+    throw new VerificationError(
+          `Mismatch: digest of bytes (${toHex(hash)}) does not match digest in CID (${toHex(cid.multihash.digest)})`)
+  }
+}
+
+/**
+ * Verifies and extracts the raw content from a CAR stream.
+ *
+ * @param {string} cidPath
+ * @param {ReadableStream|AsyncIterable} carStream
+ */
+export async function * extractVerifiedContent (cidPath, carStream) {
+  const getter = await CarBlockGetter.fromStream(carStream)
+
+  for await (const child of recursive(cidPath, getter)) {
+    for await (const chunk of child.content()) {
+      yield chunk
+    }
+  }
+}

src/utils/errors.js

@@ -0,0 +1,13 @@
+export class VerificationError extends Error {
+  constructor (message) {
+    super(message)
+    this.name = 'VerificationError'
+  }
+}
+
+export class TimeoutError extends Error {
+  constructor (message) {
+    super(message)
+    this.name = 'TimeoutError'
+  }
+}

src/utils/timers.js

@@ -3,3 +3,15 @@ export const setTimeoutPromise = async ms => {
     setTimeout(resolve, ms)
   })
 }
+
+export function promiseTimeout (promise, ms, timeoutErr) {
+  let id
+  const timeout = new Promise((resolve, reject) => {
+    id = setTimeout(() => {
+      const err = new Error('Promise timed out')
+      reject(timeoutErr || err)
+    }, ms)
+  })
+
+  return Promise.race([promise, timeout]).finally(() => clearTimeout(id))
+}

test/car.js

@@ -0,0 +1,119 @@
+import assert from 'node:assert/strict'
+import { createHash } from 'node:crypto'
+import fs from 'node:fs'
+import { describe, it } from 'node:test'
+
+import { CarReader, CarWriter } from '@ipld/car'
+import { CID } from 'multiformats/cid'
+
+import { extractVerifiedContent } from '#src/utils/car.js'
+
+async function getFileHash (itr) {
+  const hasher = createHash('sha256')
+  for await (const chunk of itr) {
+    hasher.update(chunk)
+  }
+  return hasher.digest('hex')
+}
+
+describe('CAR Verification', () => {
+  it('should extract content from a valid CAR', async () => {
+    const cidPath =
+      'bafybeiaxzcil63nfqz5z3c34iajzgve5buqxrc7xcoslxlq2ndg5kfvc2u'
+    const filepath =
+      './fixtures/bafybeiaxzcil63nfqz5z3c34iajzgve5buqxrc7xcoslxlq2ndg5kfvc2u.car'
+    const carStream = fs.createReadStream(filepath)
+
+    const contentItr = await extractVerifiedContent(cidPath, carStream)
+    const actualHash = await getFileHash(contentItr)
+    const expectedHash =
+      '983408ac1d97f5913b6a3abee741d3c0032bc465ae0802023d61ce38136e292b'
+
+    assert.strictEqual(actualHash, expectedHash)
+  })
+
+  it('should verify file paths', async () => {
+    const cidPath =
+      'bafybeihin6eiifex5e76ama6do2dhmt2da3g5dsrazm5eskwm6raczbjtu/2160.png'
+    const filepath =
+      './fixtures/bafybeihin6eiifex5e76ama6do2dhmt2da3g5dsrazm5eskwm6raczbjtu_2160_png.car'
+    const carStream = fs.createReadStream(filepath)
+
+    const contentItr = await extractVerifiedContent(cidPath, carStream)
+    const actualHash = await getFileHash(contentItr)
+    const expectedHash =
+      '64e4e4abef8095853c29cabcebb0240b986d4ff1e1498639f59ad1ef1ff2a3a9'
+
+    assert.strictEqual(actualHash, expectedHash)
+  })
+
+  it('should error if CAR is missing blocks', async () => {
+    const cidPath =
+      'bafybeiaxzcil63nfqz5z3c34iajzgve5buqxrc7xcoslxlq2ndg5kfvc2u'
+    const filepath =
+      './fixtures/bafybeiaxzcil63nfqz5z3c34iajzgve5buqxrc7xcoslxlq2ndg5kfvc2u.car'
+    const carStream = fs.createReadStream(filepath)
+
+    // Create an invalid CAR that only has 1 block instead of 4
+    const outCid = CID.parse(cidPath)
+    const { writer, out } = await CarWriter.create([outCid]);
+    (async () => {
+      // need wrapping IIFE to avoid node exiting early
+      const reader = await CarReader.fromIterable(carStream)
+      await writer.put(await reader.get(cidPath))
+      await writer.close()
+    })()
+
+    await assert.rejects(
+      async () => {
+        for await (const _ of extractVerifiedContent(cidPath, out)) {}
+      },
+      {
+        name: 'VerificationError',
+        message: 'CAR file has no more blocks.'
+      }
+    )
+  })
+
+  it('should error if CAR blocks are in the wrong traversal order', async () => {
+    const cidPath =
+      'bafybeiaxzcil63nfqz5z3c34iajzgve5buqxrc7xcoslxlq2ndg5kfvc2u'
+    const filepath =
+      './fixtures/bafybeiaxzcil63nfqz5z3c34iajzgve5buqxrc7xcoslxlq2ndg5kfvc2u.car'
+    const carStream = fs.createReadStream(filepath)
+
+    // Create an invalid CAR that has blocks in the wrong order
+    const outCid = CID.parse(cidPath)
+    const { writer, out } = await CarWriter.create([outCid]);
+    (async () => {
+      // need wrapping IIFE to avoid node exiting early
+      const reader = await CarReader.fromIterable(carStream)
+
+      const blocks = []
+      for await (const block of reader.blocks()) {
+        blocks.push(block)
+      }
+
+      const temp = blocks[0]
+      blocks[0] = blocks[1]
+      blocks[1] = temp
+
+      for (const block of blocks) {
+        await writer.put(block)
+      }
+      await writer.close()
+    })()
+
+    await assert.rejects(
+      async () => {
+        for await (const _ of extractVerifiedContent(cidPath, out)) {
+        }
+      },
+      {
+        name: 'VerificationError',
+        message:
+          'received block with cid bafkreih64b5ydhx6tz4hpfbhfhkjrlvrin4hvv5jz2bp62fmyrkzs633si, expected bafybeiaxzcil63nfqz5z3c34iajzgve5buqxrc7xcoslxlq2ndg5kfvc2u'
+      }
+    )
+  })
+})

test/test.js

@@ -2,7 +2,7 @@ import assert from 'node:assert/strict'
 import { randomUUID } from 'node:crypto'
 import { describe, it } from 'node:test'
 
-import Saturn from '../src/index.js'
+import Saturn from '#src/index.js'
 
 describe('Saturn client', () => {
   describe('constructor', () => {
@@ -53,4 +53,4 @@ describe('Saturn client', () => {
       assert.rejects(client.fetchCID('QmXjYBY478Cno4jzdCcPy4NcJYFrwHZ51xaCP8vUwN9MGm', { downloadTimeout: 0 }))
     })
   })
-})
+})