vomit_m2sync/

lib.rs

//! # vomit-m2sync
//!
//! `vomit-m2sync` provides full two-way synchronization between IMAP and
//! a local [m2dir][4]. At the moment, it has to be called periodically to
//! keep the two synchronized.
//!
//! [4]: https://sr.ht/~bitfehler/m2dir
//!
//! It uses the [log][1] crate for logging, so you can receive logs from it by
//! using any of the compatible logging libraries.
//!
//! [1]: https://crates.io/crates/log
//!
//! [m2sync][2] is small CLI wrapper around `vomit-m2sync`.
//!
//! [2]: https://crates.io/crates/m2sync
//!
//! As the name implies, `vomit-m2sync` is part of the [vomit project][3].
//!
//! [3]: https://sr.ht/~bitfehler/vomit

use imap::extensions::list_status::ExtendedNames;
use imap::ClientBuilder;
use imap::ImapConnection;
use imap::Session;
use log::warn;
use log::{debug, error, info, trace};
use std::collections::BTreeSet;
use std::fs;
use std::io;
use std::iter;
use std::iter::Iterator;
use std::path::Path;
use std::path::PathBuf;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::sync::mpsc;
use std::sync::Arc;
use std::thread;
use std::time::Instant;
use wildmatch::WildMatch;

use crate::state::SyncAction;
use crate::state::SyncJob;

mod flags;
mod mailboxes;
mod seqset;
mod state;
mod sync;

