Files
kumomta/crates/integration-tests/maildir-sink.lua
T
Wez Furlong 7f4a929ca2 message transfer (xfer) support
This commit implements a kumomta-specific message transfer
protocol that is intended to be used to migrate messages
from one kumomta node to another.

The transfer is carried out using an HTTP POST request
to the destination node's http listener.

The request includes the full message metadata and body,
in a compressed form.

An xfer request can be made via `kcli xfer` (and thus also via an HTTP API
endpoint).  It works similarly to a rebind operation; you specify the
criteria to be used to match scheduled queues, along with the target
node for the xfer, and kumomta will find matching queues, drain out the
messages, make an adjustment to the metadata to capture current
scheduling information, and then place the messages into an xfer queue.

The xfer queue has hard-coded scheduling queue configuration of its own,
with the base retry interval set to 10 seconds, which should be suitably
aggressive for the intended use case.

You may apply shaping to affect the number of concurrent requests in a
similar way to how TSA shaping is configured.

On the receiving side, the incoming xfer sanity check to prohibit
trying to xfer to itself.

The spool id of the Message is not suitable to be reused verbatim on
another node (spool ids include the local mac address and creation
timestamp information, as well as a random component), so the receiving
side will derive an id that should be suitable for use on that node.
The originating node id and spool id will be preserved in metadata to
aid in tracing.

It is possible for an xfer request to target an existing xfer queue, so
that you can correct/update the target in various circumstances. In that
situation the messages will be "simply" moved from the source queue to
the destination queue.

It is possible to cancel an xfer request via `kcli xfer-cancel` (and
thus also via an HTTP API endpoint).  You specify the target queue,
which must be an xfer queue, and it will have its messages drained and
the metadata changes that were applied when the xfer was initiated will
be reversed, allowing the messages to then be reinserted into their
originating queue.

refs: https://github.com/KumoCorp/kumomta/issues/311
2025-10-07 10:58:07 +01:00

140 lines
3.4 KiB
Lua

-- This config acts as a sink that capture all received mail into a maildir
local kumo = require 'kumo'
package.path = '../../assets/?.lua;' .. package.path
local utils = require 'policy-extras.policy_utils'
local TEST_DIR = os.getenv 'KUMOD_TEST_DIR'
kumo.on('init', function()
kumo.configure_accounting_db_path ':memory:'
local smtp_params = {
listen = '127.0.0.1:0',
relay_hosts = { '0.0.0.0/0' },
batch_handling = 'BatchByDomain',
max_recipients_per_message = 4,
}
local client_ca = os.getenv 'KUMOD_CLIENT_REQUIRED_CA'
if client_ca then
smtp_params.tls_required_client_ca = {
key_data = client_ca,
}
end
kumo.start_esmtp_listener(smtp_params)
kumo.start_http_listener {
listen = '127.0.0.1:0',
}
kumo.configure_local_logs {
log_dir = TEST_DIR .. '/logs',
max_segment_duration = '1s',
meta = {
'*',
},
}
kumo.define_spool {
name = 'data',
path = TEST_DIR .. '/data-spool',
}
kumo.define_spool {
name = 'meta',
path = TEST_DIR .. '/meta-spool',
}
end)
kumo.on('xfer_message_received', function(msg)
-- This is coupled with the xfer integration tests
msg:set_meta('sunk', 'received via xfer')
end)
local function apply_rejections(envelope_addr)
if string.find(envelope_addr.user, 'tempfail') then
kumo.reject(400, 'tempfail requested')
end
if string.find(envelope_addr.user, 'permfail') then
kumo.reject(500, 'permfail requested')
end
if utils.starts_with(envelope_addr.user, '450-') then
kumo.reject(450, 'you said ' .. envelope_addr.user)
end
if utils.starts_with(envelope_addr.user, '421-') then
kumo.reject(421, 'disconnecting ' .. envelope_addr.user)
end
if utils.starts_with(envelope_addr.user, '550-') then
kumo.reject(550, 'you said ' .. envelope_addr.user)
end
end
kumo.on('smtp_server_rcpt_to', function(recipient)
apply_rejections(recipient)
end)
kumo.on('smtp_server_mail_from', function(sender)
apply_rejections(sender)
end)
kumo.on('smtp_server_message_received', function(msg)
local sender = msg:sender().user
if utils.starts_with(sender, 'disconnect-in-data-no-421') then
kumo.disconnect(451, 'disconnecting ' .. sender, 'ForceDisconnect')
end
msg:set_meta('queue', 'maildir')
end)
kumo.on('get_queue_config', function(_domain, _tenant, _campaign)
return kumo.make_queue_config {
protocol = {
maildir_path = TEST_DIR .. '/maildir',
},
}
end)
function simple_auth_check(user, password)
local password_database = {
['daniel'] = 'tiger',
}
if password == '' then
return false
end
return password_database[user] == password
end
kumo.on('http_server_validate_auth_basic', function(user, password)
return simple_auth_check(user, password)
end)
kumo.on('smtp_server_auth_plain', function(authz, authc, password)
print(
string.format(
"AUTH PLAIN: authz='%s' authc='%s' pass='%s'",
authz,
authc,
password
)
)
return simple_auth_check(authc, password)
end)
kumo.on('smtp_server_ehlo', function(domain, conn_meta, extensions)
local revised = {}
for _, ext in ipairs(extensions) do
local include = true
if ext == '8BITMIME' and os.getenv 'KUMOD_HIDE_8BITMIME' then
include = false
end
if ext == 'SMTPUTF8' and os.getenv 'KUMOD_HIDE_SMTPUTF8' then
include = false
end
if include then
table.insert(revised, ext)
end
end
return revised
end)