add integration tests for smtp auth

refs: https://github.com/KumoCorp/kumomta/issues/10

crates/integration-tests/maildir-sink.lua

@@ -45,3 +45,29 @@ kumo.on('get_queue_config', function(domain, tenant, campaign)
     },
   }
 end)
+
+function simple_auth_check(user, password)
+  local password_database = {
+    ['scott'] = '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)

crates/integration-tests/sink.lua

@@ -29,3 +29,29 @@ kumo.on('smtp_server_message_received', function(msg)
   -- Accept and discard all messages
   msg:set_meta('queue', 'null')
 end)
+
+function simple_auth_check(user, password)
+  local password_database = {
+    ['scott'] = '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)

crates/integration-tests/source.lua

@@ -88,9 +88,21 @@ end)
 
 kumo.on('get_egress_path_config', function(domain, source_name, site_name)
   -- Allow sending to a sink
-  return kumo.make_egress_path {
+  local params = {
     enable_tls = 'OpportunisticInsecure',
     smtp_port = SINK_PORT,
     prohibited_hosts = {},
   }
+
+  local username = os.getenv 'KUMOD_SMTP_AUTH_USERNAME'
+  local password = os.getenv 'KUMOD_SMTP_AUTH_PASSWORD'
+
+  if username and password then
+    params.smtp_auth_plain_username = username
+    params.smtp_auth_plain_password = {
+      key_data = password,
+    }
+  end
+
+  return kumo.make_egress_path(params)
 end)

crates/integration-tests/src/kumod.rs

@@ -9,7 +9,7 @@ use std::net::SocketAddr;
 use std::process::Stdio;
 use std::time::Duration;
 use tempfile::TempDir;
-use tokio::io::{AsyncBufReadExt, BufReader};
+use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};
 use tokio::process::{Child, Command};
 
 #[derive(Debug, Default, Clone)]
@@ -111,11 +111,23 @@ pub struct DaemonWithMaildir {
 
 impl DaemonWithMaildir {
     pub async fn start() -> anyhow::Result<Self> {
+        Self::start_with_env(vec![]).await
+    }
+
+    pub async fn start_with_env(env: Vec<(&str, &str)>) -> anyhow::Result<Self> {
         let sink = KumoDaemon::spawn_maildir().await?;
         let smtp = sink.listener("smtp");
+
+        let mut env: Vec<(String, String)> = env
+            .into_iter()
+            .map(|(k, v)| (k.to_string(), v.to_string()))
+            .collect();
+
+        env.push(("KUMOD_SMTP_SINK_PORT".to_string(), smtp.port().to_string()));
+
         let source = KumoDaemon::spawn(KumoArgs {
             policy_file: "source.lua".to_string(),
-            env: vec![("KUMOD_SMTP_SINK_PORT".to_string(), smtp.port().to_string())],
+            env,
         })
         .await?;
 
@@ -153,8 +165,7 @@ impl DaemonWithMaildir {
                         }
                     }
 
-
-                        tokio::time::sleep(Duration::from_millis(100)).await;
+                    tokio::time::sleep(Duration::from_millis(100)).await;
                 }
             } => true,
             _ = tokio::time::sleep(timeout) => false,
@@ -240,7 +251,32 @@ impl KumoDaemon {
 
         // Send stdout to stderr
         let mut stdout = child.stdout.take().unwrap();
-        tokio::spawn(async move { tokio::io::copy(&mut stdout, &mut tokio::io::stderr()).await });
+
+        async fn copy_stream_with_line_prefix<SRC, DEST>(
+            prefix: &str,
+            src: SRC,
+            mut dest: DEST,
+        ) -> std::io::Result<()>
+        where
+            SRC: AsyncRead + Unpin,
+            DEST: AsyncWrite + Unpin,
+        {
+            let mut src = tokio::io::BufReader::new(src);
+            loop {
+                let mut line = String::new();
+                src.read_line(&mut line).await?;
+                if !line.is_empty() {
+                    dest.write_all(format!("{prefix}: {line}").as_bytes())
+                        .await?;
+                }
+            }
+        }
+
+        let stdout_prefix = format!("{} stdout", &args.policy_file);
+        tokio::spawn(async move {
+            copy_stream_with_line_prefix(&stdout_prefix, &mut stdout, &mut tokio::io::stderr())
+                .await
+        });
 
         // Wait until the server initializes, collect the information
         // about the various listeners that it starts
@@ -270,7 +306,11 @@ impl KumoDaemon {
         }
 
         // Now just pipe the output through to the test harness
-        tokio::spawn(async move { tokio::io::copy(&mut stderr, &mut tokio::io::stderr()).await });
+        let stderr_prefix = format!("{} stderr", &args.policy_file);
+        tokio::spawn(async move {
+            copy_stream_with_line_prefix(&stderr_prefix, &mut stderr, &mut tokio::io::stderr())
+                .await
+        });
 
         Ok(Self {
             child,

crates/integration-tests/src/main.rs

@@ -189,6 +189,96 @@ DeliverySummary {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn auth_deliver() -> anyhow::Result<()> {
+        let mut daemon = DaemonWithMaildir::start_with_env(vec![
+            ("KUMOD_SMTP_AUTH_USERNAME", "scott"),
+            ("KUMOD_SMTP_AUTH_PASSWORD", "tiger"),
+        ])
+        .await?;
+
+        let mut client = daemon.smtp_client().await?;
+
+        let body = generate_message_text(1024, 78);
+        let response = MailGenParams {
+            body: Some(&body),
+            ..Default::default()
+        }
+        .send(&mut client)
+        .await?;
+        anyhow::ensure!(response.code == 250);
+
+        daemon
+            .wait_for_maildir_count(1, Duration::from_secs(10))
+            .await;
+
+        daemon.stop_both().await?;
+        println!("Stopped!");
+
+        let delivery_summary = daemon.dump_logs()?;
+        k9::snapshot!(
+            delivery_summary,
+            "
+DeliverySummary {
+    source_counts: {
+        Reception: 1,
+        Delivery: 1,
+    },
+    sink_counts: {
+        Reception: 1,
+        Delivery: 1,
+    },
+}
+"
+        );
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn auth_deliver_invalid_password() -> anyhow::Result<()> {
+        let mut daemon = DaemonWithMaildir::start_with_env(vec![
+            ("KUMOD_SMTP_AUTH_USERNAME", "scott"),
+            ("KUMOD_SMTP_AUTH_PASSWORD", "incorrect-password"),
+        ])
+        .await?;
+
+        let mut client = daemon.smtp_client().await?;
+
+        let body = generate_message_text(1024, 78);
+        let response = MailGenParams {
+            body: Some(&body),
+            ..Default::default()
+        }
+        .send(&mut client)
+        .await?;
+        anyhow::ensure!(response.code == 250);
+
+        daemon
+            .wait_for_source_summary(
+                |summary| summary.get(&TransientFailure).copied().unwrap_or(0) > 0,
+                Duration::from_secs(5),
+            )
+            .await;
+
+        daemon.stop_both().await?;
+        println!("Stopped!");
+
+        let delivery_summary = daemon.dump_logs()?;
+        k9::snapshot!(
+            delivery_summary,
+            "
+DeliverySummary {
+    source_counts: {
+        Reception: 1,
+        TransientFailure: 1,
+    },
+    sink_counts: {},
+}
+"
+        );
+        Ok(())
+    }
+
     /// Verify that what we send in transits through and is delivered
     /// into the maildir at the other end with the same content,
     /// and that the webhook logging is also used and captures