#[derive(thiserror::Error, Debug)]
pub enum Error {
    #[error("IMAP error: {0}")]
    IMAP(#[from] imap::Error),
    #[error("no UIDVALIDITY provided")]
    MissingUIDValidity(),
    #[error("no HIGHESTMODSEQ provided")]
    MissingHMS(),
    #[error("APPEND did not return any UIDs")]
    InvalidAppendResponse(),
    #[error("server did not send UID, giving up")]
    MissingUID(),
    #[error("downloaded message has no body")]
    MissingBody(),
    #[error("invalid configuration: {0}")]
    Config(&'static str),
    #[error("refused to perform dangerous action - set 'force' option to override")]
    DangerousAction(),
    #[error("server does not support {0}")]
    MissingCapability(&'static str),
    #[error("found multiple hierarchy delimiters")]
    MultipleHierarchyDelimiters(),
    #[error("no remote mailboxes found")]
    NoRemoteMailboxes(),
    #[error("UID validity change with pending local changes makes resynchronization impossible")]
    UIDValidityChangeWithLocalChanges(),
    #[error("not all mailboxes were synced successfully")]
    SyncIncomplete(),
    #[error("failed to load state: {0}")]
    State(#[from] state::Error),
    #[error("IPC error: {0}")]
    IPC(#[from] spmc::SendError<state::SyncJob>),
    #[error("IO error: {0}")]
    IO(#[from] io::Error),
    #[error("interrupted")]
    Interrupted(),
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum SyncDirection {
    Pull,
    Push,
    TwoWay,
}

/// A set of options to control the synchronization process
#[derive(Clone, Debug)]
pub struct SyncOptions {
    /// A unique ID that must remain constant for all invocations that sync the same targets,
    /// but must be unique for each set of targets. This can be considered an "account ID". It
    /// can be used to allow syncing the same local directory to different remote accounts.
    pub uid: String,
    /// The local m2dir to sync to
    pub local: String,
    /// The IMAP server (and optional port) to sync from
    pub remote: String,
    /// The user for IMAP authentication
    pub user: String,
    /// The password for IMAP authentication
    pub password: String,
    /// The number of threads to use
    ///
    /// Each thread will use it's own IMAP session, so this must not be greater
    /// than the number of concurrent user sessions allowed by the IMAP server.
    pub threads: u8,
    /// Disable TLS certificate checks (e.g. for self-signed certs) (insecure)
    pub unsafe_tls: bool,
    /// Completely disable TLS (very insecure)
    pub disable_tls: bool,
    /// Only log with info level what actions (sync, create, delete) would be taken on
    /// which mailboxes (taking include/exclude into account), then exit.
    pub list_mailbox_actions: bool,
    /// A list of wildcard patterns to include only folders that match any of them.
    ///
    /// The wildcard pattern only supports `?` and `*` and must match the full mailbox name.
    pub include: Vec<String>,
    /// A list of wildcard patterns to exclude all folders that match any of them.
    ///
    /// The wildcard pattern only supports `?` and `*` and must match the full mailbox name.
    /// If used together with `include` and both match, `exclude` takes precedence.
    pub exclude: Vec<String>,
    /// Confirm execution of potentially dangerous actions (e.g. deleting mailboxes)
    pub force: bool,
}

macro_rules! measure {
    ( $m:expr, $x:expr ) => {{
        let start = Instant::now();
        let result = $x;
        let duration = start.elapsed();
        debug!("{} in {}ms", $m, duration.as_millis());
        result
    }};
}

const DATA_ITEMS_HMS: &str = "(HIGHESTMODSEQ)";

fn get_hierarchy_delimiter(names: &ExtendedNames) -> Result<String, Error> {
    let delims: BTreeSet<&str> = names
        .iter()
        .filter_map(|(name, _)| name.delimiter())
        .collect();
    if delims.len() != 1 {
        return Err(Error::MultipleHierarchyDelimiters());
    }
    Ok(String::from(delims.into_iter().next().unwrap()))
}

fn new_session(opts: &SyncOptions) -> Result<Session<Box<dyn ImapConnection>>, Error> {
    let (host, port) = match opts.remote.rsplit_once(':') {
        Some((host, port)) => (host, port),
        None => (opts.remote.as_str(), "993"),
    };

    let port = match port.parse() {
        Ok(p) => p,
        Err(e) => {
            error!("failed to parse remote port: {}", e);
            return Err(Error::Config("invalid port"));
        }
    };

    debug!("Connecting to {}:{}", host, port);

    let client = if opts.disable_tls {
        ClientBuilder::new(host, port)
            .mode(imap::ConnectionMode::Plaintext)
            .connect()?
    } else if opts.unsafe_tls {
        ClientBuilder::new(host, port)
            .danger_skip_tls_verify(true)
            .connect()?
    } else {
        ClientBuilder::new(host, port).connect()?
    };

    debug!("Logging in as {}", &opts.user);
    let mut session = client.login(&opts.user, &opts.password).map_err(|e| e.0)?;

    session.run_command_and_read_response("ENABLE QRESYNC")?;

    Ok(session)
}

fn check_server_capabilities_(opts: &SyncOptions) -> Result<(), Error> {
    let mut session = new_session(opts)?;
    let caps = session.capabilities()?;
    for cap in caps.iter() {
        trace!("Server capability: {:?}", cap);
    }
    session.logout()?;
    if !caps.has_str("QRESYNC") {
        return Err(Error::MissingCapability("QRESYNC (RFC 7162)"));
    }
    if !caps.has_str("UIDPLUS") {
        return Err(Error::MissingCapability("UIDPLUS (RFC 4315)"));
    }
    if !caps.has_str("LIST-STATUS") {
        return Err(Error::MissingCapability("LIST-STATUS (RFC 5819)"));
    }
    Ok(())
}

/// List remote and local mailboxes with log level "info".
///
/// This is intended for human consumption. It can e.g. be presented to the
/// user as a verification that all mailboxes are found as expected.
pub fn list_mailboxes(opts: &SyncOptions) -> Result<(), Error> {
    let mut session = new_session(opts)?;
    let names = session.list_status(None, Some("*"), DATA_ITEMS_HMS)?;
    if names.is_empty() {
        return Err(Error::NoRemoteMailboxes());
    }

    let delimiter = get_hierarchy_delimiter(&names)?;
    let remote_mailboxes = state::load_remote_mailboxes(&names);
    let local_mailboxes = state::load_local_mailboxes(&opts.uid, &opts.local, &delimiter)?;

    info!("Hierarchy delimiter: {}", delimiter);
    info!("Remote mailboxes:");
    for m in remote_mailboxes.keys() {
        info!("  {}", m);
    }
    info!("Local mailboxes:");
    for m in local_mailboxes.keys() {
        info!("  {}", m);
    }
    Ok(())
}

/// Check if the configured server supports the required IMAP capabilities
///
/// Currently, at least the QRESYNC capability is required ([RFC 7162][1]).
///
/// [1]: https://www.rfc-editor.org/rfc/rfc7162.html
pub fn check_server_capabilities(opts: &SyncOptions) -> Result<(), Error> {
    measure!(
        format!("Checked capabilities for {}", opts.remote),
        check_server_capabilities_(opts)
    )
}

fn sync_mailbox<T: io::Read + io::Write>(
    state_id: &str,
    m2store: impl AsRef<Path>,
    sync_job: &state::SyncJob,
    direction: &SyncDirection,
    session: &mut Session<T>,
    abort: &Arc<AtomicBool>,
) -> Result<(), Error> {
    let mailbox = &sync_job.name;
    let dirname = state::mailbox_to_dirname(mailbox, &sync_job.delimiter);
    let mbpath: PathBuf = [m2store.as_ref(), dirname.as_ref()].iter().collect();

    match sync_job.action {
        state::SyncAction::Sync => (),
        state::SyncAction::CreateLocal => {
            trace!("Creating local mailbox {}", mailbox);
            // Local dir creation will happen automatically
        }
        state::SyncAction::CreateRemote => {
            trace!("Creating remote mailbox {}", mailbox);
            session.create(mailbox)?
        }
        state::SyncAction::DeleteLocal => {
            trace!("Deleting local mailbox {}", mailbox);
            return fs::remove_dir_all(&mbpath).map_err(Error::IO);
        }
        state::SyncAction::DeleteRemote => {
            trace!("Deleting remote mailbox {}", mailbox);
            return session.delete(mailbox).map_err(Error::IMAP);
        }
    };

    let state = measure!(
        format!("Loaded state for {}", mailbox),
        state::SyncState::load(&mbpath, state_id)?
    );

    let remote = session.select(mailbox)?;

    match direction {
        SyncDirection::Pull => sync::pull(session, sync_job, remote, state, abort),
        SyncDirection::Push => sync::push(session, sync_job, remote, state, abort),
        SyncDirection::TwoWay => sync::sync(session, sync_job, remote, state, abort),
    }
}

fn worker_thread(
    i: u8,
    opts: &SyncOptions,
    direction: SyncDirection,
    rx: spmc::Receiver<state::SyncJob>,
    tx: mpsc::Sender<state::SyncJob>,
    session: Option<Session<Box<dyn ImapConnection>>>,
    abort: Arc<AtomicBool>,
) -> Result<(), Error> {
    let mut imap_session = match session {
        Some(s) => s,
        None => new_session(opts)?,
    };

    while let Ok(mut job) = rx.recv() {
        if abort.load(Ordering::Relaxed) {
            warn!("Worker thread {} exiting - interrupted", i);
            break;
        };
        trace!("Syncing {} in worker thread {}", job.name, i);
        match measure!(
            format!("Synced {}", job.name),
            sync_mailbox(
                &opts.uid,
                &opts.local,
                &job,
                &direction,
                &mut imap_session,
                &abort
            )
        ) {
            Ok(_) => job.success = true,
            Err(e) => error!("Error syncing {}: {}", job.name, e),
        }
        tx.send(job)?;
    }
    drop(tx);
    imap_session.logout()?;
    Ok(())
}

fn sync_(
    opts: &SyncOptions,
    direction: SyncDirection,
    abort: Arc<AtomicBool>,
) -> Result<(), Error> {
    info!("Syncing from {} to {}", opts.remote, opts.local);

    fs::create_dir_all(&opts.local)?;

    let root_state = state::RootState::load(&opts.local, &opts.uid)?;

    let mut imap_session = new_session(opts)?;

    let names = imap_session.list_status(None, Some("*"), DATA_ITEMS_HMS)?;

    if names.is_empty() {
        return Err(Error::NoRemoteMailboxes());
    }

    let includes: Vec<WildMatch> = opts.include.iter().map(|p| WildMatch::new(p)).collect();
    let excludes: Vec<WildMatch> = opts.exclude.iter().map(|p| WildMatch::new(p)).collect();

    let delimiter = get_hierarchy_delimiter(&names)?;
    let remote_mailboxes_with_state = state::load_remote_mailboxes(&names);
    let remote_mailboxes: BTreeSet<&String> = remote_mailboxes_with_state.keys().collect();
    let local_mailboxes_with_state = measure!(
        format!("Loaded local mailboxes"),
        state::load_local_mailboxes(&opts.uid, &opts.local, &delimiter)?
    );
    let local_mailboxes: BTreeSet<&String> = local_mailboxes_with_state.keys().collect();

    trace!("local: {:?}", local_mailboxes);
    trace!("remote: {:?}", remote_mailboxes);

    let only_local: BTreeSet<&String> = local_mailboxes
        .difference(&remote_mailboxes)
        .cloned()
        .collect();
    trace!("Only local: {:?}", only_local);
    let only_remote: BTreeSet<&String> = remote_mailboxes
        .difference(&local_mailboxes)
        .cloned()
        .collect();
    trace!("Only remote: {:?}", only_remote);
    let sync_jobs: Vec<state::SyncJob> = local_mailboxes
        .union(&remote_mailboxes)
        .cloned()
        .filter_map(|name| {
            // Apply user-provided include/exclude filter
            if !includes.is_empty() && !includes.iter().any(|e| e.matches(name)) {
                trace!("Skipping {} due to include filter", name);
                return None;
            }
            if excludes.iter().any(|e| e.matches(name)) {
                trace!("Skipping {} due to exclude filter", name);
                return None;
            }
            trace!("Processing mailbox {}", name);
            let delimiter = delimiter.clone();

            // We'll have to go through the tedious process of determining what to do
            // with mailboxes that are missing on one side.
            let name = name.clone();
            if only_local.contains(&name) {
                match direction {
                    SyncDirection::Pull => {
                        // Delete locally, also from root state
                        SyncJob::new(name, delimiter, SyncAction::DeleteLocal)
                    }
                    SyncDirection::Push => {
                        // Create remotely, add to root state
                        SyncJob::new(name, delimiter, SyncAction::CreateRemote)
                    }
                    SyncDirection::TwoWay => {
                        let dirname = state::mailbox_to_dirname(&name, &delimiter);
                        if root_state
                            .contains_subdir(&dirname)
                            .expect("failed to inspect current state")
                        {
                            // Exists locally only because removed on server side
                            // Delete locally, also from root state
                            SyncJob::new(name, delimiter, SyncAction::DeleteLocal)
                        } else {
                            // Exists locally only because created here
                            // Create remotely, add to root state
                            SyncJob::new(name, delimiter, SyncAction::CreateRemote)
                        }
                    }
                }
            } else if only_remote.contains(&name) {
                match direction {
                    SyncDirection::Pull => {
                        // Create locally (already happening), add to root state (after sync?)
                        SyncJob::new(name, delimiter, SyncAction::CreateLocal)
                    }
                    SyncDirection::Push => {
                        // Delete remotely, also from root state
                        SyncJob::new(name, delimiter, SyncAction::DeleteRemote)
                    }
                    SyncDirection::TwoWay => {
                        let dirname = state::mailbox_to_dirname(&name, &delimiter);
                        if root_state
                            .contains_subdir(&dirname)
                            .expect("failed to inspect current state")
                        {
                            // Exists remotely only because removed locally
                            // Delete remotely, also from root state
                            SyncJob::new(name, delimiter, SyncAction::DeleteRemote)
                        } else {
                            // Exists remotely only because created there
                            // Create locally (already happening), add to root state (after sync?)
                            SyncJob::new(name, delimiter, SyncAction::CreateLocal)
                        }
                    }
                }
            } else {
                // Exists on both sides, just sync
                let state = local_mailboxes_with_state.get(&name).unwrap();
                let modseq = remote_mailboxes_with_state.get(&name).unwrap();
                if !state.has_local_changes() && state.last_seen_highest_mod_seq() == *modseq {
                    trace!("Skipping {} because HIGHESTMODSEQ is in sync", name);
                    return None;
                }
                SyncJob::new(name, delimiter, state::SyncAction::Sync)
            }
        })
        .collect();

    if opts.list_mailbox_actions {
        info!("The following actions would be performed:");
        for job in sync_jobs {
            info!("  {}: {:?}", job.name, job.action);
        }
        return Ok(());
    }

    let bail = sync_jobs.iter().any(|job| match job.action {
        state::SyncAction::DeleteLocal => {
            warn!("About to delete local mailbox {}", job.name);
            true
        }
        state::SyncAction::DeleteRemote => {
            warn!("About to delete remote mailbox {}", job.name);
            true
        }
        _ => false,
    });
    if bail && !opts.force {
        return Err(Error::DangerousAction());
    }

    let thread_count = if sync_jobs.len() < opts.threads.into() {
        // Conversion must be safe here
        sync_jobs.len().try_into().unwrap()
    } else {
        opts.threads
    };

    let (mut tx_work, rx_work) = spmc::channel::<state::SyncJob>();
    let (tx_result, rx_result) = mpsc::channel::<state::SyncJob>();

    info!(
        "Using {} threads to sync {} mailboxes",
        thread_count,
        sync_jobs.len()
    );

    let mut sess = iter::once(imap_session);

    let threads: Vec<_> = (0..thread_count)
        .map(|i| {
            let rx = rx_work.clone();
            let tx = tx_result.clone();
            let opts = (*opts).clone();
            let dir = direction.clone();
            let session = sess.next();
            let abort = Arc::clone(&abort);

            thread::spawn(move || {
                if let Err(e) = worker_thread(i, &opts, dir, rx, tx, session, abort) {
                    error!("Error in worker thread {}: {}", i, e);
                };
            })
        })
        .collect();

    let expected_results = sync_jobs.len();
    let mut received_results = 0usize;

    for job in sync_jobs {
        tx_work.send(job)?;
    }

    drop(tx_work);
    drop(tx_result);

    let mut success = true;

    while let Ok(job) = rx_result.recv() {
        // update root state if needed
        trace!("root state update: {:?} {}", job.action, job.name);
        success &= job.success;
        received_results += 1;

        root_state.sync_done(job)?;
    }

    for t in threads {
        t.join().unwrap();
    }

    if success && (expected_results == received_results) {
        info!("Sync successful");
        Ok(())
    } else {
        error!("Sync failed!");
        Err(Error::SyncIncomplete())
    }
}

/// Apply all changes from IMAP to the local m2dir
///
/// This overwrites any changes that may have happened on the local side.
///
/// <div class="warning">
///
/// Synchronizing local and remote state as quickly as possible requires an
/// additional index, which itself has to be kept in sync with the actual
/// storage. Doing this all atomically is next to impossible. Any caller of this
/// function **should catch termination signals!** Setting the inner value of
/// the `abort` argument to `true` will make the function return an error as
/// early as possible, while still making sure the persisted state is
/// consistent.
///
/// </div>
///
/// # Arguments
///
/// * `opts`: options controlling the synchronization process
/// * `abort`: if the inner value becomes true at any time, the sync will abort
///   as soon as possible without corrupting the index
pub fn pull(opts: &SyncOptions, abort: Arc<AtomicBool>) -> Result<(), Error> {
    measure!(
        format!("Pulled from {}", opts.remote),
        sync_(opts, SyncDirection::Pull, abort)
    )
}

/// Apply all changes in local m2dir to IMAP
///
/// This overwrites any changes that may have happened on the IMAP side.
///
/// <div class="warning">
///
/// Synchronizing local and remote state as quickly as possible requires an
/// additional index, which itself has to be kept in sync with the actual
/// storage. Doing this all atomically is next to impossible. Any caller of this
/// function **should catch termination signals!** Setting the inner value of
/// the `abort` argument to `true` will make the function return an error as
/// early as possible, while still making sure the persisted state is
/// consistent.
///
/// </div>
///
/// # Arguments
///
/// * `opts`: options controlling the synchronization process
/// * `abort`: if the inner value becomes true at any time, the sync will abort
///   as soon as possible without corrupting the index
pub fn push(opts: &SyncOptions, abort: Arc<AtomicBool>) -> Result<(), Error> {
    measure!(
        format!("Pushed to {}", opts.remote),
        sync_(opts, SyncDirection::Push, abort)
    )
}

/// Synchronize local and remote changes between IMAP and local m2dir
///
/// This includes fetching new mail, deleting expunged mails, updating tags,
/// etc.
///
/// <div class="warning">
///
/// Synchronizing local and remote state as quickly as possible requires an
/// additional index, which itself has to be kept in sync with the actual
/// storage. Doing this all atomically is next to impossible. Any caller of this
/// function **should catch termination signals!** Setting the inner value of
/// the `abort` argument to `true` will make the function return an error as
/// early as possible, while still making sure the persisted state is
/// consistent.
///
/// </div>
///
/// # Arguments
///
/// * `opts`: options controlling the synchronization process
/// * `abort`: if the inner value becomes true at any time, the sync will abort
///   as soon as possible without corrupting the index
pub fn sync(opts: &SyncOptions, abort: Arc<AtomicBool>) -> Result<(), Error> {
    measure!(
        format!("Synced with {}", opts.remote),
        sync_(opts, SyncDirection::TwoWay, abort)
    )
}