email: outbound mail subsystem and RESETPASS

Adds a config-driven mailer: a SendEmail action the link layer pipes to
a configured command (sendmail -t, msmtp, ...) off-thread. NickServ
RESETPASS <account> emails a short-lived code to the address on file;
RESETPASS <account> <code> <newpassword> completes the reset. Codes are
node-local; the password change itself federates.

src/config.rs

@@ -9,6 +9,18 @@ pub struct Config {
     pub gossip: Option<Gossip>,
     #[serde(default)]
     pub peer: Vec<Peer>,
+    // Outbound email (password resets). Absent = email features are off.
+    #[serde(default)]
+    pub email: Option<Email>,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct Email {
+    // Sender address stamped on outgoing mail.
+    pub from: String,
+    // A shell command the message is piped to on stdin, e.g. "sendmail -t" or
+    // "msmtp -t". Run via `sh -c`, so redirection and pipes work.
+    pub command: String,
 }
 
 #[derive(Debug, Deserialize, Clone)]

src/engine/db.rs

@@ -1,9 +1,9 @@
 use std::collections::HashMap;
 use std::io::Write;
 use std::path::PathBuf;
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
-use argon2::password_hash::rand_core::OsRng;
+use argon2::password_hash::rand_core::{OsRng, RngCore};
 use argon2::password_hash::SaltString;
 use argon2::{Argon2, PasswordHash, PasswordHasher, PasswordVerifier};
 use serde::{Deserialize, Serialize};
@@ -443,6 +443,10 @@ pub struct Db {
     log: EventLog,
     // PBKDF2 cost baked into new SCRAM verifiers; lowered by tests.
     pub(crate) scram_iterations: u32,
+    // Whether outbound email is configured, so email features can gate themselves.
+    email_enabled: bool,
+    // Node-local, non-persisted password-reset codes: account -> (code, expiry).
+    reset_codes: HashMap<String, (String, Instant)>,
 }
 
 fn key(name: &str) -> String {
@@ -482,7 +486,7 @@ impl Db {
             apply(&mut accounts, &mut channels, &mut grouped, event);
         }
         tracing::info!(accounts = accounts.len(), channels = channels.len(), "account store loaded");
-        Self { accounts, channels, grouped, log, scram_iterations: scram::DEFAULT_ITERATIONS }
+        Self { accounts, channels, grouped, log, scram_iterations: scram::DEFAULT_ITERATIONS, email_enabled: false, reset_codes: HashMap::new() }
     }
 
     /// Fold an entry authored by another node into the store — the services-side
@@ -680,6 +684,34 @@ impl Db {
         self.accounts.get(&key(account)).map_or(&[], |a| a.certfps.as_slice())
     }
 
+    /// Whether outbound email is configured.
+    pub fn email_enabled(&self) -> bool {
+        self.email_enabled
+    }
+
+    pub fn set_email_enabled(&mut self, on: bool) {
+        self.email_enabled = on;
+    }
+
+    /// Issue a fresh password-reset code for `account`, valid for 15 minutes.
+    pub fn issue_reset_code(&mut self, account: &str) -> String {
+        let code = gen_code();
+        self.reset_codes.insert(key(account), (code.clone(), Instant::now() + Duration::from_secs(900)));
+        code
+    }
+
+    /// Consume a reset code for `account`: true if it matches and hasn't expired.
+    pub fn take_reset_code(&mut self, account: &str, code: &str) -> bool {
+        let k = key(account);
+        match self.reset_codes.get(&k) {
+            Some((c, deadline)) if c == code && *deadline > Instant::now() => {
+                self.reset_codes.remove(&k);
+                true
+            }
+            _ => false,
+        }
+    }
+
     /// Set (or clear) `account`'s email.
     pub fn set_email(&mut self, account: &str, email: Option<String>) -> Result<(), RegError> {
         let k = key(account);
@@ -1062,6 +1094,13 @@ fn glob_match(pattern: &str, text: &str) -> bool {
     pi == p.len()
 }
 
+// A random 6-digit code for email verification / password reset.
+fn gen_code() -> String {
+    let mut b = [0u8; 4];
+    OsRng.fill_bytes(&mut b);
+    format!("{:06}", u32::from_le_bytes(b) % 1_000_000)
+}
+
 fn hash_password(password: &str) -> Option<String> {
     let salt = SaltString::generate(&mut OsRng);
     Argon2::default().hash_password(password.as_bytes(), &salt).ok().map(|h| h.to_string())

src/engine/mod.rs

@@ -1210,6 +1210,40 @@ mod tests {
         assert!(notice(&to_ns(&mut e, "000AAAAAC", "IDENTIFY sesame"), "isn't registered"));
     }
 
+    // RESETPASS emails a code, and that code + a new password completes the reset;
+    // the new password then authenticates and the old one no longer does.
+    #[test]
+    fn nickserv_resetpass() {
+        let mut e = engine_with("nsreset", "alice", "sesame");
+        e.db.set_email_enabled(true);
+        e.db.set_email("alice", Some("alice@example.org".into())).unwrap();
+        e.handle(NetEvent::UserConnect { uid: "000AAAAAB".into(), nick: "someone".into(), host: "h".into() });
+        let to_ns = |e: &mut Engine, text: &str| e.handle(NetEvent::Privmsg { from: "000AAAAAB".into(), to: "42SAAAAAA".into(), text: text.into() });
+
+        // Request: a code is emailed to the address on file.
+        let out = to_ns(&mut e, "RESETPASS alice");
+        let (to, body) = out.iter().find_map(|a| match a {
+            NetAction::SendEmail { to, body, .. } => Some((to.clone(), body.clone())),
+            _ => None,
+        }).expect("a reset code is emailed");
+        assert_eq!(to, "alice@example.org");
+        let code = body.split("is: ").nth(1).unwrap().split_whitespace().next().unwrap().to_string();
+
+        // Complete: the code + a new password defers the change.
+        let out = to_ns(&mut e, &format!("RESETPASS alice {code} brandnew"));
+        let (account, password, agent, uid) = out.iter().find_map(|a| match a {
+            NetAction::DeferPassword { account, password, agent, uid } => Some((account.clone(), password.clone(), agent.clone(), uid.clone())),
+            _ => None,
+        }).expect("a valid code defers the new password");
+        assert_eq!(password, "brandnew");
+        e.complete_password_change(&account, Db::derive_credentials(&password, 4096), &agent, &uid);
+        assert!(e.db.authenticate("alice", "brandnew").is_some(), "new password works");
+        assert!(e.db.authenticate("alice", "sesame").is_none(), "old password rejected");
+
+        // A used/wrong code is refused.
+        assert!(to_ns(&mut e, "RESETPASS alice 000000 x").iter().any(|a| matches!(a, NetAction::Notice { text, .. } if text.contains("Invalid or expired"))));
+    }
+
     // GHOST renames off a session using a nick the caller owns.
     #[test]
     fn nickserv_ghost() {

src/engine/service.rs

@@ -57,6 +57,11 @@ impl ServiceCtx {
         });
     }
 
+    // Send an email (the link layer pipes it to the configured mail command).
+    pub fn send_email(&mut self, to: impl Into<String>, subject: impl Into<String>, body: impl Into<String>) {
+        self.actions.push(NetAction::SendEmail { to: to.into(), subject: subject.into(), body: body.into() });
+    }
+
     // Hand a password change to the engine to finish: its derivation runs off the
     // reactor, then the engine commits it and notices `uid`, sourced from `agent`.
     pub fn defer_password(&mut self, account: impl Into<String>, password: impl Into<String>, agent: impl Into<String>, uid: impl Into<String>) {

src/link.rs

@@ -12,7 +12,7 @@ use crate::proto::{NetAction, Protocol};
 // One uplink session: connect, handshake + burst, then translate lines forever.
 // The engine is shared with the gossip layer, so it is locked per operation and
 // never held across the registration key-stretching await.
-pub async fn run(mut proto: Box<dyn Protocol>, engine: Arc<Mutex<Engine>>, addr: &str, mut irc_rx: mpsc::UnboundedReceiver<NetAction>) -> Result<()> {
+pub async fn run(mut proto: Box<dyn Protocol>, engine: Arc<Mutex<Engine>>, addr: &str, mut irc_rx: mpsc::UnboundedReceiver<NetAction>, email: Option<crate::config::Email>) -> Result<()> {
     let stream = TcpStream::connect(addr).await?;
     let (read, mut write) = stream.into_split();
     let mut lines = BufReader::new(read).lines();
@@ -66,6 +66,10 @@ pub async fn run(mut proto: Box<dyn Protocol>, engine: Arc<Mutex<Engine>>, addr:
                             action => vec![action],
                         };
                         for act in outs {
+                            if let NetAction::SendEmail { to, subject, body } = act {
+                                dispatch_email(&email, to, subject, body);
+                                continue;
+                            }
                             for out in proto.serialize(&act) {
                                 send(&mut write, &out).await?;
                             }
@@ -76,8 +80,12 @@ pub async fn run(mut proto: Box<dyn Protocol>, engine: Arc<Mutex<Engine>>, addr:
             // A services-initiated action (e.g. a forced logout after a lost
             // registration conflict), pushed by the engine from the gossip path.
             Some(action) = irc_rx.recv() => {
-                for out in proto.serialize(&action) {
-                    send(&mut write, &out).await?;
+                if let NetAction::SendEmail { to, subject, body } = action {
+                    dispatch_email(&email, to, subject, body);
+                } else {
+                    for out in proto.serialize(&action) {
+                        send(&mut write, &out).await?;
+                    }
                 }
             }
         }
@@ -87,6 +95,33 @@ pub async fn run(mut proto: Box<dyn Protocol>, engine: Arc<Mutex<Engine>>, addr:
     Ok(())
 }
 
+// Fire off an email if email is configured: pipe an RFC822 message to the mail
+// command on a spawned task, so a slow MTA never stalls the link.
+fn dispatch_email(email: &Option<crate::config::Email>, to: String, subject: String, body: String) {
+    let Some(email) = email.clone() else {
+        return tracing::warn!(%to, "email requested but not configured");
+    };
+    tokio::spawn(async move {
+        let msg = format!("From: {}\r\nTo: {to}\r\nSubject: {subject}\r\n\r\n{body}\r\n", email.from);
+        let child = tokio::process::Command::new("sh")
+            .arg("-c")
+            .arg(&email.command)
+            .stdin(std::process::Stdio::piped())
+            .stdout(std::process::Stdio::null())
+            .stderr(std::process::Stdio::null())
+            .spawn();
+        let mut child = match child {
+            Ok(c) => c,
+            Err(e) => return tracing::warn!(%e, "mail command failed to start"),
+        };
+        if let Some(mut stdin) = child.stdin.take() {
+            let _ = stdin.write_all(msg.as_bytes()).await;
+            let _ = stdin.shutdown().await;
+        }
+        let _ = child.wait().await;
+    });
+}
+
 async fn send(write: &mut (impl AsyncWriteExt + Unpin), line: &str) -> Result<()> {
     tracing::debug!(dir = ">>", %line);
     write.write_all(line.as_bytes()).await?;

src/main.rs

@@ -55,6 +55,7 @@ async fn main() -> Result<()> {
     let mut db = engine::db::Db::open("fedserv.db.jsonl", &cfg.server.sid);
     db.scram_iterations = cfg.server.scram_iterations;
     db.set_outbound(gossip_tx.clone());
+    db.set_email_enabled(cfg.email.is_some());
     let engine = Arc::new(Mutex::new(Engine::new(services, db)));
 
     // Channel for services-initiated actions to reach the uplink (drained by the link loop).
@@ -81,5 +82,5 @@ async fn main() -> Result<()> {
 
     let addr = format!("{}:{}", cfg.uplink.host, cfg.uplink.port);
     tracing::info!(server = %cfg.server.name, %addr, "linking to uplink");
-    link::run(proto, engine, &addr, irc_rx).await
+    link::run(proto, engine, &addr, irc_rx, cfg.email.clone()).await
 }