netget/server/milter/
wire.rs

1//! The milter protocol (libmilter's wire format, version 6), shared by the filter (server) and
2//! the MTA side (client): length-prefixed packets of a command byte and NUL-separated fields.
3use anyhow::{bail, ensure, Context, Result};
4use std::time::Duration;
5use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
6
7/// Largest packet accepted (libmilter's own limit for a body chunk is 64 KiB, a header line
8/// is far smaller).
9pub const MAX_PACKET: usize = 256 * 1024;
10/// Largest message body collected for the end-of-message decision.
11pub const MAX_BODY: usize = 1024 * 1024;
12/// Most headers collected for one message.
13pub const MAX_HEADERS: usize = 512;
14pub const IO_TIMEOUT: Duration = Duration::from_secs(30);
15pub const VERSION: u32 = 6;
16
17// MTA → filter commands.
18pub const C_ABORT: u8 = b'A';
19pub const C_BODY: u8 = b'B';
20pub const C_CONNECT: u8 = b'C';
21pub const C_MACRO: u8 = b'D';
22pub const C_BODYEOB: u8 = b'E';
23pub const C_HELO: u8 = b'H';
24pub const C_QUIT_NC: u8 = b'K';
25pub const C_HEADER: u8 = b'L';
26pub const C_MAIL: u8 = b'M';
27pub const C_EOH: u8 = b'N';
28pub const C_OPTNEG: u8 = b'O';
29pub const C_QUIT: u8 = b'Q';
30pub const C_RCPT: u8 = b'R';
31pub const C_DATA: u8 = b'T';
32pub const C_UNKNOWN: u8 = b'U';
33
34// Filter → MTA replies.
35pub const R_ADDRCPT: u8 = b'+';
36pub const R_DELRCPT: u8 = b'-';
37pub const R_ACCEPT: u8 = b'a';
38pub const R_REPLBODY: u8 = b'b';
39pub const R_CONTINUE: u8 = b'c';
40pub const R_DISCARD: u8 = b'd';
41pub const R_ADDHEADER: u8 = b'h';
42pub const R_CHGHEADER: u8 = b'm';
43pub const R_PROGRESS: u8 = b'p';
44pub const R_QUARANTINE: u8 = b'q';
45pub const R_REJECT: u8 = b'r';
46pub const R_TEMPFAIL: u8 = b't';
47pub const R_REPLYCODE: u8 = b'y';
48pub const R_SKIP: u8 = b's';
49
50// Actions a filter may take (SMFIF_*).
51pub const F_ADDHDRS: u32 = 0x01;
52pub const F_CHGBODY: u32 = 0x02;
53pub const F_ADDRCPT: u32 = 0x04;
54pub const F_DELRCPT: u32 = 0x08;
55pub const F_CHGHDRS: u32 = 0x10;
56pub const F_QUARANTINE: u32 = 0x20;
57pub const ALL_ACTIONS: u32 =
58    F_ADDHDRS | F_CHGBODY | F_ADDRCPT | F_DELRCPT | F_CHGHDRS | F_QUARANTINE;
59
60// Protocol steps a filter may ask the MTA to leave out (SMFIP_NO*) or not to wait for a reply
61// to (SMFIP_NR_*), and SMFIP_SKIP: the filter may answer a body chunk with SMFIR_SKIP.
62pub const P_NOCONNECT: u32 = 0x01;
63pub const P_NOHELO: u32 = 0x02;
64pub const P_NOMAIL: u32 = 0x04;
65pub const P_NORCPT: u32 = 0x08;
66pub const P_NOBODY: u32 = 0x10;
67pub const P_NOHDRS: u32 = 0x20;
68pub const P_NOEOH: u32 = 0x40;
69pub const P_NR_HDR: u32 = 0x80;
70pub const P_NODATA: u32 = 0x200;
71pub const P_SKIP: u32 = 0x400;
72pub const P_NR_CONN: u32 = 0x1000;
73pub const P_NR_HELO: u32 = 0x2000;
74pub const P_NR_MAIL: u32 = 0x4000;
75pub const P_NR_RCPT: u32 = 0x8000;
76pub const P_NR_DATA: u32 = 0x10000;
77pub const P_NR_EOH: u32 = 0x40000;
78pub const P_NR_BODY: u32 = 0x80000;
79/// Every protocol step the MTA side can honour, with its name in events.
80pub const PROTOCOL_STEPS: [(u32, &str); 17] = [
81    (P_NOCONNECT, "no_connect"),
82    (P_NOHELO, "no_helo"),
83    (P_NOMAIL, "no_mail"),
84    (P_NORCPT, "no_rcpt"),
85    (P_NOBODY, "no_body"),
86    (P_NOHDRS, "no_headers"),
87    (P_NOEOH, "no_eoh"),
88    (P_NR_HDR, "no_reply_header"),
89    (P_NODATA, "no_data"),
90    (P_SKIP, "skip"),
91    (P_NR_CONN, "no_reply_connect"),
92    (P_NR_HELO, "no_reply_helo"),
93    (P_NR_MAIL, "no_reply_mail"),
94    (P_NR_RCPT, "no_reply_rcpt"),
95    (P_NR_DATA, "no_reply_data"),
96    (P_NR_EOH, "no_reply_eoh"),
97    (P_NR_BODY, "no_reply_body"),
98];
99pub const OFFERED_PROTOCOL: u32 = P_NOCONNECT
100    | P_NOHELO
101    | P_NOMAIL
102    | P_NORCPT
103    | P_NOBODY
104    | P_NOHDRS
105    | P_NOEOH
106    | P_NR_HDR
107    | P_NODATA
108    | P_SKIP
109    | P_NR_CONN
110    | P_NR_HELO
111    | P_NR_MAIL
112    | P_NR_RCPT
113    | P_NR_DATA
114    | P_NR_EOH
115    | P_NR_BODY;
116
117pub async fn read_packet<R: AsyncRead + Unpin>(
118    r: &mut R,
119    idle: Duration,
120) -> Result<Option<(u8, Vec<u8>)>> {
121    let mut len = [0u8; 4];
122    match tokio::time::timeout(idle, r.read(&mut len[..1])).await {
123        Err(_) => bail!("peer idle past its deadline"),
124        Ok(Ok(0)) => return Ok(None),
125        Ok(Ok(_)) => {}
126        Ok(Err(e)) => return Err(e.into()),
127    }
128    tokio::time::timeout(IO_TIMEOUT, async {
129        r.read_exact(&mut len[1..]).await?;
130        let n = u32::from_be_bytes(len) as usize;
131        ensure!(
132            (1..=MAX_PACKET).contains(&n),
133            "packet length {n} out of range"
134        );
135        let mut buf = vec![0u8; n];
136        r.read_exact(&mut buf).await?;
137        let cmd = buf.remove(0);
138        Ok(Some((cmd, buf)))
139    })
140    .await
141    .context("packet not completed in time")?
142}
143
144pub fn packet(cmd: u8, data: &[u8]) -> Vec<u8> {
145    let mut out = ((data.len() + 1) as u32).to_be_bytes().to_vec();
146    out.push(cmd);
147    out.extend(data);
148    out
149}
150
151pub async fn write<W: AsyncWrite + Unpin>(w: &mut W, cmd: u8, data: &[u8]) -> Result<()> {
152    tokio::time::timeout(IO_TIMEOUT, w.write_all(&packet(cmd, data)))
153        .await
154        .context("write deadline")??;
155    Ok(())
156}
157
158/// NUL-terminated strings, as fields.
159pub fn strings(data: &[u8]) -> Vec<String> {
160    let mut parts: Vec<String> = data
161        .split(|b| *b == 0)
162        .map(|s| String::from_utf8_lossy(s).into_owned())
163        .collect();
164    if data.last() == Some(&0) {
165        parts.pop();
166    }
167    parts
168}
169
170pub fn cstrings(fields: &[&str]) -> Vec<u8> {
171    let mut out = Vec::new();
172    for f in fields {
173        out.extend(f.as_bytes());
174        out.push(0);
175    }
176    out
177}
178
179/// A field from the handler: one line, no NUL.
180pub fn check_field(s: &str, what: &str) -> Result<()> {
181    ensure!(
182        s.len() <= 4096 && !s.contains(['\0', '\r', '\n']),
183        "{what} must be one line without NUL, at most 4096 bytes"
184    );
185    Ok(())
186}
187
188/// OPTNEG body: version, actions, protocol flags.
189pub fn optneg(version: u32, actions: u32, protocol: u32) -> Vec<u8> {
190    let mut out = version.to_be_bytes().to_vec();
191    out.extend(actions.to_be_bytes());
192    out.extend(protocol.to_be_bytes());
193    out
194}
195
196pub fn parse_optneg(d: &[u8]) -> Result<(u32, u32, u32)> {
197    ensure!(d.len() >= 12, "short OPTNEG");
198    let n = |i: usize| u32::from_be_bytes(d[i..i + 4].try_into().unwrap());
199    Ok((n(0), n(4), n(8)))
200}
201
202/// CONNECT body: hostname, family, port, address.
203pub fn parse_connect(d: &[u8]) -> Result<(String, char, u16, String)> {
204    let nul = d
205        .iter()
206        .position(|b| *b == 0)
207        .context("CONNECT without a hostname")?;
208    let host = String::from_utf8_lossy(&d[..nul]).into_owned();
209    let rest = &d[nul + 1..];
210    let family = *rest.first().context("CONNECT without a family")? as char;
211    if family == 'U' || rest.len() < 3 {
212        return Ok((host, family, 0, String::new()));
213    }
214    let port = u16::from_be_bytes([rest[1], rest[2]]);
215    let addr = strings(&rest[3..]).into_iter().next().unwrap_or_default();
216    Ok((host, family, port, addr))
217}
218
219pub fn connect_body(host: &str, family: char, port: u16, addr: &str) -> Vec<u8> {
220    let mut out = cstrings(&[host]);
221    out.push(family as u8);
222    out.extend(port.to_be_bytes());
223    out.extend(cstrings(&[addr]));
224    out
225}
226
227/// A reply's name as events and logs report it.
228pub fn reply_name(code: u8) -> &'static str {
229    match code {
230        R_ACCEPT => "accept",
231        R_CONTINUE => "continue",
232        R_DISCARD => "discard",
233        R_REJECT => "reject",
234        R_TEMPFAIL => "tempfail",
235        R_REPLYCODE => "replycode",
236        R_ADDHEADER => "add_header",
237        R_CHGHEADER => "change_header",
238        R_ADDRCPT => "add_rcpt",
239        R_DELRCPT => "del_rcpt",
240        R_REPLBODY => "replace_body",
241        R_QUARANTINE => "quarantine",
242        R_PROGRESS => "progress",
243        R_SKIP => "skip",
244        _ => "unknown",
245    }
246}
247
248/// A REPLYCODE body: "550 5.7.1 text".
249pub fn replycode(code: u64, xcode: Option<&str>, text: &str) -> Result<Vec<u8>> {
250    ensure!((400..600).contains(&code), "reply code must be 4xx or 5xx");
251    check_field(text, "text")?;
252    let line = match xcode {
253        Some(x) => {
254            check_field(x, "xcode")?;
255            ensure!(
256                x.starts_with(if code < 500 { '4' } else { '5' }),
257                "xcode class must match the code"
258            );
259            format!("{code} {x} {text}")
260        }
261        None => format!("{code} {text}"),
262    };
263    Ok(cstrings(&[&line]))
264}
netget/server/milter/wire.rs

netget/server/milter/
wire.rs
