22929902245

Committed 11 Mar 2026 12:02AM UTC coverage: 63.783%. First build

Build # 22929902245

Build Type

Pull #3528

github

Committed by

web-flow

Commit Message

Merge a16d0f189 into e85fd8c5b

Pull Request Pull Request #3528: Outbound: replace fs callbacks with promises and async/await

Coverage Stats

975 of 1452 branches covered (67.15%)

181 of 346 new or added lines in 5 files covered. (52.31%)

5514 of 8645 relevant lines covered (63.78%)

28.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

40.69

/server.js

'use strict'
// smtp network server

const cluster = require('node:cluster')
const fs = require('node:fs')
const os = require('node:os')
const path = require('node:path')
const tls = require('node:tls')

const daemon = require('daemon')
const constants = require('haraka-constants')

const tls_socket = require('./tls_socket')
const conn = require('./connection')
const outbound = require('./outbound')
const endpoint = require('./endpoint')

const Server = exports
Server.logger = require('./logger')
Server.config = require('haraka-config')
Server.plugins = require('./plugins')
Server.notes = {}

const { logger } = Server

// Need these here so we can run hooks
logger.add_log_methods(Server, 'server')

Server.listeners = []

Server.load_smtp_ini = () => {
    Server.cfg = Server.config.get(
        'smtp.ini',
        {
            booleans: ['-main.daemonize', '-main.graceful_shutdown'],
        },
        () => {
            Server.load_smtp_ini()
        },
    )

    if (Server.cfg.main.nodes === undefined) {
        Server.logwarn(`smtp.ini.nodes unset, using 1, see https://github.com/haraka/Haraka/wiki/Performance-Tuning`)
    }

    const defaults = {
        inactivity_timeout: 300,
        daemon_log_file: '/var/log/haraka.log',
        daemon_pid_file: '/var/run/haraka.pid',
        force_shutdown_timeout: 30,
        smtps_port: 465,
        nodes: 1,
    }

    for (const key in defaults) {
        if (Server.cfg.main[key] !== undefined) continue
        Server.cfg.main[key] = defaults[key]
    }
}

Server.load_http_ini = () => {
    Server.http = {}
    Server.http.cfg = Server.config.get('http.ini', () => {
        Server.load_http_ini()
    }).main
}

Server.load_smtp_ini()
Server.load_http_ini()

Server.daemonize = function () {
    const c = this.cfg.main
    if (!c.daemonize) return

    if (!process.env.__daemon) {
        // Remove process.on('exit') listeners otherwise
        // we get a spurious 'Exiting' log entry.
        process.removeAllListeners('exit')
        Server.lognotice('Daemonizing...')
    }

    const log_fd = fs.openSync(c.daemon_log_file, 'a')
    daemon({ cwd: process.cwd(), stdout: log_fd })

    // We are the daemon from here on...
    const npid = require('npid')
    try {
        npid.create(c.daemon_pid_file).removeOnExit()
    } catch (err) {
        Server.logerror(err.message)
        logger.dump_and_exit(1)
    }
}

Server.flushQueue = async (domain) => {
    if (!Server.cluster) {
        await outbound.flush_queue(domain)
        return
    }

    for (const id in cluster.workers) {
        cluster.workers[id].send({ event: 'outbound.flush_queue', domain })
    }
}

let graceful_in_progress = false

Server.gracefulRestart = () => {
    Server._graceful()
}

Server.stopListeners = () => {
    Server.loginfo('Shutting down listeners')
    for (const l of Server.listeners) {
        l.close()
    }
    Server.listeners = []
}

Server.performShutdown = () => {
    if (Server.cfg.main.graceful_shutdown) {
        return Server.gracefulShutdown()
    }
    Server.loginfo('Shutting down.')
    process.exit(0)
}

Server.gracefulShutdown = () => {
    Server.stopListeners()
    Server._graceful(() => {
        // log();
        Server.loginfo('Failed to shutdown naturally. Exiting.')
        process.exit(0)
    })
}

Server._graceful = async (shutdown) => {
    if (!Server.cluster && shutdown) {
        for (const module of ['outbound', 'cfreader', 'plugins']) {
            process.emit('message', { event: `${module}.shutdown` })
        }
        const t = setTimeout(shutdown, Server.cfg.main.force_shutdown_timeout * 1000)
        return t.unref()
    }

    if (graceful_in_progress) {
        Server.lognotice('Restart currently in progress - ignoring request')
        return
    }

    graceful_in_progress = true
    // TODO: Make these configurable
    const disconnect_timeout = 30
    const exit_timeout = 30
    cluster.removeAllListeners('exit')

    // we reload using eachLimit where limit = num_workers - 1
    // this kills all-but-one workers in parallel, leaving one running
    // for new connections, and then restarts that one last worker.

    const worker_ids = Object.keys(cluster.workers)
    let limit = worker_ids.length - 1
    if (limit < 2) limit = 1

    const todo = []

    for (const id of Object.keys(cluster.workers)) {
        todo.push((id) => {
            return new Promise((resolve) => {
                Server.lognotice(`Killing worker: ${id}`)
                const worker = cluster.workers[id]
                for (const module of ['outbound', 'cfreader', 'plugins']) {
                    worker.send({ event: `${module}.shutdown` })
                }
                worker.disconnect()
                let disconnect_received = false
                const disconnect_timer = setTimeout(() => {
                    if (!disconnect_received) {
                        Server.logcrit('Disconnect never received by worker. Killing.')
                        worker.kill()
                    }
                }, disconnect_timeout * 1000)

                worker.once('disconnect', () => {
                    clearTimeout(disconnect_timer)
                    disconnect_received = true
                    Server.lognotice('Disconnect complete')
                    let dead = false
                    const timer = setTimeout(() => {
                        if (!dead) {
                            Server.logcrit(`Worker ${id} failed to shutdown. Killing.`)
                            worker.kill()
                        }
                    }, exit_timeout * 1000)
                    worker.once('exit', () => {
                        dead = true
                        clearTimeout(timer)
                        if (shutdown) resolve()
                    })
                })
                if (!shutdown) {
                    const newWorker = cluster.fork()
                    newWorker.once('listening', () => {
                        Server.lognotice('Replacement worker online.')
                        newWorker.on('exit', (code, signal) => {
                            cluster_exit_listener(newWorker, code, signal)
                        })
                        resolve()
                    })
                }
            })
        })
    }

    while (todo.length) {
        // process batches of workers
        await Promise.all(todo.splice(0, limit))
    }

    if (shutdown) {
        Server.loginfo('Workers closed. Shutting down master process subsystems')
        for (const module of ['outbound', 'cfreader', 'plugins']) {
            process.emit('message', { event: `${module}.shutdown` })
        }
        const t2 = setTimeout(shutdown, Server.cfg.main.force_shutdown_timeout * 1000)
        return t2.unref()
    }
    graceful_in_progress = false
    Server.lognotice(`Reload complete, workers: ${JSON.stringify(Object.keys(cluster.workers))}`)
}

