netget/client/lmtp/
actions.rs

1use crate::llm::actions::{
2    client_trait::{Client, ClientActionResult},
3    protocol_trait::Protocol,
4    ActionDefinition, ParameterDefinition,
5};
6use crate::protocol::{ConnectContext, EventType};
7use crate::server::lmtp::{
8    actions::{action, parameter},
9    wire,
10};
11use crate::state::app_state::AppState;
12use anyhow::{bail, ensure, Result};
13use serde_json::{json, Map, Value};
14use std::sync::LazyLock;
15
16#[derive(Default)]
17pub struct LmtpClientProtocol;
18impl LmtpClientProtocol {
19    pub fn new() -> Self {
20        Self
21    }
22}
23
24pub const DEFAULT_LHLO_DOMAIN: &str = "netget.local";
25/// Largest body the client will compose.
26pub const MAX_BODY_BYTES: usize = 1024 * 1024;
27
28fn send_action() -> ActionDefinition {
29    action(
30        "lmtp_send",
31        "Deliver one message: MAIL FROM, one RCPT TO per recipient, DATA. LMTP then reports \
32         a delivery result for every accepted recipient, raised as lmtp_result.",
33        vec![
34            parameter(
35                "from",
36                "string",
37                "Sender address for MAIL FROM and the From header (may be empty for a bounce)",
38                true,
39            ),
40            parameter("to", "array", "Recipient addresses (1 to 100)", true),
41            parameter(
42                "subject",
43                "string",
44                "Subject line of the message, sent as its Subject header field",
45                false,
46            ),
47            parameter(
48                "body",
49                "string",
50                "Plain-text body; newlines are sent as CRLF",
51                true,
52            ),
53            parameter(
54                "headers",
55                "object",
56                "Extra header fields {name: value}",
57                false,
58            ),
59        ],
60        json!({"type":"lmtp_send","from":"sender@example.test","to":["alice@example.test"],"subject":"Hello","body":"Hello Alice"}),
61    )
62}
63
64fn disconnect_action() -> ActionDefinition {
65    action(
66        "disconnect",
67        "Send QUIT and close the LMTP connection",
68        vec![],
69        json!({"type":"disconnect"}),
70    )
71}
72
73fn actions() -> Vec<ActionDefinition> {
74    vec![send_action(), disconnect_action()]
75}
76
77pub static CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
78    EventType::new(
79        "lmtp_connected",
80        "Connected and LHLO accepted; send a message or disconnect",
81        send_action().example.clone(),
82    )
83    .with_parameters(vec![
84        parameter("remote_addr", "string", "LMTP server address", true),
85        parameter("greeting", "string", "Text of the 220 greeting", true),
86        parameter(
87            "capabilities",
88            "array",
89            "Extensions from the LHLO reply, such as PIPELINING or SIZE 10485760",
90            true,
91        ),
92    ])
93    .with_actions(actions())
94});
95
96pub static RESULT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
97    EventType::new(
98        "lmtp_result",
99        "Outcome of one lmtp_send: the MAIL reply, each recipient's RCPT reply and, for \
100         accepted recipients, its own delivery reply after DATA",
101        disconnect_action().example.clone(),
102    )
103    .with_parameters(vec![
104        parameter("from", "string", "Sender of the transaction", true),
105        parameter("mail", "object", "{code, text} reply to MAIL FROM", true),
106        parameter(
107            "recipients",
108            "array",
109            "[{recipient, rcpt: {code,text}, delivery: {code,text} or null, delivered: bool}]",
110            true,
111        ),
112        parameter(
113            "data",
114            "object",
115            "{code, text} reply to DATA, or null when DATA was not sent",
116            false,
117        ),
118        parameter(
119            "delivered",
120            "array",
121            "Recipients whose delivery reply was 2xx",
122            true,
123        ),
124    ])
125    .with_actions(actions())
126});
127
128/// A validated outgoing message.
129#[derive(Debug, Clone)]
130pub struct Outgoing {
131    pub from: String,
132    pub to: Vec<String>,
133    pub subject: Option<String>,
134    pub body: String,
135    pub headers: Vec<(String, String)>,
136}
137
138fn header_value_ok(value: &str) -> bool {
139    value.len() <= 900 && !value.chars().any(|c| c == '\r' || c == '\n' || c == '\0')
140}
141
142impl Outgoing {
143    pub fn from_action(v: &Value) -> Result<Self> {
144        let from = v["from"]
145            .as_str()
146            .ok_or_else(|| anyhow::anyhow!("from must be a string"))?
147            .to_string();
148        ensure!(
149            wire::valid_mailbox(&from, true),
150            "from is not a valid mailbox"
151        );
152        let to = v["to"]
153            .as_array()
154            .ok_or_else(|| anyhow::anyhow!("to must be an array of addresses"))?
155            .iter()
156            .map(|a| {
157                a.as_str()
158                    .filter(|a| wire::valid_mailbox(a, false))
159                    .map(str::to_string)
160                    .ok_or_else(|| anyhow::anyhow!("every recipient must be a non-empty mailbox"))
161            })
162            .collect::<Result<Vec<_>>>()?;
163        ensure!(
164            (1..=wire::MAX_RECIPIENTS).contains(&to.len()),
165            "to must name 1 to {} recipients",
166            wire::MAX_RECIPIENTS
167        );
168        let subject = match v.get("subject") {
169            None | Some(Value::Null) => None,
170            Some(Value::String(s)) if header_value_ok(s) => Some(s.clone()),
171            _ => bail!("subject must be a single-line string"),
172        };
173        let body = v["body"]
174            .as_str()
175            .ok_or_else(|| anyhow::anyhow!("body must be a string"))?
176            .to_string();
177        ensure!(
178            body.len() <= MAX_BODY_BYTES,
179            "body exceeds {MAX_BODY_BYTES} bytes"
180        );
181        let mut headers = Vec::new();
182        if let Some(extra) = v.get("headers").filter(|h| !h.is_null()) {
183            let extra = extra
184                .as_object()
185                .ok_or_else(|| anyhow::anyhow!("headers must be an object"))?;
186            ensure!(extra.len() <= 50, "at most 50 extra headers");
187            for (name, value) in extra {
188                let value = value
189                    .as_str()
190                    .ok_or_else(|| anyhow::anyhow!("header {name} must be a string"))?;
191                ensure!(
192                    !name.is_empty()
193                        && name.len() <= 76
194                        && name.bytes().all(|b| b.is_ascii_graphic() && b != b':'),
195                    "invalid header name {name}"
196                );
197                ensure!(
198                    header_value_ok(value),
199                    "header {name} must be a single line"
200                );
201                headers.push((name.clone(), value.to_string()));
202            }
203        }
204        Ok(Self {
205            from,
206            to,
207            subject,
208            body,
209            headers,
210        })
211    }
212
213    pub fn to_json(&self) -> Value {
214        let headers: Map<String, Value> = self
215            .headers
216            .iter()
217            .map(|(k, v)| (k.clone(), Value::String(v.clone())))
218            .collect();
219        json!({"type":"lmtp_send","from":self.from,"to":self.to,"subject":self.subject,"body":self.body,"headers":headers})
220    }
221
222    /// The RFC 5322 message: generated Date and Message-ID unless supplied.
223    pub fn compose(&self, domain: &str) -> String {
224        let has = |name: &str| {
225            self.headers
226                .iter()
227                .any(|(k, _)| k.eq_ignore_ascii_case(name))
228        };
229        let mut out = String::new();
230        if !has("Date") {
231            out.push_str(&format!("Date: {}\n", chrono::Utc::now().to_rfc2822()));
232        }
233        if !has("Message-ID") {
234            out.push_str(&format!(
235                "Message-ID: <{}@{}>\n",
236                uuid::Uuid::new_v4(),
237                domain
238            ));
239        }
240        if !has("From") {
241            out.push_str(&format!("From: <{}>\n", self.from));
242        }
243        if !has("To") {
244            let to: Vec<String> = self.to.iter().map(|a| format!("<{a}>")).collect();
245            out.push_str(&format!("To: {}\n", to.join(", ")));
246        }
247        if let Some(subject) = &self.subject {
248            if !has("Subject") {
249                out.push_str(&format!("Subject: {subject}\n"));
250            }
251        }
252        for (name, value) in &self.headers {
253            out.push_str(&format!("{name}: {value}\n"));
254        }
255        out.push('\n');
256        out.push_str(&self.body);
257        out
258    }
259}
260
261impl Protocol for LmtpClientProtocol {
262    fn protocol_name(&self) -> &'static str {
263        "LMTP"
264    }
265    fn stack_name(&self) -> &'static str {
266        "ETH>IP>TCP>LMTP"
267    }
268    fn keywords(&self) -> Vec<&'static str> {
269        vec!["lmtp", "rfc2033", "lhlo", "mail delivery"]
270    }
271    fn description(&self) -> &'static str {
272        "LMTP client that delivers messages and reports per-recipient results"
273    }
274    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
275        actions()
276    }
277    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
278        vec![]
279    }
280    fn get_event_types(&self) -> Vec<EventType> {
281        vec![CONNECTED_EVENT.clone(), RESULT_EVENT.clone()]
282    }
283    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
284        vec![ParameterDefinition {
285            name: "lhlo_domain".into(),
286            type_hint: "string".into(),
287            description: "Domain the client names in LHLO and in generated Message-IDs".into(),
288            required: false,
289            example: json!("client.example.test"),
290            default: Some(json!(DEFAULT_LHLO_DOMAIN)),
291        }]
292    }
293    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
294        use crate::protocol::metadata::*;
295        ProtocolMetadataV2::builder()
296            .state(DevelopmentState::Experimental)
297            .privilege_requirement(PrivilegeRequirement::None)
298            .well_known_port(24)
299            .implementation("Tokio TCP: greeting and LHLO on connect, then MAIL/RCPT/DATA per message with one delivery reply read per accepted recipient")
300            .llm_control("Which messages to send to which recipients, and what to do with each per-recipient result")
301            .e2e_testing("tests/client/lmtp: scripted wire fixture, injected sends, malformed replies; aiosmtpd's LMTP server as the independent peer")
302            .notes("Plain TCP only (no STARTTLS or AUTH). Replies are bounded to 1000-byte lines and 64 lines; every connect, write and reply has a 30 s deadline; bodies up to 1 MiB.")
303            .max_inbound_bytes(wire::MAX_LINE_BYTES)
304            .build()
305    }
306    fn example_prompt(&self) -> &'static str {
307        "Connect to LMTP at 127.0.0.1:2424 and deliver a test message to alice@example.test"
308    }
309    fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
310        let llm = json!({"type":"open_client","protocol":"lmtp","remote_addr":"127.0.0.1:2424","instruction":"Deliver a greeting to alice@example.test"});
311        let mut static_example = llm.clone();
312        static_example["event_handlers"] = json!([
313            {"event_pattern":"lmtp_connected","handler":{"type":"static","actions":[{"type":"lmtp_send","from":"sender@example.test","to":["alice@example.test"],"subject":"Hello","body":"Hello Alice"}]}},
314            {"event_pattern":"lmtp_result","handler":{"type":"static","actions":[{"type":"disconnect"}]}}
315        ]);
316        let mut scripted = static_example.clone();
317        scripted["event_handlers"][0]["handler"] = json!({"type":"script","language":"python","code":"import json\nprint(json.dumps({'actions':[{'type':'lmtp_send','from':'sender@example.test','to':['alice@example.test'],'body':'Hello'}]}))"});
318        crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
319    }
320    fn group_name(&self) -> &'static str {
321        "Application"
322    }
323}
324
325impl Client for LmtpClientProtocol {
326    fn connect(
327        &self,
328        ctx: ConnectContext,
329    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
330    {
331        Box::pin(super::connect(ctx))
332    }
333    fn execute_action(&self, v: Value) -> Result<ClientActionResult> {
334        match v["type"].as_str() {
335            Some("lmtp_send") => Ok(ClientActionResult::Custom {
336                name: "lmtp_send".into(),
337                data: Outgoing::from_action(&v)?.to_json(),
338            }),
339            Some("disconnect") => Ok(ClientActionResult::Disconnect),
340            _ => bail!("Unknown LMTP client action"),
341        }
342    }
343}
netget/client/lmtp/actions.rs

netget/client/lmtp/
actions.rs
