'use strict'; const packageInfo = require('../../package.json'); const EventEmitter = require('events').EventEmitter; const net = require('net'); const tls = require('tls'); const os = require('os'); const crypto = require('crypto'); const DataStream = require('./data-stream'); const PassThrough = require('stream').PassThrough; const shared = require('../shared'); // default timeout values in ms const CONNECTION_TIMEOUT = 2 * 60 * 1000; // how much to wait for the connection to be established const SOCKET_TIMEOUT = 10 * 60 * 1000; // how much to wait for socket inactivity before disconnecting the client const GREETING_TIMEOUT = 30 * 1000; // how much to wait after connection is established but SMTP greeting is not receieved /** * Generates a SMTP connection object * * Optional options object takes the following possible properties: * * * **port** - is the port to connect to (defaults to 587 or 465) * * **host** - is the hostname or IP address to connect to (defaults to 'localhost') * * **secure** - use SSL * * **ignoreTLS** - ignore server support for STARTTLS * * **requireTLS** - forces the client to use STARTTLS * * **name** - the name of the client server * * **localAddress** - outbound address to bind to (see: http://nodejs.org/api/net.html#net_net_connect_options_connectionlistener) * * **greetingTimeout** - Time to wait in ms until greeting message is received from the server (defaults to 10000) * * **connectionTimeout** - how many milliseconds to wait for the connection to establish * * **socketTimeout** - Time of inactivity until the connection is closed (defaults to 1 hour) * * **lmtp** - if true, uses LMTP instead of SMTP protocol * * **logger** - bunyan compatible logger interface * * **debug** - if true pass SMTP traffic to the logger * * **tls** - options for createCredentials * * **socket** - existing socket to use instead of creating a new one (see: http://nodejs.org/api/net.html#net_class_net_socket) * * **secured** - boolean indicates that the provided socket has already been upgraded to tls * * @constructor * @namespace SMTP Client module * @param {Object} [options] Option properties */ class SMTPConnection extends EventEmitter { constructor(options) { super(options); this.id = crypto.randomBytes(8).toString('base64').replace(/\W/g, ''); this.stage = 'init'; this.options = options || {}; this.secureConnection = !!this.options.secure; this.alreadySecured = !!this.options.secured; this.port = Number(this.options.port) || (this.secureConnection ? 465 : 587); this.host = this.options.host || 'localhost'; if (typeof this.options.secure === 'undefined' && this.port === 465) { // if secure option is not set but port is 465, then default to secure this.secureConnection = true; } this.name = this.options.name || this._getHostname(); this.logger = shared.getLogger(this.options, { component: this.options.component || 'smtp-connection', sid: this.id }); this.customAuth = new Map(); Object.keys(this.options.customAuth || {}).forEach(key => { let mapKey = (key || '').toString().trim().toUpperCase(); if (!mapKey) { return; } this.customAuth.set(mapKey, this.options.customAuth[key]); }); /** * Expose version nr, just for the reference * @type {String} */ this.version = packageInfo.version; /** * If true, then the user is authenticated * @type {Boolean} */ this.authenticated = false; /** * If set to true, this instance is no longer active * @private */ this.destroyed = false; /** * Defines if the current connection is secure or not. If not, * STARTTLS can be used if available * @private */ this.secure = !!this.secureConnection; /** * Store incomplete messages coming from the server * @private */ this._remainder = ''; /** * Unprocessed responses from the server * @type {Array} */ this._responseQueue = []; this.lastServerResponse = false; /** * The socket connecting to the server * @publick */ this._socket = false; /** * Lists supported auth mechanisms * @private */ this._supportedAuth = []; /** * Set to true, if EHLO response includes "AUTH". * If false then authentication is not tried */ this.allowsAuth = false; /** * Includes current envelope (from, to) * @private */ this._envelope = false; /** * Lists supported extensions * @private */ this._supportedExtensions = []; /** * Defines the maximum allowed size for a single message * @private */ this._maxAllowedSize = 0; /** * Function queue to run if a data chunk comes from the server * @private */ this._responseActions = []; this._recipientQueue = []; /** * Timeout variable for waiting the greeting * @private */ this._greetingTimeout = false; /** * Timeout variable for waiting the connection to start * @private */ this._connectionTimeout = false; /** * If the socket is deemed already closed * @private */ this._destroyed = false; /** * If the socket is already being closed * @private */ this._closing = false; /** * Callbacks for socket's listeners */ this._onSocketData = chunk => this._onData(chunk); this._onSocketError = error => this._onError(error, 'ESOCKET', false, 'CONN'); this._onSocketClose = () => this._onClose(); this._onSocketEnd = () => this._onEnd(); this._onSocketTimeout = () => this._onTimeout(); } /** * Creates a connection to a SMTP server and sets up connection * listener */ connect(connectCallback) { if (typeof connectCallback === 'function') { this.once('connect', () => { this.logger.debug( { tnx: 'smtp' }, 'SMTP handshake finished' ); connectCallback(); }); const isDestroyedMessage = this._isDestroyedMessage('connect'); if (isDestroyedMessage) { return connectCallback(this._formatError(isDestroyedMessage, 'ECONNECTION', false, 'CONN')); } } let opts = { port: this.port, host: this.host }; if (this.options.localAddress) { opts.localAddress = this.options.localAddress; } let setupConnectionHandlers = () => { this._connectionTimeout = setTimeout(() => { this._onError('Connection timeout', 'ETIMEDOUT', false, 'CONN'); }, this.options.connectionTimeout || CONNECTION_TIMEOUT); this._socket.on('error', this._onSocketError); }; if (this.options.connection) { // connection is already opened this._socket = this.options.connection; if (this.secureConnection && !this.alreadySecured) { setImmediate(() => this._upgradeConnection(err => { if (err) { this._onError(new Error('Error initiating TLS - ' + (err.message || err)), 'ETLS', false, 'CONN'); return; } this._onConnect(); }) ); } else { setImmediate(() => this._onConnect()); } return; } else if (this.options.socket) { // socket object is set up but not yet connected this._socket = this.options.socket; return shared.resolveHostname(opts, (err, resolved) => { if (err) { return setImmediate(() => this._onError(err, 'EDNS', false, 'CONN')); } this.logger.debug( { tnx: 'dns', source: opts.host, resolved: resolved.host, cached: !!resolved._cached }, 'Resolved %s as %s [cache %s]', opts.host, resolved.host, resolved._cached ? 'hit' : 'miss' ); Object.keys(resolved).forEach(key => { if (key.charAt(0) !== '_' && resolved[key]) { opts[key] = resolved[key]; } }); try { this._socket.connect(this.port, this.host, () => { this._socket.setKeepAlive(true); this._onConnect(); }); setupConnectionHandlers(); } catch (E) { return setImmediate(() => this._onError(E, 'ECONNECTION', false, 'CONN')); } }); } else if (this.secureConnection) { // connect using tls if (this.options.tls) { Object.keys(this.options.tls).forEach(key => { opts[key] = this.options.tls[key]; }); } return shared.resolveHostname(opts, (err, resolved) => { if (err) { return setImmediate(() => this._onError(err, 'EDNS', false, 'CONN')); } this.logger.debug( { tnx: 'dns', source: opts.host, resolved: resolved.host, cached: !!resolved._cached }, 'Resolved %s as %s [cache %s]', opts.host, resolved.host, resolved._cached ? 'hit' : 'miss' ); Object.keys(resolved).forEach(key => { if (key.charAt(0) !== '_' && resolved[key]) { opts[key] = resolved[key]; } }); try { this._socket = tls.connect(opts, () => { this._socket.setKeepAlive(true); this._onConnect(); }); setupConnectionHandlers(); } catch (E) { return setImmediate(() => this._onError(E, 'ECONNECTION', false, 'CONN')); } }); } else { // connect using plaintext return shared.resolveHostname(opts, (err, resolved) => { if (err) { return setImmediate(() => this._onError(err, 'EDNS', false, 'CONN')); } this.logger.debug( { tnx: 'dns', source: opts.host, resolved: resolved.host, cached: !!resolved._cached }, 'Resolved %s as %s [cache %s]', opts.host, resolved.host, resolved._cached ? 'hit' : 'miss' ); Object.keys(resolved).forEach(key => { if (key.charAt(0) !== '_' && resolved[key]) { opts[key] = resolved[key]; } }); try { this._socket = net.connect(opts, () => { this._socket.setKeepAlive(true); this._onConnect(); }); setupConnectionHandlers(); } catch (E) { return setImmediate(() => this._onError(E, 'ECONNECTION', false, 'CONN')); } }); } } /** * Sends QUIT */ quit() { this._sendCommand('QUIT'); this._responseActions.push(this.close); } /** * Closes the connection to the server */ close() { clearTimeout(this._connectionTimeout); clearTimeout(this._greetingTimeout); this._responseActions = []; // allow to run this function only once if (this._closing) { return; } this._closing = true; let closeMethod = 'end'; if (this.stage === 'init') { // Close the socket immediately when connection timed out closeMethod = 'destroy'; } this.logger.debug( { tnx: 'smtp' }, 'Closing connection to the server using "%s"', closeMethod ); let socket = (this._socket && this._socket.socket) || this._socket; if (socket && !socket.destroyed) { try { this._socket[closeMethod](); } catch (E) { // just ignore } } this._destroy(); } /** * Authenticate user */ login(authData, callback) { const isDestroyedMessage = this._isDestroyedMessage('login'); if (isDestroyedMessage) { return callback(this._formatError(isDestroyedMessage, 'ECONNECTION', false, 'API')); } this._auth = authData || {}; // Select SASL authentication method this._authMethod = (this._auth.method || '').toString().trim().toUpperCase() || false; if (!this._authMethod && this._auth.oauth2 && !this._auth.credentials) { this._authMethod = 'XOAUTH2'; } else if (!this._authMethod || (this._authMethod === 'XOAUTH2' && !this._auth.oauth2)) { // use first supported this._authMethod = (this._supportedAuth[0] || 'PLAIN').toUpperCase().trim(); } if (this._authMethod !== 'XOAUTH2' && (!this._auth.credentials || !this._auth.credentials.user || !this._auth.credentials.pass)) { if (this._auth.user && this._auth.pass) { this._auth.credentials = { user: this._auth.user, pass: this._auth.pass, options: this._auth.options }; } else { return callback(this._formatError('Missing credentials for "' + this._authMethod + '"', 'EAUTH', false, 'API')); } } if (this.customAuth.has(this._authMethod)) { let handler = this.customAuth.get(this._authMethod); let lastResponse; let returned = false; let resolve = () => { if (returned) { return; } returned = true; this.logger.info( { tnx: 'smtp', username: this._auth.user, action: 'authenticated', method: this._authMethod }, 'User %s authenticated', JSON.stringify(this._auth.user) ); this.authenticated = true; callback(null, true); }; let reject = err => { if (returned) { return; } returned = true; callback(this._formatError(err, 'EAUTH', lastResponse, 'AUTH ' + this._authMethod)); }; let handlerResponse = handler({ auth: this._auth, method: this._authMethod, extensions: [].concat(this._supportedExtensions), authMethods: [].concat(this._supportedAuth), maxAllowedSize: this._maxAllowedSize || false, sendCommand: (cmd, done) => { let promise; if (!done) { promise = new Promise((resolve, reject) => { done = shared.callbackPromise(resolve, reject); }); } this._responseActions.push(str => { lastResponse = str; let codes = str.match(/^(\d+)(?:\s(\d+\.\d+\.\d+))?\s/); let data = { command: cmd, response: str }; if (codes) { data.status = Number(codes[1]) || 0; if (codes[2]) { data.code = codes[2]; } data.text = str.substr(codes[0].length); } else { data.text = str; data.status = 0; // just in case we need to perform numeric comparisons } done(null, data); }); setImmediate(() => this._sendCommand(cmd)); return promise; }, resolve, reject }); if (handlerResponse && typeof handlerResponse.catch === 'function') { // a promise was returned handlerResponse.then(resolve).catch(reject); } return; } switch (this._authMethod) { case 'XOAUTH2': this._handleXOauth2Token(false, callback); return; case 'LOGIN': this._responseActions.push(str => { this._actionAUTH_LOGIN_USER(str, callback); }); this._sendCommand('AUTH LOGIN'); return; case 'PLAIN': this._responseActions.push(str => { this._actionAUTHComplete(str, callback); }); this._sendCommand( 'AUTH PLAIN ' + Buffer.from( //this._auth.user+'\u0000'+ '\u0000' + // skip authorization identity as it causes problems with some servers this._auth.credentials.user + '\u0000' + this._auth.credentials.pass, 'utf-8' ).toString('base64') ); return; case 'CRAM-MD5': this._responseActions.push(str => { this._actionAUTH_CRAM_MD5(str, callback); }); this._sendCommand('AUTH CRAM-MD5'); return; } return callback(this._formatError('Unknown authentication method "' + this._authMethod + '"', 'EAUTH', false, 'API')); } /** * Sends a message * * @param {Object} envelope Envelope object, {from: addr, to: [addr]} * @param {Object} message String, Buffer or a Stream * @param {Function} callback Callback to return once sending is completed */ send(envelope, message, done) { if (!message) { return done(this._formatError('Empty message', 'EMESSAGE', false, 'API')); } const isDestroyedMessage = this._isDestroyedMessage('send message'); if (isDestroyedMessage) { return done(this._formatError(isDestroyedMessage, 'ECONNECTION', false, 'API')); } // reject larger messages than allowed if (this._maxAllowedSize && envelope.size > this._maxAllowedSize) { return setImmediate(() => { done(this._formatError('Message size larger than allowed ' + this._maxAllowedSize, 'EMESSAGE', false, 'MAIL FROM')); }); } // ensure that callback is only called once let returned = false; let callback = function () { if (returned) { return; } returned = true; done(...arguments); }; if (typeof message.on === 'function') { message.on('error', err => callback(this._formatError(err, 'ESTREAM', false, 'API'))); } let startTime = Date.now(); this._setEnvelope(envelope, (err, info) => { if (err) { return callback(err); } let envelopeTime = Date.now(); let stream = this._createSendStream((err, str) => { if (err) { return callback(err); } info.envelopeTime = envelopeTime - startTime; info.messageTime = Date.now() - envelopeTime; info.messageSize = stream.outByteCount; info.response = str; return callback(null, info); }); if (typeof message.pipe === 'function') { message.pipe(stream); } else { stream.write(message); stream.end(); } }); } /** * Resets connection state * * @param {Function} callback Callback to return once connection is reset */ reset(callback) { this._sendCommand('RSET'); this._responseActions.push(str => { if (str.charAt(0) !== '2') { return callback(this._formatError('Could not reset session state. response=' + str, 'EPROTOCOL', str, 'RSET')); } this._envelope = false; return callback(null, true); }); } /** * Connection listener that is run when the connection to * the server is opened * * @event */ _onConnect() { clearTimeout(this._connectionTimeout); this.logger.info( { tnx: 'network', localAddress: this._socket.localAddress, localPort: this._socket.localPort, remoteAddress: this._socket.remoteAddress, remotePort: this._socket.remotePort }, '%s established to %s:%s', this.secure ? 'Secure connection' : 'Connection', this._socket.remoteAddress, this._socket.remotePort ); if (this._destroyed) { // Connection was established after we already had canceled it this.close(); return; } this.stage = 'connected'; // clear existing listeners for the socket this._socket.removeListener('data', this._onSocketData); this._socket.removeListener('timeout', this._onSocketTimeout); this._socket.removeListener('close', this._onSocketClose); this._socket.removeListener('end', this._onSocketEnd); this._socket.on('data', this._onSocketData); this._socket.once('close', this._onSocketClose); this._socket.once('end', this._onSocketEnd); this._socket.setTimeout(this.options.socketTimeout || SOCKET_TIMEOUT); this._socket.on('timeout', this._onSocketTimeout); this._greetingTimeout = setTimeout(() => { // if still waiting for greeting, give up if (this._socket && !this._destroyed && this._responseActions[0] === this._actionGreeting) { this._onError('Greeting never received', 'ETIMEDOUT', false, 'CONN'); } }, this.options.greetingTimeout || GREETING_TIMEOUT); this._responseActions.push(this._actionGreeting); // we have a 'data' listener set up so resume socket if it was paused this._socket.resume(); } /** * 'data' listener for data coming from the server * * @event * @param {Buffer} chunk Data chunk coming from the server */ _onData(chunk) { if (this._destroyed || !chunk || !chunk.length) { return; } let data = (chunk || '').toString('binary'); let lines = (this._remainder + data).split(/\r?\n/); let lastline; this._remainder = lines.pop(); for (let i = 0, len = lines.length; i < len; i++) { if (this._responseQueue.length) { lastline = this._responseQueue[this._responseQueue.length - 1]; if (/^\d+-/.test(lastline.split('\n').pop())) { this._responseQueue[this._responseQueue.length - 1] += '\n' + lines[i]; continue; } } this._responseQueue.push(lines[i]); } if (this._responseQueue.length) { lastline = this._responseQueue[this._responseQueue.length - 1]; if (/^\d+-/.test(lastline.split('\n').pop())) { return; } } this._processResponse(); } /** * 'error' listener for the socket * * @event * @param {Error} err Error object * @param {String} type Error name */ _onError(err, type, data, command) { clearTimeout(this._connectionTimeout); clearTimeout(this._greetingTimeout); if (this._destroyed) { // just ignore, already closed // this might happen when a socket is canceled because of reached timeout // but the socket timeout error itself receives only after return; } err = this._formatError(err, type, data, command); this.logger.error(data, err.message); this.emit('error', err); this.close(); } _formatError(message, type, response, command) { let err; if (/Error\]$/i.test(Object.prototype.toString.call(message))) { err = message; } else { err = new Error(message); } if (type && type !== 'Error') { err.code = type; } if (response) { err.response = response; err.message += ': ' + response; } let responseCode = (typeof response === 'string' && Number((response.match(/^\d+/) || [])[0])) || false; if (responseCode) { err.responseCode = responseCode; } if (command) { err.command = command; } return err; } /** * 'close' listener for the socket * * @event */ _onClose() { this.logger.info( { tnx: 'network' }, 'Connection closed' ); if (this.upgrading && !this._destroyed) { return this._onError(new Error('Connection closed unexpectedly'), 'ETLS', false, 'CONN'); } else if (![this._actionGreeting, this.close].includes(this._responseActions[0]) && !this._destroyed) { return this._onError(new Error('Connection closed unexpectedly'), 'ECONNECTION', false, 'CONN'); } this._destroy(); } /** * 'end' listener for the socket * * @event */ _onEnd() { if (this._socket && !this._socket.destroyed) { this._socket.destroy(); } } /** * 'timeout' listener for the socket * * @event */ _onTimeout() { return this._onError(new Error('Timeout'), 'ETIMEDOUT', false, 'CONN'); } /** * Destroys the client, emits 'end' */ _destroy() { if (this._destroyed) { return; } this._destroyed = true; this.emit('end'); } /** * Upgrades the connection to TLS * * @param {Function} callback Callback function to run when the connection * has been secured */ _upgradeConnection(callback) { // do not remove all listeners or it breaks node v0.10 as there's // apparently a 'finish' event set that would be cleared as well // we can safely keep 'error', 'end', 'close' etc. events this._socket.removeListener('data', this._onSocketData); // incoming data is going to be gibberish from this point onwards this._socket.removeListener('timeout', this._onSocketTimeout); // timeout will be re-set for the new socket object let socketPlain = this._socket; let opts = { socket: this._socket, host: this.host }; Object.keys(this.options.tls || {}).forEach(key => { opts[key] = this.options.tls[key]; }); this.upgrading = true; // tls.connect is not an asynchronous function however it may still throw errors and requires to be wrapped with try/catch try { this._socket = tls.connect(opts, () => { this.secure = true; this.upgrading = false; this._socket.on('data', this._onSocketData); socketPlain.removeListener('close', this._onSocketClose); socketPlain.removeListener('end', this._onSocketEnd); return callback(null, true); }); } catch (err) { return callback(err); } this._socket.on('error', this._onSocketError); this._socket.once('close', this._onSocketClose); this._socket.once('end', this._onSocketEnd); this._socket.setTimeout(this.options.socketTimeout || SOCKET_TIMEOUT); // 10 min. this._socket.on('timeout', this._onSocketTimeout); // resume in case the socket was paused socketPlain.resume(); } /** * Processes queued responses from the server * * @param {Boolean} force If true, ignores _processing flag */ _processResponse() { if (!this._responseQueue.length) { return false; } let str = (this.lastServerResponse = (this._responseQueue.shift() || '').toString()); if (/^\d+-/.test(str.split('\n').pop())) { // keep waiting for the final part of multiline response return; } if (this.options.debug || this.options.transactionLog) { this.logger.debug( { tnx: 'server' }, str.replace(/\r?\n$/, '') ); } if (!str.trim()) { // skip unexpected empty lines setImmediate(() => this._processResponse(true)); } let action = this._responseActions.shift(); if (typeof action === 'function') { action.call(this, str); setImmediate(() => this._processResponse(true)); } else { return this._onError(new Error('Unexpected Response'), 'EPROTOCOL', str, 'CONN'); } } /** * Send a command to the server, append \r\n * * @param {String} str String to be sent to the server */ _sendCommand(str) { if (this._destroyed) { // Connection already closed, can't send any more data return; } if (this._socket.destroyed) { return this.close(); } if (this.options.debug || this.options.transactionLog) { this.logger.debug( { tnx: 'client' }, (str || '').toString().replace(/\r?\n$/, '') ); } this._socket.write(Buffer.from(str + '\r\n', 'utf-8')); } /** * Initiates a new message by submitting envelope data, starting with * MAIL FROM: command * * @param {Object} envelope Envelope object in the form of * {from:'...', to:['...']} * or * {from:{address:'...',name:'...'}, to:[address:'...',name:'...']} */ _setEnvelope(envelope, callback) { let args = []; let useSmtpUtf8 = false; this._envelope = envelope || {}; this._envelope.from = ((this._envelope.from && this._envelope.from.address) || this._envelope.from || '').toString().trim(); this._envelope.to = [].concat(this._envelope.to || []).map(to => ((to && to.address) || to || '').toString().trim()); if (!this._envelope.to.length) { return callback(this._formatError('No recipients defined', 'EENVELOPE', false, 'API')); } if (this._envelope.from && /[\r\n<>]/.test(this._envelope.from)) { return callback(this._formatError('Invalid sender ' + JSON.stringify(this._envelope.from), 'EENVELOPE', false, 'API')); } // check if the sender address uses only ASCII characters, // otherwise require usage of SMTPUTF8 extension if (/[\x80-\uFFFF]/.test(this._envelope.from)) { useSmtpUtf8 = true; } for (let i = 0, len = this._envelope.to.length; i < len; i++) { if (!this._envelope.to[i] || /[\r\n<>]/.test(this._envelope.to[i])) { return callback(this._formatError('Invalid recipient ' + JSON.stringify(this._envelope.to[i]), 'EENVELOPE', false, 'API')); } // check if the recipients addresses use only ASCII characters, // otherwise require usage of SMTPUTF8 extension if (/[\x80-\uFFFF]/.test(this._envelope.to[i])) { useSmtpUtf8 = true; } } // clone the recipients array for latter manipulation this._envelope.rcptQueue = JSON.parse(JSON.stringify(this._envelope.to || [])); this._envelope.rejected = []; this._envelope.rejectedErrors = []; this._envelope.accepted = []; if (this._envelope.dsn) { try { this._envelope.dsn = this._setDsnEnvelope(this._envelope.dsn); } catch (err) { return callback(this._formatError('Invalid DSN ' + err.message, 'EENVELOPE', false, 'API')); } } this._responseActions.push(str => { this._actionMAIL(str, callback); }); // If the server supports SMTPUTF8 and the envelope includes an internationalized // email address then append SMTPUTF8 keyword to the MAIL FROM command if (useSmtpUtf8 && this._supportedExtensions.includes('SMTPUTF8')) { args.push('SMTPUTF8'); this._usingSmtpUtf8 = true; } // If the server supports 8BITMIME and the message might contain non-ascii bytes // then append the 8BITMIME keyword to the MAIL FROM command if (this._envelope.use8BitMime && this._supportedExtensions.includes('8BITMIME')) { args.push('BODY=8BITMIME'); this._using8BitMime = true; } if (this._envelope.size && this._supportedExtensions.includes('SIZE')) { args.push('SIZE=' + this._envelope.size); } // If the server supports DSN and the envelope includes an DSN prop // then append DSN params to the MAIL FROM command if (this._envelope.dsn && this._supportedExtensions.includes('DSN')) { if (this._envelope.dsn.ret) { args.push('RET=' + shared.encodeXText(this._envelope.dsn.ret)); } if (this._envelope.dsn.envid) { args.push('ENVID=' + shared.encodeXText(this._envelope.dsn.envid)); } } this._sendCommand('MAIL FROM:<' + this._envelope.from + '>' + (args.length ? ' ' + args.join(' ') : '')); } _setDsnEnvelope(params) { let ret = (params.ret || params.return || '').toString().toUpperCase() || null; if (ret) { switch (ret) { case 'HDRS': case 'HEADERS': ret = 'HDRS'; break; case 'FULL': case 'BODY': ret = 'FULL'; break; } } if (ret && !['FULL', 'HDRS'].includes(ret)) { throw new Error('ret: ' + JSON.stringify(ret)); } let envid = (params.envid || params.id || '').toString() || null; let notify = params.notify || null; if (notify) { if (typeof notify === 'string') { notify = notify.split(','); } notify = notify.map(n => n.trim().toUpperCase()); let validNotify = ['NEVER', 'SUCCESS', 'FAILURE', 'DELAY']; let invaliNotify = notify.filter(n => !validNotify.includes(n)); if (invaliNotify.length || (notify.length > 1 && notify.includes('NEVER'))) { throw new Error('notify: ' + JSON.stringify(notify.join(','))); } notify = notify.join(','); } let orcpt = (params.orcpt || params.recipient || '').toString() || null; if (orcpt && orcpt.indexOf(';') < 0) { orcpt = 'rfc822;' + orcpt; } return { ret, envid, notify, orcpt }; } _getDsnRcptToArgs() { let args = []; // If the server supports DSN and the envelope includes an DSN prop // then append DSN params to the RCPT TO command if (this._envelope.dsn && this._supportedExtensions.includes('DSN')) { if (this._envelope.dsn.notify) { args.push('NOTIFY=' + shared.encodeXText(this._envelope.dsn.notify)); } if (this._envelope.dsn.orcpt) { args.push('ORCPT=' + shared.encodeXText(this._envelope.dsn.orcpt)); } } return args.length ? ' ' + args.join(' ') : ''; } _createSendStream(callback) { let dataStream = new DataStream(); let logStream; if (this.options.lmtp) { this._envelope.accepted.forEach((recipient, i) => { let final = i === this._envelope.accepted.length - 1; this._responseActions.push(str => { this._actionLMTPStream(recipient, final, str, callback); }); }); } else { this._responseActions.push(str => { this._actionSMTPStream(str, callback); }); } dataStream.pipe(this._socket, { end: false }); if (this.options.debug) { logStream = new PassThrough(); logStream.on('readable', () => { let chunk; while ((chunk = logStream.read())) { this.logger.debug( { tnx: 'message' }, chunk.toString('binary').replace(/\r?\n$/, '') ); } }); dataStream.pipe(logStream); } dataStream.once('end', () => { this.logger.info( { tnx: 'message', inByteCount: dataStream.inByteCount, outByteCount: dataStream.outByteCount }, '<%s bytes encoded mime message (source size %s bytes)>', dataStream.outByteCount, dataStream.inByteCount ); }); return dataStream; } /** ACTIONS **/ /** * Will be run after the connection is created and the server sends * a greeting. If the incoming message starts with 220 initiate * SMTP session by sending EHLO command * * @param {String} str Message from the server */ _actionGreeting(str) { clearTimeout(this._greetingTimeout); if (str.substr(0, 3) !== '220') { this._onError(new Error('Invalid greeting. response=' + str), 'EPROTOCOL', str, 'CONN'); return; } if (this.options.lmtp) { this._responseActions.push(this._actionLHLO); this._sendCommand('LHLO ' + this.name); } else { this._responseActions.push(this._actionEHLO); this._sendCommand('EHLO ' + this.name); } } /** * Handles server response for LHLO command. If it yielded in * error, emit 'error', otherwise treat this as an EHLO response * * @param {String} str Message from the server */ _actionLHLO(str) { if (str.charAt(0) !== '2') { this._onError(new Error('Invalid LHLO. response=' + str), 'EPROTOCOL', str, 'LHLO'); return; } this._actionEHLO(str); } /** * Handles server response for EHLO command. If it yielded in * error, try HELO instead, otherwise initiate TLS negotiation * if STARTTLS is supported by the server or move into the * authentication phase. * * @param {String} str Message from the server */ _actionEHLO(str) { let match; if (str.substr(0, 3) === '421') { this._onError(new Error('Server terminates connection. response=' + str), 'ECONNECTION', str, 'EHLO'); return; } if (str.charAt(0) !== '2') { if (this.options.requireTLS) { this._onError(new Error('EHLO failed but HELO does not support required STARTTLS. response=' + str), 'ECONNECTION', str, 'EHLO'); return; } // Try HELO instead this._responseActions.push(this._actionHELO); this._sendCommand('HELO ' + this.name); return; } // Detect if the server supports STARTTLS if (!this.secure && !this.options.ignoreTLS && (/[ -]STARTTLS\b/im.test(str) || this.options.requireTLS)) { this._sendCommand('STARTTLS'); this._responseActions.push(this._actionSTARTTLS); return; } // Detect if the server supports SMTPUTF8 if (/[ -]SMTPUTF8\b/im.test(str)) { this._supportedExtensions.push('SMTPUTF8'); } // Detect if the server supports DSN if (/[ -]DSN\b/im.test(str)) { this._supportedExtensions.push('DSN'); } // Detect if the server supports 8BITMIME if (/[ -]8BITMIME\b/im.test(str)) { this._supportedExtensions.push('8BITMIME'); } // Detect if the server supports PIPELINING if (/[ -]PIPELINING\b/im.test(str)) { this._supportedExtensions.push('PIPELINING'); } // Detect if the server supports AUTH if (/[ -]AUTH\b/i.test(str)) { this.allowsAuth = true; } // Detect if the server supports PLAIN auth if (/[ -]AUTH(?:(\s+|=)[^\n]*\s+|\s+|=)PLAIN/i.test(str)) { this._supportedAuth.push('PLAIN'); } // Detect if the server supports LOGIN auth if (/[ -]AUTH(?:(\s+|=)[^\n]*\s+|\s+|=)LOGIN/i.test(str)) { this._supportedAuth.push('LOGIN'); } // Detect if the server supports CRAM-MD5 auth if (/[ -]AUTH(?:(\s+|=)[^\n]*\s+|\s+|=)CRAM-MD5/i.test(str)) { this._supportedAuth.push('CRAM-MD5'); } // Detect if the server supports XOAUTH2 auth if (/[ -]AUTH(?:(\s+|=)[^\n]*\s+|\s+|=)XOAUTH2/i.test(str)) { this._supportedAuth.push('XOAUTH2'); } // Detect if the server supports SIZE extensions (and the max allowed size) if ((match = str.match(/[ -]SIZE(?:[ \t]+(\d+))?/im))) { this._supportedExtensions.push('SIZE'); this._maxAllowedSize = Number(match[1]) || 0; } this.emit('connect'); } /** * Handles server response for HELO command. If it yielded in * error, emit 'error', otherwise move into the authentication phase. * * @param {String} str Message from the server */ _actionHELO(str) { if (str.charAt(0) !== '2') { this._onError(new Error('Invalid HELO. response=' + str), 'EPROTOCOL', str, 'HELO'); return; } // assume that authentication is enabled (most probably is not though) this.allowsAuth = true; this.emit('connect'); } /** * Handles server response for STARTTLS command. If there's an error * try HELO instead, otherwise initiate TLS upgrade. If the upgrade * succeedes restart the EHLO * * @param {String} str Message from the server */ _actionSTARTTLS(str) { if (str.charAt(0) !== '2') { if (this.options.opportunisticTLS) { this.logger.info( { tnx: 'smtp' }, 'Failed STARTTLS upgrade, continuing unencrypted' ); return this.emit('connect'); } this._onError(new Error('Error upgrading connection with STARTTLS'), 'ETLS', str, 'STARTTLS'); return; } this._upgradeConnection((err, secured) => { if (err) { this._onError(new Error('Error initiating TLS - ' + (err.message || err)), 'ETLS', false, 'STARTTLS'); return; } this.logger.info( { tnx: 'smtp' }, 'Connection upgraded with STARTTLS' ); if (secured) { // restart session if (this.options.lmtp) { this._responseActions.push(this._actionLHLO); this._sendCommand('LHLO ' + this.name); } else { this._responseActions.push(this._actionEHLO); this._sendCommand('EHLO ' + this.name); } } else { this.emit('connect'); } }); } /** * Handle the response for AUTH LOGIN command. We are expecting * '334 VXNlcm5hbWU6' (base64 for 'Username:'). Data to be sent as * response needs to be base64 encoded username. We do not need * exact match but settle with 334 response in general as some * hosts invalidly use a longer message than VXNlcm5hbWU6 * * @param {String} str Message from the server */ _actionAUTH_LOGIN_USER(str, callback) { if (!/^334[ -]/.test(str)) { // expecting '334 VXNlcm5hbWU6' callback(this._formatError('Invalid login sequence while waiting for "334 VXNlcm5hbWU6"', 'EAUTH', str, 'AUTH LOGIN')); return; } this._responseActions.push(str => { this._actionAUTH_LOGIN_PASS(str, callback); }); this._sendCommand(Buffer.from(this._auth.credentials.user + '', 'utf-8').toString('base64')); } /** * Handle the response for AUTH CRAM-MD5 command. We are expecting * '334 '. Data to be sent as response needs to be * base64 decoded challenge string, MD5 hashed using the password as * a HMAC key, prefixed by the username and a space, and finally all * base64 encoded again. * * @param {String} str Message from the server */ _actionAUTH_CRAM_MD5(str, callback) { let challengeMatch = str.match(/^334\s+(.+)$/); let challengeString = ''; if (!challengeMatch) { return callback(this._formatError('Invalid login sequence while waiting for server challenge string', 'EAUTH', str, 'AUTH CRAM-MD5')); } else { challengeString = challengeMatch[1]; } // Decode from base64 let base64decoded = Buffer.from(challengeString, 'base64').toString('ascii'), hmac_md5 = crypto.createHmac('md5', this._auth.credentials.pass); hmac_md5.update(base64decoded); let hex_hmac = hmac_md5.digest('hex'); let prepended = this._auth.credentials.user + ' ' + hex_hmac; this._responseActions.push(str => { this._actionAUTH_CRAM_MD5_PASS(str, callback); }); this._sendCommand(Buffer.from(prepended).toString('base64')); } /** * Handles the response to CRAM-MD5 authentication, if there's no error, * the user can be considered logged in. Start waiting for a message to send * * @param {String} str Message from the server */ _actionAUTH_CRAM_MD5_PASS(str, callback) { if (!str.match(/^235\s+/)) { return callback(this._formatError('Invalid login sequence while waiting for "235"', 'EAUTH', str, 'AUTH CRAM-MD5')); } this.logger.info( { tnx: 'smtp', username: this._auth.user, action: 'authenticated', method: this._authMethod }, 'User %s authenticated', JSON.stringify(this._auth.user) ); this.authenticated = true; callback(null, true); } /** * Handle the response for AUTH LOGIN command. We are expecting * '334 UGFzc3dvcmQ6' (base64 for 'Password:'). Data to be sent as * response needs to be base64 encoded password. * * @param {String} str Message from the server */ _actionAUTH_LOGIN_PASS(str, callback) { if (!/^334[ -]/.test(str)) { // expecting '334 UGFzc3dvcmQ6' return callback(this._formatError('Invalid login sequence while waiting for "334 UGFzc3dvcmQ6"', 'EAUTH', str, 'AUTH LOGIN')); } this._responseActions.push(str => { this._actionAUTHComplete(str, callback); }); this._sendCommand(Buffer.from(this._auth.credentials.pass + '', 'utf-8').toString('base64')); } /** * Handles the response for authentication, if there's no error, * the user can be considered logged in. Start waiting for a message to send * * @param {String} str Message from the server */ _actionAUTHComplete(str, isRetry, callback) { if (!callback && typeof isRetry === 'function') { callback = isRetry; isRetry = false; } if (str.substr(0, 3) === '334') { this._responseActions.push(str => { if (isRetry || this._authMethod !== 'XOAUTH2') { this._actionAUTHComplete(str, true, callback); } else { // fetch a new OAuth2 access token setImmediate(() => this._handleXOauth2Token(true, callback)); } }); this._sendCommand(''); return; } if (str.charAt(0) !== '2') { this.logger.info( { tnx: 'smtp', username: this._auth.user, action: 'authfail', method: this._authMethod }, 'User %s failed to authenticate', JSON.stringify(this._auth.user) ); return callback(this._formatError('Invalid login', 'EAUTH', str, 'AUTH ' + this._authMethod)); } this.logger.info( { tnx: 'smtp', username: this._auth.user, action: 'authenticated', method: this._authMethod }, 'User %s authenticated', JSON.stringify(this._auth.user) ); this.authenticated = true; callback(null, true); } /** * Handle response for a MAIL FROM: command * * @param {String} str Message from the server */ _actionMAIL(str, callback) { let message, curRecipient; if (Number(str.charAt(0)) !== 2) { if (this._usingSmtpUtf8 && /^550 /.test(str) && /[\x80-\uFFFF]/.test(this._envelope.from)) { message = 'Internationalized mailbox name not allowed'; } else { message = 'Mail command failed'; } return callback(this._formatError(message, 'EENVELOPE', str, 'MAIL FROM')); } if (!this._envelope.rcptQueue.length) { return callback(this._formatError('Can\x27t send mail - no recipients defined', 'EENVELOPE', false, 'API')); } else { this._recipientQueue = []; if (this._supportedExtensions.includes('PIPELINING')) { while (this._envelope.rcptQueue.length) { curRecipient = this._envelope.rcptQueue.shift(); this._recipientQueue.push(curRecipient); this._responseActions.push(str => { this._actionRCPT(str, callback); }); this._sendCommand('RCPT TO:<' + curRecipient + '>' + this._getDsnRcptToArgs()); } } else { curRecipient = this._envelope.rcptQueue.shift(); this._recipientQueue.push(curRecipient); this._responseActions.push(str => { this._actionRCPT(str, callback); }); this._sendCommand('RCPT TO:<' + curRecipient + '>' + this._getDsnRcptToArgs()); } } } /** * Handle response for a RCPT TO: command * * @param {String} str Message from the server */ _actionRCPT(str, callback) { let message, err, curRecipient = this._recipientQueue.shift(); if (Number(str.charAt(0)) !== 2) { // this is a soft error if (this._usingSmtpUtf8 && /^553 /.test(str) && /[\x80-\uFFFF]/.test(curRecipient)) { message = 'Internationalized mailbox name not allowed'; } else { message = 'Recipient command failed'; } this._envelope.rejected.push(curRecipient); // store error for the failed recipient err = this._formatError(message, 'EENVELOPE', str, 'RCPT TO'); err.recipient = curRecipient; this._envelope.rejectedErrors.push(err); } else { this._envelope.accepted.push(curRecipient); } if (!this._envelope.rcptQueue.length && !this._recipientQueue.length) { if (this._envelope.rejected.length < this._envelope.to.length) { this._responseActions.push(str => { this._actionDATA(str, callback); }); this._sendCommand('DATA'); } else { err = this._formatError('Can\x27t send mail - all recipients were rejected', 'EENVELOPE', str, 'RCPT TO'); err.rejected = this._envelope.rejected; err.rejectedErrors = this._envelope.rejectedErrors; return callback(err); } } else if (this._envelope.rcptQueue.length) { curRecipient = this._envelope.rcptQueue.shift(); this._recipientQueue.push(curRecipient); this._responseActions.push(str => { this._actionRCPT(str, callback); }); this._sendCommand('RCPT TO:<' + curRecipient + '>' + this._getDsnRcptToArgs()); } } /** * Handle response for a DATA command * * @param {String} str Message from the server */ _actionDATA(str, callback) { // response should be 354 but according to this issue https://github.com/eleith/emailjs/issues/24 // some servers might use 250 instead, so lets check for 2 or 3 as the first digit if (!/^[23]/.test(str)) { return callback(this._formatError('Data command failed', 'EENVELOPE', str, 'DATA')); } let response = { accepted: this._envelope.accepted, rejected: this._envelope.rejected }; if (this._envelope.rejectedErrors.length) { response.rejectedErrors = this._envelope.rejectedErrors; } callback(null, response); } /** * Handle response for a DATA stream when using SMTP * We expect a single response that defines if the sending succeeded or failed * * @param {String} str Message from the server */ _actionSMTPStream(str, callback) { if (Number(str.charAt(0)) !== 2) { // Message failed return callback(this._formatError('Message failed', 'EMESSAGE', str, 'DATA')); } else { // Message sent succesfully return callback(null, str); } } /** * Handle response for a DATA stream * We expect a separate response for every recipient. All recipients can either * succeed or fail separately * * @param {String} recipient The recipient this response applies to * @param {Boolean} final Is this the final recipient? * @param {String} str Message from the server */ _actionLMTPStream(recipient, final, str, callback) { let err; if (Number(str.charAt(0)) !== 2) { // Message failed err = this._formatError('Message failed for recipient ' + recipient, 'EMESSAGE', str, 'DATA'); err.recipient = recipient; this._envelope.rejected.push(recipient); this._envelope.rejectedErrors.push(err); for (let i = 0, len = this._envelope.accepted.length; i < len; i++) { if (this._envelope.accepted[i] === recipient) { this._envelope.accepted.splice(i, 1); } } } if (final) { return callback(null, str); } } _handleXOauth2Token(isRetry, callback) { this._auth.oauth2.getToken(isRetry, (err, accessToken) => { if (err) { this.logger.info( { tnx: 'smtp', username: this._auth.user, action: 'authfail', method: this._authMethod }, 'User %s failed to authenticate', JSON.stringify(this._auth.user) ); return callback(this._formatError(err, 'EAUTH', false, 'AUTH XOAUTH2')); } this._responseActions.push(str => { this._actionAUTHComplete(str, isRetry, callback); }); this._sendCommand('AUTH XOAUTH2 ' + this._auth.oauth2.buildXOAuth2Token(accessToken)); }); } /** * * @param {string} command * @private */ _isDestroyedMessage(command) { if (this._destroyed) { return 'Cannot ' + command + ' - smtp connection is already destroyed.'; } if (this._socket) { if (this._socket.destroyed) { return 'Cannot ' + command + ' - smtp connection socket is already destroyed.'; } if (!this._socket.writable) { return 'Cannot ' + command + ' - smtp connection socket is already half-closed.'; } } } _getHostname() { // defaul hostname is machine hostname or [IP] let defaultHostname = os.hostname() || ''; // ignore if not FQDN if (defaultHostname.indexOf('.') < 0) { defaultHostname = '[127.0.0.1]'; } // IP should be enclosed in [] if (defaultHostname.match(/^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$/)) { defaultHostname = '[' + defaultHostname + ']'; } return defaultHostname; } } module.exports = SMTPConnection;