rspamd: add native integration with file header support

[vstakhov]

Add native Rspamd integration using rspamd-client crate v0.4.0.

Features:

• File header optimization for local scanning (avoids message body transfer)
• ZSTD compression enabled by default
• HTTPCrypt encryption support
• Automatic metadata extraction (IP, HELO, sender, recipient)
• Flexible action handling (reject, tag, quarantine)

New components:

• mod-rspamd: Rust module with Lua bindings
• policy-extras/rspamd.lua: high-level Lua helper with setup() function
• examples/rspamd-integration.lua: comprehensive usage examples

The file header optimization could be particularly beneficial when Rspamd runs on the same host as KumoMTA, avoiding transmission of message bodies over the socket for significant performance improvement.

[wez]

Thank you for this!
I have quite a few comments/suggestions on how to integrate this better, and some are because of changes that only got pushed this week, likely while you weren't looking!

For testing, I'd suggest looking at the redis and amqp integrations.

For amqp, we leverage https://docs.rs/testcontainers-modules/0.13.0/testcontainers_modules/rabbitmq/struct.RabbitMq.html to run a server in a docker container:

kumomta/crates/integration-tests/src/rabbit.rs

Lines 97 to 101 in f582d60

	#[tokio::test]
	async fn test_amqprs_rabbit() -> anyhow::Result<()> {
	if std::env::var("KUMOD_TESTCONTAINERS").unwrap_or_else(|_| String::new()) != "1" {
	return Ok(());
	}

For redis, we launch an isolated copy of the redis server ourselves (no need for a container), on a port that we allocate dynamically:

kumomta/crates/mod-redis/src/test.rs

Lines 31 to 49 in f582d60

	pub async fn spawn(extra_config: &str) -> anyhow::Result<Self> {
	let mut errors = vec![];
	for _ in 0..2 {
	let port = allocate_port();
	match timeout(
	Duration::from_secs(5),
	Self::spawn_with_port(port, extra_config),
	)
	.await?
	{
	Ok(me) => return Ok(me),
	Err(err) => {
	errors.push(format!("{err:#}"));
	}
	}
	}
	anyhow::bail!("failed to spawn redis-server: {}", errors.join(". "));
	}

What I think we should have for rspamd testing is similar to those: if we find rspamd installed in a reasonable path, then we can spawn it with an isolated configuration from a unit test, then run with a policy file that will trigger rspamd testing appropriately. Look at:

kumomta/crates/integration-tests/src/test/queue_ndr.rs

Lines 13 to 17 in f582d60

	let mut daemon = DaemonWithMaildirOptions::new()
	.policy_file("ndr.lua")
	.start()
	.await
	.context("DaemonWithMaildir::start")?;

for an example of an integration test that uses a custom policy. That particular one loads in https://github.com/KumoCorp/kumomta/blob/main/crates/integration-tests/ndr.lua

[vstakhov]

Ok, it seems to be more or less functional now I hope.

[wez]

This is looking really good, thank you!
A few more comments below!

[vstakhov]

@wez it seems there might be an issue with headers rewriting in the Rust. Or maybe I don't understand the original logic...

--- a/crates/kumod/src/smtp_server.rs
+++ b/crates/kumod/src/smtp_server.rs
@@ -2619,9 +2619,12 @@ impl SmtpServerSession {
                     )
                 };

-                let mut body = Vec::with_capacity(data.len() + received.len());
+                // Use base_message.get_data() instead of original data to preserve
+                // any modifications made in smtp_server_data hook (e.g., Rspamd headers)
+                let base_data = base_message.get_data();
+                let mut body = Vec::with_capacity(base_data.len() + received.len());
                 body.extend_from_slice(received.as_bytes());
-                body.extend_from_slice(&data);
+                body.extend_from_slice(&base_data);
                 Arc::new(body.into_boxed_slice())
             } else {
                 base_message.get_data()

This change looks logical to me but please tell me if I'm wrong.

[wez]

@wez it seems there might be an issue with headers rewriting in the Rust. Or maybe I don't understand the original logic...

--- a/crates/kumod/src/smtp_server.rs
+++ b/crates/kumod/src/smtp_server.rs
@@ -2619,9 +2619,12 @@ impl SmtpServerSession {
                     )
                 };

-                let mut body = Vec::with_capacity(data.len() + received.len());
+                // Use base_message.get_data() instead of original data to preserve
+                // any modifications made in smtp_server_data hook (e.g., Rspamd headers)
+                let base_data = base_message.get_data();
+                let mut body = Vec::with_capacity(base_data.len() + received.len());
                 body.extend_from_slice(received.as_bytes());
-                body.extend_from_slice(&data);
+                body.extend_from_slice(&base_data);
                 Arc::new(body.into_boxed_slice())
             } else {
                 base_message.get_data()

This change looks logical to me but please tell me if I'm wrong.

Oh, good catch! I pushed a fix for this in 42b65b2
Thanks!

[wez]

Just submitting a couple of comments for now; I can see that you're active so I'll review again a little later!

[vstakhov]

Any updates?

[wez]

Thanks for this; sorry for the delay in reviewing since the last time, we've had a lot going on!

Most of my comments are rather nit-picky around the tests, but the core looks good.
There's a bit of tidying up to do to match the current rust-centric implementation as compared to the original more lua-centric approach.

There's something that I think we should do to enable "connection pooling" (or rather, client-object-caching), but I think that we can do that in a follow-up PR: I'd like to get this one buttoned up so that the caching can be in its own smaller PR.