wildduck/lib/maildropper.js

'use strict';

const SeqIndex = require('seq-index');
const RelaxedBody = require('nodemailer/lib/dkim/relaxed-body');
const MessageSplitter = require('./message-splitter');
const seqIndex = new SeqIndex();
const GridFSBucket = require('mongodb').GridFSBucket;
const uuid = require('uuid');
const os = require('os');
const hostname = os.hostname().toLowerCase();
const addressparser = require('nodemailer/lib/addressparser');
const punycode = require('punycode/');
const crypto = require('crypto');
const tools = require('./tools');
const plugins = require('./plugins');
const PassThrough = require('stream').PassThrough;
const util = require('util');

class Maildropper {
    constructor(options) {
        this.options = options || {};
        this.db = options.db;
        this.zone = options.zone;
        this.collection = options.collection;
        this.gfs = options.gfs;

        this.gridstore =
            options.gridstore ||
            new GridFSBucket(this.db.senderDb, {
                bucketName: this.gfs
            });
    }

    checkLoop(envelope, deliveries) {
        if (envelope.reason !== 'forward') {
            return false;
        }

        if (envelope.headers.get('Received').length >= 30) {
            envelope.looped = true;
            return true;
        }

        if (!this.options.loopSecret) {
            return false;
        }

        const loopKey = 'X-WildDuck-Seen';
        const algo = 'sha256';
        const secret = this.options.loopSecret;

        const targetStr = JSON.stringify(deliveries);

        const loopFields = envelope.headers.getDecoded(loopKey);

        // check existing loop headers (max 100 to avoid checking too many hashes)
        for (let i = 0, len = Math.min(loopFields.length, 100); i < len; i++) {
            let field = (loopFields[i].value || '').toLowerCase().trim();
            let salt = field.substr(0, 12);
            let hash = field.substr(12);
            let hmac = crypto.createHmac(algo, secret);
            hmac.update(salt);
            hmac.update(targetStr);
            let result = hmac.digest('hex');
            if (result.toLowerCase() === hash) {
                // Loop detected!
                envelope.looped = true;
                return true;
            }
        }

        const salt = crypto.randomBytes(6).toString('hex').toLowerCase();
        const loopHeader = salt + crypto.createHmac(algo, secret).update(salt).update(targetStr).digest('hex').toLocaleLowerCase();
        envelope.headers.add(loopKey, loopHeader);

        return false;
    }