Server.sendToMaster = (command, params) => {
    // console.log("Send to master: ", command);
    if (Server.cluster) {
        if (Server.cluster.isMaster) {
            Server.receiveAsMaster(command, params)
        } else {
            process.send({ cmd: command, params })
        }
    } else {
        Server.receiveAsMaster(command, params)
    }
}

Server.receiveAsMaster = (command, params) => {
    if (!Server[command]) {
        Server.logerror(`Invalid command: ${command}`)
        return
    }
    Server[command].apply(Server, params)
}

function messageHandler(worker, msg, handle) {
    // sunset Haraka v3 (Node < 6)
    if (arguments.length === 2) {
        handle = msg
        msg = worker
        worker = undefined
    }
    // console.log("received cmd: ", msg);
    if (msg?.cmd) {
        Server.receiveAsMaster(msg.cmd, msg.params)
    }
}

Server.get_listen_addrs = (cfg, port) => {
    if (!port) port = 25
    let listeners = []
    if (cfg?.listen) {
        listeners = cfg.listen.split(/\s*,\s*/)
        if (listeners[0] === '') listeners = []
        for (let i = 0; i < listeners.length; i++) {
            const ep = endpoint(listeners[i], port)
            if (ep instanceof Error) continue
            listeners[i] = ep.toString()
        }
    }
    if (cfg.port) {
        let host = cfg.listen_host
        if (!host) {
            host = '[::0]'
            Server.default_host = true
        }
        listeners.unshift(`${host}:${cfg.port}`)
    }
    if (listeners.length) return listeners

    Server.default_host = true
    listeners.push(`[::0]:${port}`)

    return listeners
}

Server.createServer = (params) => {
    const c = Server.cfg.main
    for (const key in params) {
        if (typeof params[key] === 'function') continue
        c[key] = params[key]
    }

    Server.notes = {}
    Server.plugins.server = Server
    Server.plugins.load_plugins()

    const inactivity_timeout = (c.inactivity_timeout || 300) * 1000

    if (!cluster || !c.nodes) {
        Server.daemonize(c)
        Server.setup_smtp_listeners(Server.plugins, 'master', inactivity_timeout)
        return
    }

    // Cluster
    Server.cluster = cluster

    // Cluster Workers
    if (!cluster.isMaster) {
        Server.setup_smtp_listeners(Server.plugins, 'child', inactivity_timeout)
        return
    } else {
        // console.log("Setting up message handler");
        cluster.on('message', messageHandler)
    }

    // Cluster Master
    // We fork workers in init_master_respond so that plugins
    // can put handlers on cluster events before they are emitted.
    Server.plugins.run_hooks('init_master', Server)
}

Server.load_default_tls_config = (done) => {
    // this fn exists solely for testing
    if (Server.config.root_path != tls_socket.config.root_path) {
        Server.loginfo(`resetting tls_config.config path to ${Server.config.root_path}`)
        tls_socket.config = tls_socket.config.module_config(path.dirname(Server.config.root_path))
    }
    tls_socket.getSocketOpts('*').then((opts) => {
        done(opts)
    })
}

Server.get_smtp_server = async (ep, inactivity_timeout) => {
    let server

    function onConnect(client) {
        client.setTimeout(inactivity_timeout)
        const connection = conn.createConnection(client, server, Server.cfg)

        if (!server.has_tls) return

        const cipher = client.getCipher()
        cipher.version = client.getProtocol() // replace min with actual

        connection.setTLS({
            cipher,
            verified: client.authorized,
            verifyError: client.authorizationError,
            peerCertificate: client.getPeerCertificate(),
        })
    }

    if (ep.port === parseInt(Server.cfg.main.smtps_port, 10)) {
        Server.loginfo('getting SocketOpts for SMTPS server')
        const opts = await tls_socket.getSocketOpts('*')
        Server.loginfo(`Creating TLS server on ${ep}`)

        opts.rejectUnauthorized = tls_socket.get_rejectUnauthorized(
            opts.rejectUnauthorized,
            ep.port,
            tls_socket.cfg.main.requireAuthorized,
        )

        server = tls.createServer(opts, onConnect)
        tls_socket.addOCSP(server)
        server.has_tls = true
        server.on('resumeSession', (id, rsDone) => {
            Server.loginfo('client requested TLS resumeSession')
            rsDone(null, null)
        })
        Server.listeners.push(server)
        return server
    } else {
        server = tls_socket.createServer(onConnect)
        server.has_tls = false
        const opts = await tls_socket.getSocketOpts('*')
        Server.listeners.push(server)
        return server
    }
}

