//! Minimal async IMAP4rev1/rev2 client: just what a migration needs. use super::proto::{self, Arg, Kind, Part, Response, Status, Value}; use super::utf7; use crate::config::{Endpoint, TlsMode}; use anyhow::{Context, Result, anyhow, bail}; use base64::Engine; use std::collections::{BTreeMap, HashSet}; use std::sync::Arc; use std::time::Duration; use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufStream}; use tokio::net::TcpStream; pub trait Stream: AsyncRead + AsyncWrite + Unpin + Send {} impl Stream for T {} pub struct Client { io: BufStream>, tag: u32, pub caps: HashSet, pub label: String, timeout: Duration, } #[derive(Debug, Clone)] pub struct Mailbox { /// Name exactly as on the wire (modified UTF-7). pub wire: String, /// Decoded Unicode name. pub name: String, pub delim: Option, /// Attributes, e.g. `\Noselect`, `\Sent` (as sent by the server). pub attrs: Vec, } impl Mailbox { pub fn has_attr(&self, a: &str) -> bool { self.attrs.iter().any(|x| x.eq_ignore_ascii_case(a)) } pub fn selectable(&self) -> bool { !self.has_attr("\\Noselect") && !self.has_attr("\\NonExistent") } pub fn special_use(&self) -> Option<&'static str> { ["\\Sent", "\\Trash", "\\Junk", "\\Drafts", "\\Archive", "\\All", "\\Flagged"] .into_iter() .find(|s| self.has_attr(s)) } } #[derive(Debug, Clone, Default)] pub struct Namespaces { /// (prefix, delimiter) of the personal namespace(s). pub personal: Vec<(String, Option)>, /// Prefixes of other-users and shared namespaces (to skip). pub foreign: Vec, } #[derive(Debug, Clone, Default)] pub struct SelectInfo { pub exists: u32, pub uidvalidity: Option, pub permanent_flags: Option>, /// EXAMINE / [READ-ONLY]: PERMANENTFLAGS is then empty and says nothing /// about what APPEND will keep. pub read_only: bool, } impl SelectInfo { /// Whether the mailbox accepts this flag permanently. pub fn accepts_flag(&self, f: &str) -> bool { match &self.permanent_flags { None => true, Some(_) if self.read_only => true, Some(p) => { p.iter().any(|x| x.eq_ignore_ascii_case(f)) || (!f.starts_with('\\') && p.iter().any(|x| x == "\\*")) } } } } #[derive(Debug, Clone, Default)] pub struct MsgMeta { pub uid: u32, pub flags: Vec, pub internaldate: Option, pub size: u64, pub headers: Vec, } #[derive(Debug, Clone, Default)] pub struct Fetched { pub uid: u32, pub flags: Vec, pub internaldate: Option, pub body: Vec, } pub struct CmdResult { pub untagged: Vec, pub status: Status, } /// A NO/BAD completion, kept typed so callers can inspect the response code. #[derive(Debug)] pub struct ImapNo { pub cmd: String, pub status: Status, } impl std::fmt::Display for ImapNo { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "{} -> {}", self.cmd, self.status.word)?; if let Some(c) = &self.status.code { write!(f, " [{c}]")?; } write!(f, " {}", self.status.text) } } impl std::error::Error for ImapNo {} pub enum Login<'a> { Plain { user: &'a str, password: &'a str }, Authzid { authzid: &'a str, authcid: &'a str, password: &'a str }, } fn tls_config(ep: &Endpoint) -> Result> { if let Some(ca) = &ep.ca_file { use rustls_pki_types::pem::PemObject; let mut roots = rustls::RootCertStore::empty(); for cert in rustls_pki_types::CertificateDer::pem_file_iter(ca) .with_context(|| format!("reading CA file {}", ca.display()))? { roots.add(cert?)?; } Ok(Arc::new(rustls::ClientConfig::builder().with_root_certificates(roots).with_no_client_auth())) } else { use rustls_platform_verifier::ConfigVerifierExt; Ok(Arc::new(rustls::ClientConfig::with_platform_verifier()?)) } } async fn tls_wrap(ep: &Endpoint, tcp: TcpStream) -> Result> { let name = ep.tls_server_name.clone().unwrap_or_else(|| ep.host.clone()); let server_name = rustls_pki_types::ServerName::try_from(name.clone()) .with_context(|| format!("invalid TLS server name {name}"))?; let connector = tokio_rustls::TlsConnector::from(tls_config(ep)?); let tls = connector.connect(server_name, tcp).await.with_context(|| format!("TLS handshake with {}", ep.host))?; Ok(Box::new(tls)) } impl Client { pub async fn connect(ep: &Endpoint, label: &str) -> Result { let addr = (ep.host.as_str(), ep.port()); let timeout = Duration::from_secs(ep.timeout); let tcp = tokio::time::timeout(Duration::from_secs(30), TcpStream::connect(addr)) .await .map_err(|_| anyhow!("{label}: connect to {}:{} timed out", ep.host, ep.port()))? .with_context(|| format!("{label}: connect to {}:{}", ep.host, ep.port()))?; tcp.set_nodelay(true).ok(); let stream: Box = match ep.tls { TlsMode::Implicit => tls_wrap(ep, tcp).await?, TlsMode::None => Box::new(tcp), TlsMode::Starttls => { let mut tcp = tcp; starttls_preamble(&mut tcp, label, timeout).await?; tls_wrap(ep, tcp).await? } }; let mut c = Client { io: BufStream::new(stream), tag: 0, caps: HashSet::new(), label: label.to_string(), timeout }; if ep.tls != TlsMode::Starttls { c.greeting().await?; } c.capability().await?; Ok(c) } async fn greeting(&mut self) -> Result<()> { let r = self.read_response().await?; match r.status() { Some(s) if s.word == "OK" || s.word == "PREAUTH" => { if let Some(code) = &s.code { self.set_caps_from_code(code); } Ok(()) } _ => bail!("{}: unexpected greeting: {}", self.label, r.summary()), } } fn set_caps_from_code(&mut self, code: &str) { let mut it = code.split_whitespace(); if it.next().is_some_and(|w| w.eq_ignore_ascii_case("CAPABILITY")) { self.caps = it.map(|c| c.to_ascii_uppercase()).collect(); } } pub fn has_cap(&self, c: &str) -> bool { self.caps.contains(&c.to_ascii_uppercase()) } async fn read_response(&mut self) -> Result { let fut = async { let mut parts = Vec::new(); loop { let mut line = Vec::new(); let n = self.io.read_until(b'\n', &mut line).await?; if n == 0 { bail!("{}: connection closed by server", self.label); } while matches!(line.last(), Some(b'\n' | b'\r')) { line.pop(); } if let Some(len) = proto::literal_len(&line) { parts.push(Part::Text(line)); let mut buf = vec![0u8; len]; self.io.read_exact(&mut buf).await?; parts.push(Part::Lit(buf)); } else { parts.push(Part::Text(line)); return Ok(Response { parts }); } } }; let r: Result = tokio::time::timeout(self.timeout, fut) .await .map_err(|_| anyhow!("{}: timed out waiting for server", self.label))?; let r = r?; tracing::trace!("{} S: {}", self.label, r.summary()); Ok(r) } /// Send a command made of raw text and literals; collect untagged /// responses until the tagged completion. pub async fn run(&mut self, args: Vec, log_as: &str) -> Result { self.tag += 1; let tag = format!("m{}", self.tag); tracing::trace!("{} C: {tag} {log_as}", self.label); let mut untagged = Vec::new(); self.io.write_all(tag.as_bytes()).await?; self.io.write_all(b" ").await?; for a in args { match a { Arg::Raw(s) => self.io.write_all(s.as_bytes()).await?, Arg::Lit(data) => { let nonsync = self.has_cap("LITERAL+") || (self.has_cap("LITERAL-") && data.len() <= 4096); if nonsync { self.io.write_all(format!("{{{}+}}\r\n", data.len()).as_bytes()).await?; } else { self.io.write_all(format!("{{{}}}\r\n", data.len()).as_bytes()).await?; self.io.flush().await?; loop { let r = self.read_response().await?; match r.kind() { Kind::Continuation => break, Kind::Tagged(t) if t == tag => { let status = r.status().unwrap_or(Status { word: "BAD".into(), code: None, text: r.summary(), }); return Err(ImapNo { cmd: log_as.to_string(), status }.into()); } _ => untagged.push(r), } } } self.io.write_all(&data).await?; } } } self.io.write_all(b"\r\n").await?; self.io.flush().await?; loop { let r = self.read_response().await?; match r.kind() { Kind::Tagged(t) if t == tag => { let status = r.status().ok_or_else(|| anyhow!("{}: malformed completion: {}", self.label, r.summary()))?; if status.word != "OK" { return Err(ImapNo { cmd: log_as.to_string(), status }.into()); } return Ok(CmdResult { untagged, status }); } Kind::Continuation => bail!("{}: unexpected continuation for {log_as}", self.label), _ => { if r.status().is_some_and(|s| s.word == "BYE") && !log_as.starts_with("LOGOUT") { tracing::debug!("{}: server said BYE: {}", self.label, r.summary()); } untagged.push(r) } } } } pub async fn simple(&mut self, cmd: &str) -> Result { self.run(vec![Arg::Raw(cmd.to_string())], cmd).await } pub async fn capability(&mut self) -> Result<()> { let r = self.simple("CAPABILITY").await?; for u in &r.untagged { let v = u.values()?; if v.get(1).is_some_and(|x| x.is_atom("CAPABILITY")) { self.caps = v[2..].iter().filter_map(|x| x.text()).map(|c| c.to_ascii_uppercase()).collect(); } } Ok(()) } pub async fn login(&mut self, how: Login<'_>) -> Result<()> { let res = match how { Login::Plain { user, password } => { if self.has_cap("LOGINDISABLED") { bail!("{}: server has LOGINDISABLED (use TLS)", self.label); } let mut args = vec![Arg::Raw("LOGIN ".into())]; args.push(proto::astring(user)); args.push(Arg::Raw(" ".into())); args.push(proto::astring(password)); self.run(args, &format!("LOGIN {user} ****")).await } Login::Authzid { authzid, authcid, password } => { let raw = format!("{authzid}\0{authcid}\0{password}"); let b64 = base64::engine::general_purpose::STANDARD.encode(raw.as_bytes()); if self.has_cap("SASL-IR") { self.run( vec![Arg::Raw(format!("AUTHENTICATE PLAIN {b64}"))], &format!("AUTHENTICATE PLAIN {authzid}/{authcid} ****"), ) .await } else { self.authenticate_continuation(&b64, authzid, authcid).await } } }; let r = res.map_err(|e| anyhow!("{}: login failed: {e}", self.label))?; match r.status.code.as_deref() { Some(code) if code.to_ascii_uppercase().starts_with("CAPABILITY") => self.set_caps_from_code(code), _ => self.capability().await?, } Ok(()) } async fn authenticate_continuation(&mut self, b64: &str, authzid: &str, authcid: &str) -> Result { self.tag += 1; let tag = format!("m{}", self.tag); self.io.write_all(format!("{tag} AUTHENTICATE PLAIN\r\n").as_bytes()).await?; self.io.flush().await?; let log_as = format!("AUTHENTICATE PLAIN {authzid}/{authcid} ****"); let r = self.read_response().await?; if r.kind() != Kind::Continuation { bail!("{}: AUTHENTICATE PLAIN refused: {}", self.label, r.summary()); } self.io.write_all(format!("{b64}\r\n").as_bytes()).await?; self.io.flush().await?; let mut untagged = Vec::new(); loop { let r = self.read_response().await?; match r.kind() { Kind::Tagged(t) if t == tag => { let status = r.status().ok_or_else(|| anyhow!("malformed completion"))?; if status.word != "OK" { return Err(ImapNo { cmd: log_as, status }.into()); } return Ok(CmdResult { untagged, status }); } _ => untagged.push(r), } } } pub async fn namespace(&mut self) -> Result> { if !self.has_cap("NAMESPACE") && !self.has_cap("IMAP4REV2") { return Ok(None); } let r = self.simple("NAMESPACE").await?; for u in &r.untagged { let v = u.values()?; if v.get(1).is_some_and(|x| x.is_atom("NAMESPACE")) && v.len() >= 5 { let mut ns = Namespaces::default(); let entries = |val: &Value| -> Vec<(String, Option)> { val.list() .unwrap_or(&[]) .iter() .filter_map(|e| { let l = e.list()?; let prefix = l.first()?.text()?; let delim = l.get(1).and_then(|d| d.text()).and_then(|d| d.chars().next()); Some((prefix, delim)) }) .collect() }; ns.personal = entries(&v[2]); ns.foreign = entries(&v[3]).into_iter().chain(entries(&v[4])).map(|(p, _)| p).collect(); return Ok(Some(ns)); } } Ok(None) } pub async fn list(&mut self) -> Result> { let cmd = if self.has_cap("LIST-EXTENDED") && self.has_cap("SPECIAL-USE") { r#"LIST "" "*" RETURN (SPECIAL-USE)"# } else { r#"LIST "" "*""# }; let r = self.simple(cmd).await?; let mut out: Vec = Vec::new(); for u in &r.untagged { let v = u.values()?; if !v.get(1).is_some_and(|x| x.is_atom("LIST")) || v.len() < 5 { continue; } let attrs: Vec = v[2].list().unwrap_or(&[]).iter().filter_map(|a| a.text()).collect(); let delim = v[3].text().and_then(|d| d.chars().next()); let wire = v[4].text().unwrap_or_default(); let wire = if wire.eq_ignore_ascii_case("INBOX") { "INBOX".to_string() } else { wire }; if let Some(existing) = out.iter_mut().find(|m| m.wire == wire) { existing.attrs.extend(attrs); continue; } out.push(Mailbox { name: utf7::decode(&wire), wire, delim, attrs }); } Ok(out) } pub async fn select(&mut self, wire: &str, readonly: bool) -> Result { let verb = if readonly { "EXAMINE" } else { "SELECT" }; let mut args = vec![Arg::Raw(format!("{verb} "))]; args.push(mailbox_arg(wire)); let r = self.run(args, &format!("{verb} {wire}")).await?; let mut info = SelectInfo { read_only: readonly || r.status.code.as_deref().is_some_and(|c| c.eq_ignore_ascii_case("READ-ONLY")), ..Default::default() }; for u in &r.untagged { if let Some(s) = u.status() { if let Some(code) = s.code { let up = code.to_ascii_uppercase(); if let Some(rest) = up.strip_prefix("UIDVALIDITY ") { info.uidvalidity = rest.trim().parse().ok(); } else if up.starts_with("PERMANENTFLAGS") { let open = code.find('(').unwrap_or(code.len()); let close = code.rfind(')').unwrap_or(code.len()); if open < close { info.permanent_flags = Some(code[open + 1..close].split_whitespace().map(|s| s.to_string()).collect()); } } } continue; } let v = u.values()?; if v.len() >= 3 && v[2].is_atom("EXISTS") { info.exists = v[1].number().unwrap_or(0) as u32; } } Ok(info) } /// Run a UID FETCH and merge the items of all FETCH responses by UID. async fn uid_fetch(&mut self, set: &str, items: &str) -> Result>> { let cmd = format!("UID FETCH {set} ({items})"); let r = self.simple(&cmd).await?; let mut out: BTreeMap> = BTreeMap::new(); for u in r.untagged { if u.status().is_some() { continue; } let v = u.values()?; if v.len() < 4 || !v[2].is_atom("FETCH") { continue; } let Some(list) = v[3].list() else { continue }; let mut pairs = Vec::new(); let mut uid = None; for kv in list.chunks(2) { if let [k, val] = kv { let key = k.text().unwrap_or_default().to_ascii_uppercase(); if key == "UID" { uid = val.number().map(|n| n as u32); } pairs.push((key, val.clone())); } } if let Some(uid) = uid { out.entry(uid).or_default().extend(pairs); } } Ok(out) } /// UID, flags, date, size and identifying headers of every message in the /// selected mailbox. pub async fn fetch_meta(&mut self) -> Result> { let items = "UID FLAGS INTERNALDATE RFC822.SIZE BODY.PEEK[HEADER.FIELDS (MESSAGE-ID DATE FROM SUBJECT)]"; let map = self.uid_fetch("1:*", items).await?; Ok(map .into_iter() .map(|(uid, pairs)| { let mut m = MsgMeta { uid, ..Default::default() }; for (k, v) in pairs { match k.as_str() { "FLAGS" => m.flags = flags_of(&v), "INTERNALDATE" => m.internaldate = v.text(), "RFC822.SIZE" => m.size = v.number().unwrap_or(0), k if k.starts_with("BODY[") => m.headers = v.bytes().unwrap_or(&[]).to_vec(), _ => {} } } m }) .collect()) } pub async fn fetch_bodies(&mut self, uids: &[u32]) -> Result> { if uids.is_empty() { return Ok(Vec::new()); } let set = uid_set(uids); let map = self.uid_fetch(&set, "UID FLAGS INTERNALDATE BODY.PEEK[]").await?; let mut out = Vec::new(); for (uid, pairs) in map { let mut f = Fetched { uid, ..Default::default() }; let mut got_body = false; for (k, v) in pairs { match k.as_str() { "FLAGS" => f.flags = flags_of(&v), "INTERNALDATE" => f.internaldate = v.text(), k if k.starts_with("BODY[") => { f.body = v.bytes().unwrap_or(&[]).to_vec(); got_body = true; } _ => {} } } if got_body { out.push(f); } } Ok(out) } pub async fn append(&mut self, wire: &str, flags: &[String], date: Option<&str>, body: &[u8]) -> Result<()> { let mut head = String::from("APPEND "); let mut args = Vec::new(); args.push(Arg::Raw(std::mem::take(&mut head))); args.push(mailbox_arg(wire)); if !flags.is_empty() { args.push(Arg::Raw(format!(" ({})", flags.join(" ")))); } if let Some(d) = date { args.push(Arg::Raw(format!(" {}", proto::quote(d)))); } args.push(Arg::Raw(" ".into())); args.push(Arg::Lit(body.to_vec())); self.run(args, &format!("APPEND {wire} <{} bytes>", body.len())).await?; Ok(()) } /// CREATE, treating "already exists" as success. pub async fn create(&mut self, wire: &str) -> Result<()> { let args = vec![Arg::Raw("CREATE ".into()), mailbox_arg(wire)]; match self.run(args, &format!("CREATE {wire}")).await { Ok(_) => Ok(()), Err(e) => match e.downcast_ref::() { Some(no) if no.status.code.as_deref().is_some_and(|c| c.eq_ignore_ascii_case("ALREADYEXISTS")) || no.status.text.to_ascii_lowercase().contains("exist") => { Ok(()) } _ => Err(e), }, } } pub async fn subscribe(&mut self, wire: &str) -> Result<()> { let args = vec![Arg::Raw("SUBSCRIBE ".into()), mailbox_arg(wire)]; self.run(args, &format!("SUBSCRIBE {wire}")).await.map(|_| ()) } pub async fn uid_store_flags(&mut self, uid: u32, flags: &[String]) -> Result<()> { self.simple(&format!("UID STORE {uid} FLAGS.SILENT ({})", flags.join(" "))).await.map(|_| ()) } pub async fn logout(mut self) { let _ = tokio::time::timeout(Duration::from_secs(5), self.simple("LOGOUT")).await; } } fn mailbox_arg(wire: &str) -> Arg { // Always quote mailbox names; INBOX and ASCII names are safe quoted. if wire.bytes().all(|c| (0x20..0x7f).contains(&c)) { Arg::Raw(proto::quote(wire)) } else { Arg::Lit(wire.as_bytes().to_vec()) } } fn flags_of(v: &Value) -> Vec { v.list().unwrap_or(&[]).iter().filter_map(|f| f.text()).collect() } /// Compress sorted UIDs into an IMAP sequence set ("1:3,7,9:10"). pub fn uid_set(uids: &[u32]) -> String { let mut sorted = uids.to_vec(); sorted.sort_unstable(); sorted.dedup(); let mut out = String::new(); let mut i = 0; while i < sorted.len() { let start = sorted[i]; let mut end = start; while i + 1 < sorted.len() && sorted[i + 1] == end + 1 { i += 1; end = sorted[i]; } if !out.is_empty() { out.push(','); } if start == end { out.push_str(&start.to_string()); } else { out.push_str(&format!("{start}:{end}")); } i += 1; } out } /// Plain-text greeting + STARTTLS on the raw socket, before the TLS wrap. async fn starttls_preamble(tcp: &mut TcpStream, label: &str, timeout: Duration) -> Result<()> { let fut = async { let mut r = tokio::io::BufReader::new(&mut *tcp); let mut line = String::new(); r.read_line(&mut line).await?; if !line.to_ascii_uppercase().starts_with("* OK") { bail!("{label}: unexpected greeting: {}", line.trim_end()); } r.write_all(b"s0 STARTTLS\r\n").await?; r.flush().await?; loop { line.clear(); if r.read_line(&mut line).await? == 0 { bail!("{label}: connection closed during STARTTLS"); } if let Some(rest) = line.strip_prefix("s0 ") { if !rest.to_ascii_uppercase().starts_with("OK") { bail!("{label}: STARTTLS refused: {}", line.trim_end()); } break; } } if !r.buffer().is_empty() { bail!("{label}: server sent data after STARTTLS (possible injection)"); } Ok(()) }; tokio::time::timeout(timeout, fut).await.map_err(|_| anyhow!("{label}: STARTTLS timed out"))? } #[cfg(test)] mod tests { use super::*; #[test] fn uid_sets() { assert_eq!(uid_set(&[1, 2, 3, 7, 9, 10]), "1:3,7,9:10"); assert_eq!(uid_set(&[5]), "5"); assert_eq!(uid_set(&[10, 1, 2]), "1:2,10"); } #[test] fn permanent_flags() { let s = SelectInfo { permanent_flags: Some(vec!["\\Seen".into(), "\\*".into()]), ..Default::default() }; assert!(s.accepts_flag("\\Seen")); assert!(s.accepts_flag("$Forwarded")); assert!(!s.accepts_flag("\\Flagged")); let s = SelectInfo { permanent_flags: Some(vec!["\\Seen".into()]), ..Default::default() }; assert!(!s.accepts_flag("$Forwarded")); // EXAMINE: PERMANENTFLAGS () is meaningless for APPEND let s = SelectInfo { permanent_flags: Some(vec![]), read_only: true, ..Default::default() }; assert!(s.accepts_flag("\\Seen")); } }