    push(options, callback) {
        let id = options.id || seqIndex.get();
        let seq = 0;
        let documents = [];

        let envelope = {
            id,

            from: options.from || '',
            to: Array.isArray(options.to) ? options.to : [].concat(options.to || []),

            interface: options.interface || 'maildrop',
            transtype: 'API',
            time: Date.now(),

            dkim: {
                hashAlgo: 'sha256'
            }
        };

        if (options.user) {
            envelope.userId = options.user.toString();
        }

        if (options.userEmail) {
            // prefer email address as the username
            envelope.user = options.userEmail;
        }

        if (options.parentId) {
            envelope.parentId = options.parentId;
        }

        if (options.reason) {
            envelope.reason = options.reason;
        }

        envelope.origin = options.origin || '127.0.0.1';

        let messageInfo = {
            'message-id': '<>',
            from: envelope.from || '<>',
            to: [].concat(envelope.to || []).join(',') || '<>',
            src: envelope.origin,
            format() {
                let values = [];
                Object.keys(this).forEach(key => {
                    if (typeof this[key] === 'function' || typeof this[key] === 'undefined') {
                        return;
                    }
                    values.push(util.format('%s=%s', key, !/^"/.test(this[key]) && /\s/.test(this[key]) ? JSON.stringify(this[key]) : this[key]));
                });
                return values.join(' ');
            },
            keys() {
                let data = {};
                Object.keys(this).forEach(key => {
                    if (typeof this[key] === 'function' || typeof this[key] === 'undefined') {
                        return;
                    }
                    data[key] = this[key];
                });
                return data;
            }
        };

        let deliveries = [];

        if (options.targets) {
            options.targets.forEach(target => {
                switch (target.type) {
                    case 'mail':
                        deliveries.push({
                            to: target.value,
                            forwardedFor: target.recipient
                        });
                        break;

                    case 'relay':
                        {
                            let recipients = new Set([].concat(options.to || []).concat(target.recipient || []));
                            recipients.forEach(to => {
                                let relayData = target.value;
                                if (typeof relayData === 'string') {
                                    relayData = tools.getRelayData(relayData);
                                }
                                deliveries.push({
                                    to,
                                    mx: relayData.mx,
                                    mxPort: relayData.mxPort,
                                    mxAuth: relayData.mxAuth,
                                    mxSecure: relayData.mxSecure,
                                    skipSRS: true,
                                    skipSTS: true,
                                    forwardedFor: target.recipient
                                });
                            });
                        }
                        break;

                    case 'http':
                        {
                            let recipients = new Set([].concat(options.to || []).concat(target.recipient || []));
                            recipients.forEach(to => {
                                deliveries.push({
                                    to,
                                    http: true,
                                    targetUrl: target.value,
                                    skipSRS: true,
                                    skipSTS: true,
                                    forwardedFor: target.recipient
                                });
                            });
                        }
                        break;
                }
            });
        }

        if (!deliveries.length) {
            deliveries = envelope.to.map(to => ({
                to
            }));
        }

        if (!deliveries.length) {
            let err = new Error('No valid recipients');
            err.responseCode = 400;
            err.code = 'ENORECIPIENTS';
            setImmediate(() => callback(err));
            return false;
        }

        let messageSplitter = new MessageSplitter();
        let dkimStream = new RelaxedBody(envelope.dkim);

        messageSplitter.once('headers', headers => {
            envelope.headers = headers;
            this.updateHeaders(envelope, options);
        });

        dkimStream.on('hash', bodyHash => {
            // store relaxed body hash for signing
            envelope.dkim.bodyHash = bodyHash;
            envelope.bodySize = dkimStream.byteLength;
        });

        messageSplitter.once('error', err => dkimStream.emit('error', err));

        plugins.handler.runHooks('message:store', [envelope, dkimStream], err => {
            if (err) {
                if (dkimStream.readable) {
                    dkimStream.resume(); // let the original stream to end normally before displaying the error message
                }
                return setImmediate(() => callback(err));
            }

            this.store(id, dkimStream, err => {
                if (err) {
                    return callback(err);
                }

                if (this.checkLoop(envelope, deliveries)) {
                    // looped message
                    let err = new Error('Message loop detected');
                    err.responseCode = 500;
                    err.code = 'ELOOP';
                    return this.removeMessage(id, () => callback(err));
                }

                plugins.handler.runHooks('message:queue', [envelope, messageInfo], err => {
                    if (err) {
                        return setImmediate(() => this.removeMessage(id, () => callback(err)));
                    }

                    envelope.headers = envelope.headers.getList();
                    this.setMeta(id, envelope, err => {
                        if (err) {
                            return this.removeMessage(id, () => callback(err));
                        }

                        let date = new Date();

                        for (let i = 0, len = deliveries.length; i < len; i++) {
                            let recipient = deliveries[i];

                            let deliveryZone = options.zone || this.zone || 'default';
                            let recipientDomain = recipient.to.substr(recipient.to.lastIndexOf('@') + 1).replace(/[[\]]/g, '');

                            seq++;
                            let deliverySeq = (seq < 0x100 ? '0' : '') + (seq < 0x10 ? '0' : '') + seq.toString(16);
                            let delivery = {
                                id,
                                seq: deliverySeq,

                                // Actual delivery data
                                domain: recipientDomain,
                                sendingZone: deliveryZone,

                                assigned: 'no',

                                // actual recipient address
                                recipient: recipient.to,

                                locked: false,
                                lockTime: 0,

                                // earliest time to attempt delivery, defaults to now
                                queued: options.sendTime || date,

                                // queued date might change but created should not
                                created: date
                            };

                            if (recipient.http) {
                                delivery.http = recipient.http;
                                delivery.targetUrl = recipient.targetUrl;
                            }

                            ['mx', 'mxPort', 'mxAuth', 'mxSecure'].forEach(key => {
                                if (recipient[key]) {
                                    delivery[key] = recipient[key];
                                }
                            });

                            if (recipient.skipSRS) {
                                delivery.skipSRS = true;
                            }

                            if (recipient.skipSTS) {
                                delivery.skipSTS = true;
                            }

                            documents.push(delivery);
                        }

                        this.db.senderDb.collection(this.collection).insertMany(
                            documents,
                            {
                                writeConcern: 1,
                                ordered: false
                            },
                            err => {
                                if (err) {
                                    return callback(err);
                                }

                                callback(null, envelope);
                            }
                        );
                    });
                });
            });
        });

        if (options.runPlugins) {
            // message submissions
            let source = new PassThrough();
            let raw = new PassThrough();

            plugins.handler.runAnalyzerHooks(envelope, source, raw);
            raw.pipe(messageSplitter);
            messageSplitter.pipe(dkimStream);

            source.on('error', err => raw.emit('error', err));
            raw.on('error', err => messageSplitter.emit('error', err));

            return source;
        } else {
            // default, no plugins (autoreplies etc.)
            messageSplitter.pipe(dkimStream);
            return messageSplitter;
        }
    }

    convertAddresses(addresses, withNames, addressList) {
        addressList = addressList || new Map();

        this.flatten(addresses || []).forEach(address => {
            if (address.address) {
                let normalized = this.normalizeAddress(address, withNames);
                let key = typeof normalized === 'string' ? normalized : normalized.address;
                addressList.set(key, normalized);
            } else if (address.group) {
                this.convertAddresses(address.group, withNames, addressList);
            }
        });

        return addressList;
    }

    parseAddressList(headers, key, withNames) {
        return this.parseAddressses(
            headers.getDecoded(key).map(header => header.value),
            withNames
        );
    }

    parseAddressses(headerList, withNames) {
        let map = this.convertAddresses(
            headerList.map(address => {
                if (typeof address === 'string') {
                    address = addressparser(address);
                }
                return address;
            }),
            withNames
        );
        return Array.from(map).map(entry => entry[1]);
    }

    normalizeDomain(domain) {
        domain = domain.toLowerCase().trim();
        try {
            domain = punycode.toASCII(domain);
        } catch (E) {
            // ignore
        }
        return domain;
    }

    // helper function to flatten arrays
    flatten(arr) {
        let flat = [].concat(...arr);
        return flat.some(Array.isArray) ? this.flatten(flat) : flat;
    }

    normalizeAddress(address, withNames) {
        if (typeof address === 'string') {
            address = {
                address
            };
        }
        if (!address || !address.address) {
            return '';
        }
        let user = address.address.substr(0, address.address.lastIndexOf('@'));
        let domain = address.address.substr(address.address.lastIndexOf('@') + 1);
        let addr = user.trim() + '@' + this.normalizeDomain(domain);

        if (withNames) {
            return {
                name: address.name || '',
                address: addr
            };
        }

        return addr;
    }

    updateHeaders(envelope, options) {
        let updateDate = options && options.updateDate;
        // Fetch sender and receiver addresses
        envelope.parsedEnvelope = {
            from: this.parseAddressList(envelope.headers, 'from').shift() || false,
            to: this.parseAddressList(envelope.headers, 'to'),
            cc: this.parseAddressList(envelope.headers, 'cc'),
            bcc: this.parseAddressList(envelope.headers, 'bcc'),
            replyTo: this.parseAddressList(envelope.headers, 'reply-to').shift() || false,
            sender: this.parseAddressList(envelope.headers, 'sender').shift() || false
        };

        // Check Message-ID: value. Add if missing
        let mId = envelope.headers.getFirst('message-id');
        if (!mId) {
            mId = '<' + uuid.v4() + '@' + (envelope.from.substr(envelope.from.lastIndexOf('@') + 1) || hostname) + '>';

            envelope.headers.remove('message-id'); // in case there's an empty value
            envelope.headers.add('Message-ID', mId);
        }
        envelope.messageId = mId;

        // Check Date: value. Add if missing or invalid or future date
        let date = envelope.headers.getFirst('date');
        let dateVal = new Date(date);

        if (updateDate || !date || dateVal.toString() === 'Invalid Date' || dateVal < new Date(1000)) {
            date = new Date().toUTCString().replace(/GMT/, '+0000');
            envelope.headers.remove('date'); // remove old empty or invalid values
            envelope.headers.add('Date', date);
        }

        envelope.date = date;

        // Remove BCC if present
        envelope.headers.remove('bcc');
    }

    store(id, stream, callback) {
        let returned = false;
        let store = this.gridstore.openUploadStream('message ' + id, {
            fsync: true,
            contentType: 'message/rfc822',
            metadata: {
                created: new Date()
            }
        });

        stream.once('error', err => {
            if (returned) {
                return;
            }
            returned = true;

            store.once('finish', () => {
                this.removeMessage(id, () => callback(err));
            });

            store.end();
        });

        store.once('error', err => {
            if (returned) {
                return;
            }
            returned = true;
            callback(err);
        });

        store.once('finish', () => {
            if (returned) {
                return;
            }
            returned = true;

            return callback(null, id);
        });

        stream.pipe(store);
    }

    removeMessage(id, callback) {
        this.db.senderDb.collection(this.gfs + '.files').findOne(
            {
                filename: 'message ' + id
            },
            (err, entry) => {
                if (err) {
                    return callback(err);
                }
                if (!entry) {
                    return callback(null, false);
                }
                this.gridstore.delete(entry._id, callback);
            }
        );
    }

    setMeta(id, data, callback) {
        this.db.senderDb.collection(this.gfs + '.files').findOneAndUpdate(
            {
                filename: 'message ' + id
            },
            {
                $set: {
                    'metadata.data': data
                }
            },
            {},
            err => {
                if (err) {
                    return callback(err);
                }
                return callback();
            }
        );
    }

    async removeFromQueue(id, user) {
        let queueFile = await this.db.senderDb.collection(this.gfs + '.files').findOne({
            filename: 'message ' + id
        });

        if (!queueFile) {
            return { success: false, code: 'NoSuchQueueEntry' };
        }

        if (user && queueFile.metadata.data.userId && user.toString() !== queueFile.metadata.data.userId.toString()) {
            // message does not belong to us
            return { success: false, code: 'NotEnoughPrivileges' };
        }

        // delete message entries that are not locked
        let queueDelRes = await this.db.senderDb.collection(this.collection).deleteMany({ id, locked: false });
        let deleted = queueDelRes.deletedCount;

        let stillQueued = await this.db.senderDb.collection(this.collection).countDocuments({ id });
        if (!stillQueued) {
            // delete gridstore file as there are no more queued queue entries
            let gsFile = await this.db.senderDb.collection(this.gfs + '.files').findOne({
                filename: 'message ' + id
            });
            if (gsFile) {
                await this.gridstore.delete(gsFile._id);
            }
        }
        // delete queue entries

        return { success: true, queueId: id, deleted };
    }
}

module.exports = Maildropper;