Server.setup_smtp_listeners = async (plugins2, type, inactivity_timeout) => {
    const errors = []

    for (const [ifName, ifObj] of Object.entries(os.networkInterfaces())) {
        for (const addr of ifObj) {
            if (addr.family === 'IPv6') {
                if (!Server.notes.IPv6) Server.notes.IPv6 = true
            } else if (addr.family === 'IPv4') {
                if (!Server.notes.IPv4) Server.notes.IPv4 = true
            } else {
                console.error(addr)
            }
        }
    }

    for (const listen_address of Server.get_listen_addrs(Server.cfg.main)) {
        const ep = endpoint(listen_address, 25)

        if (ep instanceof Error) {
            Server.logerror(`Invalid "listen" format in smtp.ini: ${listen_address}`)
            continue
        }

        const server = await Server.get_smtp_server(ep, inactivity_timeout)
        if (!server) continue

        server.notes = Server.notes
        if (Server.cluster) server.cluster = Server.cluster

        server
            .on('listening', function () {
                const addr = this.address()
                Server.lognotice(`Listening on ${endpoint(addr)}`)
            })
            .on('close', () => {
                Server.loginfo(`Listener ${ep} stopped`)
            })
            .on('error', (e) => {
                errors.push(e)
                Server.logerror(`Failed to setup listeners: ${e.message}`)
                if (e.code !== 'EAFNOSUPPORT') {
                    Server.logerror(e)
                    return
                }
                // Fallback from IPv6 to IPv4 if not supported
                // But only if we supplied the default of [::0]:25
                if (/^::0/.test(ep.host) && Server.default_host) {
                    server.listen(ep.port, '0.0.0.0', 0)
                    return
                }
                // Pass error to callback
                Server.logerror(e)
            })

        await ep.bind(server, { backlog: 0 })
    }

    if (errors.length) {
        for (const e of errors) {
            Server.logerror(`Failed to setup listeners: ${e.message}`)
        }
        return logger.dump_and_exit(-1)
    }
    Server.listening()
    plugins2.run_hooks(`init_${type}`, Server)
}

Server.setup_http_listeners = async () => {
    if (!Server.http?.cfg?.listen) return

    const listeners = Server.get_listen_addrs(Server.http.cfg, 80)
    if (!listeners.length) return

    try {
        Server.http.express = require('express')
        Server.loginfo('express loaded at Server.http.express')
    } catch (err) {
        Server.logerror('express failed to load. No http server. Install express with: npm install -g express')
        return
    }

    const app = Server.http.express()
    Server.http.app = app
    Server.loginfo('express app is at Server.http.app')

    for (const listen_address of listeners) {
        const ep = endpoint(listen_address, 80)
        if (ep instanceof Error) {
            Server.logerror(`Invalid format for listen in http.ini: ${listen_address}`)
            continue
        }

        if (443 == ep.port) {
            const tlsOpts = { ...tls_socket.certsByHost['*'] }
            tlsOpts.requestCert = false // not appropriate for HTTPS
            Server.http.server = require('https').createServer(tlsOpts, app)
        } else {
            Server.http.server = require('http').createServer(app)
        }

        Server.listeners.push(Server.http.server)

        Server.http.server.on('listening', function () {
            Server.lognotice(`Listening on ${endpoint(this.address())}`)
        })

        Server.http.server.on('error', (e) => {
            Server.logerror(e)
        })

        await ep.bind(Server.http.server, { backlog: 0 })
    }

    Server.plugins.run_hooks('init_http', Server)
    app.use(Server.http.express.static(Server.get_http_docroot()))
    app.use(Server.handle404)
}

Server.init_master_respond = async (retval, msg) => {
    if (!(retval === constants.ok || retval === constants.cont)) {
        Server.logerror(`init_master returned error${msg ? `: ${msg}` : ''}`)
        return logger.dump_and_exit(1)
    }

    const c = Server.cfg.main
    Server.ready = 1

    // Load the queue if we're just one process
    if (!(cluster && c.nodes)) {
        try {
            await outbound.init_queue()
        } catch (err) {
            Server.logcrit('Loading queue failed. Shutting down.')
            return logger.dump_and_exit(1)
        }
        Server.setup_http_listeners()
        return
    }

    // Running under cluster, fork children here, so that
    // cluster events can be registered in init_master hooks.
    try {
        const pids = await outbound.scan_queue_pids()
        Server.daemonize()
        // Fork workers
        const workers = c.nodes === 'cpus' ? os.cpus().length : c.nodes
        const new_workers = []
        for (let i = 0; i < workers; i++) {
            new_workers.push(cluster.fork({ CLUSTER_MASTER_PID: process.pid }))
        }
        for (let j = 0; j < pids.length; j++) {
            new_workers[j % new_workers.length].send({
                event: 'outbound.load_pid_queue',
                data: pids[j],
            })
        }
        cluster.on('online', (worker) => {
            Server.lognotice('worker started', {
                worker: worker.id,
                pid: worker.process.pid,
            })
        })
        cluster.on('listening', (worker, address) => {
            Server.lognotice(`worker ${worker.id} listening on ${endpoint(address)}`)
        })
        cluster.on('exit', cluster_exit_listener)
    } catch (err) {
        Server.logcrit('Scanning queue failed. Shutting down.')
        logger.dump_and_exit(1)
    }
}

function cluster_exit_listener(worker, code, signal) {
    if (signal) {
        Server.lognotice(`worker ${worker.id} killed by signal ${signal}`)
    } else if (code !== 0) {
        Server.lognotice(`worker ${worker.id} exited with error code: ${code}`)
    }
    if (signal || code !== 0) {
        // Restart worker
        const new_worker = cluster.fork({
            CLUSTER_MASTER_PID: process.pid,
        })
        new_worker.send({
            event: 'outbound.load_pid_queue',
            data: worker.process.pid,
        })
    }
}

