Update imap-flow, clean IDLE

Cargo.lock

@@ -1965,7 +1965,7 @@ dependencies = [
 [[package]]
 name = "imap-codec"
 version = "2.0.0"
-source = "git+https://github.com/superboum/imap-codec?branch=custom/aerogramme#0adcc244282c64cc7874ffa9cd22e4a451ee19f8"
+source = "git+https://github.com/superboum/imap-codec?branch=custom/aerogramme#d8a5afc03fb771232e94c73af6a05e79dc80bbed"
 dependencies = [
  "abnf-core",
  "base64 0.21.7",
@@ -1980,7 +1980,7 @@ dependencies = [
 [[package]]
 name = "imap-flow"
 version = "0.1.0"
-source = "git+https://github.com/superboum/imap-flow.git?branch=custom/aerogramme#2bea066da1e09ad04bb5fb71b0dd8d6e5d9e3d19"
+source = "git+https://github.com/duesee/imap-flow.git?branch=main#68c1da5d1c56dbe543d9736de9683259d1d28191"
 dependencies = [
  "bounded-static",
  "bytes",
@@ -1994,7 +1994,7 @@ dependencies = [
 [[package]]
 name = "imap-types"
 version = "2.0.0"
-source = "git+https://github.com/superboum/imap-codec?branch=custom/aerogramme#0adcc244282c64cc7874ffa9cd22e4a451ee19f8"
+source = "git+https://github.com/superboum/imap-codec?branch=custom/aerogramme#d8a5afc03fb771232e94c73af6a05e79dc80bbed"
 dependencies = [
  "base64 0.21.7",
  "bounded-static",

Cargo.toml

@@ -64,7 +64,7 @@ eml-codec = "0.1.2"
 smtp-message = { git = "http://github.com/Alexis211/kannader", branch = "feature/lmtp" }
 smtp-server = { git = "http://github.com/Alexis211/kannader", branch = "feature/lmtp" }
 imap-codec = { version = "2.0.0", features = ["bounded-static", "ext_condstore_qresync"] }
-imap-flow = { git = "https://github.com/superboum/imap-flow.git", branch = "custom/aerogramme" }
+imap-flow = { git = "https://github.com/duesee/imap-flow.git", branch = "main" }
 thiserror = "1.0.56"
 
 [dev-dependencies]

src/imap/capability.rs

@@ -1,5 +1,5 @@
 use imap_codec::imap_types::command::{FetchModifier, SelectExamineModifier, StoreModifier};
-use imap_codec::imap_types::core::NonEmptyVec;
+use imap_codec::imap_types::core::Vec1;
 use imap_codec::imap_types::extensions::enable::{CapabilityEnable, Utf8Kind};
 use imap_codec::imap_types::response::Capability;
 use std::collections::HashSet;
@@ -49,7 +49,7 @@ impl Default for ServerCapability {
 }
 
 impl ServerCapability {
-    pub fn to_vec(&self) -> NonEmptyVec<Capability<'static>> {
+    pub fn to_vec(&self) -> Vec1<Capability<'static>> {
         self.0
             .iter()
             .map(|v| v.clone())

src/imap/command/authenticated.rs

@@ -6,7 +6,7 @@ use anyhow::{anyhow, bail, Result};
 use imap_codec::imap_types::command::{
     Command, CommandBody, ListReturnItem, SelectExamineModifier,
 };
-use imap_codec::imap_types::core::{Atom, Literal, NonEmptyVec, QuotedChar};
+use imap_codec::imap_types::core::{Atom, Literal, Vec1, QuotedChar};
 use imap_codec::imap_types::datetime::DateTime;
 use imap_codec::imap_types::extensions::enable::CapabilityEnable;
 use imap_codec::imap_types::flag::{Flag, FlagNameAttribute};
@@ -584,7 +584,7 @@ impl<'a> AuthenticatedContext<'a> {
 
     fn enable(
         self,
-        cap_enable: &NonEmptyVec<CapabilityEnable<'static>>,
+        cap_enable: &Vec1<CapabilityEnable<'static>>,
     ) -> Result<(Response<'static>, flow::Transition)> {
         let mut response_builder = Response::build().to_req(self.req);
         let capabilities = self.client_capabilities.try_enable(cap_enable.as_ref());

src/imap/command/selected.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 
 use anyhow::Result;
 use imap_codec::imap_types::command::{Command, CommandBody, FetchModifier, StoreModifier};
-use imap_codec::imap_types::core::Charset;
+use imap_codec::imap_types::core::{Vec1, Charset};
 use imap_codec::imap_types::fetch::MacroOrMessageDataItemNames;
 use imap_codec::imap_types::flag::{Flag, StoreResponse, StoreType};
 use imap_codec::imap_types::mailbox::Mailbox as MailboxCodec;
@@ -54,11 +54,12 @@ pub async fn dispatch<'a>(
             ctx.fetch(sequence_set, macro_or_item_names, modifiers, uid)
                 .await
         }
+        //@FIXME SearchKey::And is a legacy hack, should be refactored
         CommandBody::Search {
             charset,
             criteria,
             uid,
-        } => ctx.search(charset, criteria, uid).await,
+        } => ctx.search(charset, &SearchKey::And(criteria.clone()), uid).await,
         CommandBody::Expunge {
             // UIDPLUS (rfc4315)
             uid_sequence_set,
@@ -88,15 +89,6 @@ pub async fn dispatch<'a>(
         // UNSELECT extension (rfc3691)
         CommandBody::Unselect => ctx.unselect().await,
 
-        // IDLE extension (rfc2177)
-        CommandBody::Idle => Ok((
-            Response::build()
-                .to_req(ctx.req)
-                .message("DUMMY command due to anti-pattern in the code")
-                .ok()?,
-            flow::Transition::Idle(ctx.req.tag.clone(), tokio::sync::Notify::new()),
-        )),
-
         // In selected mode, we fallback to authenticated when needed
         _ => {
             authenticated::dispatch(authenticated::AuthenticatedContext {

src/imap/flow.rs

@@ -1,11 +1,12 @@
 use std::error::Error as StdError;
 use std::fmt;
 use std::sync::Arc;
+
 use tokio::sync::Notify;
+use imap_codec::imap_types::core::Tag;
 
 use crate::imap::mailbox_view::MailboxView;
 use crate::mail::user::User;
-use imap_codec::imap_types::core::Tag;
 
 #[derive(Debug)]
 pub enum Error {
@@ -31,6 +32,14 @@ pub enum State {
     ),
     Logout,
 }
+impl State {
+    pub fn notify(&self) -> Option<Arc<Notify>> {
+        match self {
+            Self::Idle(_, _, _, _, anotif) => Some(anotif.clone()),
+            _ => None,
+        }
+    }
+}
 impl fmt::Display for State {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         use State::*;

src/imap/mailbox_view.rs

@@ -6,7 +6,7 @@ use anyhow::{anyhow, Error, Result};
 
 use futures::stream::{StreamExt, TryStreamExt};
 
-use imap_codec::imap_types::core::Charset;
+use imap_codec::imap_types::core::{Charset, Vec1};
 use imap_codec::imap_types::fetch::MessageDataItem;
 use imap_codec::imap_types::flag::{Flag, FlagFetch, FlagPerm, StoreResponse, StoreType};
 use imap_codec::imap_types::response::{Code, CodeOther, Data, Status};
@@ -629,7 +629,7 @@ impl MailboxView {
 mod tests {
     use super::*;
     use imap_codec::encode::Encoder;
-    use imap_codec::imap_types::core::NonEmptyVec;
+    use imap_codec::imap_types::core::Vec1;
     use imap_codec::imap_types::fetch::Section;
     use imap_codec::imap_types::fetch::{MacroOrMessageDataItemNames, MessageDataItemName};
     use imap_codec::imap_types::response::Response;
@@ -749,7 +749,7 @@ mod tests {
 
             let test_repr = Response::Data(Data::Fetch {
                 seq: NonZeroU32::new(1).unwrap(),
-                items: NonEmptyVec::from(MessageDataItem::Body(mime_view::bodystructure(
+                items: Vec1::from(MessageDataItem::Body(mime_view::bodystructure(
                     &message.child,
                     false,
                 )?)),

src/imap/mime_view.rs

@@ -8,7 +8,7 @@ use imap_codec::imap_types::body::{
     BasicFields, Body as FetchBody, BodyStructure, MultiPartExtensionData, SinglePartExtensionData,
     SpecificFields,
 };
-use imap_codec::imap_types::core::{AString, IString, NString, NonEmptyVec};
+use imap_codec::imap_types::core::{AString, IString, NString, Vec1};
 use imap_codec::imap_types::fetch::{Part as FetchPart, Section as FetchSection};
 
 use eml_codec::{
@@ -141,8 +141,8 @@ impl<'a> NodeMime<'a> {
 enum SubsettedSection<'a> {
     Part,
     Header,
-    HeaderFields(&'a NonEmptyVec<AString<'a>>),
+    HeaderFields(&'a Vec1<AString<'a>>),
-    HeaderFieldsNot(&'a NonEmptyVec<AString<'a>>),
+    HeaderFieldsNot(&'a Vec1<AString<'a>>),
     Text,
     Mime,
 }
@@ -238,7 +238,7 @@ impl<'a> SelectedMime<'a> {
     /// case-insensitive but otherwise exact.
     fn header_fields(
         &self,
-        fields: &'a NonEmptyVec<AString<'a>>,
+        fields: &'a Vec1<AString<'a>>,
         invert: bool,
     ) -> Result<ExtractedFull<'a>> {
         // Build a lowercase ascii hashset with the fields to fetch
@@ -398,8 +398,8 @@ impl<'a> NodeMult<'a> {
             .filter_map(|inner| NodeMime(&inner).structure(is_ext).ok())
             .collect::<Vec<_>>();
 
-        NonEmptyVec::validate(&inner_bodies)?;
+        Vec1::validate(&inner_bodies)?;
-        let bodies = NonEmptyVec::unvalidated(inner_bodies);
+        let bodies = Vec1::unvalidated(inner_bodies);
 
         Ok(BodyStructure::Multi {
             bodies,

src/imap/mod.rs

@@ -15,7 +15,7 @@ mod session;
 
 use std::net::SocketAddr;
 
-use anyhow::{bail, Result};
+use anyhow::{anyhow, bail, Result, Context};
 use futures::stream::{FuturesUnordered, StreamExt};
 
 use tokio::net::TcpListener;
@@ -144,13 +144,6 @@ use tokio_util::bytes::BytesMut;
 
 const PIPELINABLE_COMMANDS: usize = 64;
 
-#[derive(Debug)]
-enum LoopMode {
-    Quit,
-    Interactive,
-    Idle(BytesMut, Arc<Notify>),
-}
-
 // @FIXME a full refactor of this part of the code will be needed sooner or later
 struct NetLoop {
     ctx: ClientContext,
@@ -163,7 +156,7 @@ impl NetLoop {
     async fn handler(ctx: ClientContext, sock: AnyStream) {
         let addr = ctx.addr.clone();
 
-        let nl = match Self::new(ctx, sock).await {
+        let mut nl = match Self::new(ctx, sock).await {
             Ok(nl) => {
                 tracing::debug!(addr=?addr, "netloop successfully initialized");
                 nl
@@ -241,27 +234,15 @@ impl NetLoop {
         tracing::info!("runner is quitting");
     }
 
-    async fn core(mut self) -> Result<()> {
+    async fn core(&mut self) -> Result<()> {
-        tracing::trace!("Starting the core loop");
+        let mut maybe_idle: Option<Arc<Notify>> = None;
-        let mut mode = LoopMode::Interactive;
         loop {
-            tracing::trace!(mode=?mode, "Core loop iter");
-            mode = match mode {
-                LoopMode::Interactive => self.interactive_mode().await?,
-                LoopMode::Idle(buff, stop) => self.idle_mode(buff, stop).await?,
-                LoopMode::Quit => break,
-            }
-        }
-        Ok(())
-    }
-
-    async fn interactive_mode(&mut self) -> Result<LoopMode> {
             tokio::select! {
                 // Managing imap_flow stuff
                 srv_evt = self.server.progress() =>  match srv_evt? {
                     ServerFlowEvent::ResponseSent { handle: _handle, response } => {
                         match response {
-                        Response::Status(Status::Bye(_)) => return Ok(LoopMode::Quit),
+                            Response::Status(Status::Bye(_)) => return Ok(()),
                             _ => tracing::trace!("sent to {} content {:?}", self.ctx.addr, response),
                         }
                     },
@@ -278,6 +259,24 @@ impl NetLoop {
                             }
                         }
                     },
+                    ServerFlowEvent::IdleCommandReceived { tag } => {
+                        match self.cmd_tx.try_send(Request::IdleStart(tag)) {
+                            Ok(_) => (),
+                            Err(mpsc::error::TrySendError::Full(_)) => {
+                                self.server.enqueue_status(Status::bye(None, "Too fast").unwrap());
+                                tracing::error!("client {:?} is sending commands too fast, closing.", self.ctx.addr);
+                            }
+                            _ => {
+                                self.server.enqueue_status(Status::bye(None, "Internal session exited").unwrap());
+                                tracing::error!("session task exited for {:?}, quitting", self.ctx.addr);
+                            }
+                        }
+                    }
+                    ServerFlowEvent::IdleDoneReceived => {
+                        tracing::trace!("client sent DONE and want to stop IDLE");
+                        maybe_idle.ok_or(anyhow!("Received IDLE done but not idling currently"))?.notify_one();
+                        maybe_idle = None;
+                    }
                     flow => {
                         self.server.enqueue_status(Status::bye(None, "Unsupported server flow event").unwrap());
                         tracing::error!("session task exited for {:?} due to unsupported flow {:?}", self.ctx.addr, flow);
@@ -296,12 +295,31 @@ impl NetLoop {
                         }
                         self.server.enqueue_status(response.completion);
                     },
-                Some(ResponseOrIdle::StartIdle(stop)) => {
+                    Some(ResponseOrIdle::IdleAccept(stop)) => {
                         tracing::trace!("Interactive, server agreed to switch in idle mode");
                         let cr = CommandContinuationRequest::basic(None, "Idling")?;
-                    self.server.enqueue_continuation(cr);
+                        self.server.idle_accept(cr).or(Err(anyhow!("refused continuation for idle accept")))?;
-                    self.cmd_tx.try_send(Request::Idle)?;
+                        self.cmd_tx.try_send(Request::IdlePoll)?;
-                    return Ok(LoopMode::Idle(BytesMut::new(), stop))
+                        if maybe_idle.is_some() {
+                            bail!("Can't start IDLE if already idling");
+                        }
+                        maybe_idle = Some(stop);
+                    },
+                    Some(ResponseOrIdle::IdleEvent(elems)) => {
+                        tracing::trace!("server imap session has some change to communicate to the client");
+                        for body_elem in elems.into_iter() {
+                            let _handle = match body_elem {
+                                Body::Data(d) => self.server.enqueue_data(d),
+                                Body::Status(s) => self.server.enqueue_status(s),
+                            };
+                        }
+                        self.cmd_tx.try_send(Request::IdlePoll)?;
+                    },
+                    Some(ResponseOrIdle::IdleReject(response)) => {
+                        tracing::trace!("inform client that session rejected idle");
+                        self.server
+                            .idle_reject(response.completion)
+                            .or(Err(anyhow!("wrong reject command")))?;
                     },
                     None => {
                         self.server.enqueue_status(Status::bye(None, "Internal session exited").unwrap());
@@ -317,9 +335,10 @@ impl NetLoop {
                     self.server.enqueue_status(Status::bye(None, "Server is being shutdown").unwrap());
                 },
             };
-        Ok(LoopMode::Interactive)
+        }
     }
 
+    /*
     async fn idle_mode(&mut self, mut buff: BytesMut, stop: Arc<Notify>) -> Result<LoopMode> {
         // Flush send
         loop {
@@ -398,5 +417,5 @@ impl NetLoop {
                 return Ok(LoopMode::Interactive)
             },
         };
-    }
+    }*/
 }

src/imap/request.rs

@@ -1,7 +1,9 @@
 use imap_codec::imap_types::command::Command;
+use imap_codec::imap_types::core::Tag;
 
 #[derive(Debug)]
 pub enum Request {
     ImapCommand(Command<'static>),
-    Idle,
+    IdleStart(Tag<'static>),
+    IdlePoll,
 }

src/imap/response.rs

@@ -118,6 +118,7 @@ impl<'a> Response<'a> {
 #[derive(Debug)]
 pub enum ResponseOrIdle {
     Response(Response<'static>),
-    StartIdle(Arc<Notify>),
+    IdleAccept(Arc<Notify>),
+    IdleReject(Response<'static>),
     IdleEvent(Vec<Body<'static>>),
 }

src/imap/search.rs

@@ -1,6 +1,6 @@
 use std::num::{NonZeroU32, NonZeroU64};
 
-use imap_codec::imap_types::core::NonEmptyVec;
+use imap_codec::imap_types::core::Vec1;
 use imap_codec::imap_types::search::{MetadataItemSearch, SearchKey};
 use imap_codec::imap_types::sequence::{SeqOrUid, Sequence, SequenceSet};
 
@@ -48,7 +48,7 @@ impl<'a> Criteria<'a> {
                         let mut new_vec = base.0.into_inner();
                         new_vec.extend_from_slice(ext.0.as_ref());
                         let seq = SequenceSet(
-                            NonEmptyVec::try_from(new_vec)
+                            Vec1::try_from(new_vec)
                                 .expect("merging non empty vec lead to non empty vec"),
                         );
                         (seq, x)

src/imap/session.rs

@@ -4,8 +4,8 @@ use crate::imap::flow;
 use crate::imap::request::Request;
 use crate::imap::response::{Response, ResponseOrIdle};
 use crate::login::ArcLoginProvider;
-use anyhow::{anyhow, bail, Result};
+use anyhow::{anyhow, bail, Result, Context};
-use imap_codec::imap_types::command::Command;
+use imap_codec::imap_types::{core::Tag, command::Command};
 
 //-----
 pub struct Instance {
@@ -27,13 +27,44 @@ impl Instance {
 
     pub async fn request(&mut self, req: Request) -> ResponseOrIdle {
         match req {
-            Request::Idle => self.idle().await,
+            Request::IdleStart(tag) => self.idle_init(tag),
+            Request::IdlePoll => self.idle_poll().await,
             Request::ImapCommand(cmd) => self.command(cmd).await,
         }
     }
 
-    pub async fn idle(&mut self) -> ResponseOrIdle {
+    pub fn idle_init(&mut self, tag: Tag<'static>) -> ResponseOrIdle {
-        match self.idle_happy().await {
+        // Build transition
+        //@FIXME the notifier should be hidden inside the state and thus not part of the transition!
+        let transition = flow::Transition::Idle(tag.clone(), tokio::sync::Notify::new());
+
+        // Try to apply the transition and get the stop notifier
+        let maybe_stop = self
+            .state
+            .apply(transition)
+            .context("IDLE transition failed")
+            .and_then(|_| self.state.notify().ok_or(anyhow!("IDLE state has no Notify object")));
+
+        // Build an appropriate response
+        match maybe_stop {
+            Ok(stop) => ResponseOrIdle::IdleAccept(stop),
+            Err(e) => {
+                tracing::error!(err=?e, "unable to init idle due to a transition error");
+                //ResponseOrIdle::IdleReject(tag)
+                let no = Response::build()
+                    .tag(tag)
+                    .message(
+                        "Internal error, processing command triggered an illegal IMAP state transition",
+                    )
+                    .no()
+                    .unwrap();
+                ResponseOrIdle::IdleReject(no)
+            }
+        }
+    }
+
+    pub async fn idle_poll(&mut self) -> ResponseOrIdle {
+        match self.idle_poll_happy().await {
             Ok(r) => r,
             Err(e) => {
                 tracing::error!(err=?e, "something bad happened in idle");
@@ -42,7 +73,7 @@ impl Instance {
         }
     }
 
-    pub async fn idle_happy(&mut self) -> Result<ResponseOrIdle> {
+    pub async fn idle_poll_happy(&mut self) -> Result<ResponseOrIdle> {
         let (mbx, tag, stop) = match &mut self.state {
             flow::State::Idle(_, ref mut mbx, _, tag, stop) => (mbx, tag.clone(), stop.clone()),
             _ => bail!("Invalid session state, can't idle"),
@@ -128,10 +159,11 @@ impl Instance {
                 .bad()
                 .unwrap());
         }
+        ResponseOrIdle::Response(resp)
 
-        match &self.state {
+        /*match &self.state {
             flow::State::Idle(_, _, _, _, n) => ResponseOrIdle::StartIdle(n.clone()),
             _ => ResponseOrIdle::Response(resp),
-        }
+        }*/
     }
 }