feat(dkim-sign): Added new Transfor stream class DkimSignStream to si…

…gn emails in a stream processing pipeline

README.md

@@ -157,6 +157,52 @@ DKIM-Signature: a=rsa-sha256; v=1; c=relaxed/relaxed; d=tahvel.info;
 From: ...
 ```
 
+### Signing as a PassThrough Stream
+
+Use `DkimSignStream` stream if you want to use DKIM signing as part of a stream processing pipeline.
+
+```js
+const { DkimSignStream } = require('mailauth/lib/dkim/sign');
+
+const dkimSignStream = new DkimSignStream({
+    // Optional, default canonicalization, default is "relaxed/relaxed"
+    canonicalization: 'relaxed/relaxed', // c=
+
+    // Optional, default signing and hashing algorithm
+    // Mostly useful when you want to use rsa-sha1, otherwise no need to set
+    algorithm: 'rsa-sha256',
+
+    // Optional, default is current time
+    signTime: new Date(), // t=
+
+    // Keys for one or more signatures
+    // Different signatures can use different algorithms (mostly useful when
+    // you want to sign a message both with RSA and Ed25519)
+    signatureData: [
+        {
+            signingDomain: 'tahvel.info', // d=
+            selector: 'test.rsa', // s=
+            // supported key types: RSA, Ed25519
+            privateKey: fs.readFileSync('./test/fixtures/private-rsa.pem'),
+
+            // Optional algorithm, default is derived from the key.
+            // Overrides whatever was set in parent object
+            algorithm: 'rsa-sha256',
+
+            // Optional signature specifc canonicalization, overrides whatever was set in parent object
+            canonicalization: 'relaxed/relaxed' // c=
+
+            // Maximum number of canonicalized body bytes to sign (eg. the "l=" tag).
+            // Do not use though. This is available only for compatibility testing.
+            // maxBodyLength: 12345
+        }
+    ]
+});
+
+// Writes a signed message to the output
+process.stdin.pipe(dkimSignStream).pipe(process.stdout);
+```
+
 ### Verifying
 
 ```js

examples/dkim-sign-stream.js

@@ -0,0 +1,73 @@
+'use strict';
+
+// sign and verify:
+//   $ node sign-and-verify.js /path/to/message.eml
+
+const fs = require('node:fs');
+const { Buffer } = require('node:buffer');
+const { DkimSignStream } = require('../lib/dkim/sign');
+
+let file = process.argv[2];
+let eml = fs.createReadStream(file);
+
+let algo = process.argv[3] || false; // allowed: 'rsa-sha256', 'rsa-sha1', 'ed25519-sha256'
+
+let signer = new DkimSignStream({
+    canonicalization: 'relaxed/relaxed',
+    signTime: Date.now(),
+    //expires: Date.now() + 1000,
+    signatureData: [
+        {
+            algorithm: algo,
+            signingDomain: 'tahvel.info',
+            selector: 'test.invalid',
+            privateKey: fs.readFileSync('./test/fixtures/private-rsa.pem')
+        },
+
+        {
+            algorithm: algo,
+            signingDomain: 'tahvel.info',
+            selector: 'test.rsa',
+            privateKey: fs.readFileSync('./test/fixtures/private-rsa.pem')
+        },
+
+        {
+            algorithm: algo,
+            signingDomain: 'tahvel.info',
+            selector: 'test.small',
+            privateKey: fs.readFileSync('./test/fixtures/private-small.pem')
+        },
+
+        {
+            algorithm: algo,
+            signingDomain: 'tahvel.info',
+            selector: 'test.small',
+            privateKey: fs.readFileSync('./test/fixtures/private-small.pem'),
+            maxBodyLength: 12
+        },
+
+        {
+            // PEM
+            //canonicalization: 'relaxed/relaxed',
+            algorithm: algo,
+            signingDomain: 'tahvel.info',
+            selector: 'test.ed25519',
+            privateKey: fs.readFileSync('./test/fixtures/private-ed25519.pem')
+        },
+
+        {
+            // Raw key
+            //canonicalization: 'relaxed/relaxed',
+            algorithm: algo,
+            signingDomain: 'tahvel.info',
+            selector: 'test.ed25519',
+            privateKey: Buffer.from('YgsMTASxKi7M+Rxg+h9H4UTUNOGsAer6LaQgCwcl3mY=', 'base64')
+        }
+    ]
+});
+
+eml.on('error', err => {
+    console.error(err);
+});
+
+eml.pipe(signer).pipe(process.stdout);

lib/dkim/sign.js

@@ -2,6 +2,7 @@
 
 const { writeToStream } = require('../../lib/tools');
 const { DkimSigner } = require('./dkim-signer');
+const { Transform } = require('node:stream');
 
 const dkimSign = async (input, options) => {
     let dkimSigner = new DkimSigner(options);
@@ -10,4 +11,67 @@ const dkimSign = async (input, options) => {
     return { signatures: dkimSigner.signatureHeaders.join('\r\n') + '\r\n', arc: dkimSigner.arc, errors: dkimSigner.errors };
 };
 
-module.exports = { dkimSign };
+class DkimSignStream extends Transform {
+    constructor(options) {
+        super(options);
+        this.signer = new DkimSigner(options);
+
+        this.chunks = [];
+        this.chunklen = 0;
+
+        this.errors = null;
+
+        this.finished = false;
+        this.finishCb = null;
+        this.signer.on('end', () => this.finishStream());
+        this.signer.on('finish', () => this.finishStream());
+        this.signer.on('error', err => {
+            this.finished = true;
+            this.destroy(err);
+        });
+    }
+
+    finishStream() {
+        if (this.finished || !this.finishCb) {
+            return;
+        }
+        this.finished = true;
+        let done = this.finishCb;
+        this.finishCb = null;
+
+        this.errors = this.signer.errors;
+
+        this.push(Buffer.from(this.signer.signatureHeaders.join('\r\n') + '\r\n'));
+        this.push(Buffer.concat(this.chunks, this.chunklen));
+        done();
+    }
+
+    _transform(chunk, encoding, done) {
+        if (!chunk || !chunk.length || this.finished) {
+            return done();
+        }
+
+        if (typeof chunk === 'string') {
+            chunk = Buffer.from(chunk, encoding);
+        }
+
+        this.chunks.push(chunk);
+        this.chunklen += chunk.length;
+
+        if (this.signer.write(chunk) === false) {
+            // wait for drain
+            return this.signer.once('drain', done);
+        }
+        done();
+    }
+
+    _flush(done) {
+        if (this.finished) {
+            return done();
+        }
+        this.finishCb = done;
+        this.signer.end();
+    }
+}
+
+module.exports = { dkimSign, DkimSignStream };

lib/gatherer-stream.js

@@ -41,9 +41,9 @@ class GathererStream extends Transform {
         return stream;
     }
 
-    _transform(chunk, encodng, done) {
+    _transform(chunk, encoding, done) {
         if (typeof chunk === 'string') {
-            chunk = Buffer.from(chunk, encodng);
+            chunk = Buffer.from(chunk, encoding);
         }
 
         if (!chunk || !chunk.length) {