Server.init_child_respond = (retval, msg) => {
    switch (retval) {
        case constants.ok:
        case constants.cont:
            Server.setup_http_listeners()
            return
    }

    const pid = process.env.CLUSTER_MASTER_PID
    Server.logerror(`init_child returned error ${msg ? `: ${msg}` : ''}`)
    try {
        if (pid) {
            process.kill(pid)
            Server.logerror(`Killing master (pid=${pid})`)
        }
    } catch (err) {
        Server.logerror('Terminating child')
    }
    logger.dump_and_exit(1)
}

Server.listening = () => {
    const c = Server.cfg.main

    // Drop privileges
    if (c.group) {
        Server.lognotice(`Switching from current gid: ${process.getgid()}`)
        process.setgid(c.group)
        Server.lognotice(`New gid: ${process.getgid()}`)
    }
    if (c.user) {
        Server.lognotice(`Switching from current uid: ${process.getuid()}`)
        process.setuid(c.user)
        Server.lognotice(`New uid: ${process.getuid()}`)
    }

    Server.ready = 1
}

Server.init_http_respond = () => {
    Server.loginfo('init_http_respond')

    let WebSocketServer
    try {
        WebSocketServer = require('ws').Server
    } catch (e) {
        Server.logerror(`unable to load ws.\n  did you: npm install -g ws?`)
        return
    }

    if (!WebSocketServer) {
        Server.logerror('ws failed to load')
        return
    }

    Server.http.wss = new WebSocketServer({ server: Server.http.server })
    Server.loginfo('Server.http.wss loaded')

    Server.plugins.run_hooks('init_wss', Server)
}

Server.init_wss_respond = () => {
    Server.loginfo('init_wss_respond')
}

Server.get_http_docroot = () => {
    if (Server.http.cfg.docroot) return Server.http.cfg.docroot

    Server.http.cfg.docroot = path.join(process.env.HARAKA || __dirname, 'http', 'html')
    Server.loginfo(`using html docroot: ${Server.http.cfg.docroot}`)
    return Server.http.cfg.docroot
}

Server.handle404 = (req, res) => {
    // abandon all hope, serve up a 404
    const docroot = Server.get_http_docroot()

    // respond with html page
    if (req.accepts('html')) {
        res.status(404).sendFile('404.html', { root: docroot })
        return
    }

    // respond with json
    if (req.accepts('json')) {
        res.status(404).send({ err: 'Not found' })
        return
    }

    res.status(404).send('Not found!')
}

