push.rocks/smartmta
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

BREAKING CHANGE(security): implement resilience and lifecycle management for RustSecurityBridge (auto-restart, health checks, state machine and eventing); remove legacy TS SMTP test helper and DNSManager; remove deliverability IP-warmup/sender-reputation integrations and related types; drop unused dependencies

Browse Source
This commit is contained in:
Jürgen Kunz
2026-02-10 23:23:00 +00:00
parent d43fc15d8e
commit 36006191fc
13 changed files with 495 additions and 2778 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ts/00_commitinfo_data.ts
View File

@@ -3,6 +3,6 @@
*/
export const commitinfo = {
name: '@push.rocks/smartmta',
version: '2.4.0',
version: '3.0.0',
description: 'A high-performance, enterprise-grade Mail Transfer Agent (MTA) built from scratch in TypeScript with Rust acceleration.'
}

22
ts/mail/delivery/classes.emailsendjob.ts
View File

@@ -225,28 +225,6 @@ export class EmailSendJob {
this.log(`Connecting to ${mxServer}:25`);
try {
// Check if IP warmup is enabled and get an IP to use
let localAddress: string | undefined = undefined;
try {
const fromDomain = this.email.getFromDomain();
const bestIP = this.emailServerRef.getBestIPForSending({
from: this.email.from,
to: this.email.getAllRecipients(),
domain: fromDomain,
isTransactional: this.email.priority === 'high'
});
if (bestIP) {
this.log(`Using warmed-up IP ${bestIP} for sending`);
localAddress = bestIP;
// Record the send for warm-up tracking
this.emailServerRef.recordIPSend(bestIP);
}
} catch (error) {
this.log(`Error selecting IP address: ${error.message}`);
}
// Get SMTP client from UnifiedEmailServer
const smtpClient = this.emailServerRef.getSmtpClient(mxServer, 25);

1414
ts/mail/delivery/classes.smtp.client.legacy.ts
View File

File diff suppressed because it is too large Load Diff

124
ts/mail/delivery/interfaces.ts
View File

@@ -2,8 +2,6 @@
* SMTP and email delivery interface definitions
*/
import type { Email } from '../core/classes.email.js';
/**
* SMTP session state enumeration
*/
@@ -167,125 +165,3 @@ export interface ISmtpAuth {
password: string;
}
/**
* SMTP server options
*/
export interface ISmtpServerOptions {
/**
* Port to listen on
*/
port: number;
/**
* TLS private key (PEM format)
*/
key: string;
/**
* TLS certificate (PEM format)
*/
cert: string;
/**
* Server hostname for SMTP banner
*/
hostname?: string;
/**
* Host address to bind to (defaults to all interfaces)
*/
host?: string;
/**
* Secure port for dedicated TLS connections
*/
securePort?: number;
/**
* CA certificates for TLS (PEM format)
*/
ca?: string;
/**
* Maximum size of messages in bytes
*/
maxSize?: number;
/**
* Maximum number of concurrent connections
*/
maxConnections?: number;
/**
* Authentication options
*/
auth?: {
/**
* Whether authentication is required
*/
required: boolean;
/**
* Allowed authentication methods
*/
methods: ('PLAIN' | 'LOGIN' | 'OAUTH2')[];
};
/**
* Socket timeout in milliseconds (default: 5 minutes / 300000ms)
*/
socketTimeout?: number;
/**
* Initial connection timeout in milliseconds (default: 30 seconds / 30000ms)
*/
connectionTimeout?: number;
/**
* Interval for checking idle sessions in milliseconds (default: 5 seconds / 5000ms)
* For testing, can be set lower (e.g. 1000ms) to detect timeouts more quickly
*/
cleanupInterval?: number;
/**
* Maximum number of recipients allowed per message (default: 100)
*/
maxRecipients?: number;
/**
* Maximum message size in bytes (default: 10MB / 10485760 bytes)
* This is advertised in the EHLO SIZE extension
*/
size?: number;
/**
* Timeout for the DATA command in milliseconds (default: 60000ms / 1 minute)
* This controls how long to wait for the complete email data
*/
dataTimeout?: number;
}
/**
* Result of SMTP transaction
*/
export interface ISmtpTransactionResult {
/**
* Whether the transaction was successful
*/
success: boolean;
/**
* Error message if failed
*/
error?: string;
/**
* Message ID if successful
*/
messageId?: string;
/**
* Resulting email if successful
*/
email?: Email;
}

559
ts/mail/routing/classes.dnsmanager.ts
View File

@@ -1,559 +0,0 @@
import * as plugins from '../../plugins.js';
import * as paths from '../../paths.js';
import { DKIMCreator } from '../security/classes.dkimcreator.js';
/**
* Interface for DNS record information
*/
export interface IDnsRecord {
name: string;
type: string;
value: string;
ttl?: number;
dnsSecEnabled?: boolean;
}
/**
* Interface for DNS lookup options
*/
export interface IDnsLookupOptions {
/** Cache time to live in milliseconds, 0 to disable caching */
cacheTtl?: number;
/** Timeout for DNS queries in milliseconds */
timeout?: number;
}
/**
* Interface for DNS verification result
*/
export interface IDnsVerificationResult {
record: string;
found: boolean;
valid: boolean;
value?: string;
expectedValue?: string;
error?: string;
}
/**
* Manager for DNS-related operations, including record lookups, verification, and generation
*/
export class DNSManager {
public dkimCreator: DKIMCreator;
private cache: Map<string, { data: any; expires: number }> = new Map();
private defaultOptions: IDnsLookupOptions = {
cacheTtl: 300000, // 5 minutes
timeout: 5000 // 5 seconds
};
constructor(dkimCreatorArg: DKIMCreator, options?: IDnsLookupOptions) {
this.dkimCreator = dkimCreatorArg;
if (options) {
this.defaultOptions = {
...this.defaultOptions,
...options
};
}
// Ensure the DNS records directory exists
plugins.fs.mkdirSync(paths.dnsRecordsDir, { recursive: true });
}
/**
* Lookup MX records for a domain
* @param domain Domain to look up
* @param options Lookup options
* @returns Array of MX records sorted by priority
*/
public async lookupMx(domain: string, options?: IDnsLookupOptions): Promise<plugins.dns.MxRecord[]> {
const lookupOptions = { ...this.defaultOptions, ...options };
const cacheKey = `mx:${domain}`;
// Check cache first
const cached = this.getFromCache<plugins.dns.MxRecord[]>(cacheKey);
if (cached) {
return cached;
}
try {
const records = await this.dnsResolveMx(domain, lookupOptions.timeout);
// Sort by priority
records.sort((a, b) => a.priority - b.priority);
// Cache the result
this.setInCache(cacheKey, records, lookupOptions.cacheTtl);
return records;
} catch (error) {
console.error(`Error looking up MX records for ${domain}:`, error);
throw new Error(`Failed to lookup MX records for ${domain}: ${error.message}`);
}
}
/**
* Lookup TXT records for a domain
* @param domain Domain to look up
* @param options Lookup options
* @returns Array of TXT records
*/
public async lookupTxt(domain: string, options?: IDnsLookupOptions): Promise<string[][]> {
const lookupOptions = { ...this.defaultOptions, ...options };
const cacheKey = `txt:${domain}`;
// Check cache first
const cached = this.getFromCache<string[][]>(cacheKey);
if (cached) {
return cached;
}
try {
const records = await this.dnsResolveTxt(domain, lookupOptions.timeout);
// Cache the result
this.setInCache(cacheKey, records, lookupOptions.cacheTtl);
return records;
} catch (error) {
console.error(`Error looking up TXT records for ${domain}:`, error);
throw new Error(`Failed to lookup TXT records for ${domain}: ${error.message}`);
}
}
/**
* Find specific TXT record by subdomain and prefix
* @param domain Base domain
* @param subdomain Subdomain prefix (e.g., "dkim._domainkey")
* @param prefix Record prefix to match (e.g., "v=DKIM1")
* @param options Lookup options
* @returns Matching TXT record or null if not found
*/
public async findTxtRecord(
domain: string,
subdomain: string = '',
prefix: string = '',
options?: IDnsLookupOptions
): Promise<string | null> {
const fullDomain = subdomain ? `${subdomain}.${domain}` : domain;
try {
const records = await this.lookupTxt(fullDomain, options);
for (const recordArray of records) {
// TXT records can be split into chunks, join them
const record = recordArray.join('');
if (!prefix || record.startsWith(prefix)) {
return record;
}
}
return null;
} catch (error) {
// Domain might not exist or no TXT records
console.log(`No matching TXT record found for ${fullDomain} with prefix ${prefix}`);
return null;
}
}
/**
* Verify if a domain has a valid SPF record
* @param domain Domain to verify
* @returns Verification result
*/
public async verifySpfRecord(domain: string): Promise<IDnsVerificationResult> {
const result: IDnsVerificationResult = {
record: 'SPF',
found: false,
valid: false
};
try {
const spfRecord = await this.findTxtRecord(domain, '', 'v=spf1');
if (spfRecord) {
result.found = true;
result.value = spfRecord;
// Basic validation - check if it contains all, include, ip4, ip6, or mx mechanisms
const isValid = /v=spf1\s+([-~?+]?(all|include:|ip4:|ip6:|mx|a|exists:))/.test(spfRecord);
result.valid = isValid;
if (!isValid) {
result.error = 'SPF record format is invalid';
}
} else {
result.error = 'No SPF record found';
}
} catch (error) {
result.error = `Error verifying SPF: ${error.message}`;
}
return result;
}
/**
* Verify if a domain has a valid DKIM record
* @param domain Domain to verify
* @param selector DKIM selector (usually "mta" in our case)
* @returns Verification result
*/
public async verifyDkimRecord(domain: string, selector: string = 'mta'): Promise<IDnsVerificationResult> {
const result: IDnsVerificationResult = {
record: 'DKIM',
found: false,
valid: false
};
try {
const dkimSelector = `${selector}._domainkey`;
const dkimRecord = await this.findTxtRecord(domain, dkimSelector, 'v=DKIM1');
if (dkimRecord) {
result.found = true;
result.value = dkimRecord;
// Basic validation - check for required fields
const hasP = dkimRecord.includes('p=');
result.valid = dkimRecord.includes('v=DKIM1') && hasP;
if (!result.valid) {
result.error = 'DKIM record is missing required fields';
} else if (dkimRecord.includes('p=') && !dkimRecord.match(/p=[a-zA-Z0-9+/]+/)) {
result.valid = false;
result.error = 'DKIM record has invalid public key format';
}
} else {
result.error = `No DKIM record found for selector ${selector}`;
}
} catch (error) {
result.error = `Error verifying DKIM: ${error.message}`;
}
return result;
}
/**
* Verify if a domain has a valid DMARC record
* @param domain Domain to verify
* @returns Verification result
*/
public async verifyDmarcRecord(domain: string): Promise<IDnsVerificationResult> {
const result: IDnsVerificationResult = {
record: 'DMARC',
found: false,
valid: false
};
try {
const dmarcDomain = `_dmarc.${domain}`;
const dmarcRecord = await this.findTxtRecord(dmarcDomain, '', 'v=DMARC1');
if (dmarcRecord) {
result.found = true;
result.value = dmarcRecord;
// Basic validation - check for required fields
const hasPolicy = dmarcRecord.includes('p=');
result.valid = dmarcRecord.includes('v=DMARC1') && hasPolicy;
if (!result.valid) {
result.error = 'DMARC record is missing required fields';
}
} else {
result.error = 'No DMARC record found';
}
} catch (error) {
result.error = `Error verifying DMARC: ${error.message}`;
}
return result;
}
/**
* Check all email authentication records (SPF, DKIM, DMARC) for a domain
* @param domain Domain to check
* @param dkimSelector DKIM selector
* @returns Object with verification results for each record type
*/
public async verifyEmailAuthRecords(domain: string, dkimSelector: string = 'mta'): Promise<{
spf: IDnsVerificationResult;
dkim: IDnsVerificationResult;
dmarc: IDnsVerificationResult;
}> {
const [spf, dkim, dmarc] = await Promise.all([
this.verifySpfRecord(domain),
this.verifyDkimRecord(domain, dkimSelector),
this.verifyDmarcRecord(domain)
]);
return { spf, dkim, dmarc };
}
/**
* Generate a recommended SPF record for a domain
* @param domain Domain name
* @param options Configuration options for the SPF record
* @returns Generated SPF record
*/
public generateSpfRecord(domain: string, options: {
includeMx?: boolean;
includeA?: boolean;
includeIps?: string[];
includeSpf?: string[];
policy?: 'none' | 'neutral' | 'softfail' | 'fail' | 'reject';
} = {}): IDnsRecord {
const {
includeMx = true,
includeA = true,
includeIps = [],
includeSpf = [],
policy = 'softfail'
} = options;
let value = 'v=spf1';
if (includeMx) {
value += ' mx';
}
if (includeA) {
value += ' a';
}
// Add IP addresses
for (const ip of includeIps) {
if (ip.includes(':')) {
value += ` ip6:${ip}`;
} else {
value += ` ip4:${ip}`;
}
}
// Add includes
for (const include of includeSpf) {
value += ` include:${include}`;
}
// Add policy
const policyMap = {
'none': '?all',
'neutral': '~all',
'softfail': '~all',
'fail': '-all',
'reject': '-all'
};
value += ` ${policyMap[policy]}`;
return {
name: domain,
type: 'TXT',
value: value
};
}
/**
* Generate a recommended DMARC record for a domain
* @param domain Domain name
* @param options Configuration options for the DMARC record
* @returns Generated DMARC record
*/
public generateDmarcRecord(domain: string, options: {
policy?: 'none' | 'quarantine' | 'reject';
subdomainPolicy?: 'none' | 'quarantine' | 'reject';
pct?: number;
rua?: string;
ruf?: string;
daysInterval?: number;
} = {}): IDnsRecord {
const {
policy = 'none',
subdomainPolicy,
pct = 100,
rua,
ruf,
daysInterval = 1
} = options;
let value = 'v=DMARC1; p=' + policy;
if (subdomainPolicy) {
value += `; sp=${subdomainPolicy}`;
}
if (pct !== 100) {
value += `; pct=${pct}`;
}
if (rua) {
value += `; rua=mailto:${rua}`;
}
if (ruf) {
value += `; ruf=mailto:${ruf}`;
}
if (daysInterval !== 1) {
value += `; ri=${daysInterval * 86400}`;
}
// Add reporting format and ADKIM/ASPF alignment
value += '; fo=1; adkim=r; aspf=r';
return {
name: `_dmarc.${domain}`,
type: 'TXT',
value: value
};
}
/**
* Save DNS record recommendations to a file
* @param domain Domain name
* @param records DNS records to save
*/
public async saveDnsRecommendations(domain: string, records: IDnsRecord[]): Promise<void> {
try {
const filePath = plugins.path.join(paths.dnsRecordsDir, `${domain}.recommendations.json`);
await plugins.smartfs.file(filePath).write(JSON.stringify(records, null, 2));
console.log(`DNS recommendations for ${domain} saved to ${filePath}`);
} catch (error) {
console.error(`Error saving DNS recommendations for ${domain}:`, error);
}
}
/**
* Get cache key value
* @param key Cache key
* @returns Cached value or undefined if not found or expired
*/
private getFromCache<T>(key: string): T | undefined {
const cached = this.cache.get(key);
if (cached && cached.expires > Date.now()) {
return cached.data as T;
}
// Remove expired entry
if (cached) {
this.cache.delete(key);
}
return undefined;
}
/**
* Set cache key value
* @param key Cache key
* @param data Data to cache
* @param ttl TTL in milliseconds
*/
private setInCache<T>(key: string, data: T, ttl: number = this.defaultOptions.cacheTtl): void {
if (ttl <= 0) return; // Don't cache if TTL is disabled
this.cache.set(key, {
data,
expires: Date.now() + ttl
});
}
/**
* Clear the DNS cache
* @param key Optional specific key to clear, or all cache if not provided
*/
public clearCache(key?: string): void {
if (key) {
this.cache.delete(key);
} else {
this.cache.clear();
}
}
/**
* Promise-based wrapper for dns.resolveMx
* @param domain Domain to resolve
* @param timeout Timeout in milliseconds
* @returns Promise resolving to MX records
*/
private dnsResolveMx(domain: string, timeout: number = 5000): Promise<plugins.dns.MxRecord[]> {
return new Promise((resolve, reject) => {
const timeoutId = setTimeout(() => {
reject(new Error(`DNS MX lookup timeout for ${domain}`));
}, timeout);
plugins.dns.resolveMx(domain, (err, addresses) => {
clearTimeout(timeoutId);
if (err) {
reject(err);
} else {
resolve(addresses);
}
});
});
}
/**
* Promise-based wrapper for dns.resolveTxt
* @param domain Domain to resolve
* @param timeout Timeout in milliseconds
* @returns Promise resolving to TXT records
*/
private dnsResolveTxt(domain: string, timeout: number = 5000): Promise<string[][]> {
return new Promise((resolve, reject) => {
const timeoutId = setTimeout(() => {
reject(new Error(`DNS TXT lookup timeout for ${domain}`));
}, timeout);
plugins.dns.resolveTxt(domain, (err, records) => {
clearTimeout(timeoutId);
if (err) {
reject(err);
} else {
resolve(records);
}
});
});
}
/**
* Generate all recommended DNS records for proper email authentication
* @param domain Domain to generate records for
* @returns Array of recommended DNS records
*/
public async generateAllRecommendedRecords(domain: string): Promise<IDnsRecord[]> {
const records: IDnsRecord[] = [];
// Get DKIM record (already created by DKIMCreator)
try {
// Call the DKIM creator directly
const dkimRecord = await this.dkimCreator.getDNSRecordForDomain(domain);
records.push(dkimRecord);
} catch (error) {
console.error(`Error getting DKIM record for ${domain}:`, error);
}
// Generate SPF record
const spfRecord = this.generateSpfRecord(domain, {
includeMx: true,
includeA: true,
policy: 'softfail'
});
records.push(spfRecord);
// Generate DMARC record
const dmarcRecord = this.generateDmarcRecord(domain, {
policy: 'none', // Start with monitoring mode
rua: `dmarc@${domain}` // Replace with appropriate report address
});
records.push(dmarcRecord);
// Save recommendations
await this.saveDnsRecommendations(domain, records);
return records;
}
}

410
ts/mail/routing/classes.unified.email.server.ts
View File

@@ -11,35 +11,6 @@ import { DKIMCreator } from '../security/classes.dkimcreator.js';
import { IPReputationChecker } from '../../security/classes.ipreputationchecker.js';
import { RustSecurityBridge } from '../../security/classes.rustsecuritybridge.js';
import type { IEmailReceivedEvent, IAuthRequestEvent, IEmailData } from '../../security/classes.rustsecuritybridge.js';
// Deliverability types (IPWarmupManager and SenderReputationMonitor are optional external modules)
interface IIPWarmupConfig {
enabled?: boolean;
ips?: string[];
[key: string]: any;
}
interface IReputationMonitorConfig {
enabled?: boolean;
domains?: string[];
[key: string]: any;
}
interface IPWarmupManager {
getWarmupStatus(ip: string): any;
addIPToWarmup(ip: string, config?: any): void;
removeIPFromWarmup(ip: string): void;
updateMetrics(ip: string, metrics: any): void;
canSendMoreToday(ip: string): boolean;
canSendMoreThisHour(ip: string): boolean;
getBestIPForSending(...args: any[]): string | null;
setActiveAllocationPolicy(policy: string): void;
recordSend(...args: any[]): void;
}
interface SenderReputationMonitor {
getReputationData(domain: string): any;
getReputationSummary(): any;
addDomain(domain: string): void;
removeDomain(domain: string): void;
recordSendEvent(domain: string, event: any): void;
}
import { EmailRouter } from './classes.email.router.js';
import type { IEmailRoute, IEmailAction, IEmailContext, IEmailDomainConfig } from './interfaces.js';
import { Email } from '../core/classes.email.js';
@@ -128,10 +99,6 @@ export interface IUnifiedEmailServerOptions {
// Rate limiting (global limits, can be overridden per domain)
rateLimits?: IHierarchicalRateLimits;
// Deliverability options
ipWarmupConfig?: IIPWarmupConfig;
reputationMonitorConfig?: IReputationMonitorConfig;
}
@@ -196,8 +163,6 @@ export class UnifiedEmailServer extends EventEmitter {
private rustBridge: RustSecurityBridge;
private ipReputationChecker: IPReputationChecker;
private bounceManager: BounceManager;
private ipWarmupManager: IPWarmupManager | null;
private senderReputationMonitor: SenderReputationMonitor | null;
public deliveryQueue: UnifiedDeliveryQueue;
public deliverySystem: MultiModeDeliverySystem;
private rateLimiter: UnifiedRateLimiter; // TODO: Implement rate limiting in SMTP server handlers
@@ -239,11 +204,6 @@ export class UnifiedEmailServer extends EventEmitter {
storageManager: dcRouter.storageManager
});
// IP warmup manager and sender reputation monitor are optional
// They will be initialized when the deliverability module is available
this.ipWarmupManager = null;
this.senderReputationMonitor = null;
// Initialize domain registry
this.domainRegistry = new DomainRegistry(options.domains, options.defaults);
@@ -373,6 +333,13 @@ export class UnifiedEmailServer extends EventEmitter {
}
logger.log('info', 'Rust security bridge started — Rust is the primary security backend');
// Listen for bridge state changes to propagate resilience events
this.rustBridge.on('stateChange', ({ oldState, newState }: { oldState: string; newState: string }) => {
if (newState === 'failed') this.emit('bridgeFailed');
else if (newState === 'restarting') this.emit('bridgeRestarting');
else if (newState === 'running' && oldState === 'restarting') this.emit('bridgeRecovered');
});
// Set up DKIM for all domains
await this.setupDkimForDomains();
logger.log('info', 'DKIM configuration completed for all domains');
@@ -414,13 +381,17 @@ export class UnifiedEmailServer extends EventEmitter {
await this.handleRustEmailReceived(data);
} catch (err) {
logger.log('error', `Error handling email from Rust SMTP: ${(err as Error).message}`);
// Send rejection back to Rust
await this.rustBridge.sendEmailProcessingResult({
correlationId: data.correlationId,
accepted: false,
smtpCode: 451,
smtpMessage: 'Internal processing error',
});
// Send rejection back to Rust (may fail if bridge is restarting)
try {
await this.rustBridge.sendEmailProcessingResult({
correlationId: data.correlationId,
accepted: false,
smtpCode: 451,
smtpMessage: 'Internal processing error',
});
} catch (sendErr) {
logger.log('warn', `Could not send rejection back to Rust: ${(sendErr as Error).message}`);
}
}
});
@@ -429,11 +400,15 @@ export class UnifiedEmailServer extends EventEmitter {
await this.handleRustAuthRequest(data);
} catch (err) {
logger.log('error', `Error handling auth from Rust SMTP: ${(err as Error).message}`);
await this.rustBridge.sendAuthResult({
correlationId: data.correlationId,
success: false,
message: 'Internal auth error',
});
try {
await this.rustBridge.sendAuthResult({
correlationId: data.correlationId,
success: false,
message: 'Internal auth error',
});
} catch (sendErr) {
logger.log('warn', `Could not send auth rejection back to Rust: ${(sendErr as Error).message}`);
}
}
});
@@ -495,7 +470,8 @@ export class UnifiedEmailServer extends EventEmitter {
// Clear the servers array - servers will be garbage collected
this.servers = [];
// Stop Rust security bridge
// Remove bridge state change listener and stop bridge
this.rustBridge.removeAllListeners('stateChange');
await this.rustBridge.stop();
// Stop the delivery system
@@ -653,7 +629,7 @@ export class UnifiedEmailServer extends EventEmitter {
logger.log('info', 'Using pre-computed security results from Rust in-process pipeline');
result = precomputed;
} else {
// Fallback: IPC round-trip to Rust (for backward compat / handleSocket mode)
// Fallback: IPC round-trip to Rust (for backward compat)
const rawMessage = session.emailData || email.toRFC822String();
result = await this.rustBridge.verifyEmail({
rawMessage,
@@ -967,171 +943,6 @@ export class UnifiedEmailServer extends EventEmitter {
throw error;
}
/**
* Handle email in MTA mode (programmatic processing)
*/
private async _handleMtaMode(email: Email, session: IExtendedSmtpSession): Promise<void> {
logger.log('info', `Handling email in MTA mode for session ${session.id}`);
try {
// Apply MTA rule options if provided
if (session.matchedRoute?.action.options?.mtaOptions) {
const options = session.matchedRoute.action.options.mtaOptions;
// Apply DKIM signing if enabled
if (options.dkimSign && options.dkimOptions) {
const dkimDomain = options.dkimOptions.domainName;
const dkimSelector = options.dkimOptions.keySelector || 'mta';
logger.log('info', `Signing email with DKIM for domain ${dkimDomain}`);
await this.handleDkimSigning(email, dkimDomain, dkimSelector);
}
}
// Get email content for logging/processing
const subject = email.subject;
const recipients = email.getAllRecipients().join(', ');
logger.log('info', `Email processed by MTA: ${subject} to ${recipients}`);
SecurityLogger.getInstance().logEvent({
level: SecurityLogLevel.INFO,
type: SecurityEventType.EMAIL_PROCESSING,
message: 'Email processed by MTA',
ipAddress: session.remoteAddress,
details: {
sessionId: session.id,
ruleName: session.matchedRoute?.name || 'default',
subject,
recipients
},
success: true
});
} catch (error) {
logger.log('error', `Failed to process email in MTA mode: ${error.message}`);
SecurityLogger.getInstance().logEvent({
level: SecurityLogLevel.ERROR,
type: SecurityEventType.EMAIL_PROCESSING,
message: 'MTA processing failed',
ipAddress: session.remoteAddress,
details: {
sessionId: session.id,
ruleName: session.matchedRoute?.name || 'default',
error: error.message
},
success: false
});
throw error;
}
}
/**
* Handle email in process mode (store-and-forward with scanning)
*/
private async _handleProcessMode(email: Email, session: IExtendedSmtpSession): Promise<void> {
logger.log('info', `Handling email in process mode for session ${session.id}`);
try {
const route = session.matchedRoute;
// Apply content scanning if enabled
if (route?.action.options?.contentScanning && route.action.options.scanners && route.action.options.scanners.length > 0) {
logger.log('info', 'Performing content scanning');
// Apply each scanner
for (const scanner of route.action.options.scanners) {
switch (scanner.type) {
case 'spam':
logger.log('info', 'Scanning for spam content');
// Implement spam scanning
break;
case 'virus':
logger.log('info', 'Scanning for virus content');
// Implement virus scanning
break;
case 'attachment':
logger.log('info', 'Scanning attachments');
// Check for blocked extensions
if (scanner.blockedExtensions && scanner.blockedExtensions.length > 0) {
for (const attachment of email.attachments) {
const ext = this.getFileExtension(attachment.filename);
if (scanner.blockedExtensions.includes(ext)) {
if (scanner.action === 'reject') {
throw new Error(`Blocked attachment type: ${ext}`);
} else { // tag
email.addHeader('X-Attachment-Warning', `Potentially unsafe attachment: ${attachment.filename}`);
}
}
}
}
break;
}
}
}
// Apply transformations if defined
if (route?.action.options?.transformations && route.action.options.transformations.length > 0) {
logger.log('info', 'Applying email transformations');
for (const transform of route.action.options.transformations) {
switch (transform.type) {
case 'addHeader':
if (transform.header && transform.value) {
email.addHeader(transform.header, transform.value);
}
break;
}
}
}
logger.log('info', `Email successfully processed in store-and-forward mode`);
SecurityLogger.getInstance().logEvent({
level: SecurityLogLevel.INFO,
type: SecurityEventType.EMAIL_PROCESSING,
message: 'Email processed and queued',
ipAddress: session.remoteAddress,
details: {
sessionId: session.id,
ruleName: route?.name || 'default',
contentScanning: route?.action.options?.contentScanning || false,
subject: email.subject
},
success: true
});
} catch (error) {
logger.log('error', `Failed to process email: ${error.message}`);
SecurityLogger.getInstance().logEvent({
level: SecurityLogLevel.ERROR,
type: SecurityEventType.EMAIL_PROCESSING,
message: 'Email processing failed',
ipAddress: session.remoteAddress,
details: {
sessionId: session.id,
ruleName: session.matchedRoute?.name || 'default',
error: error.message
},
success: false
});
throw error;
}
}
/**
* Get file extension from filename
*/
private getFileExtension(filename: string): string {
return filename.substring(filename.lastIndexOf('.')).toLowerCase();
}
/**
* Set up DKIM configuration for all domains
*/
@@ -1474,44 +1285,6 @@ export class UnifiedEmailServer extends EventEmitter {
}
}
// IP warmup handling
let ipAddress = options?.ipAddress;
// If no specific IP was provided, use IP warmup manager to find the best IP
if (!ipAddress) {
const domain = email.from.split('@')[1];
ipAddress = this.getBestIPForSending({
from: email.from,
to: email.to,
domain,
isTransactional: options?.isTransactional
});
if (ipAddress) {
logger.log('info', `Selected IP ${ipAddress} for sending based on warmup status`);
}
}
// If an IP is provided or selected by warmup manager, check its capacity
if (ipAddress) {
// Check if the IP can send more today
if (!this.canIPSendMoreToday(ipAddress)) {
logger.log('warn', `IP ${ipAddress} has reached its daily sending limit, email will be queued for later delivery`);
}
// Check if the IP can send more this hour
if (!this.canIPSendMoreThisHour(ipAddress)) {
logger.log('warn', `IP ${ipAddress} has reached its hourly sending limit, email will be queued for later delivery`);
}
// Record the send for IP warmup tracking
this.recordIPSend(ipAddress);
// Add IP header to the email
email.addHeader('X-Sending-IP', ipAddress);
}
// Check if the sender domain has DKIM keys and sign the email if needed
if (mode === 'mta' && route?.action.options?.mtaOptions?.dkimSign) {
const domain = email.from.split('@')[1];
@@ -1794,125 +1567,8 @@ export class UnifiedEmailServer extends EventEmitter {
}
/**
* Get the status of IP warmup process
* @param ipAddress Optional specific IP to check
* @returns Status of IP warmup
*/
public getIPWarmupStatus(ipAddress?: string): any {
return this.ipWarmupManager.getWarmupStatus(ipAddress);
}
/**
* Add a new IP address to the warmup process
* @param ipAddress IP address to add
*/
public addIPToWarmup(ipAddress: string): void {
this.ipWarmupManager.addIPToWarmup(ipAddress);
}
/**
* Remove an IP address from the warmup process
* @param ipAddress IP address to remove
*/
public removeIPFromWarmup(ipAddress: string): void {
this.ipWarmupManager.removeIPFromWarmup(ipAddress);
}
/**
* Update metrics for an IP in the warmup process
* @param ipAddress IP address
* @param metrics Metrics to update
*/
public updateIPWarmupMetrics(
ipAddress: string,
metrics: { openRate?: number; bounceRate?: number; complaintRate?: number }
): void {
this.ipWarmupManager.updateMetrics(ipAddress, metrics);
}
/**
* Check if an IP can send more emails today
* @param ipAddress IP address to check
* @returns Whether the IP can send more today
*/
public canIPSendMoreToday(ipAddress: string): boolean {
return this.ipWarmupManager.canSendMoreToday(ipAddress);
}
/**
* Check if an IP can send more emails in the current hour
* @param ipAddress IP address to check
* @returns Whether the IP can send more this hour
*/
public canIPSendMoreThisHour(ipAddress: string): boolean {
return this.ipWarmupManager.canSendMoreThisHour(ipAddress);
}
/**
* Get the best IP to use for sending an email based on warmup status
* @param emailInfo Information about the email being sent
* @returns Best IP to use or null
*/
public getBestIPForSending(emailInfo: {
from: string;
to: string[];
domain: string;
isTransactional?: boolean;
}): string | null {
return this.ipWarmupManager.getBestIPForSending(emailInfo);
}
/**
* Set the active IP allocation policy for warmup
* @param policyName Name of the policy to set
*/
public setIPAllocationPolicy(policyName: string): void {
this.ipWarmupManager.setActiveAllocationPolicy(policyName);
}
/**
* Record that an email was sent using a specific IP
* @param ipAddress IP address used for sending
*/
public recordIPSend(ipAddress: string): void {
this.ipWarmupManager.recordSend(ipAddress);
}
/**
* Get reputation data for a domain
* @param domain Domain to get reputation for
* @returns Domain reputation metrics
*/
public getDomainReputationData(domain: string): any {
return this.senderReputationMonitor.getReputationData(domain);
}
/**
* Get summary reputation data for all monitored domains
* @returns Summary data for all domains
*/
public getReputationSummary(): any {
return this.senderReputationMonitor.getReputationSummary();
}
/**
* Add a domain to the reputation monitoring system
* @param domain Domain to add
*/
public addDomainToMonitoring(domain: string): void {
this.senderReputationMonitor.addDomain(domain);
}
/**
* Remove a domain from the reputation monitoring system
* @param domain Domain to remove
*/
public removeDomainFromMonitoring(domain: string): void {
this.senderReputationMonitor.removeDomain(domain);
}
/**
* Record an email event for domain reputation tracking
* Record an email event for domain reputation tracking.
* Currently a no-op — the sender reputation monitor is not yet implemented.
* @param domain Domain sending the email
* @param event Event details
*/
@@ -1922,7 +1578,7 @@ export class UnifiedEmailServer extends EventEmitter {
hardBounce?: boolean;
receivingDomain?: string;
}): void {
this.senderReputationMonitor.recordSendEvent(domain, event);
logger.log('debug', `Reputation event for ${domain}: ${event.type}`);
}
/**

275
ts/security/classes.rustsecuritybridge.ts
View File

@@ -1,6 +1,7 @@
import * as plugins from '../plugins.js';
import * as paths from '../paths.js';
import { logger } from '../logger.js';
import { EventEmitter } from 'events';
// ---------------------------------------------------------------------------
// IPC command type map — mirrors the methods in mailer-bin's management mode
@@ -213,6 +214,35 @@ type TMailerCommands = {
};
};
// ---------------------------------------------------------------------------
// Bridge state machine
// ---------------------------------------------------------------------------
export enum BridgeState {
Idle = 'idle',
Starting = 'starting',
Running = 'running',
Restarting = 'restarting',
Failed = 'failed',
Stopped = 'stopped',
}
export interface IBridgeResilienceConfig {
maxRestartAttempts: number;
healthCheckIntervalMs: number;
restartBackoffBaseMs: number;
restartBackoffMaxMs: number;
healthCheckTimeoutMs: number;
}
const DEFAULT_RESILIENCE_CONFIG: IBridgeResilienceConfig = {
maxRestartAttempts: 5,
healthCheckIntervalMs: 30_000,
restartBackoffBaseMs: 1_000,
restartBackoffMaxMs: 30_000,
healthCheckTimeoutMs: 5_000,
};
// ---------------------------------------------------------------------------
// RustSecurityBridge — singleton wrapper around smartrust.RustBridge
// ---------------------------------------------------------------------------
@@ -222,14 +252,26 @@ type TMailerCommands = {
*
* Uses `@push.rocks/smartrust` for JSON-over-stdin/stdout IPC.
* Singleton — access via `RustSecurityBridge.getInstance()`.
*
* Features resilience via auto-restart with exponential backoff,
* periodic health checks, and a state machine that tracks the
* bridge lifecycle.
*/
export class RustSecurityBridge {
export class RustSecurityBridge extends EventEmitter {
private static instance: RustSecurityBridge | null = null;
private static _resilienceConfig: IBridgeResilienceConfig = { ...DEFAULT_RESILIENCE_CONFIG };
private bridge: InstanceType<typeof plugins.smartrust.RustBridge<TMailerCommands>>;
private _running = false;
private _state: BridgeState = BridgeState.Idle;
private _restartAttempts = 0;
private _restartTimer: ReturnType<typeof setTimeout> | null = null;
private _healthCheckTimer: ReturnType<typeof setInterval> | null = null;
private _deliberateStop = false;
private _smtpServerConfig: ISmtpServerConfig | null = null;
private constructor() {
super();
this.bridge = new plugins.smartrust.RustBridge<TMailerCommands>({
binaryName: 'mailer-bin',
cliArgs: ['--management'],
@@ -252,6 +294,13 @@ export class RustSecurityBridge {
this.bridge.on('exit', (code: number | null, signal: string | null) => {
this._running = false;
logger.log('warn', `Rust security bridge exited (code=${code}, signal=${signal})`);
if (this._deliberateStop) {
this.setState(BridgeState.Stopped);
} else if (this._state === BridgeState.Running) {
// Unexpected exit — attempt restart
this.attemptRestart();
}
});
this.bridge.on('stderr', (line: string) => {
@@ -259,6 +308,10 @@ export class RustSecurityBridge {
});
}
// -----------------------------------------------------------------------
// Static configuration & singleton
// -----------------------------------------------------------------------
/** Get or create the singleton instance. */
public static getInstance(): RustSecurityBridge {
if (!RustSecurityBridge.instance) {
@@ -267,11 +320,73 @@ export class RustSecurityBridge {
return RustSecurityBridge.instance;
}
/** Reset the singleton instance (for testing). */
public static resetInstance(): void {
if (RustSecurityBridge.instance) {
RustSecurityBridge.instance.stopHealthCheck();
if (RustSecurityBridge.instance._restartTimer) {
clearTimeout(RustSecurityBridge.instance._restartTimer);
RustSecurityBridge.instance._restartTimer = null;
}
RustSecurityBridge.instance.removeAllListeners();
}
RustSecurityBridge.instance = null;
}
/** Configure resilience parameters. Can be called before or after getInstance(). */
public static configure(config: Partial<IBridgeResilienceConfig>): void {
RustSecurityBridge._resilienceConfig = {
...RustSecurityBridge._resilienceConfig,
...config,
};
}
// -----------------------------------------------------------------------
// State management
// -----------------------------------------------------------------------
/** Current bridge state. */
public get state(): BridgeState {
return this._state;
}
/** Whether the Rust process is currently running and accepting commands. */
public get running(): boolean {
return this._running;
}
private setState(newState: BridgeState): void {
const oldState = this._state;
if (oldState === newState) return;
this._state = newState;
logger.log('info', `Rust bridge state: ${oldState} -> ${newState}`);
this.emit('stateChange', { oldState, newState });
}
/**
* Throws a descriptive error if the bridge is not in Running state.
* Called at the top of every command method.
*/
private ensureRunning(): void {
if (this._state === BridgeState.Running && this._running) {
return;
}
switch (this._state) {
case BridgeState.Idle:
throw new Error('Rust bridge has not been started yet. Call start() first.');
case BridgeState.Starting:
throw new Error('Rust bridge is still starting. Wait for start() to resolve.');
case BridgeState.Restarting:
throw new Error('Rust bridge is restarting after a crash. Commands will resume once it recovers.');
case BridgeState.Failed:
throw new Error('Rust bridge has failed after exhausting all restart attempts.');
case BridgeState.Stopped:
throw new Error('Rust bridge has been stopped. Call start() to restart it.');
default:
throw new Error(`Rust bridge is not running (state=${this._state}).`);
}
}
// -----------------------------------------------------------------------
// Lifecycle
// -----------------------------------------------------------------------
@@ -281,55 +396,195 @@ export class RustSecurityBridge {
* @returns `true` if the binary started successfully, `false` otherwise.
*/
public async start(): Promise<boolean> {
if (this._running) {
if (this._running && this._state === BridgeState.Running) {
return true;
}
this._deliberateStop = false;
this._restartAttempts = 0;
this.setState(BridgeState.Starting);
try {
const ok = await this.bridge.spawn();
this._running = ok;
if (ok) {
this.setState(BridgeState.Running);
this.startHealthCheck();
logger.log('info', 'Rust security bridge started');
} else {
this.setState(BridgeState.Failed);
logger.log('warn', 'Rust security bridge failed to start (binary not found or timeout)');
}
return ok;
} catch (err) {
this.setState(BridgeState.Failed);
logger.log('error', `Failed to start Rust security bridge: ${(err as Error).message}`);
return false;
}
}
/** Kill the Rust process. */
/** Kill the Rust process deliberately. */
public async stop(): Promise<void> {
this._deliberateStop = true;
// Cancel any pending restart
if (this._restartTimer) {
clearTimeout(this._restartTimer);
this._restartTimer = null;
}
this.stopHealthCheck();
this._smtpServerConfig = null;
if (!this._running) {
this.setState(BridgeState.Stopped);
return;
}
try {
this.bridge.kill();
this._running = false;
this.setState(BridgeState.Stopped);
logger.log('info', 'Rust security bridge stopped');
} catch (err) {
logger.log('error', `Error stopping Rust security bridge: ${(err as Error).message}`);
}
}
// -----------------------------------------------------------------------
// Auto-restart with exponential backoff
// -----------------------------------------------------------------------
private attemptRestart(): void {
const config = RustSecurityBridge._resilienceConfig;
this._restartAttempts++;
if (this._restartAttempts > config.maxRestartAttempts) {
logger.log('error', `Rust bridge exceeded max restart attempts (${config.maxRestartAttempts}). Giving up.`);
this.setState(BridgeState.Failed);
return;
}
this.setState(BridgeState.Restarting);
this.stopHealthCheck();
const delay = Math.min(
config.restartBackoffBaseMs * Math.pow(2, this._restartAttempts - 1),
config.restartBackoffMaxMs,
);
logger.log('info', `Rust bridge restart attempt ${this._restartAttempts}/${config.maxRestartAttempts} in ${delay}ms`);
this._restartTimer = setTimeout(async () => {
this._restartTimer = null;
// Guard: if stop() was called while we were waiting, don't restart
if (this._deliberateStop) {
this.setState(BridgeState.Stopped);
return;
}
try {
const ok = await this.bridge.spawn();
this._running = ok;
if (ok) {
logger.log('info', 'Rust bridge restarted successfully');
this._restartAttempts = 0;
this.setState(BridgeState.Running);
this.startHealthCheck();
await this.restoreAfterRestart();
} else {
logger.log('warn', 'Rust bridge restart failed (spawn returned false)');
this.attemptRestart();
}
} catch (err) {
logger.log('error', `Rust bridge restart failed: ${(err as Error).message}`);
this.attemptRestart();
}
}, delay);
}
/**
* Restore state after a successful restart:
* - Re-send startSmtpServer command if the SMTP server was running
*/
private async restoreAfterRestart(): Promise<void> {
if (this._smtpServerConfig) {
try {
logger.log('info', 'Restoring SMTP server after bridge restart');
const result = await this.bridge.sendCommand('startSmtpServer', this._smtpServerConfig);
if (result?.started) {
logger.log('info', 'SMTP server restored after bridge restart');
} else {
logger.log('warn', 'SMTP server failed to restore after bridge restart');
}
} catch (err) {
logger.log('error', `Failed to restore SMTP server after restart: ${(err as Error).message}`);
}
}
}
// -----------------------------------------------------------------------
// Health check
// -----------------------------------------------------------------------
private startHealthCheck(): void {
this.stopHealthCheck();
const config = RustSecurityBridge._resilienceConfig;
this._healthCheckTimer = setInterval(async () => {
if (this._state !== BridgeState.Running || !this._running) {
return;
}
try {
const pongPromise = this.bridge.sendCommand('ping', {} as any);
const timeoutPromise = new Promise<never>((_, reject) =>
setTimeout(() => reject(new Error('Health check timeout')), config.healthCheckTimeoutMs),
);
const res = await Promise.race([pongPromise, timeoutPromise]);
if (!(res as any)?.pong) {
throw new Error('Health check: unexpected ping response');
}
} catch (err) {
logger.log('warn', `Rust bridge health check failed: ${(err as Error).message}. Killing process to trigger restart.`);
try {
this.bridge.kill();
} catch {
// Already dead
}
// The exit handler will trigger attemptRestart()
}
}, config.healthCheckIntervalMs);
}
private stopHealthCheck(): void {
if (this._healthCheckTimer) {
clearInterval(this._healthCheckTimer);
this._healthCheckTimer = null;
}
}
// -----------------------------------------------------------------------
// Commands — thin typed wrappers over sendCommand
// -----------------------------------------------------------------------
/** Ping the Rust process. */
public async ping(): Promise<boolean> {
this.ensureRunning();
const res = await this.bridge.sendCommand('ping', {} as any);
return res?.pong === true;
}
/** Get version information for all Rust crates. */
public async getVersion(): Promise<IVersionInfo> {
this.ensureRunning();
return this.bridge.sendCommand('version', {} as any);
}
/** Validate an email address. */
public async validateEmail(email: string): Promise<IValidationResult> {
this.ensureRunning();
return this.bridge.sendCommand('validateEmail', { email });
}
@@ -339,6 +594,7 @@ export class RustSecurityBridge {
diagnosticCode?: string;
statusCode?: string;
}): Promise<IBounceDetection> {
this.ensureRunning();
return this.bridge.sendCommand('detectBounce', opts);
}
@@ -349,16 +605,19 @@ export class RustSecurityBridge {
htmlBody?: string;
attachmentNames?: string[];
}): Promise<IContentScanResult> {
this.ensureRunning();
return this.bridge.sendCommand('scanContent', opts);
}
/** Check IP reputation via DNSBL. */
public async checkIpReputation(ip: string): Promise<IReputationResult> {
this.ensureRunning();
return this.bridge.sendCommand('checkIpReputation', { ip });
}
/** Verify DKIM signatures on a raw email message. */
public async verifyDkim(rawMessage: string): Promise<IDkimVerificationResult[]> {
this.ensureRunning();
return this.bridge.sendCommand('verifyDkim', { rawMessage });
}
@@ -369,6 +628,7 @@ export class RustSecurityBridge {
selector?: string;
privateKey: string;
}): Promise<{ header: string; signedMessage: string }> {
this.ensureRunning();
return this.bridge.sendCommand('signDkim', opts);
}
@@ -379,6 +639,7 @@ export class RustSecurityBridge {
hostname?: string;
mailFrom: string;
}): Promise<ISpfResult> {
this.ensureRunning();
return this.bridge.sendCommand('checkSpf', opts);
}
@@ -395,6 +656,7 @@ export class RustSecurityBridge {
hostname?: string;
mailFrom: string;
}): Promise<IEmailSecurityResult> {
this.ensureRunning();
return this.bridge.sendCommand('verifyEmail', opts);
}
@@ -408,12 +670,16 @@ export class RustSecurityBridge {
* emailReceived and authRequest that must be handled by the caller.
*/
public async startSmtpServer(config: ISmtpServerConfig): Promise<boolean> {
this.ensureRunning();
this._smtpServerConfig = config;
const result = await this.bridge.sendCommand('startSmtpServer', config);
return result?.started === true;
}
/** Stop the Rust SMTP server. */
public async stopSmtpServer(): Promise<void> {
this.ensureRunning();
this._smtpServerConfig = null;
await this.bridge.sendCommand('stopSmtpServer', {} as any);
}
@@ -428,6 +694,7 @@ export class RustSecurityBridge {
smtpCode?: number;
smtpMessage?: string;
}): Promise<void> {
this.ensureRunning();
await this.bridge.sendCommand('emailProcessingResult', opts);
}
@@ -439,11 +706,13 @@ export class RustSecurityBridge {
success: boolean;
message?: string;
}): Promise<void> {
this.ensureRunning();
await this.bridge.sendCommand('authResult', opts);
}
/** Update rate limit configuration at runtime. */
public async configureRateLimits(config: IRateLimitConfig): Promise<void> {
this.ensureRunning();
await this.bridge.sendCommand('configureRateLimits', config);
}

2
ts/security/index.ts
View File

@@ -22,6 +22,8 @@ export {
export {
RustSecurityBridge,
BridgeState,
type IBridgeResilienceConfig,
type IDkimVerificationResult,
type ISpfResult,
type IDmarcResult,