imap_client/
session.rs

1//! Type-state IMAP session. Compile-time enforcement of valid state
2//! transitions plus credential-vs-cleartext separation.
3//!
4//! ## States
5//!
6//! | State          | Allowed commands |
7//! |----------------|------------------|
8//! | Unauthenticated | `STARTTLS` (PlainText only), `LOGIN` / `AUTHENTICATE` (Tls only), `CAPABILITY`, `LOGOUT`, `NOOP` |
9//! | Authenticated  | `SELECT` / `EXAMINE`, `LIST`, `LSUB`, `STATUS`, `LOGOUT`, `NOOP`, `CAPABILITY`, `ENABLE` |
10//! | Selected       | All Authenticated commands plus `FETCH`, `STORE`, `SEARCH`, `EXPUNGE`, `IDLE`, `CLOSE`, `UNSELECT`, `MOVE` |
11//!
12//! ## Capability re-fetch
13//!
14//! Per RFC 3501 §6.1.1 / RFC 9051 §6.2 capabilities MUST be re-evaluated
15//! after STARTTLS and after a successful authentication. The session does
16//! this automatically: if the server's tagged OK response carries a
17//! `[CAPABILITY …]` response code we use it, otherwise we issue an
18//! explicit `CAPABILITY` round-trip.
19
20use std::marker::PhantomData;
21use std::time::Duration;
22
23use base64::Engine as _;
24use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
25use tokio::sync::broadcast;
26
27use crate::capabilities::Capabilities;
28use crate::client::RawClient;
29use crate::credentials::{Password, imap_quoted};
30use crate::error::ClientError;
31use crate::flags::{Flag, StoreAction};
32use crate::idle::IdleHandle;
33use crate::search::SearchQuery;
34
35use imap_core::ast::{DataResponse, FetchAttribute, Response};
36use imap_core::parser::parse_response;
37
38// --- State markers -----------------------------------------------------
39
40pub struct Unauthenticated;
41pub struct Authenticated;
42pub struct Selected;
43
44// --- Transport markers -------------------------------------------------
45
46pub struct PlainText;
47pub struct Tls;
48
49/// Generic session enforcing compile-time state transitions.
50pub struct Session<State, Transport> {
51    raw: RawClient,
52    pub capabilities: Capabilities,
53    _state: PhantomData<State>,
54    _transport: PhantomData<Transport>,
55}
56
57#[derive(Debug, Clone)]
58pub struct FetchResult {
59    pub seq: u32,
60    pub uid: Option<u32>,
61    pub body: Option<Vec<u8>>,
62}
63
64// --- State-agnostic helpers --------------------------------------------
65
66impl<S, T> Session<S, T> {
67    fn transition_state<NewState>(self) -> Session<NewState, T> {
68        Session {
69            raw: self.raw,
70            capabilities: self.capabilities,
71            _state: PhantomData,
72            _transport: PhantomData,
73        }
74    }
75
76    pub fn transition_transport<NewTransport>(self) -> Session<S, NewTransport> {
77        Session {
78            raw: self.raw,
79            capabilities: self.capabilities,
80            _state: PhantomData,
81            _transport: PhantomData,
82        }
83    }
84
85    pub fn events(&self) -> broadcast::Receiver<Vec<u8>> {
86        self.raw.events()
87    }
88
89    /// Issue an explicit `CAPABILITY` command and update the cached set.
90    /// Used after STARTTLS and after authentication when the OK response
91    /// did not carry a `[CAPABILITY …]` response code.
92    pub async fn refresh_capabilities(&mut self) -> Result<(), ClientError> {
93        let mut events = self.raw.events();
94        let _resp = self.raw.execute_command("CAPABILITY").await?;
95        // The CAPABILITY response is broadcast as an untagged data frame.
96        while let Ok(event) = events.try_recv() {
97            if let Ok((_, response)) = parse_response(&event)
98                && self.capabilities.try_update_from(&response)
99            {
100                return Ok(());
101            }
102        }
103        Ok(())
104    }
105
106    /// If `frame` carries a `[CAPABILITY …]` response code, use it; else
107    /// issue an explicit `CAPABILITY` round-trip.
108    async fn refresh_capabilities_from_frame(&mut self, frame: &[u8]) -> Result<(), ClientError> {
109        if let Ok((_, response)) = parse_response(frame)
110            && self.capabilities.try_update_from(&response)
111        {
112            return Ok(());
113        }
114        self.refresh_capabilities().await
115    }
116}
117
118impl<T> Session<Unauthenticated, T> {
119    pub fn new(raw: RawClient, capabilities: Capabilities) -> Self {
120        Self::new_in_state(raw, capabilities)
121    }
122}
123
124impl<S, T> Session<S, T> {
125    pub(crate) fn new_in_state(raw: RawClient, capabilities: Capabilities) -> Self {
126        Self {
127            raw,
128            capabilities,
129            _state: PhantomData,
130            _transport: PhantomData,
131        }
132    }
133
134    /// `LOGOUT` is valid in any state; the session is consumed because the
135    /// connection is closed afterwards.
136    pub async fn logout(mut self) -> Result<(), ClientError> {
137        // Server replies `* BYE` (broadcast) then a tagged OK. Both BYE and
138        // ConnectionClosed are acceptable outcomes.
139        match self.raw.execute_command("LOGOUT").await {
140            Ok(_) | Err(ClientError::ConnectionClosed) => Ok(()),
141            Err(e) => Err(e),
142        }
143    }
144
145    /// `NOOP` keeps the connection alive and triggers status updates.
146    pub async fn noop(&mut self) -> Result<(), ClientError> {
147        self.raw.execute_command("NOOP").await.map(|_| ())
148    }
149}
150
151// --- Unauthenticated/Tls — login & SASL ---------------------------------
152
153impl Session<Unauthenticated, Tls> {
154    /// `LOGIN` over TLS. The username and password are sent as IMAP
155    /// quoted strings with proper escaping (`"`, `\`).
156    ///
157    /// Returns [`ClientError::CommandFailed`] if the credentials cannot be
158    /// represented as quoted strings (8-bit, control bytes, CR, LF). Use
159    /// [`Self::authenticate_plain`] in that case.
160    pub async fn login(
161        mut self,
162        user: &str,
163        pass: Password,
164    ) -> Result<Session<Authenticated, Tls>, ClientError> {
165        let user_q = imap_quoted(user)?;
166        let pass_q = pass.as_imap_quoted()?;
167        let cmd = format!("LOGIN {} {}", user_q, pass_q);
168
169        let frame = self.raw.execute_command(&cmd).await?;
170        self.refresh_capabilities_from_frame(&frame).await?;
171        Ok(self.transition_state())
172    }
173
174    /// `AUTHENTICATE PLAIN` (RFC 4616). Credentials travel as base64 over
175    /// the SASL exchange — no quoting limitations and 8-bit safe.
176    pub async fn authenticate_plain(
177        mut self,
178        user: &str,
179        pass: &Password,
180    ) -> Result<Session<Authenticated, Tls>, ClientError> {
181        if user.as_bytes().contains(&0) {
182            return Err(ClientError::CommandFailed(
183                "username must not contain NUL".into(),
184            ));
185        }
186        if pass.as_str().as_bytes().contains(&0) {
187            return Err(ClientError::CommandFailed(
188                "password must not contain NUL".into(),
189            ));
190        }
191
192        let mut events = self.raw.events();
193        let (_tag, reply_rx) = self.raw.send_command_async("AUTHENTICATE PLAIN").await?;
194
195        // Wait for `+ ...` continuation request.
196        wait_for_continuation(&mut events, Duration::from_secs(30)).await?;
197
198        // SASL PLAIN: \0<authzid>\0<authcid>\0<password>; we use empty authzid.
199        let mut sasl = Vec::with_capacity(2 + user.len() + pass.as_str().len());
200        sasl.push(0);
201        sasl.extend_from_slice(user.as_bytes());
202        sasl.push(0);
203        sasl.extend_from_slice(pass.as_str().as_bytes());
204        let mut payload = BASE64_STANDARD.encode(&sasl).into_bytes();
205        payload.extend_from_slice(b"\r\n");
206        self.raw.send_raw(payload).await?;
207
208        let frame = match reply_rx.await {
209            Ok(r) => r?,
210            Err(_) => return Err(ClientError::ConnectionClosed),
211        };
212        self.refresh_capabilities_from_frame(&frame).await?;
213        Ok(self.transition_state())
214    }
215}
216
217// --- Authenticated -----------------------------------------------------
218
219impl<T> Session<Authenticated, T> {
220    /// `SELECT` — open a mailbox read-write and transition to `Selected`.
221    pub async fn select(mut self, mailbox: &str) -> Result<Session<Selected, T>, ClientError> {
222        let mb = imap_quoted(mailbox)?;
223        let cmd = format!("SELECT {}", mb);
224        self.raw.execute_command(&cmd).await?;
225        Ok(self.transition_state())
226    }
227
228    /// `EXAMINE` — open a mailbox read-only and transition to `Selected`.
229    pub async fn examine(mut self, mailbox: &str) -> Result<Session<Selected, T>, ClientError> {
230        let mb = imap_quoted(mailbox)?;
231        let cmd = format!("EXAMINE {}", mb);
232        self.raw.execute_command(&cmd).await?;
233        Ok(self.transition_state())
234    }
235
236    /// `LIST` — list mailboxes matching `mailbox_mask` under `reference`.
237    pub async fn list(
238        &mut self,
239        reference: &str,
240        mailbox_mask: &str,
241    ) -> Result<Vec<u8>, ClientError> {
242        let cmd = format!(
243            "LIST {} {}",
244            imap_quoted(reference)?,
245            imap_quoted(mailbox_mask)?
246        );
247        self.raw.execute_command(&cmd).await
248    }
249}
250
251// --- Selected ----------------------------------------------------------
252
253impl<T> Session<Selected, T> {
254    /// `FETCH` returning the raw tagged-response bytes.
255    pub async fn fetch_raw(
256        &mut self,
257        sequence_set: &str,
258        items: &str,
259    ) -> Result<Vec<u8>, ClientError> {
260        let cmd = format!("FETCH {} {}", sequence_set, items);
261        self.raw.execute_command(&cmd).await
262    }
263
264    /// `FETCH` returning structured [`FetchResult`] entries derived from
265    /// the broadcast untagged FETCH frames.
266    pub async fn fetch(
267        &mut self,
268        sequence_set: &str,
269        items: &str,
270    ) -> Result<Vec<FetchResult>, ClientError> {
271        let mut events = self.raw.events();
272        let cmd = format!("FETCH {} {}", sequence_set, items);
273        let _resp = self.raw.execute_command(&cmd).await?;
274
275        let mut results = Vec::new();
276        while let Ok(event) = events.try_recv() {
277            if let Ok((_, Response::Data(DataResponse::Fetch { seq, attributes }))) =
278                parse_response(&event)
279            {
280                let mut uid = None;
281                let mut body = None;
282                for attr in attributes {
283                    match attr {
284                        FetchAttribute::Uid(u) => uid = Some(u),
285                        FetchAttribute::BodySection { data: Some(d), .. } => {
286                            body = Some(d.to_vec())
287                        }
288                        FetchAttribute::Body(b) => body = Some(b.to_vec()),
289                        FetchAttribute::Rfc822(b) => body = Some(b.to_vec()),
290                        _ => {}
291                    }
292                }
293                results.push(FetchResult { seq, uid, body });
294            }
295        }
296        Ok(results)
297    }
298
299    /// Convenience: fetch the body of the first message in `sequence_set`.
300    pub async fn fetch_body(&mut self, sequence_set: &str) -> Result<Option<String>, ClientError> {
301        let results = self.fetch(sequence_set, "BODY[]").await?;
302        if let Some(res) = results.first()
303            && let Some(body) = &res.body
304        {
305            return Ok(Some(String::from_utf8_lossy(body).into_owned()));
306        }
307        Ok(None)
308    }
309
310    pub async fn search(&mut self, query: SearchQuery) -> Result<Vec<u32>, ClientError> {
311        self.run_search(&format!("SEARCH {}", query.build())).await
312    }
313
314    pub async fn uid_search(&mut self, query: SearchQuery) -> Result<Vec<u32>, ClientError> {
315        self.run_search(&format!("UID SEARCH {}", query.build()))
316            .await
317    }
318
319    async fn run_search(&mut self, cmd: &str) -> Result<Vec<u32>, ClientError> {
320        let mut events = self.raw.events();
321        let _resp = self.raw.execute_command(cmd).await?;
322
323        let mut all_ids = Vec::new();
324        while let Ok(event) = events.try_recv() {
325            if let Ok((_, Response::Data(DataResponse::Search(ids)))) = parse_response(&event) {
326                all_ids.extend(ids);
327            }
328        }
329        Ok(all_ids)
330    }
331
332    pub async fn store(
333        &mut self,
334        sequence_set: &str,
335        action: StoreAction,
336        flags: &[Flag],
337    ) -> Result<Vec<u8>, ClientError> {
338        let flags_str = flags
339            .iter()
340            .map(|f| f.to_string())
341            .collect::<Vec<_>>()
342            .join(" ");
343        let cmd = format!(
344            "STORE {} {} ({})",
345            sequence_set,
346            action.to_imap_prefix(false),
347            flags_str
348        );
349        self.raw.execute_command(&cmd).await
350    }
351
352    pub async fn uid_store(
353        &mut self,
354        uid_set: &str,
355        action: StoreAction,
356        flags: &[Flag],
357    ) -> Result<Vec<u8>, ClientError> {
358        let flags_str = flags
359            .iter()
360            .map(|f| f.to_string())
361            .collect::<Vec<_>>()
362            .join(" ");
363        let cmd = format!(
364            "UID STORE {} {} ({})",
365            uid_set,
366            action.to_imap_prefix(false),
367            flags_str
368        );
369        self.raw.execute_command(&cmd).await
370    }
371
372    pub async fn expunge(&mut self) -> Result<Vec<u8>, ClientError> {
373        self.raw.execute_command("EXPUNGE").await
374    }
375
376    /// `CLOSE` — implicitly expunges deleted messages and transitions back
377    /// to `Authenticated`.
378    pub async fn close_mailbox(mut self) -> Result<Session<Authenticated, T>, ClientError> {
379        self.raw.execute_command("CLOSE").await?;
380        Ok(self.transition_state())
381    }
382
383    /// `UNSELECT` (RFC 3691) — like `CLOSE` but without expunging. Errors
384    /// if the server has not advertised the `UNSELECT` capability.
385    pub async fn unselect(mut self) -> Result<Session<Authenticated, T>, ClientError> {
386        if !self.capabilities.unselect {
387            return Err(ClientError::UnsupportedCapability("UNSELECT"));
388        }
389        self.raw.execute_command("UNSELECT").await?;
390        Ok(self.transition_state())
391    }
392
393    /// `CHECK` — implementation-defined housekeeping checkpoint.
394    pub async fn check(&mut self) -> Result<(), ClientError> {
395        self.raw.execute_command("CHECK").await.map(|_| ())
396    }
397
398    /// `IDLE` (RFC 2177). Returns an [`IdleHandle`]; call `stop()` on it
399    /// to gracefully terminate. Callers must re-issue IDLE at least every
400    /// ~28 minutes — see [`crate::idle`] for details.
401    pub async fn idle(&mut self) -> Result<IdleHandle, ClientError> {
402        if !self.capabilities.idle {
403            return Err(ClientError::UnsupportedCapability("IDLE"));
404        }
405        let mut events = self.raw.events();
406        let writer = self.raw.writer();
407        let (_tag, reply_rx) = self.raw.send_command_async("IDLE").await?;
408        wait_for_continuation(&mut events, Duration::from_secs(30)).await?;
409        Ok(IdleHandle::new(writer, reply_rx))
410    }
411
412    /// `MOVE` (RFC 6851). Errors if the server has not advertised `MOVE`.
413    pub async fn move_messages(
414        &mut self,
415        sequence_set: &str,
416        mailbox: &str,
417    ) -> Result<Vec<u8>, ClientError> {
418        if !self.capabilities.move_ext {
419            return Err(ClientError::UnsupportedCapability("MOVE"));
420        }
421        let cmd = format!("MOVE {} {}", sequence_set, imap_quoted(mailbox)?);
422        self.raw.execute_command(&cmd).await
423    }
424}
425
426// --- Helpers -----------------------------------------------------------
427
428/// Wait for the next `+ ...` continuation request on the broadcast
429/// channel, dropping any other untagged frames in the interim.
430async fn wait_for_continuation(
431    events: &mut broadcast::Receiver<Vec<u8>>,
432    timeout: Duration,
433) -> Result<(), ClientError> {
434    let deadline = tokio::time::Instant::now() + timeout;
435    loop {
436        let now = tokio::time::Instant::now();
437        if now >= deadline {
438            return Err(ClientError::Timeout);
439        }
440        match tokio::time::timeout(deadline - now, events.recv()).await {
441            Ok(Ok(frame)) => {
442                if frame.starts_with(b"+") {
443                    return Ok(());
444                }
445                // Otherwise it's an unrelated untagged frame — keep waiting.
446            }
447            Ok(Err(broadcast::error::RecvError::Closed)) => {
448                return Err(ClientError::ConnectionClosed);
449            }
450            Ok(Err(broadcast::error::RecvError::Lagged(_))) => {
451                // We may have missed the continuation — the dispatcher's
452                // bound is generous, but treat lag as a protocol error
453                // rather than guessing.
454                return Err(ClientError::CommandFailed(
455                    "broadcast lagged; continuation may have been missed".into(),
456                ));
457            }
458            Err(_) => return Err(ClientError::Timeout),
459        }
460    }
461}
462
463// --- Tests -------------------------------------------------------------
464
465#[cfg(test)]
466mod tests {
467    use super::*;
468    use crate::{Flag, SearchQuery, StoreAction, Tls};
469    use tokio::io::{AsyncReadExt, AsyncWriteExt, duplex};
470
471    async fn read_cmd_tag(server: &mut tokio::io::DuplexStream) -> String {
472        let mut buf = [0u8; 1024];
473        let n = server.read(&mut buf).await.unwrap();
474        String::from_utf8_lossy(&buf[..n])
475            .split_whitespace()
476            .next()
477            .unwrap()
478            .to_owned()
479    }
480
481    fn unauth_session(client_io: tokio::io::DuplexStream) -> Session<Unauthenticated, Tls> {
482        let raw = RawClient::new(client_io);
483        Session::<Unauthenticated, Tls>::new_in_state(raw, Capabilities::default())
484    }
485
486    fn auth_session(client_io: tokio::io::DuplexStream) -> Session<Authenticated, Tls> {
487        let raw = RawClient::new(client_io);
488        Session::<Authenticated, Tls>::new_in_state(raw, Capabilities::default())
489    }
490
491    fn selected_session(client_io: tokio::io::DuplexStream) -> Session<Selected, Tls> {
492        let raw = RawClient::new(client_io);
493        Session::<Selected, Tls>::new_in_state(raw, Capabilities::default())
494    }
495
496    fn selected_session_with_caps(
497        client_io: tokio::io::DuplexStream,
498        caps: Capabilities,
499    ) -> Session<Selected, Tls> {
500        let raw = RawClient::new(client_io);
501        Session::<Selected, Tls>::new_in_state(raw, caps)
502    }
503
504    #[tokio::test]
505    async fn test_session_login() {
506        let (client_io, mut server_io) = duplex(1024);
507        let session = unauth_session(client_io);
508
509        let login_task =
510            tokio::spawn(async move { session.login("user", Password::new("pass")).await });
511
512        // Server reads `A0001 LOGIN "user" "pass"\r\n`
513        let mut buf = [0u8; 1024];
514        let n = server_io.read(&mut buf).await.unwrap();
515        let cmd = String::from_utf8_lossy(&buf[..n]).into_owned();
516        assert!(cmd.contains("LOGIN \"user\" \"pass\""));
517        let tag = cmd.split_whitespace().next().unwrap().to_owned();
518
519        server_io
520            .write_all(
521                format!("{} OK [CAPABILITY IMAP4rev2 IDLE] LOGIN completed\r\n", tag).as_bytes(),
522            )
523            .await
524            .unwrap();
525
526        let auth = login_task.await.unwrap().unwrap();
527        // No second CAPABILITY round-trip — caps came from the response code.
528        assert!(auth.capabilities.imap4rev2);
529        assert!(auth.capabilities.idle);
530    }
531
532    #[tokio::test]
533    async fn test_session_login_falls_back_to_capability_round_trip() {
534        let (client_io, mut server_io) = duplex(1024);
535        let session = unauth_session(client_io);
536
537        let login_task =
538            tokio::spawn(async move { session.login("user", Password::new("pass")).await });
539
540        let tag = read_cmd_tag(&mut server_io).await;
541        server_io
542            .write_all(format!("{} OK LOGIN completed\r\n", tag).as_bytes())
543            .await
544            .unwrap();
545
546        // Server now sees CAPABILITY round-trip.
547        let tag2 = read_cmd_tag(&mut server_io).await;
548        server_io
549            .write_all(b"* CAPABILITY IMAP4rev2 STARTTLS UNSELECT\r\n")
550            .await
551            .unwrap();
552        server_io
553            .write_all(format!("{} OK CAPABILITY done\r\n", tag2).as_bytes())
554            .await
555            .unwrap();
556
557        let auth = login_task.await.unwrap().unwrap();
558        assert!(auth.capabilities.imap4rev2);
559        assert!(auth.capabilities.unselect);
560    }
561
562    #[tokio::test]
563    async fn test_login_escapes_quote_and_backslash() {
564        let (client_io, mut server_io) = duplex(1024);
565        let session = unauth_session(client_io);
566
567        let login_task = tokio::spawn(async move {
568            session
569                .login("user\"with\\specials", Password::new("p@ss\"\\"))
570                .await
571        });
572
573        let mut buf = [0u8; 1024];
574        let n = server_io.read(&mut buf).await.unwrap();
575        let cmd = String::from_utf8_lossy(&buf[..n]).into_owned();
576        // Both quote and backslash must be escaped on the wire.
577        assert!(cmd.contains(r#"LOGIN "user\"with\\specials" "p@ss\"\\""#));
578        let tag = cmd.split_whitespace().next().unwrap().to_owned();
579        // Include [CAPABILITY] response code to skip the explicit refresh.
580        server_io
581            .write_all(format!("{} OK [CAPABILITY IMAP4rev2] done\r\n", tag).as_bytes())
582            .await
583            .unwrap();
584        let _ = login_task.await.unwrap().unwrap();
585    }
586
587    #[tokio::test]
588    async fn test_login_rejects_8bit_password() {
589        let (client_io, _server_io) = duplex(1024);
590        let session = unauth_session(client_io);
591        let r = session.login("user", Password::new("café")).await;
592        assert!(matches!(r, Err(ClientError::CommandFailed(_))));
593    }
594
595    #[tokio::test]
596    async fn test_login_failure() {
597        let (client_io, mut server_io) = duplex(1024);
598        let session = unauth_session(client_io);
599        let task = tokio::spawn(async move { session.login("user", Password::new("pass")).await });
600        let tag = read_cmd_tag(&mut server_io).await;
601        server_io
602            .write_all(format!("{} NO authentication failed\r\n", tag).as_bytes())
603            .await
604            .unwrap();
605        match task.await.unwrap() {
606            Err(ClientError::CommandFailed(t)) => assert!(t.contains("authentication")),
607            _ => panic!("unexpected variant"),
608        }
609    }
610
611    #[tokio::test]
612    async fn test_authenticate_plain() {
613        let (client_io, mut server_io) = duplex(1024);
614        let session = unauth_session(client_io);
615        let task = tokio::spawn(async move {
616            session
617                .authenticate_plain("user", &Password::new("pass"))
618                .await
619        });
620
621        // Server sees `A0001 AUTHENTICATE PLAIN\r\n`
622        let mut buf = [0u8; 1024];
623        let n = server_io.read(&mut buf).await.unwrap();
624        let cmd = String::from_utf8_lossy(&buf[..n]).into_owned();
625        assert!(cmd.contains("AUTHENTICATE PLAIN"));
626        let tag = cmd.split_whitespace().next().unwrap().to_owned();
627
628        // Send continuation
629        server_io.write_all(b"+ \r\n").await.unwrap();
630
631        // Read base64 SASL payload
632        let n = server_io.read(&mut buf).await.unwrap();
633        let payload_b64 = String::from_utf8_lossy(&buf[..n]).trim_end().to_string();
634        let payload = base64::engine::general_purpose::STANDARD
635            .decode(payload_b64)
636            .unwrap();
637        assert_eq!(payload, b"\0user\0pass");
638
639        server_io
640            .write_all(format!("{} OK [CAPABILITY IMAP4rev2] auth done\r\n", tag).as_bytes())
641            .await
642            .unwrap();
643
644        let _auth = task.await.unwrap().unwrap();
645    }
646
647    #[tokio::test]
648    async fn test_authenticate_plain_failure() {
649        let (client_io, mut server_io) = duplex(1024);
650        let session = unauth_session(client_io);
651        let task = tokio::spawn(async move {
652            session
653                .authenticate_plain("user", &Password::new("pass"))
654                .await
655        });
656
657        let tag = read_cmd_tag(&mut server_io).await;
658        server_io.write_all(b"+ \r\n").await.unwrap();
659        let mut buf = [0u8; 1024];
660        let _ = server_io.read(&mut buf).await.unwrap();
661        server_io
662            .write_all(format!("{} NO bad creds\r\n", tag).as_bytes())
663            .await
664            .unwrap();
665
666        assert!(matches!(
667            task.await.unwrap(),
668            Err(ClientError::CommandFailed(_))
669        ));
670    }
671
672    #[tokio::test]
673    async fn test_session_select() {
674        let (client_io, mut server_io) = duplex(1024);
675        let session = auth_session(client_io);
676        let task = tokio::spawn(async move { session.select("INBOX").await });
677        let tag = read_cmd_tag(&mut server_io).await;
678        server_io
679            .write_all(format!("{} OK SELECT completed\r\n", tag).as_bytes())
680            .await
681            .unwrap();
682        let _selected = task.await.unwrap().unwrap();
683    }
684
685    #[tokio::test]
686    async fn test_session_examine() {
687        let (client_io, mut server_io) = duplex(1024);
688        let session = auth_session(client_io);
689        let task = tokio::spawn(async move { session.examine("INBOX").await });
690        let mut buf = [0u8; 1024];
691        let n = server_io.read(&mut buf).await.unwrap();
692        let cmd = String::from_utf8_lossy(&buf[..n]).into_owned();
693        assert!(cmd.contains("EXAMINE \"INBOX\""));
694        let tag = cmd.split_whitespace().next().unwrap();
695        server_io
696            .write_all(format!("{} OK EXAMINE completed\r\n", tag).as_bytes())
697            .await
698            .unwrap();
699        let _ = task.await.unwrap().unwrap();
700    }
701
702    #[tokio::test]
703    async fn test_session_select_failure() {
704        let (client_io, mut server_io) = duplex(1024);
705        let session = auth_session(client_io);
706        let task = tokio::spawn(async move { session.select("BadBox").await });
707        let tag = read_cmd_tag(&mut server_io).await;
708        server_io
709            .write_all(format!("{} NO no such mailbox\r\n", tag).as_bytes())
710            .await
711            .unwrap();
712        assert!(matches!(
713            task.await.unwrap(),
714            Err(ClientError::CommandFailed(_))
715        ));
716    }
717
718    #[tokio::test]
719    async fn test_session_logout() {
720        let (client_io, mut server_io) = duplex(1024);
721        let session = auth_session(client_io);
722        let task = tokio::spawn(async move { session.logout().await });
723        let tag = read_cmd_tag(&mut server_io).await;
724        server_io.write_all(b"* BYE goodbye\r\n").await.unwrap();
725        server_io
726            .write_all(format!("{} OK LOGOUT completed\r\n", tag).as_bytes())
727            .await
728            .unwrap();
729        task.await.unwrap().unwrap();
730    }
731
732    #[tokio::test]
733    async fn test_session_noop() {
734        let (client_io, mut server_io) = duplex(1024);
735        let mut session = auth_session(client_io);
736        let task = tokio::spawn(async move {
737            let r = session.noop().await;
738            (session, r)
739        });
740        let tag = read_cmd_tag(&mut server_io).await;
741        server_io
742            .write_all(format!("{} OK NOOP done\r\n", tag).as_bytes())
743            .await
744            .unwrap();
745        let (_session, r) = task.await.unwrap();
746        r.unwrap();
747    }
748
749    #[tokio::test]
750    async fn test_session_close() {
751        let (client_io, mut server_io) = duplex(1024);
752        let session = selected_session(client_io);
753        let task = tokio::spawn(async move { session.close_mailbox().await });
754        let tag = read_cmd_tag(&mut server_io).await;
755        server_io
756            .write_all(format!("{} OK CLOSE done\r\n", tag).as_bytes())
757            .await
758            .unwrap();
759        let _ = task.await.unwrap().unwrap();
760    }
761
762    #[tokio::test]
763    async fn test_session_unselect_unsupported() {
764        let (client_io, _server_io) = duplex(1024);
765        let session = selected_session(client_io);
766        // Default Capabilities has unselect=false.
767        match session.unselect().await {
768            Err(ClientError::UnsupportedCapability("UNSELECT")) => {}
769            _ => panic!("unexpected variant"),
770        }
771    }
772
773    #[tokio::test]
774    async fn test_session_unselect_supported() {
775        let (client_io, mut server_io) = duplex(1024);
776        let caps = Capabilities {
777            unselect: true,
778            ..Default::default()
779        };
780        let session = selected_session_with_caps(client_io, caps);
781        let task = tokio::spawn(async move { session.unselect().await });
782        let tag = read_cmd_tag(&mut server_io).await;
783        server_io
784            .write_all(format!("{} OK UNSELECT done\r\n", tag).as_bytes())
785            .await
786            .unwrap();
787        let _ = task.await.unwrap().unwrap();
788    }
789
790    #[tokio::test]
791    async fn test_session_check() {
792        let (client_io, mut server_io) = duplex(1024);
793        let mut session = selected_session(client_io);
794        let task = tokio::spawn(async move {
795            let r = session.check().await;
796            (session, r)
797        });
798        let tag = read_cmd_tag(&mut server_io).await;
799        server_io
800            .write_all(format!("{} OK CHECK done\r\n", tag).as_bytes())
801            .await
802            .unwrap();
803        let (_, r) = task.await.unwrap();
804        r.unwrap();
805    }
806
807    #[tokio::test]
808    async fn test_session_idle_unsupported() {
809        let (client_io, _server_io) = duplex(1024);
810        let mut session = selected_session(client_io);
811        match session.idle().await {
812            Err(ClientError::UnsupportedCapability("IDLE")) => {}
813            _ => panic!("unexpected variant"),
814        }
815    }
816
817    #[tokio::test]
818    async fn test_session_idle_flow() {
819        let (client_io, mut server_io) = duplex(1024);
820        let caps = Capabilities {
821            idle: true,
822            ..Default::default()
823        };
824        let mut session = selected_session_with_caps(client_io, caps);
825        let task = tokio::spawn(async move {
826            let h = session.idle().await.unwrap();
827            (session, h)
828        });
829
830        let tag = read_cmd_tag(&mut server_io).await;
831        server_io.write_all(b"+ idling\r\n").await.unwrap();
832
833        let (_session, handle) = task.await.unwrap();
834
835        let stop_task = tokio::spawn(async move { handle.stop().await });
836        let mut buf = [0u8; 1024];
837        let n = server_io.read(&mut buf).await.unwrap();
838        assert_eq!(&buf[..n], b"DONE\r\n");
839        server_io
840            .write_all(format!("{} OK IDLE done\r\n", tag).as_bytes())
841            .await
842            .unwrap();
843        stop_task.await.unwrap().unwrap();
844    }
845
846    #[tokio::test]
847    async fn test_session_search() {
848        let (client_io, mut server_io) = duplex(1024);
849        let mut session = selected_session(client_io);
850        let task = tokio::spawn(async move { session.search(SearchQuery::subject("test")).await });
851        let tag = read_cmd_tag(&mut server_io).await;
852        server_io.write_all(b"* SEARCH 1 2 3\r\n").await.unwrap();
853        server_io
854            .write_all(format!("{} OK SEARCH completed\r\n", tag).as_bytes())
855            .await
856            .unwrap();
857        assert_eq!(task.await.unwrap().unwrap(), vec![1, 2, 3]);
858    }
859
860    #[tokio::test]
861    async fn test_session_uid_search() {
862        let (client_io, mut server_io) = duplex(1024);
863        let mut session = selected_session(client_io);
864        let task = tokio::spawn(async move { session.uid_search(SearchQuery::subject("t")).await });
865        let mut buf = [0u8; 1024];
866        let n = server_io.read(&mut buf).await.unwrap();
867        let cmd = String::from_utf8_lossy(&buf[..n]).into_owned();
868        assert!(cmd.contains("UID SEARCH"));
869        let tag = cmd.split_whitespace().next().unwrap();
870        server_io.write_all(b"* SEARCH 4 5 6\r\n").await.unwrap();
871        server_io
872            .write_all(format!("{} OK done\r\n", tag).as_bytes())
873            .await
874            .unwrap();
875        assert_eq!(task.await.unwrap().unwrap(), vec![4, 5, 6]);
876    }
877
878    #[tokio::test]
879    async fn test_session_search_failure() {
880        let (client_io, mut server_io) = duplex(1024);
881        let mut session = selected_session(client_io);
882        let task = tokio::spawn(async move { session.search(SearchQuery::subject("t")).await });
883        let tag = read_cmd_tag(&mut server_io).await;
884        server_io
885            .write_all(format!("{} NO failed\r\n", tag).as_bytes())
886            .await
887            .unwrap();
888        assert!(task.await.unwrap().is_err());
889    }
890
891    #[tokio::test]
892    async fn test_session_run_search_multiple_events() {
893        let (client_io, mut server_io) = duplex(1024);
894        let mut session = selected_session(client_io);
895        let task = tokio::spawn(async move { session.search(SearchQuery::subject("t")).await });
896        let tag = read_cmd_tag(&mut server_io).await;
897        server_io.write_all(b"* SEARCH 1 2\r\n").await.unwrap();
898        server_io.write_all(b"* SEARCH 3 4\r\n").await.unwrap();
899        server_io
900            .write_all(format!("{} OK done\r\n", tag).as_bytes())
901            .await
902            .unwrap();
903        assert_eq!(task.await.unwrap().unwrap(), vec![1, 2, 3, 4]);
904    }
905
906    #[tokio::test]
907    async fn test_session_store_and_uid_store() {
908        for (kind, expected_prefix) in [("STORE", "STORE"), ("UID STORE", "UID STORE")] {
909            let (client_io, mut server_io) = duplex(1024);
910            let mut session = selected_session(client_io);
911            let task = tokio::spawn(async move {
912                if kind == "STORE" {
913                    session.store("1", StoreAction::Add, &[Flag::Seen]).await
914                } else {
915                    session
916                        .uid_store("1", StoreAction::Add, &[Flag::Seen])
917                        .await
918                }
919            });
920            let mut buf = [0u8; 1024];
921            let n = server_io.read(&mut buf).await.unwrap();
922            assert!(String::from_utf8_lossy(&buf[..n]).contains(expected_prefix));
923            let tag = String::from_utf8_lossy(&buf[..n])
924                .split_whitespace()
925                .next()
926                .unwrap()
927                .to_owned();
928            server_io
929                .write_all(format!("{} OK done\r\n", tag).as_bytes())
930                .await
931                .unwrap();
932            let _ = task.await.unwrap().unwrap();
933        }
934    }
935
936    #[tokio::test]
937    async fn test_session_expunge() {
938        let (client_io, mut server_io) = duplex(1024);
939        let mut session = selected_session(client_io);
940        let task = tokio::spawn(async move { session.expunge().await });
941        let tag = read_cmd_tag(&mut server_io).await;
942        server_io
943            .write_all(format!("{} OK done\r\n", tag).as_bytes())
944            .await
945            .unwrap();
946        let _ = task.await.unwrap().unwrap();
947    }
948
949    #[tokio::test]
950    async fn test_session_move_unsupported() {
951        let (client_io, _server_io) = duplex(1024);
952        let mut session = selected_session(client_io);
953        match session.move_messages("1", "Archive").await {
954            Err(ClientError::UnsupportedCapability("MOVE")) => {}
955            _ => panic!("unexpected variant"),
956        }
957    }
958
959    #[tokio::test]
960    async fn test_session_move_supported() {
961        let (client_io, mut server_io) = duplex(1024);
962        let caps = Capabilities {
963            move_ext: true,
964            ..Default::default()
965        };
966        let mut session = selected_session_with_caps(client_io, caps);
967        let task = tokio::spawn(async move { session.move_messages("1", "Archive").await });
968        let mut buf = [0u8; 1024];
969        let n = server_io.read(&mut buf).await.unwrap();
970        let cmd = String::from_utf8_lossy(&buf[..n]).into_owned();
971        assert!(cmd.contains(r#"MOVE 1 "Archive""#));
972        let tag = cmd.split_whitespace().next().unwrap();
973        server_io
974            .write_all(format!("{} OK done\r\n", tag).as_bytes())
975            .await
976            .unwrap();
977        let _ = task.await.unwrap().unwrap();
978    }
979
980    #[tokio::test]
981    async fn test_session_list() {
982        let (client_io, mut server_io) = duplex(1024);
983        let mut session = auth_session(client_io);
984        let task = tokio::spawn(async move { session.list("", "*").await });
985        let tag = read_cmd_tag(&mut server_io).await;
986        server_io
987            .write_all(format!("{} OK done\r\n", tag).as_bytes())
988            .await
989            .unwrap();
990        let _ = task.await.unwrap().unwrap();
991    }
992
993    #[tokio::test]
994    async fn test_session_list_failure() {
995        let (client_io, mut server_io) = duplex(1024);
996        let mut session = auth_session(client_io);
997        let task = tokio::spawn(async move { session.list("", "*").await });
998        let tag = read_cmd_tag(&mut server_io).await;
999        server_io
1000            .write_all(format!("{} NO failed\r\n", tag).as_bytes())
1001            .await
1002            .unwrap();
1003        assert!(task.await.unwrap().is_err());
1004    }
1005
1006    #[tokio::test]
1007    async fn test_session_fetch_raw() {
1008        let (client_io, mut server_io) = duplex(1024);
1009        let mut session = selected_session(client_io);
1010        let task = tokio::spawn(async move { session.fetch_raw("1", "ALL").await });
1011        let tag = read_cmd_tag(&mut server_io).await;
1012        server_io
1013            .write_all(format!("{} OK done\r\n", tag).as_bytes())
1014            .await
1015            .unwrap();
1016        let _ = task.await.unwrap().unwrap();
1017    }
1018
1019    #[tokio::test]
1020    async fn test_session_fetch_structured() {
1021        let (client_io, mut server_io) = duplex(1024);
1022        let mut session = selected_session(client_io);
1023        let task = tokio::spawn(async move { session.fetch("1", "BODY[]").await });
1024        let tag = read_cmd_tag(&mut server_io).await;
1025        server_io
1026            .write_all(b"* 1 FETCH (BODY[] {10}\r\n0123456789 UID 123)\r\n")
1027            .await
1028            .unwrap();
1029        server_io
1030            .write_all(format!("{} OK done\r\n", tag).as_bytes())
1031            .await
1032            .unwrap();
1033        let results = task.await.unwrap().unwrap();
1034        assert_eq!(results.len(), 1);
1035        assert_eq!(results[0].seq, 1);
1036        assert_eq!(results[0].uid, Some(123));
1037        assert_eq!(results[0].body.as_deref(), Some(&b"0123456789"[..]));
1038    }
1039
1040    #[tokio::test]
1041    async fn test_session_fetch_body_helper() {
1042        let (client_io, mut server_io) = duplex(1024);
1043        let mut session = selected_session(client_io);
1044        let task = tokio::spawn(async move { session.fetch_body("1").await });
1045        let tag = read_cmd_tag(&mut server_io).await;
1046        server_io
1047            .write_all(b"* 1 FETCH (BODY[] {10}\r\n0123456789)\r\n")
1048            .await
1049            .unwrap();
1050        server_io
1051            .write_all(format!("{} OK done\r\n", tag).as_bytes())
1052            .await
1053            .unwrap();
1054        assert_eq!(task.await.unwrap().unwrap(), Some("0123456789".to_string()));
1055    }
1056
1057    #[tokio::test]
1058    async fn test_session_transition_transport() {
1059        let (client_io, _server_io) = duplex(1024);
1060        let session = unauth_session(client_io);
1061        let _ = session.transition_transport::<crate::PlainText>();
1062    }
1063
1064    #[tokio::test]
1065    async fn test_session_events() {
1066        let (client_io, _server_io) = duplex(1024);
1067        let session = unauth_session(client_io);
1068        let _ = session.events();
1069    }
1070
1071    #[tokio::test]
1072    async fn test_refresh_capabilities_explicit() {
1073        let (client_io, mut server_io) = duplex(1024);
1074        let mut session = unauth_session(client_io);
1075        let task = tokio::spawn(async move {
1076            let r = session.refresh_capabilities().await;
1077            (session, r)
1078        });
1079        let tag = read_cmd_tag(&mut server_io).await;
1080        server_io
1081            .write_all(b"* CAPABILITY IMAP4rev2 IDLE\r\n")
1082            .await
1083            .unwrap();
1084        server_io
1085            .write_all(format!("{} OK done\r\n", tag).as_bytes())
1086            .await
1087            .unwrap();
1088        let (session, r) = task.await.unwrap();
1089        r.unwrap();
1090        assert!(session.capabilities.imap4rev2);
1091        assert!(session.capabilities.idle);
1092    }
1093}
imap_client/session.rs

imap_client/
session.rs