1	'use strict'	1✔
2	// smtp network server	1✔
3		1✔
4	const cluster = require('node:cluster')	1✔
5	const fs = require('node:fs')	1✔
6	const os = require('node:os')	1✔
7	const path = require('node:path')	1✔
8	const tls = require('node:tls')	1✔
9		1✔
10	const daemon = require('daemon')	1✔
11	const constants = require('haraka-constants')	1✔
12		1✔
13	const tls_socket = require('./tls_socket')	1✔
14	const conn = require('./connection')	1✔
15	const outbound = require('./outbound')	1✔
16	const endpoint = require('./endpoint')	1✔
17		1✔
18	const Server = exports	1✔
19	Server.logger = require('./logger')	1✔
20	Server.config = require('haraka-config')	1✔
21	Server.plugins = require('./plugins')	1✔
22	Server.notes = {}	1✔
23		1✔
24	const { logger } = Server	1✔
25		1✔
26	// Need these here so we can run hooks	1✔
27	logger.add_log_methods(Server, 'server')	1✔
28		1✔
29	Server.listeners = []	1✔
30		1✔
31	Server.load_smtp_ini = () => {	1✔
32	Server.cfg = Server.config.get(	9✔
33	'smtp.ini',	9✔
34	{	9✔
35	booleans: ['-main.daemonize', '-main.graceful_shutdown'],	9✔
36	},	9✔
37	() => {	9✔
38	Server.load_smtp_ini()	×
39	},	9✔
40	)	9✔
41		9✔
42	if (Server.cfg.main.nodes === undefined) {	9✔
43	Server.logwarn(`smtp.ini.nodes unset, using 1, see https://github.com/haraka/Haraka/wiki/Performance-Tuning`)	2✔
44	}	2✔
45		9✔
46	const defaults = {	9✔
47	inactivity_timeout: 300,	9✔
48	daemon_log_file: '/var/log/haraka.log',	9✔
49	daemon_pid_file: '/var/run/haraka.pid',	9✔
50	force_shutdown_timeout: 30,	9✔
51	smtps_port: 465,	9✔
52	nodes: 1,	9✔
53	}	9✔
54		9✔
55	for (const key in defaults) {	9✔
56	if (Server.cfg.main[key] !== undefined) continue	54✔
57	Server.cfg.main[key] = defaults[key]	47✔
58	}	47✔
59	}	9✔
60		1✔
61	Server.load_http_ini = () => {	1✔
62	Server.http = {}	1✔
63	Server.http.cfg = Server.config.get('http.ini', () => {	1✔
64	Server.load_http_ini()	×
65	}).main	1✔
66	}	1✔
67		1✔
68	Server.load_smtp_ini()	1✔
69	Server.load_http_ini()	1✔
70		1✔
71	Server.daemonize = function () {	1✔
72	const c = this.cfg.main	7✔
73	if (!c.daemonize) return	7✔
74		×
75	if (!process.env.__daemon) {	×
76	// Remove process.on('exit') listeners otherwise	×
77	// we get a spurious 'Exiting' log entry.	×
78	process.removeAllListeners('exit')	×
79	Server.lognotice('Daemonizing...')	×
80	}	×
81		×
82	const log_fd = fs.openSync(c.daemon_log_file, 'a')	×
83	daemon({ cwd: process.cwd(), stdout: log_fd })	×
84		×
85	// We are the daemon from here on...	×
86	const npid = require('npid')	×
87	try {	×
88	npid.create(c.daemon_pid_file).removeOnExit()	×
89	} catch (err) {	×
90	Server.logerror(err.message)	×
91	logger.dump_and_exit(1)	×
92	}	×
93	}	7✔
94		1✔
95	Server.flushQueue = async (domain) => {	1✔
96	if (!Server.cluster) {	×
NEW 97	await outbound.flush_queue(domain)	×
98	return	×
99	}	×
100		×
101	for (const id in cluster.workers) {	×
102	cluster.workers[id].send({ event: 'outbound.flush_queue', domain })	×
103	}	×
104	}	×
105		1✔
106	let graceful_in_progress = false	1✔
107		1✔
108	Server.gracefulRestart = () => {	1✔
109	Server._graceful()	×
110	}	×
111		1✔
112	Server.stopListeners = () => {	1✔
113	Server.loginfo('Shutting down listeners')	6✔
114	for (const l of Server.listeners) {	6✔
115	l.close()	8✔
116	}	8✔
117	Server.listeners = []	6✔
118	}	6✔
119		1✔
120	Server.performShutdown = () => {	1✔
121	if (Server.cfg.main.graceful_shutdown) {	×
122	return Server.gracefulShutdown()	×
123	}	×
124	Server.loginfo('Shutting down.')	×
125	process.exit(0)	×
126	}	×
127		1✔
128	Server.gracefulShutdown = () => {	1✔
129	Server.stopListeners()	×
130	Server._graceful(() => {	×
131	// log();	×
132	Server.loginfo('Failed to shutdown naturally. Exiting.')	×
133	process.exit(0)	×
134	})	×
135	}	×
136		1✔
137	Server._graceful = async (shutdown) => {	1✔
138	if (!Server.cluster && shutdown) {	×
139	for (const module of ['outbound', 'cfreader', 'plugins']) {	×
140	process.emit('message', { event: `${module}.shutdown` })	×
141	}	×
142	const t = setTimeout(shutdown, Server.cfg.main.force_shutdown_timeout * 1000)	×
143	return t.unref()	×
144	}	×
145		×
146	if (graceful_in_progress) {	×
147	Server.lognotice('Restart currently in progress - ignoring request')	×
148	return	×
149	}	×
150		×
151	graceful_in_progress = true	×
152	// TODO: Make these configurable	×
153	const disconnect_timeout = 30	×
154	const exit_timeout = 30	×
155	cluster.removeAllListeners('exit')	×
156		×
157	// we reload using eachLimit where limit = num_workers - 1	×
158	// this kills all-but-one workers in parallel, leaving one running	×
159	// for new connections, and then restarts that one last worker.	×
160		×
161	const worker_ids = Object.keys(cluster.workers)	×
162	let limit = worker_ids.length - 1	×
163	if (limit < 2) limit = 1	×
164		×
165	const todo = []	×
166		×
167	for (const id of Object.keys(cluster.workers)) {	×
168	todo.push((id) => {	×
169	return new Promise((resolve) => {	×
170	Server.lognotice(`Killing worker: ${id}`)	×
171	const worker = cluster.workers[id]	×
172	for (const module of ['outbound', 'cfreader', 'plugins']) {	×
173	worker.send({ event: `${module}.shutdown` })	×
174	}	×
175	worker.disconnect()	×
176	let disconnect_received = false	×
177	const disconnect_timer = setTimeout(() => {	×
178	if (!disconnect_received) {	×
179	Server.logcrit('Disconnect never received by worker. Killing.')	×
180	worker.kill()	×
181	}	×
182	}, disconnect_timeout * 1000)	×
183		×
184	worker.once('disconnect', () => {	×
185	clearTimeout(disconnect_timer)	×
186	disconnect_received = true	×
187	Server.lognotice('Disconnect complete')	×
188	let dead = false	×
189	const timer = setTimeout(() => {	×
190	if (!dead) {	×
191	Server.logcrit(`Worker ${id} failed to shutdown. Killing.`)	×
192	worker.kill()	×
193	}	×
194	}, exit_timeout * 1000)	×
195	worker.once('exit', () => {	×
196	dead = true	×
197	clearTimeout(timer)	×
198	if (shutdown) resolve()	×
199	})	×
200	})	×
201	if (!shutdown) {	×
202	const newWorker = cluster.fork()	×
203	newWorker.once('listening', () => {	×
204	Server.lognotice('Replacement worker online.')	×
205	newWorker.on('exit', (code, signal) => {	×
206	cluster_exit_listener(newWorker, code, signal)	×
207	})	×
208	resolve()	×
209	})	×
210	}	×
211	})	×
212	})	×
213	}	×
214		×
215	while (todo.length) {	×
216	// process batches of workers	×
217	await Promise.all(todo.splice(0, limit))	×
218	}	×
219		×
220	if (shutdown) {	×
221	Server.loginfo('Workers closed. Shutting down master process subsystems')	×
222	for (const module of ['outbound', 'cfreader', 'plugins']) {	×
223	process.emit('message', { event: `${module}.shutdown` })	×
224	}	×
225	const t2 = setTimeout(shutdown, Server.cfg.main.force_shutdown_timeout * 1000)	×
226	return t2.unref()	×
227	}	×
228	graceful_in_progress = false	×
229	Server.lognotice(`Reload complete, workers: ${JSON.stringify(Object.keys(cluster.workers))}`)	×
230	}	×
231		1✔
232	Server.sendToMaster = (command, params) => {	1✔
233	// console.log("Send to master: ", command);	×
234	if (Server.cluster) {	×
235	if (Server.cluster.isMaster) {	×
236	Server.receiveAsMaster(command, params)	×
237	} else {	×
238	process.send({ cmd: command, params })	×
239	}	×
240	} else {	×
241	Server.receiveAsMaster(command, params)	×
242	}	×
243	}	×
244		1✔
245	Server.receiveAsMaster = (command, params) => {	1✔
246	if (!Server[command]) {	×
247	Server.logerror(`Invalid command: ${command}`)	×
248	return	×
249	}	×
250	Server[command].apply(Server, params)	×
251	}	×
252		1✔
253	function messageHandler(worker, msg, handle) {	×
254	// sunset Haraka v3 (Node < 6)	×
255	if (arguments.length === 2) {	×
256	handle = msg	×
257	msg = worker	×
258	worker = undefined	×
259	}	×
260	// console.log("received cmd: ", msg);	×
261	if (msg?.cmd) {	×
262	Server.receiveAsMaster(msg.cmd, msg.params)	×
263	}	×
264	}	×
265		1✔
266	Server.get_listen_addrs = (cfg, port) => {	1✔
267	if (!port) port = 25	16✔
268	let listeners = []	16✔
269	if (cfg?.listen) {	16✔
270	listeners = cfg.listen.split(/\s,\s/)	16✔
271	if (listeners[0] === '') listeners = []	16!
272	for (let i = 0; i < listeners.length; i++) {	16✔
273	const ep = endpoint(listeners[i], port)	19✔
274	if (ep instanceof Error) continue	19!
275	listeners[i] = ep.toString()	19✔
276	}	19✔
277	}	16✔
278	if (cfg.port) {	16!
279	let host = cfg.listen_host	×
280	if (!host) {	×
281	host = '[::0]'	×
282	Server.default_host = true	×
283	}	×
284	listeners.unshift(`${host}:${cfg.port}`)	×
285	}	×
286	if (listeners.length) return listeners	16✔
287		×
288	Server.default_host = true	×
289	listeners.push(`[::0]:${port}`)	×
290		×
291	return listeners	×
292	}	16✔
293		1✔
294	Server.createServer = (params) => {	1✔
295	const c = Server.cfg.main	7✔
296	for (const key in params) {	7!
297	if (typeof params[key] === 'function') continue	×
298	c[key] = params[key]	×
299	}	×
300		7✔
301	Server.notes = {}	7✔
302	Server.plugins.server = Server	7✔
303	Server.plugins.load_plugins()	7✔
304		7✔
305	const inactivity_timeout = (c.inactivity_timeout \|\| 300) * 1000	7!
306		7✔
307	if (!cluster \|\| !c.nodes) {	7✔
308	Server.daemonize(c)	7✔
309	Server.setup_smtp_listeners(Server.plugins, 'master', inactivity_timeout)	7✔
310	return	7✔
311	}	7✔
312		×
313	// Cluster	×
314	Server.cluster = cluster	×
315		×
316	// Cluster Workers	×
317	if (!cluster.isMaster) {	×
318	Server.setup_smtp_listeners(Server.plugins, 'child', inactivity_timeout)	×
319	return	×
320	} else {	×
321	// console.log("Setting up message handler");	×
322	cluster.on('message', messageHandler)	×
323	}	×
324		×
325	// Cluster Master	×
326	// We fork workers in init_master_respond so that plugins	×
327	// can put handlers on cluster events before they are emitted.	×
328	Server.plugins.run_hooks('init_master', Server)	×
329	}	7✔
330		1✔
331	Server.load_default_tls_config = (done) => {	1✔
332	// this fn exists solely for testing	9✔
333	if (Server.config.root_path != tls_socket.config.root_path) {	9✔
334	Server.loginfo(`resetting tls_config.config path to ${Server.config.root_path}`)	1✔
335	tls_socket.config = tls_socket.config.module_config(path.dirname(Server.config.root_path))	1✔
336	}	1✔
337	tls_socket.getSocketOpts('*').then((opts) => {	9✔
338	done(opts)	9✔
339	})	9✔
340	}	9✔
341		1✔
342	Server.get_smtp_server = async (ep, inactivity_timeout) => {	1✔
343	let server	9✔
344		9✔
345	function onConnect(client) {	9✔
346	client.setTimeout(inactivity_timeout)	5✔
347	const connection = conn.createConnection(client, server, Server.cfg)	5✔
348		5✔
349	if (!server.has_tls) return	5✔
350		×
351	const cipher = client.getCipher()	×
352	cipher.version = client.getProtocol() // replace min with actual	×
353		×
354	connection.setTLS({	×
355	cipher,	×
356	verified: client.authorized,	×
357	verifyError: client.authorizationError,	×
358	peerCertificate: client.getPeerCertificate(),	×
359	})	×
360	}	5✔
361		9✔
362	if (ep.port === parseInt(Server.cfg.main.smtps_port, 10)) {	9✔
363	Server.loginfo('getting SocketOpts for SMTPS server')	2✔
364	const opts = await tls_socket.getSocketOpts('*')	2✔
365	Server.loginfo(`Creating TLS server on ${ep}`)	2✔
366		2✔
367	opts.rejectUnauthorized = tls_socket.get_rejectUnauthorized(	2✔
368	opts.rejectUnauthorized,	2✔
369	ep.port,	2✔
370	tls_socket.cfg.main.requireAuthorized,	2✔
371	)	2✔
372		2✔
373	server = tls.createServer(opts, onConnect)	2✔
374	tls_socket.addOCSP(server)	2✔
375	server.has_tls = true	2✔
376	server.on('resumeSession', (id, rsDone) => {	2✔
377	Server.loginfo('client requested TLS resumeSession')	1✔
378	rsDone(null, null)	1✔
379	})	2✔
380	Server.listeners.push(server)	2✔
381	return server	2✔
382	} else {	9✔
383	server = tls_socket.createServer(onConnect)	7✔
384	server.has_tls = false	7✔
385	const opts = await tls_socket.getSocketOpts('*')	7✔
386	Server.listeners.push(server)	7✔
387	return server	7✔
388	}	7✔
389	}	9✔
390		1✔
391	Server.setup_smtp_listeners = async (plugins2, type, inactivity_timeout) => {	1✔
392	const errors = []	7✔
393		7✔
394	for (const [ifName, ifObj] of Object.entries(os.networkInterfaces())) {	7✔
395	for (const addr of ifObj) {	14✔
396	if (addr.family === 'IPv6') {	28✔
397	if (!Server.notes.IPv6) Server.notes.IPv6 = true	14✔
398	} else if (addr.family === 'IPv4') {	14✔
399	if (!Server.notes.IPv4) Server.notes.IPv4 = true	14✔
400	} else {	14!
401	console.error(addr)	×
402	}	×
403	}	28✔
404	}	14✔
405		7✔
406	for (const listen_address of Server.get_listen_addrs(Server.cfg.main)) {	7✔
407	const ep = endpoint(listen_address, 25)	7✔
408		7✔
409	if (ep instanceof Error) {	7!
410	Server.logerror(`Invalid "listen" format in smtp.ini: ${listen_address}`)	×
411	continue	×
412	}	×
413		7✔
414	const server = await Server.get_smtp_server(ep, inactivity_timeout)	7✔
415	if (!server) continue	7!
416		7✔
417	server.notes = Server.notes	7✔
418	if (Server.cluster) server.cluster = Server.cluster	7!
419		7✔
420	server	7✔
421	.on('listening', function () {	7✔
422	const addr = this.address()	7✔
423	Server.lognotice(`Listening on ${endpoint(addr)}`)	7✔
424	})	7✔
425	.on('close', () => {	7✔
426	Server.loginfo(`Listener ${ep} stopped`)	6✔
427	})	7✔
428	.on('error', (e) => {	7✔
429	errors.push(e)	×
430	Server.logerror(`Failed to setup listeners: ${e.message}`)	×
431	if (e.code !== 'EAFNOSUPPORT') {	×
432	Server.logerror(e)	×
433	return	×
434	}	×
435	// Fallback from IPv6 to IPv4 if not supported	×
436	// But only if we supplied the default of [::0]:25	×
437	if (/^::0/.test(ep.host) && Server.default_host) {	×
438	server.listen(ep.port, '0.0.0.0', 0)	×
439	return	×
440	}	×
441	// Pass error to callback	×
442	Server.logerror(e)	×
443	})	7✔
444		7✔
445	await ep.bind(server, { backlog: 0 })	7✔
446	}	7✔
447		7✔
448	if (errors.length) {	7!
449	for (const e of errors) {	×
450	Server.logerror(`Failed to setup listeners: ${e.message}`)	×
451	}	×
452	return logger.dump_and_exit(-1)	×
453	}	×
454	Server.listening()	7✔
455	plugins2.run_hooks(`init_${type}`, Server)	7✔
456	}	7✔
457		1✔
458	Server.setup_http_listeners = async () => {	1✔
459	if (!Server.http?.cfg?.listen) return	7✔
460		×
461	const listeners = Server.get_listen_addrs(Server.http.cfg, 80)	×
462	if (!listeners.length) return	×
463		×
464	try {	×
465	Server.http.express = require('express')	×
466	Server.loginfo('express loaded at Server.http.express')	×
467	} catch (err) {	×
468	Server.logerror('express failed to load. No http server. Install express with: npm install -g express')	×
469	return	×
470	}	×
471		×
472	const app = Server.http.express()	×
473	Server.http.app = app	×
474	Server.loginfo('express app is at Server.http.app')	×
475		×
476	for (const listen_address of listeners) {	×
477	const ep = endpoint(listen_address, 80)	×
478	if (ep instanceof Error) {	×
479	Server.logerror(`Invalid format for listen in http.ini: ${listen_address}`)	×
480	continue	×
481	}	×
482		×
483	if (443 == ep.port) {	×
484	const tlsOpts = { ...tls_socket.certsByHost['*'] }	×
485	tlsOpts.requestCert = false // not appropriate for HTTPS	×
486	Server.http.server = require('https').createServer(tlsOpts, app)	×
487	} else {	×
488	Server.http.server = require('http').createServer(app)	×
489	}	×
490		×
491	Server.listeners.push(Server.http.server)	×
492		×
493	Server.http.server.on('listening', function () {	×
494	Server.lognotice(`Listening on ${endpoint(this.address())}`)	×
495	})	×
496		×
497	Server.http.server.on('error', (e) => {	×
498	Server.logerror(e)	×
499	})	×
500		×
501	await ep.bind(Server.http.server, { backlog: 0 })	×
502	}	×
503		×
504	Server.plugins.run_hooks('init_http', Server)	×
505	app.use(Server.http.express.static(Server.get_http_docroot()))	×
506	app.use(Server.handle404)	×
507	}	7✔
508		1✔
509	Server.init_master_respond = async (retval, msg) => {	1✔
510	if (!(retval === constants.ok \|\| retval === constants.cont)) {	7!
511	Server.logerror(`init_master returned error${msg ? `: ${msg}` : ''}`)	×
512	return logger.dump_and_exit(1)	×
513	}	×
514		7✔
515	const c = Server.cfg.main	7✔
516	Server.ready = 1	7✔
517		7✔
518	// Load the queue if we're just one process	7✔
519	if (!(cluster && c.nodes)) {	7✔
520	try {	7✔
521	await outbound.init_queue()	7✔
522	} catch (err) {	7!
NEW 523	Server.logcrit('Loading queue failed. Shutting down.')	×
NEW 524	return logger.dump_and_exit(1)	×
NEW 525	}	×
526	Server.setup_http_listeners()	7✔
527	return	7✔
528	}	7✔
529		×
530	// Running under cluster, fork children here, so that	×
531	// cluster events can be registered in init_master hooks.	×
NEW 532	try {	×
NEW 533	const pids = await outbound.scan_queue_pids()	×
534	Server.daemonize()	×
535	// Fork workers	×
536	const workers = c.nodes === 'cpus' ? os.cpus().length : c.nodes	7!
537	const new_workers = []	7✔
538	for (let i = 0; i < workers; i++) {	7!
539	new_workers.push(cluster.fork({ CLUSTER_MASTER_PID: process.pid }))	×
540	}	×
541	for (let j = 0; j < pids.length; j++) {	×
542	new_workers[j % new_workers.length].send({	×
543	event: 'outbound.load_pid_queue',	×
544	data: pids[j],	×
545	})	×
546	}	×
547	cluster.on('online', (worker) => {	×
548	Server.lognotice('worker started', {	×
549	worker: worker.id,	×
550	pid: worker.process.pid,	×
551	})	×
552	})	×
553	cluster.on('listening', (worker, address) => {	×
554	Server.lognotice(`worker ${worker.id} listening on ${endpoint(address)}`)	×
555	})	×
556	cluster.on('exit', cluster_exit_listener)	×
NEW 557	} catch (err) {	×
NEW 558	Server.logcrit('Scanning queue failed. Shutting down.')	×
NEW 559	logger.dump_and_exit(1)	×
NEW 560	}	×
561	}	7✔
562		1✔
563	function cluster_exit_listener(worker, code, signal) {	×
564	if (signal) {	×
565	Server.lognotice(`worker ${worker.id} killed by signal ${signal}`)	×
566	} else if (code !== 0) {	×
567	Server.lognotice(`worker ${worker.id} exited with error code: ${code}`)	×
568	}	×
569	if (signal \|\| code !== 0) {	×
570	// Restart worker	×
571	const new_worker = cluster.fork({	×
572	CLUSTER_MASTER_PID: process.pid,	×
573	})	×
574	new_worker.send({	×
575	event: 'outbound.load_pid_queue',	×
576	data: worker.process.pid,	×
577	})	×
578	}	×
579	}	×
580		1✔
581	Server.init_child_respond = (retval, msg) => {	1✔
582	switch (retval) {	×
583	case constants.ok:	×
584	case constants.cont:	×
585	Server.setup_http_listeners()	×
586	return	×
587	}	×
588		×
589	const pid = process.env.CLUSTER_MASTER_PID	×
590	Server.logerror(`init_child returned error ${msg ? `: ${msg}` : ''}`)	×
591	try {	×
592	if (pid) {	×
593	process.kill(pid)	×
594	Server.logerror(`Killing master (pid=${pid})`)	×
595	}	×
596	} catch (err) {	×
597	Server.logerror('Terminating child')	×
598	}	×
599	logger.dump_and_exit(1)	×
600	}	×
601		1✔
602	Server.listening = () => {	1✔
603	const c = Server.cfg.main	7✔
604		7✔
605	// Drop privileges	7✔
606	if (c.group) {	7!
607	Server.lognotice(`Switching from current gid: ${process.getgid()}`)	×
608	process.setgid(c.group)	×
609	Server.lognotice(`New gid: ${process.getgid()}`)	×
610	}	×
611	if (c.user) {	7!
612	Server.lognotice(`Switching from current uid: ${process.getuid()}`)	×
613	process.setuid(c.user)	×
614	Server.lognotice(`New uid: ${process.getuid()}`)	×
615	}	×
616		7✔
617	Server.ready = 1	7✔
618	}	7✔
619		1✔
620	Server.init_http_respond = () => {	1✔
621	Server.loginfo('init_http_respond')	×
622		×
623	let WebSocketServer	×
624	try {	×
625	WebSocketServer = require('ws').Server	×
626	} catch (e) {	×
627	Server.logerror(`unable to load ws.\n did you: npm install -g ws?`)	×
628	return	×
629	}	×
630		×
631	if (!WebSocketServer) {	×
632	Server.logerror('ws failed to load')	×
633	return	×
634	}	×
635		×
636	Server.http.wss = new WebSocketServer({ server: Server.http.server })	×
637	Server.loginfo('Server.http.wss loaded')	×
638		×
639	Server.plugins.run_hooks('init_wss', Server)	×
640	}	×
641		1✔
642	Server.init_wss_respond = () => {	1✔
643	Server.loginfo('init_wss_respond')	×
644	}	×
645		1✔
646	Server.get_http_docroot = () => {	1✔
647	if (Server.http.cfg.docroot) return Server.http.cfg.docroot	1!
648		1✔
649	Server.http.cfg.docroot = path.join(process.env.HARAKA \|\| __dirname, 'http', 'html')	1✔
650	Server.loginfo(`using html docroot: ${Server.http.cfg.docroot}`)	1✔
651	return Server.http.cfg.docroot	1✔
652	}	1✔
653		1✔
654	Server.handle404 = (req, res) => {	1✔
655	// abandon all hope, serve up a 404	×
656	const docroot = Server.get_http_docroot()	×
657		×
658	// respond with html page	×
659	if (req.accepts('html')) {	×
660	res.status(404).sendFile('404.html', { root: docroot })	×
661	return	×
662	}	×
663		×
664	// respond with json	×
665	if (req.accepts('json')) {	×
666	res.status(404).send({ err: 'Not found' })	×
667	return	×
668	}	×
669		×
670	res.status(404).send('Not found!')	×
671	}	×

haraka / Haraka / 22929902245

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous