RXJS 加密大文件并发送到 AWS S3答案

【问题标题】：RXJS Encrypting large files and sending to AWS S3RXJS 加密大文件并发送到 AWS S3
【发布时间】：2021-04-30 13:27:48
【问题描述】：

我对 RXJS 比较陌生，它的学习曲线相当陡峭。我正在通过 GRPC 流式传输文件，我想用 AES 加密它们，然后将它们存储在 S3 存储桶中。目前我正在将缓冲区存储在内存中，如果我开始上传大文件，这将很快成为一个问题。

我想知道如何使用 RXJS 流，通过加密将其通过管道传输到 S3 存储桶（即我不想将整个文件存储在内存中）

这是可能的还是我误解了什么？

这是我目前的实现

  @GrpcStreamMethod()
  upload(data$: Observable<FileUploadRequest>): Observable<FileUploadResponse> {
    let fileBuffer: Buffer;
    let metaBuffer: Buffer;
    let storageBucket: string;
    let publicKey: string;
    let publicKeyHash: string;

    const response$ = new Subject<FileUploadResponse>();
    let headersDeliverd = false;

    const symetricKeys = this.service.generateEncryptionKeys(); //returns key + iv
    const fileCipher = this.service.getCypher(symetricKeys); //returns cipher
    const metaCipher = this.service.getCypher(symetricKeys); //returns cipher

    data$.subscribe({
      next: (data: FileUploadRequest) => {
        const content = data?.file?.content;
        const meta: Meta = data?.metadata;

        // Metadata has been delivered
        if (meta && !headersDeliverd) {
          headersDeliverd = true;

          publicKey = meta.owner;
          publicKeyHash = this.service.createHash(meta.owner);
          storageBucket = join(FILE_STORE, publicKeyHash);
          const metaBuff = this.service.metaToBuffer(meta);
          metaBuffer = metaCipher.update(metaBuff);
        }

        if (content && !headersDeliverd) {
          console.log('SOME ERROR');
        }

        if (content && headersDeliverd) {
          fileBuffer = fileCipher.update(content);
        }
      },
      complete: async () => {
        const encryptedSymetricKey = await this.service.encryptSymetricKey(
          symetricKeys,
          publicKey,
        );

        fileBuffer = Buffer.concat([fileBuffer, fileCipher.final()]);
        metaBuffer = Buffer.concat([metaBuffer, metaCipher.final()]);

        const documentId = uuidv4();
        const fileName = `${documentId}.file`;
        const metaName = `${documentId}.meta`;
        const keys = `${documentId}.access`;

        await this.service.uploadToAws(
          metaBuffer,
          `${publicKeyHash}/${metaName}`,
        );
        await this.service.uploadToAws(
          fileBuffer,
          `${publicKeyHash}/${fileName}`,
        );
        await this.service.uploadToAws(
          JSON.stringify({
            [publicKeyHash]: encryptedSymetricKey,
          }),
          `${publicKeyHash}/${keys}`,
        );

        response$.next({
          status: Status.SUCCESS,
          path: storageBucket,
        });
        return response$.complete();
      },
    });

    return response$.asObservable();
  }

【问题讨论】：

您不能使用 Node.js 流的任何特殊原因，它可以很好地与加密货币配合使用？例如，crypto.createCipheriv() 的返回值是 Stream.Transform。您基本上可以将文件直接通过管道输入、通过和输出到 S3。 RxJS 在这里听起来像是一个障碍。
你的正确是一个障碍。我在这里使用 RXJS 的唯一真正原因是因为我使用的框架使用 RXJS 来处理 GRPC 流docs.nestjs.com/microservices/grpc#grpc-streaming 你会建议我弄清楚如何摆脱使用 RXJS 吗？
我想出了一个解决方案，当我完成后我会在这里发布。基本上我只是创建一个读取流并将块推送到流中，然后我可以通过加密将其通过管道传输，然后传输到 aws。就这样实现了next: (data: FileUploadRequest) => readableStream.push('data')

标签： node.js amazon-s3 rxjs aes

【解决方案1】：

我想出的解决方案是创建一个流并将数据也推送给它。

uploadToAws(key) {
    const pass = new PassThrough();

    return {
      writeStream: pass,
      promise: this.s3.send(
        new PutObjectCommand({
          Bucket: '...',
          Key: key,
          Body: pass,
          ServerSideEncryption: '...',
          ContentLength: 37, //If you do not put content lendth will error
        }),
      ),
    };
}

const stream = new Readable({
  read(data) {
    return data;
  },
});

data$.subscribe({
    next: (data) => stream.push(data),
    error: (error) => stream.destroy(),
    complete: async () => stream.destroy()
});

stream.pipe(encryptFunc).pipe(uploadToAws);

【讨论】：