netget/server/lmtp/
mod.rs

1//! LMTP (RFC 2033) delivery server. Rust owns the SMTP-style state machine, framing, bounds
2//! and the one-reply-per-recipient rule after DATA; handlers decide which recipients exist
3//! and what happened to each delivery. Nothing is stored here.
4pub mod actions;
5pub mod wire;
6
7use crate::llm::action_helper::call_llm;
8use crate::llm::actions::protocol_trait::ActionResult;
9use crate::logging::emit::Log;
10use crate::protocol::{Event, SpawnContext};
11use crate::server::connection::ConnectionId;
12use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
13use anyhow::{Context, Result};
14use serde_json::{json, Value};
15use std::{collections::HashMap, net::SocketAddr, time::Duration};
16use tokio::{
17    io::{AsyncWriteExt, BufReader},
18    net::{TcpListener, TcpStream},
19};
20use wire::Line;
21
22/// RFC 5321 4.5.3.2.7 suggests a five-minute server timeout between commands.
23pub const IDLE_TIMEOUT: Duration = Duration::from_secs(300);
24pub const MAX_IDLE_TIMEOUT_SECS: u64 = 3600;
25pub const DEFAULT_HOSTNAME: &str = "netget";
26/// Consecutive refused commands before the session is closed with 421.
27pub const MAX_BAD_COMMANDS: usize = 20;
28
29#[derive(Clone)]
30struct Config {
31    hostname: String,
32    max_message: u64,
33    idle: Duration,
34}
35
36fn config(ctx: &SpawnContext) -> Result<Config> {
37    let params = ctx.startup_params.as_ref();
38    let hostname = params
39        .map(|p| p.get_optional_string("hostname"))
40        .transpose()?
41        .flatten()
42        .unwrap_or_else(|| DEFAULT_HOSTNAME.to_string());
43    anyhow::ensure!(
44        !hostname.is_empty()
45            && hostname.len() <= 255
46            && hostname.chars().all(|c| c.is_ascii_graphic()),
47        "hostname must be 1..=255 printable ASCII characters without spaces"
48    );
49    let max_message = params
50        .map(|p| p.get_optional_u64("max_message_bytes"))
51        .transpose()?
52        .flatten()
53        .unwrap_or(wire::DEFAULT_MAX_MESSAGE_BYTES);
54    anyhow::ensure!(
55        (1..=wire::MAX_MESSAGE_BYTES_LIMIT).contains(&max_message),
56        "max_message_bytes must be between 1 and {}",
57        wire::MAX_MESSAGE_BYTES_LIMIT
58    );
59    let idle = params
60        .map(|p| p.get_optional_u64("idle_timeout_secs"))
61        .transpose()?
62        .flatten()
63        .unwrap_or(IDLE_TIMEOUT.as_secs());
64    anyhow::ensure!(
65        (1..=MAX_IDLE_TIMEOUT_SECS).contains(&idle),
66        "idle_timeout_secs must be between 1 and {MAX_IDLE_TIMEOUT_SECS}"
67    );
68    Ok(Config {
69        hostname,
70        max_message,
71        idle: Duration::from_secs(idle),
72    })
73}
74
75pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
76    let cfg = config(&ctx)?;
77    let listener = TcpListener::bind(ctx.legacy_listen_addr()).await?;
78    let addr = listener.local_addr()?;
79    Log::new(Some(&ctx.status_tx)).info(format!("LMTP listening on {addr}"));
80    let state = ctx.state.clone();
81    let server_id = ctx.server_id;
82    let accept = tokio::spawn(async move {
83        let limiter = crate::server::accept_bounded::ConnectionLimiter::new(
84            crate::server::accept_bounded::DEFAULT_MAX_CONNECTIONS,
85        );
86        loop {
87            let (socket, peer, permit) = match crate::server::accept_bounded::accept_bounded(
88                &listener,
89                &limiter,
90                b"421 4.3.2 Too many connections, try again later\r\n",
91                "LMTP",
92                Some(&ctx.status_tx),
93            )
94            .await
95            {
96                Ok(v) => v,
97                Err(_) => break,
98            };
99            let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
100            let now = crate::utils::clock::Instant::now();
101            ctx.state
102                .add_connection_to_server(
103                    server_id,
104                    ConnectionState {
105                        id,
106                        remote_addr: peer,
107                        local_addr: addr,
108                        bytes_sent: 0,
109                        bytes_received: 0,
110                        packets_sent: 0,
111                        packets_received: 0,
112                        last_activity: now,
113                        status: ConnectionStatus::Active,
114                        status_changed_at: now,
115                        protocol_info: ProtocolConnectionInfo::empty(),
116                    },
117                )
118                .await;
119            let child = ctx.clone();
120            let cfg = cfg.clone();
121            ctx.state
122                .spawn_server_task(server_id, async move {
123                    let _permit = permit;
124                    if let Err(e) = session(&child, id, socket, &cfg).await {
125                        Log::new(Some(&child.status_tx))
126                            .warn(format!("LMTP connection {id} ended: {e}"));
127                    }
128                    child
129                        .state
130                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
131                        .await;
132                    let _ = child.status_tx.send("__UPDATE_UI__".into());
133                })
134                .await;
135        }
136    });
137    state.register_server_task(server_id, accept).await;
138    Ok(addr)
139}
140
141async fn write<W: tokio::io::AsyncWrite + Unpin>(
142    ctx: &SpawnContext,
143    id: ConnectionId,
144    w: &mut W,
145    reply: &str,
146) -> Result<()> {
147    tokio::time::timeout(wire::IO_TIMEOUT, w.write_all(reply.as_bytes()))
148        .await
149        .context("LMTP write deadline")??;
150    ctx.state
151        .update_connection_stats(
152            ctx.server_id,
153            id,
154            None,
155            Some(reply.len() as u64),
156            None,
157            Some(1),
158        )
159        .await;
160    Ok(())
161}
162
163fn outcome(ctx: &SpawnContext, id: ConnectionId, operation: &str, decision: &str) {
164    let summary = format!("LMTP connection {id} operation={operation} decision={decision}");
165    let log = Log::new(Some(&ctx.status_tx));
166    if decision.starts_with("fail_closed_") || decision == "model_silent" {
167        log.error(summary);
168    } else {
169        log.info(summary);
170    }
171}
172
173/// Ask the handler and return the single expected action, logging how a failure happened.
174async fn decision(
175    ctx: &SpawnContext,
176    id: ConnectionId,
177    event: Event,
178    expected: &str,
179) -> Result<Value> {
180    let result = match call_llm(
181        &ctx.llm_client,
182        &ctx.state,
183        ctx.server_id,
184        Some(id),
185        &event,
186        &actions::LmtpProtocol,
187    )
188    .await
189    {
190        Ok(result) => result,
191        Err(error) => {
192            outcome(ctx, id, event.id(), "fail_closed_llm_error");
193            return Err(error);
194        }
195    };
196    if !result.failures.is_empty() {
197        outcome(ctx, id, event.id(), "fail_closed_invalid_reply");
198        anyhow::bail!("LMTP handler supplied an invalid action");
199    }
200    let mut found = None;
201    let mut pending = result.protocol_results;
202    while let Some(result) = pending.pop() {
203        match result {
204            ActionResult::Custom { name, data } if name == expected => {
205                if found.is_some() {
206                    outcome(ctx, id, event.id(), "fail_closed_invalid_reply");
207                    anyhow::bail!("Multiple LMTP answers to one event");
208                }
209                found = Some(data);
210            }
211            ActionResult::Multiple(items) => pending.extend(items),
212            _ => {}
213        }
214    }
215    if found.is_none() {
216        outcome(ctx, id, event.id(), "model_silent");
217    }
218    found.context("Handler did not answer the LMTP event")
219}
220
221const TEMPORARY_FAILURE: &str = "451 4.3.0 Temporary local failure, try again later\r\n";
222
223/// One reply per accepted recipient, in RCPT order (RFC 2033 4.2).
224fn delivery_replies(
225    ctx: &SpawnContext,
226    id: ConnectionId,
227    recipients: &[String],
228    answer: &Value,
229) -> String {
230    let mut listed: HashMap<String, &Value> = HashMap::new();
231    if let Some(results) = answer["results"].as_array() {
232        for result in results {
233            if let Some(recipient) = result["recipient"].as_str() {
234                listed
235                    .entry(recipient.to_ascii_lowercase())
236                    .or_insert(result);
237            }
238        }
239    }
240    let mut replies = String::new();
241    for recipient in recipients {
242        let reply = match listed.get(&recipient.to_ascii_lowercase()) {
243            Some(result) => {
244                let reason = result["reason"].as_str();
245                if result["delivered"] == true {
246                    outcome(ctx, id, "lmtp_message", "model_answer");
247                    wire::render(
248                        250,
249                        "2.0.0",
250                        reason.unwrap_or(&format!("<{recipient}> delivered")),
251                    )
252                } else if result["temporary"] == true {
253                    outcome(ctx, id, "lmtp_message", "model_reject");
254                    wire::render(
255                        450,
256                        "4.2.0",
257                        reason.unwrap_or("Mailbox temporarily unavailable"),
258                    )
259                } else {
260                    outcome(ctx, id, "lmtp_message", "model_reject");
261                    wire::render(550, "5.0.0", reason.unwrap_or("Delivery refused"))
262                }
263            }
264            None => match answer["deliver_all"].as_bool() {
265                Some(true) => {
266                    outcome(ctx, id, "lmtp_message", "model_answer");
267                    wire::render(250, "2.0.0", &format!("<{recipient}> delivered"))
268                }
269                Some(false) => {
270                    outcome(ctx, id, "lmtp_message", "model_reject");
271                    wire::render(550, "5.0.0", "Delivery refused")
272                }
273                None => {
274                    outcome(ctx, id, "lmtp_message", "fail_closed_incomplete_reply");
275                    TEMPORARY_FAILURE.to_string()
276                }
277            },
278        };
279        replies.push_str(&reply);
280    }
281    replies
282}
283
284enum Data {
285    Message(Vec<u8>),
286    TooBig,
287    BadLine,
288}
289
290/// Read DATA through the terminating dot, keeping at most `max` bytes.
291async fn read_data<R: tokio::io::AsyncBufRead + Unpin>(
292    reader: &mut R,
293    cfg: &Config,
294) -> Result<(Data, u64)> {
295    let mut message = Vec::new();
296    let mut total: u64 = 0;
297    let mut too_big = false;
298    let mut bad_line = false;
299    loop {
300        match wire::read_line(reader, cfg.idle).await? {
301            Line::Text(line) => {
302                if line == "." {
303                    break;
304                }
305                let line = line.strip_prefix('.').unwrap_or(&line);
306                total = total.saturating_add(line.len() as u64 + 2);
307                if total > cfg.max_message {
308                    too_big = true;
309                    message = Vec::new();
310                } else if !too_big {
311                    message.extend_from_slice(line.as_bytes());
312                    message.extend_from_slice(b"\r\n");
313                }
314            }
315            Line::TooLong => {
316                bad_line = true;
317                total = total.saturating_add(wire::MAX_LINE_BYTES as u64);
318            }
319            Line::Eof => anyhow::bail!("LMTP peer closed the connection during DATA"),
320        }
321    }
322    let data = if bad_line {
323        Data::BadLine
324    } else if too_big {
325        Data::TooBig
326    } else {
327        Data::Message(message)
328    };
329    Ok((data, total))
330}
331
332async fn session(
333    ctx: &SpawnContext,
334    id: ConnectionId,
335    socket: TcpStream,
336    cfg: &Config,
337) -> Result<()> {
338    let (read, mut out) = tokio::io::split(socket);
339    let mut reader = BufReader::new(read);
340    write(
341        ctx,
342        id,
343        &mut out,
344        &wire::render(220, "", &format!("{} LMTP NetGet ready", cfg.hostname)),
345    )
346    .await?;
347    let mut lhlo: Option<String> = None;
348    let mut from: Option<String> = None;
349    let mut recipients: Vec<String> = Vec::new();
350    let mut bad = 0usize;
351    loop {
352        let line = match wire::read_line(&mut reader, cfg.idle).await {
353            Ok(Line::Text(line)) => line,
354            Ok(Line::TooLong) => {
355                bad += 1;
356                write(
357                    ctx,
358                    id,
359                    &mut out,
360                    &wire::render(500, "5.5.2", "Line too long"),
361                )
362                .await?;
363                if bad >= MAX_BAD_COMMANDS {
364                    write(
365                        ctx,
366                        id,
367                        &mut out,
368                        &wire::render(421, "4.7.0", "Too many errors"),
369                    )
370                    .await?;
371                    return Ok(());
372                }
373                continue;
374            }
375            Ok(Line::Eof) => return Ok(()),
376            Err(error) => {
377                let _ = write(
378                    ctx,
379                    id,
380                    &mut out,
381                    &wire::render(421, "4.4.2", "Idle timeout, closing"),
382                )
383                .await;
384                return Err(error);
385            }
386        };
387        ctx.state
388            .update_connection_stats(
389                ctx.server_id,
390                id,
391                Some(line.len() as u64 + 2),
392                None,
393                Some(1),
394                None,
395            )
396            .await;
397        let (verb, argument) = match line.split_once(' ') {
398            Some((verb, rest)) => (verb.to_ascii_uppercase(), rest.trim().to_string()),
399            None => (line.trim().to_ascii_uppercase(), String::new()),
400        };
401        let reply = match verb.as_str() {
402            "LHLO" => {
403                if argument.is_empty() || argument.contains(' ') {
404                    wire::render(501, "5.5.4", "Syntax: LHLO domain")
405                } else {
406                    lhlo = Some(argument);
407                    from = None;
408                    recipients.clear();
409                    wire::render_multi(
410                        250,
411                        &[
412                            cfg.hostname.clone(),
413                            "PIPELINING".into(),
414                            "ENHANCEDSTATUSCODES".into(),
415                            "8BITMIME".into(),
416                            format!("SIZE {}", cfg.max_message),
417                        ],
418                    )
419                }
420            }
421            "HELO" | "EHLO" => wire::render(500, "5.5.1", "This is an LMTP server; use LHLO"),
422            "MAIL" => {
423                if lhlo.is_none() {
424                    wire::render(503, "5.5.1", "Send LHLO first")
425                } else if from.is_some() {
426                    wire::render(503, "5.5.1", "Nested MAIL command")
427                } else {
428                    match wire::parse_path(&argument, "FROM:") {
429                        None => wire::render(501, "5.5.4", "Syntax: MAIL FROM:<address>"),
430                        Some((address, params)) => {
431                            let mut refusal = None;
432                            for param in &params {
433                                let (key, value) = param.split_once('=').unwrap_or((param, ""));
434                                match key.to_ascii_uppercase().as_str() {
435                                    "SIZE" => match value.parse::<u64>() {
436                                        Ok(size) if size > cfg.max_message => {
437                                            refusal = Some(wire::render(
438                                                552,
439                                                "5.3.4",
440                                                "Message size exceeds fixed limit",
441                                            ))
442                                        }
443                                        Ok(_) => {}
444                                        Err(_) => {
445                                            refusal = Some(wire::render(
446                                                501,
447                                                "5.5.4",
448                                                "Invalid SIZE parameter",
449                                            ))
450                                        }
451                                    },
452                                    "BODY"
453                                        if matches!(
454                                            value.to_ascii_uppercase().as_str(),
455                                            "7BIT" | "8BITMIME"
456                                        ) => {}
457                                    _ => {
458                                        refusal = Some(wire::render(
459                                            555,
460                                            "5.5.4",
461                                            "MAIL parameter not recognized",
462                                        ))
463                                    }
464                                }
465                            }
466                            match refusal {
467                                Some(refusal) => refusal,
468                                None => {
469                                    from = Some(address);
470                                    wire::render(250, "2.1.0", "Sender OK")
471                                }
472                            }
473                        }
474                    }
475                }
476            }
477            "RCPT" => match (&from, wire::parse_path(&argument, "TO:")) {
478                (None, _) => wire::render(503, "5.5.1", "Need MAIL before RCPT"),
479                (_, None) => wire::render(501, "5.5.4", "Syntax: RCPT TO:<address>"),
480                (Some(_), Some((address, _))) if address.is_empty() => {
481                    wire::render(501, "5.1.3", "Empty recipient address")
482                }
483                (Some(_), Some(_)) if recipients.len() >= wire::MAX_RECIPIENTS => {
484                    wire::render(452, "4.5.3", "Too many recipients")
485                }
486                (Some(sender), Some((address, _))) => {
487                    let event = Event::new(
488                        &actions::RECIPIENT_EVENT,
489                        json!({
490                            "recipient": address,
491                            "mail_from": sender,
492                            "lhlo": lhlo.clone().unwrap_or_default(),
493                            "accepted_so_far": recipients.len(),
494                        }),
495                    );
496                    match decision(ctx, id, event, "lmtp_recipient_reply").await {
497                        Ok(answer) => {
498                            let reason = answer["reason"].as_str();
499                            if answer["accept"] == true {
500                                outcome(ctx, id, "lmtp_recipient", "model_answer");
501                                let reply = wire::render(
502                                    250,
503                                    "2.1.5",
504                                    reason.unwrap_or(&format!("<{address}> recipient OK")),
505                                );
506                                recipients.push(address);
507                                reply
508                            } else if answer["temporary"] == true {
509                                outcome(ctx, id, "lmtp_recipient", "model_reject");
510                                wire::render(
511                                    450,
512                                    "4.2.1",
513                                    reason.unwrap_or("Mailbox temporarily unavailable"),
514                                )
515                            } else {
516                                outcome(ctx, id, "lmtp_recipient", "model_reject");
517                                wire::render(550, "5.1.1", reason.unwrap_or("No such user here"))
518                            }
519                        }
520                        Err(_) => TEMPORARY_FAILURE.to_string(),
521                    }
522                }
523            },
524            "DATA" => {
525                if !argument.is_empty() {
526                    wire::render(501, "5.5.4", "DATA takes no argument")
527                } else if from.is_none() {
528                    wire::render(503, "5.5.1", "Need MAIL before DATA")
529                } else if recipients.is_empty() {
530                    wire::render(503, "5.5.1", "No valid recipients")
531                } else {
532                    write(
533                        ctx,
534                        id,
535                        &mut out,
536                        &wire::render(354, "", "End data with <CR><LF>.<CR><LF>"),
537                    )
538                    .await?;
539                    let (data, size) = read_data(&mut reader, cfg).await?;
540                    ctx.state
541                        .update_connection_stats(ctx.server_id, id, Some(size), None, Some(1), None)
542                        .await;
543                    let replies = match data {
544                        Data::TooBig => wire::render(552, "5.3.4", "Message too big for system")
545                            .repeat(recipients.len()),
546                        Data::BadLine => wire::render(500, "5.5.2", "Line too long in message")
547                            .repeat(recipients.len()),
548                        Data::Message(message) => {
549                            let (headers, body, truncated) = wire::summarize(&message);
550                            let subject = headers.get("subject").cloned().unwrap_or(Value::Null);
551                            let event = Event::new(
552                                &actions::MESSAGE_EVENT,
553                                json!({
554                                    "mail_from": from.clone().unwrap_or_default(),
555                                    "recipients": recipients,
556                                    "size": message.len(),
557                                    "headers": headers,
558                                    "subject": subject,
559                                    "body": body,
560                                    "body_truncated": truncated,
561                                }),
562                            );
563                            match decision(ctx, id, event, "lmtp_delivery").await {
564                                Ok(answer) => delivery_replies(ctx, id, &recipients, &answer),
565                                Err(_) => TEMPORARY_FAILURE.repeat(recipients.len()),
566                            }
567                        }
568                    };
569                    from = None;
570                    recipients.clear();
571                    replies
572                }
573            }
574            "RSET" => {
575                from = None;
576                recipients.clear();
577                wire::render(250, "2.0.0", "Flushed")
578            }
579            "NOOP" => wire::render(250, "2.0.0", "OK"),
580            "VRFY" => wire::render(
581                252,
582                "2.5.0",
583                "Cannot VRFY user, but will accept message and attempt delivery",
584            ),
585            "HELP" => wire::render(
586                214,
587                "2.0.0",
588                "Commands: LHLO MAIL RCPT DATA RSET NOOP VRFY HELP QUIT",
589            ),
590            "QUIT" => {
591                write(
592                    ctx,
593                    id,
594                    &mut out,
595                    &wire::render(
596                        221,
597                        "2.0.0",
598                        &format!("{} closing connection", cfg.hostname),
599                    ),
600                )
601                .await?;
602                let _ = out.shutdown().await;
603                return Ok(());
604            }
605            "STARTTLS" | "AUTH" | "BDAT" | "ETRN" | "EXPN" | "TURN" => {
606                wire::render(502, "5.5.1", "Command not implemented")
607            }
608            _ => wire::render(500, "5.5.2", "Command unrecognized"),
609        };
610        let refused = matches!(reply.get(..3), Some("500" | "501" | "502" | "503" | "555"));
611        bad = if refused { bad + 1 } else { 0 };
612        write(ctx, id, &mut out, &reply).await?;
613        if bad >= MAX_BAD_COMMANDS {
614            write(
615                ctx,
616                id,
617                &mut out,
618                &wire::render(421, "4.7.0", "Too many errors, closing"),
619            )
620            .await?;
621            let _ = out.shutdown().await;
622            return Ok(());
623        }
624    }
625}
netget/server/lmtp/mod.rs

netget/server/lmtp/
mod.rs
