node.js - 使用 openpgp 在转换后加密流

Question

我正在使用以下管道从 Aurora 流式传输数据，将其转换为 csv，然后将其发送到 S3。

可读的 knex 流:

const getQueryStream = (organizationId) => 
  db.select('*')
    .from('users')
    .where('organization_id', organizationId)
    .stream();

转换数据:

const toCSVTransform = (fields) => new stream.Transform({
  objectMode: true,
  transform: (row, encoding, callback) => {
    let rowAsArr = [];
    for(let i = 0; i < fields.length; i++) {
      rowAsArr.push(row[fields[i]]);
    }
    callback(null, `${rowAsArr.join(',')}\n`);
  }
});

管道:

stream.pipeline(
    dbStream,
    toCSVTransform(['first_name', 'last_name', 'email']),
    s3WritableStream,
    (err) => {
        if (err) {
            console.error('Pipeline failed.', err)
        } else {
            console.log('Pipeline succeeded.')
        }
    }
)

这按原样工作，但我们被要求使用 PGP 加密来加密文件。我的想法是在 toCSVTransform 之后在管道中增加一个步骤来进行加密。 npm 包 openpgp 支持流，但我不确定如何将它应用到管道中。

从 openpgp 文档中，这里有一个示例，说明如何将可读流传递给 openpgp.encrypt 函数:

const readableStream = new ReadableStream({
    start(controller) {
        controller.enqueue('Hello, world!');
        controller.close();
    }
});

const encrypted = await openpgp.encrypt({
    message: await openpgp.createMessage({ text: readableStream }), // input as Message object
    encryptionKeys: publicKey,
    signingKeys: privateKey // optional
});

我见过的所有示例都只是将可读流传递给加密函数。但我需要在将数据发送到 s3 之前转换数据。

有没有办法将 toCSVTransform 流传递给 openpgp.encrypt 方法？

我似乎想将可读的 dbStream 和转换流 toCSVTransform 组合成一个流，并将其传递给 openpgp.encrypt 函数。

我注意到 node.js 有一个 stream.compose 方法，但它目前只是实验性的，所以它不是一个真正的选择。

**** 编辑:可能的解决方案看起来我可以在将流传递给 openpgp.encrypt 方法之前使用 pipe() 来转换流:

const encrypted = await openpgp.encrypt({
    message: await openpgp.createMessage({ text: dbStream.pipe(toCSVTransform) }), // input as Message object
    encryptionKeys: publicKey,
    signingKeys: privateKey // optional
});

Answer 1

您所拥有的大致正确，但encrypted 将是一个 Stream。

这会起作用:

const encryptedPrivateKey = await openpgp.readPrivateKey({armoredKey});
const signingKey = await openpgp.decryptKey({
  privateKey: encryptedPrivateKey,
  passphrase,
})

const encrypt = async (encryptionKeys, signingKeys, readableStream) => await openpgp.encrypt({
  message: await openpgp.createMessage({text: readableStream}),
  encryptionKeys,
  signingKeys,
});

stream.pipeline(
    await encrypt(encryptionKey, signingKey, stream.pipeline(
      dbStream,
      toCSVTransform(['first_name', 'last_name', 'email']),
    )),
    s3WritableStream,
    (err) => {
        if (err) {
            console.error('Pipeline failed.', err)
        } else {
            console.log('Pipeline succeeded.')
        }
    }
)

不幸的是，没有(简单的)方法来包装 openpgp 以便它可以直接插入到管道中。

如果您对对称加密没问题，那么更简洁的解决方案是使用 crypto:

const encrypter = crypto.createCipheriv(algo, key, iv)

stream.pipeline(
    dbStream,
    toCSVTransform(['first_name', 'last_name', 'email']),
    encrypter,
    s3WritableStream,
    (err) => {
        if (err) {
            console.error('Pipeline failed.', err)
        } else {
            console.log('Pipeline succeeded.')
        }
    }
)