import * as plugins from '../../plugins.js'; import { logger } from '../../logger.js'; import { SecurityLogger, SecurityLogLevel, SecurityEventType } from '../../security/index.js'; import { MtaConnectionError, MtaAuthenticationError, MtaDeliveryError, MtaConfigurationError, MtaTimeoutError, MtaProtocolError } from '../../errors/index.js'; import { Email } from '../core/classes.email.js'; import type { EmailProcessingMode } from './interfaces.js'; // Custom error type extension interface NodeNetworkError extends Error { code?: string; } /** * SMTP client connection options */ export type ISmtpClientOptions = { /** * Hostname of the SMTP server */ host: string; /** * Port to connect to */ port: number; /** * Whether to use TLS for the connection */ secure?: boolean; /** * Connection timeout in milliseconds */ connectionTimeout?: number; /** * Socket timeout in milliseconds */ socketTimeout?: number; /** * Command timeout in milliseconds */ commandTimeout?: number; /** * TLS options */ tls?: { /** * Whether to verify certificates */ rejectUnauthorized?: boolean; /** * Minimum TLS version */ minVersion?: string; /** * CA certificate path */ ca?: string; }; /** * Authentication options */ auth?: { /** * Authentication user */ user: string; /** * Authentication password */ pass: string; /** * Authentication method */ method?: 'PLAIN' | 'LOGIN' | 'OAUTH2'; }; /** * Domain name for EHLO */ domain?: string; /** * DKIM options for signing outgoing emails */ dkim?: { /** * Whether to sign emails with DKIM */ enabled: boolean; /** * Domain name for DKIM */ domain: string; /** * Selector for DKIM */ selector: string; /** * Private key for DKIM signing */ privateKey: string; /** * Headers to sign */ headers?: string[]; }; }; /** * SMTP delivery result */ export type ISmtpDeliveryResult = { /** * Whether the delivery was successful */ success: boolean; /** * Message ID if successful */ messageId?: string; /** * Error message if failed */ error?: string; /** * SMTP response code */ responseCode?: string; /** * Recipients successfully delivered to */ acceptedRecipients: string[]; /** * Recipients rejected during delivery */ rejectedRecipients: string[]; /** * Server response */ response?: string; /** * Timestamp of the delivery attempt */ timestamp: number; /** * Whether DKIM signing was applied */ dkimSigned?: boolean; /** * Whether this was a TLS secured delivery */ secure?: boolean; /** * Whether authentication was used */ authenticated?: boolean; }; /** * SMTP client for sending emails to remote mail servers */ export class SmtpClient { private options: ISmtpClientOptions; private connected: boolean = false; private socket?: plugins.net.Socket | plugins.tls.TLSSocket; private supportedExtensions: Set = new Set(); /** * Create a new SMTP client instance * @param options SMTP client connection options */ constructor(options: ISmtpClientOptions) { // Set default options this.options = { ...options, connectionTimeout: options.connectionTimeout || 30000, // 30 seconds socketTimeout: options.socketTimeout || 60000, // 60 seconds commandTimeout: options.commandTimeout || 30000, // 30 seconds secure: options.secure || false, domain: options.domain || 'localhost', tls: { rejectUnauthorized: options.tls?.rejectUnauthorized !== false, // Default to true minVersion: options.tls?.minVersion || 'TLSv1.2' } }; } /** * Connect to the SMTP server */ public async connect(): Promise { if (this.connected && this.socket) { return; } try { logger.log('info', `Connecting to SMTP server ${this.options.host}:${this.options.port}`); // Create socket const socket = new plugins.net.Socket(); // Set timeouts socket.setTimeout(this.options.socketTimeout); // Connect to the server await new Promise((resolve, reject) => { // Handle connection events socket.once('connect', () => { logger.log('debug', `Connected to ${this.options.host}:${this.options.port}`); resolve(); }); socket.once('timeout', () => { reject(MtaConnectionError.timeout( this.options.host, this.options.port, this.options.connectionTimeout )); }); socket.once('error', (err: NodeNetworkError) => { if (err.code === 'ECONNREFUSED') { reject(MtaConnectionError.refused( this.options.host, this.options.port )); } else if (err.code === 'ENOTFOUND') { reject(MtaConnectionError.dnsError( this.options.host, err )); } else { reject(new MtaConnectionError( `Connection error to ${this.options.host}:${this.options.port}: ${err.message}`, { data: { host: this.options.host, port: this.options.port, error: err.message, code: err.code } } )); } }); // Connect to the server const connectOptions = { host: this.options.host, port: this.options.port }; // For direct TLS connections if (this.options.secure) { const tlsSocket = plugins.tls.connect({ ...connectOptions, rejectUnauthorized: this.options.tls.rejectUnauthorized, minVersion: this.options.tls.minVersion as any, ca: this.options.tls.ca ? [this.options.tls.ca] : undefined } as plugins.tls.ConnectionOptions); tlsSocket.once('secureConnect', () => { logger.log('debug', `Secure connection established to ${this.options.host}:${this.options.port}`); this.socket = tlsSocket; resolve(); }); tlsSocket.once('error', (err: NodeNetworkError) => { reject(new MtaConnectionError( `TLS connection error to ${this.options.host}:${this.options.port}: ${err.message}`, { data: { host: this.options.host, port: this.options.port, error: err.message, code: err.code } } )); }); tlsSocket.setTimeout(this.options.socketTimeout); tlsSocket.once('timeout', () => { reject(MtaConnectionError.timeout( this.options.host, this.options.port, this.options.connectionTimeout )); }); } else { socket.connect(connectOptions); this.socket = socket; } }); // Wait for server greeting const greeting = await this.readResponse(); if (!greeting.startsWith('220')) { throw new MtaConnectionError( `Unexpected greeting from server: ${greeting}`, { data: { host: this.options.host, port: this.options.port, greeting } } ); } // Send EHLO await this.sendEhlo(); // Start TLS if not secure and supported if (!this.options.secure && this.supportedExtensions.has('STARTTLS')) { await this.startTls(); // Send EHLO again after STARTTLS await this.sendEhlo(); } // Authenticate if credentials provided if (this.options.auth) { await this.authenticate(); } this.connected = true; logger.log('info', `Successfully connected to SMTP server ${this.options.host}:${this.options.port}`); // Set up error handling for the socket this.socket.on('error', (err) => { logger.log('error', `Socket error: ${err.message}`); this.connected = false; this.socket = undefined; }); this.socket.on('close', () => { logger.log('debug', 'Socket closed'); this.connected = false; this.socket = undefined; }); this.socket.on('timeout', () => { logger.log('error', 'Socket timeout'); this.connected = false; if (this.socket) { this.socket.destroy(); this.socket = undefined; } }); } catch (error) { // Clean up socket if connection failed if (this.socket) { this.socket.destroy(); this.socket = undefined; } logger.log('error', `Failed to connect to SMTP server: ${error.message}`); throw error; } } /** * Send EHLO command to the server */ private async sendEhlo(): Promise { // Clear previous extensions this.supportedExtensions.clear(); // Send EHLO - don't allow pipelining for this command const response = await this.sendCommand(`EHLO ${this.options.domain}`, false); // Parse supported extensions const lines = response.split('\r\n'); for (let i = 1; i < lines.length; i++) { const line = lines[i]; if (line.startsWith('250-') || line.startsWith('250 ')) { const extension = line.substring(4).split(' ')[0]; this.supportedExtensions.add(extension); } } // Check if server supports pipelining this.supportsPipelining = this.supportedExtensions.has('PIPELINING'); logger.log('debug', `Server supports extensions: ${Array.from(this.supportedExtensions).join(', ')}`); if (this.supportsPipelining) { logger.log('info', 'Server supports PIPELINING - will use for improved performance'); } } /** * Start TLS negotiation */ private async startTls(): Promise { logger.log('debug', 'Starting TLS negotiation'); // Send STARTTLS command const response = await this.sendCommand('STARTTLS'); if (!response.startsWith('220')) { throw new MtaConnectionError( `Failed to start TLS: ${response}`, { data: { host: this.options.host, port: this.options.port, response } } ); } if (!this.socket) { throw new MtaConnectionError( 'No socket available for TLS upgrade', { data: { host: this.options.host, port: this.options.port } } ); } // Upgrade socket to TLS const currentSocket = this.socket; this.socket = await this.upgradeTls(currentSocket); } /** * Upgrade socket to TLS * @param socket Original socket */ private async upgradeTls(socket: plugins.net.Socket): Promise { return new Promise((resolve, reject) => { const tlsOptions: plugins.tls.ConnectionOptions = { socket, servername: this.options.host, rejectUnauthorized: this.options.tls.rejectUnauthorized, minVersion: this.options.tls.minVersion as any, ca: this.options.tls.ca ? [this.options.tls.ca] : undefined }; const tlsSocket = plugins.tls.connect(tlsOptions); tlsSocket.once('secureConnect', () => { logger.log('debug', 'TLS negotiation successful'); resolve(tlsSocket); }); tlsSocket.once('error', (err: NodeNetworkError) => { reject(new MtaConnectionError( `TLS error: ${err.message}`, { data: { host: this.options.host, port: this.options.port, error: err.message, code: err.code } } )); }); tlsSocket.setTimeout(this.options.socketTimeout); tlsSocket.once('timeout', () => { reject(MtaTimeoutError.commandTimeout( 'STARTTLS', this.options.host, this.options.socketTimeout )); }); }); } /** * Authenticate with the server */ private async authenticate(): Promise { if (!this.options.auth) { return; } const { user, pass, method = 'LOGIN' } = this.options.auth; logger.log('debug', `Authenticating as ${user} using ${method}`); try { switch (method) { case 'PLAIN': await this.authPlain(user, pass); break; case 'LOGIN': await this.authLogin(user, pass); break; case 'OAUTH2': await this.authOAuth2(user, pass); break; default: throw new MtaAuthenticationError( `Authentication method ${method} not supported by client`, { data: { method } } ); } logger.log('info', `Successfully authenticated as ${user}`); } catch (error) { logger.log('error', `Authentication failed: ${error.message}`); throw error; } } /** * Authenticate using PLAIN method * @param user Username * @param pass Password */ private async authPlain(user: string, pass: string): Promise { // PLAIN authentication format: \0username\0password const authString = Buffer.from(`\0${user}\0${pass}`).toString('base64'); const response = await this.sendCommand(`AUTH PLAIN ${authString}`); if (!response.startsWith('235')) { throw MtaAuthenticationError.invalidCredentials( this.options.host, user ); } } /** * Authenticate using LOGIN method * @param user Username * @param pass Password */ private async authLogin(user: string, pass: string): Promise { // Start LOGIN authentication const response = await this.sendCommand('AUTH LOGIN'); if (!response.startsWith('334')) { throw new MtaAuthenticationError( `Server did not accept AUTH LOGIN: ${response}`, { data: { host: this.options.host, response } } ); } // Send username (base64) const userResponse = await this.sendCommand(Buffer.from(user).toString('base64')); if (!userResponse.startsWith('334')) { throw MtaAuthenticationError.invalidCredentials( this.options.host, user ); } // Send password (base64) const passResponse = await this.sendCommand(Buffer.from(pass).toString('base64')); if (!passResponse.startsWith('235')) { throw MtaAuthenticationError.invalidCredentials( this.options.host, user ); } } /** * Authenticate using OAuth2 method * @param user Username * @param token OAuth2 token */ private async authOAuth2(user: string, token: string): Promise { // XOAUTH2 format const authString = `user=${user}\x01auth=Bearer ${token}\x01\x01`; const response = await this.sendCommand(`AUTH XOAUTH2 ${Buffer.from(authString).toString('base64')}`); if (!response.startsWith('235')) { throw MtaAuthenticationError.invalidCredentials( this.options.host, user ); } } /** * Send an email through the SMTP client * @param email Email to send * @param processingMode Optional processing mode */ public async sendMail(email: Email, processingMode?: EmailProcessingMode): Promise { // Ensure we're connected if (!this.connected || !this.socket) { await this.connect(); } const startTime = Date.now(); const result: ISmtpDeliveryResult = { success: false, acceptedRecipients: [], rejectedRecipients: [], timestamp: startTime, secure: this.options.secure || this.socket instanceof plugins.tls.TLSSocket, authenticated: !!this.options.auth }; try { logger.log('info', `Sending email to ${email.getAllRecipients().join(', ')}`); // Apply DKIM signing if configured if (this.options.dkim?.enabled) { await this.applyDkimSignature(email); result.dkimSigned = true; } // Get envelope and recipients const envelope_from = email.getEnvelopeFrom() || email.from; const recipients = email.getAllRecipients(); // Check if we can use pipelining for MAIL FROM and RCPT TO commands if (this.supportsPipelining && recipients.length > 0) { logger.log('debug', 'Using SMTP pipelining for sending'); // Send MAIL FROM command first (always needed) const mailFromCmd = `MAIL FROM:<${envelope_from}> SIZE=${this.getEmailSize(email)}`; let mailFromResponse: string; try { mailFromResponse = await this.sendCommand(mailFromCmd); if (!mailFromResponse.startsWith('250')) { throw new MtaDeliveryError( `MAIL FROM command failed: ${mailFromResponse}`, { data: { command: mailFromCmd, response: mailFromResponse } } ); } } catch (error) { logger.log('error', `MAIL FROM failed: ${error.message}`); throw error; } // Pipeline all RCPT TO commands const rcptPromises = recipients.map(recipient => { return this.sendCommand(`RCPT TO:<${recipient}>`) .then(response => { if (response.startsWith('250')) { result.acceptedRecipients.push(recipient); return { recipient, accepted: true, response }; } else { result.rejectedRecipients.push(recipient); logger.log('warn', `Recipient ${recipient} rejected: ${response}`); return { recipient, accepted: false, response }; } }) .catch(error => { result.rejectedRecipients.push(recipient); logger.log('warn', `Recipient ${recipient} rejected with error: ${error.message}`); return { recipient, accepted: false, error: error.message }; }); }); // Wait for all RCPT TO commands to complete await Promise.all(rcptPromises); } else { // Fall back to sequential commands if pipelining not supported logger.log('debug', 'Using sequential SMTP commands for sending'); // Send MAIL FROM await this.sendCommand(`MAIL FROM:<${envelope_from}> SIZE=${this.getEmailSize(email)}`); // Send RCPT TO for each recipient for (const recipient of recipients) { try { await this.sendCommand(`RCPT TO:<${recipient}>`); result.acceptedRecipients.push(recipient); } catch (error) { logger.log('warn', `Recipient ${recipient} rejected: ${error.message}`); result.rejectedRecipients.push(recipient); } } } // Check if at least one recipient was accepted if (result.acceptedRecipients.length === 0) { throw new MtaDeliveryError( 'All recipients were rejected', { data: { recipients, rejectedRecipients: result.rejectedRecipients } } ); } // Send DATA const dataResponse = await this.sendCommand('DATA'); if (!dataResponse.startsWith('354')) { throw new MtaProtocolError( `Failed to start DATA phase: ${dataResponse}`, { data: { response: dataResponse } } ); } // Format email content efficiently const emailContent = await this.getFormattedEmail(email); // Send email content const finalResponse = await this.sendCommand(emailContent + '\r\n.'); // Extract message ID if available const messageIdMatch = finalResponse.match(/\[(.*?)\]/); if (messageIdMatch) { result.messageId = messageIdMatch[1]; } result.success = true; result.response = finalResponse; logger.log('info', `Email sent successfully to ${result.acceptedRecipients.join(', ')}`); // Log security event SecurityLogger.getInstance().logEvent({ level: SecurityLogLevel.INFO, type: SecurityEventType.EMAIL_DELIVERY, message: 'Email sent successfully', details: { recipients: result.acceptedRecipients, rejectedRecipients: result.rejectedRecipients, messageId: result.messageId, secure: result.secure, authenticated: result.authenticated, server: `${this.options.host}:${this.options.port}`, dkimSigned: result.dkimSigned }, success: true }); return result; } catch (error) { logger.log('error', `Failed to send email: ${error.message}`); // Format error for result result.error = error.message; // Extract SMTP code if available if (error.context?.data?.statusCode) { result.responseCode = error.context.data.statusCode; } // Log security event SecurityLogger.getInstance().logEvent({ level: SecurityLogLevel.ERROR, type: SecurityEventType.EMAIL_DELIVERY, message: 'Email delivery failed', details: { error: error.message, server: `${this.options.host}:${this.options.port}`, recipients: email.getAllRecipients(), acceptedRecipients: result.acceptedRecipients, rejectedRecipients: result.rejectedRecipients, secure: result.secure, authenticated: result.authenticated }, success: false }); return result; } } /** * Apply DKIM signature to email * @param email Email to sign */ private async applyDkimSignature(email: Email): Promise { if (!this.options.dkim?.enabled || !this.options.dkim?.privateKey) { return; } try { logger.log('debug', `Signing email with DKIM for domain ${this.options.dkim.domain}`); // Format email for DKIM signing const { dkimSign } = plugins; const emailContent = await this.getFormattedEmail(email); // Sign email const signOptions = { domainName: this.options.dkim.domain, keySelector: this.options.dkim.selector, privateKey: this.options.dkim.privateKey, headerFieldNames: this.options.dkim.headers || [ 'from', 'to', 'subject', 'date', 'message-id' ] }; const signedEmail = await dkimSign(emailContent, signOptions); // Replace headers in original email const dkimHeader = signedEmail.substring(0, signedEmail.indexOf('\r\n\r\n')).split('\r\n') .find(line => line.startsWith('DKIM-Signature: ')); if (dkimHeader) { email.addHeader('DKIM-Signature', dkimHeader.substring('DKIM-Signature: '.length)); } logger.log('debug', 'DKIM signature applied successfully'); } catch (error) { logger.log('error', `Failed to apply DKIM signature: ${error.message}`); throw error; } } /** * Format email for SMTP transmission * @param email Email to format */ private async getFormattedEmail(email: Email): Promise { // This is a simplified implementation // In a full implementation, this would use proper MIME formatting let content = ''; // Add headers content += `From: ${email.from}\r\n`; content += `To: ${email.to.join(', ')}\r\n`; content += `Subject: ${email.subject}\r\n`; content += `Date: ${new Date().toUTCString()}\r\n`; content += `Message-ID: <${plugins.uuid.v4()}@${this.options.domain}>\r\n`; // Add additional headers for (const [name, value] of Object.entries(email.headers || {})) { content += `${name}: ${value}\r\n`; } // Add content type for multipart if (email.attachments && email.attachments.length > 0) { const boundary = `----_=_NextPart_${Math.random().toString(36).substr(2)}`; content += `MIME-Version: 1.0\r\n`; content += `Content-Type: multipart/mixed; boundary="${boundary}"\r\n`; content += `\r\n`; // Add text part content += `--${boundary}\r\n`; content += `Content-Type: text/plain; charset="UTF-8"\r\n`; content += `\r\n`; content += `${email.text}\r\n`; // Add HTML part if present if (email.html) { content += `--${boundary}\r\n`; content += `Content-Type: text/html; charset="UTF-8"\r\n`; content += `\r\n`; content += `${email.html}\r\n`; } // Add attachments for (const attachment of email.attachments) { content += `--${boundary}\r\n`; content += `Content-Type: ${attachment.contentType || 'application/octet-stream'}; name="${attachment.filename}"\r\n`; content += `Content-Disposition: attachment; filename="${attachment.filename}"\r\n`; content += `Content-Transfer-Encoding: base64\r\n`; content += `\r\n`; // Add base64 encoded content const base64Content = attachment.content.toString('base64'); // Split into lines of 76 characters for (let i = 0; i < base64Content.length; i += 76) { content += base64Content.substring(i, i + 76) + '\r\n'; } } // End boundary content += `--${boundary}--\r\n`; } else { // Simple email with just text content += `Content-Type: text/plain; charset="UTF-8"\r\n`; content += `\r\n`; content += `${email.text}\r\n`; } return content; } /** * Get size of email in bytes * @param email Email to measure */ private getEmailSize(email: Email): number { // Simplified size estimation let size = 0; // Headers size += `From: ${email.from}\r\n`.length; size += `To: ${email.to.join(', ')}\r\n`.length; size += `Subject: ${email.subject}\r\n`.length; // Body size += (email.text?.length || 0) + 2; // +2 for CRLF // HTML part if present if (email.html) { size += email.html.length + 2; } // Attachments for (const attachment of email.attachments || []) { size += attachment.content.length; } // Add overhead for MIME boundaries and headers const overhead = email.attachments?.length ? 1000 + (email.attachments.length * 200) : 200; return size + overhead; } /** * Send SMTP command and wait for response * @param command SMTP command to send */ // Queue for command pipelining private commandQueue: Array<{ command: string; resolve: (response: string) => void; reject: (error: any) => void; timeout: NodeJS.Timeout; }> = []; // Flag to indicate if we're currently processing commands private processingCommands = false; // Flag to indicate if server supports pipelining private supportsPipelining = false; /** * Send an SMTP command and wait for response * @param command SMTP command to send * @param allowPipelining Whether this command can be pipelined */ private async sendCommand(command: string, allowPipelining = true): Promise { if (!this.socket) { throw new MtaConnectionError( 'Not connected to server', { data: { host: this.options.host, port: this.options.port } } ); } // Log command if not sensitive if (!command.startsWith('AUTH')) { logger.log('debug', `> ${command}`); } else { logger.log('debug', '> AUTH ***'); } return new Promise((resolve, reject) => { // Set up timeout for command const timeout = setTimeout(() => { // Remove this command from the queue if it times out const index = this.commandQueue.findIndex(item => item.command === command); if (index !== -1) { this.commandQueue.splice(index, 1); } reject(MtaTimeoutError.commandTimeout( command.split(' ')[0], this.options.host, this.options.commandTimeout )); }, this.options.commandTimeout); // Add command to the queue this.commandQueue.push({ command, resolve, reject, timeout }); // Process command queue if we can pipeline or if not currently processing commands if ((this.supportsPipelining && allowPipelining) || !this.processingCommands) { this.processCommandQueue(); } }); } /** * Process the command queue - either one by one or pipelined if supported */ private processCommandQueue(): void { if (this.processingCommands || this.commandQueue.length === 0 || !this.socket) { return; } this.processingCommands = true; try { // If pipelining is supported, send all commands at once if (this.supportsPipelining) { // Send all commands in queue at once const commands = this.commandQueue.map(item => item.command).join('\r\n') + '\r\n'; this.socket.write(commands, (err) => { if (err) { // Handle write error for all commands const error = new MtaConnectionError( `Failed to send commands: ${err.message}`, { data: { error: err.message } } ); // Fail all pending commands while (this.commandQueue.length > 0) { const item = this.commandQueue.shift(); clearTimeout(item.timeout); item.reject(error); } this.processingCommands = false; } }); // Process responses one by one in order this.processResponses(); } else { // Process commands one by one if pipelining not supported this.processNextCommand(); } } catch (error) { logger.log('error', `Error processing command queue: ${error.message}`); this.processingCommands = false; } } /** * Process the next command in the queue (non-pipelined mode) */ private processNextCommand(): void { if (this.commandQueue.length === 0 || !this.socket) { this.processingCommands = false; return; } const currentCommand = this.commandQueue[0]; this.socket.write(currentCommand.command + '\r\n', (err) => { if (err) { // Handle write error const error = new MtaConnectionError( `Failed to send command: ${err.message}`, { data: { command: currentCommand.command.split(' ')[0], error: err.message } } ); // Remove from queue this.commandQueue.shift(); clearTimeout(currentCommand.timeout); currentCommand.reject(error); // Continue with next command this.processNextCommand(); return; } // Read response this.readResponse() .then((response) => { // Remove from queue and resolve this.commandQueue.shift(); clearTimeout(currentCommand.timeout); currentCommand.resolve(response); // Process next command this.processNextCommand(); }) .catch((err) => { // Remove from queue and reject this.commandQueue.shift(); clearTimeout(currentCommand.timeout); currentCommand.reject(err); // Process next command this.processNextCommand(); }); }); } /** * Process responses for pipelined commands */ private async processResponses(): Promise { try { // Process responses for each command in order while (this.commandQueue.length > 0) { const currentCommand = this.commandQueue[0]; try { // Wait for response const response = await this.readResponse(); // Remove from queue and resolve this.commandQueue.shift(); clearTimeout(currentCommand.timeout); currentCommand.resolve(response); } catch (error) { // Remove from queue and reject this.commandQueue.shift(); clearTimeout(currentCommand.timeout); currentCommand.reject(error); // Stop processing if this is a critical error if ( error instanceof MtaConnectionError && (error.message.includes('Connection closed') || error.message.includes('Not connected')) ) { break; } } } } catch (error) { logger.log('error', `Error processing responses: ${error.message}`); } finally { this.processingCommands = false; } } /** * Read response from the server */ private async readResponse(): Promise { if (!this.socket) { throw new MtaConnectionError( 'Not connected to server', { data: { host: this.options.host, port: this.options.port } } ); } return new Promise((resolve, reject) => { // Use an array to collect response chunks instead of string concatenation const responseChunks: Buffer[] = []; // Single function to clean up all listeners const cleanupListeners = () => { if (!this.socket) return; this.socket.removeListener('data', onData); this.socket.removeListener('error', onError); this.socket.removeListener('close', onClose); this.socket.removeListener('end', onEnd); }; const onData = (data: Buffer) => { // Store buffer directly, avoiding unnecessary string conversion responseChunks.push(data); // Convert to string only for response checking const responseData = Buffer.concat(responseChunks).toString(); // Check if this is a complete response if (this.isCompleteResponse(responseData)) { // Clean up listeners cleanupListeners(); const trimmedResponse = responseData.trim(); logger.log('debug', `< ${trimmedResponse}`); // Check if this is an error response if (this.isErrorResponse(responseData)) { const code = responseData.substring(0, 3); reject(this.createErrorFromResponse(trimmedResponse, code)); } else { resolve(trimmedResponse); } } }; const onError = (err: Error) => { cleanupListeners(); reject(new MtaConnectionError( `Socket error while waiting for response: ${err.message}`, { data: { error: err.message } } )); }; const onClose = () => { cleanupListeners(); const responseData = Buffer.concat(responseChunks).toString(); reject(new MtaConnectionError( 'Connection closed while waiting for response', { data: { partialResponse: responseData } } )); }; const onEnd = () => { cleanupListeners(); const responseData = Buffer.concat(responseChunks).toString(); reject(new MtaConnectionError( 'Connection ended while waiting for response', { data: { partialResponse: responseData } } )); }; // Set up listeners this.socket.on('data', onData); this.socket.once('error', onError); this.socket.once('close', onClose); this.socket.once('end', onEnd); }); } /** * Check if the response is complete * @param response Response to check */ private isCompleteResponse(response: string): boolean { // Check if it's a multi-line response const lines = response.split('\r\n'); const lastLine = lines[lines.length - 2]; // Second to last because of the trailing CRLF // Check if the last line starts with a code followed by a space // If it does, this is a complete response if (lastLine && /^\d{3} /.test(lastLine)) { return true; } // For single line responses if (lines.length === 2 && lines[0].length >= 3 && /^\d{3} /.test(lines[0])) { return true; } return false; } /** * Check if the response is an error * @param response Response to check */ private isErrorResponse(response: string): boolean { // Get the status code (first 3 characters) const code = response.substring(0, 3); // 4xx and 5xx are error codes return code.startsWith('4') || code.startsWith('5'); } /** * Create appropriate error from response * @param response Error response * @param code SMTP status code */ private createErrorFromResponse(response: string, code: string): Error { // Extract message part const message = response.substring(4).trim(); switch (code.charAt(0)) { case '4': // Temporary errors return MtaDeliveryError.temporary( message, 'recipient', code, response ); case '5': // Permanent errors return MtaDeliveryError.permanent( message, 'recipient', code, response ); default: return new MtaDeliveryError( `Unexpected error response: ${response}`, { data: { response, code } } ); } } /** * Close the connection to the server */ public async close(): Promise { if (!this.connected || !this.socket) { return; } try { // Send QUIT await this.sendCommand('QUIT'); } catch (error) { logger.log('warn', `Error sending QUIT command: ${error.message}`); } finally { // Close socket this.socket.destroy(); this.socket = undefined; this.connected = false; logger.log('info', 'SMTP connection closed'); } } /** * Checks if the connection is active */ public isConnected(): boolean { return this.connected && !!this.socket; } /** * Update SMTP client options * @param options New options */ public updateOptions(options: Partial): void { this.options = { ...this.options, ...options }; logger.log('info', 'SMTP client options updated'); } }