korax/korax-public
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

sync: mailcore 0.2.1-beta

Browse Source
This commit is contained in:
Eugen Kaparulin
2026-06-22 21:54:48 +03:00
parent c76d1197d0
commit c48a14d4e8
10 changed files with 827 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
mailcore/src/imap/client.rs
View File

@@ -1,4 +1,4 @@
use crate::types::{FolderInfo, FolderType, MailServerConfig};
use crate::types::{AuthCredential, FolderInfo, FolderType, MailServerConfig};
use async_imap::extensions::idle::IdleResponse;
use async_imap::Client;
use futures::stream::TryStreamExt;
@@ -10,6 +10,28 @@ use tokio::net::TcpStream;
use tokio_rustls::client::TlsStream;
use tokio_rustls::TlsConnector;
/// XOAUTH2 authenticator for async-imap.
/// The response is the pre-built `user=<email>\x01auth=Bearer <token>\x01\x01` string.
struct XOAuth2 {
response: Vec<u8>,
}
impl async_imap::Authenticator for XOAuth2 {
type Response = Vec<u8>;
fn process(&mut self, challenge: &[u8]) -> Self::Response {
// On first call challenge is empty — send the XOAUTH2 payload.
// On subsequent calls Gmail has sent a base64-encoded error; send an
// empty response to abort so the server returns NO instead of hanging.
if challenge.is_empty() {
self.response.clone()
} else {
tracing::warn!("XOAUTH2: received non-empty challenge (auth error): {:?}",
std::str::from_utf8(challenge).unwrap_or("<binary>"));
Vec::new()
}
}
}
/// Returned by `ImapClient::sync_mailbox`. Contains data needed for post-sync operations
/// that must run outside the shared session lock.
pub struct SyncResult {
@@ -25,18 +47,20 @@ pub struct ImapClient {
session: Option<async_imap::Session<TlsStream<TcpStream>>>,
config: MailServerConfig,
email: String,
auth: AuthCredential,
}
impl ImapClient {
pub fn new(config: MailServerConfig, email: String) -> Self {
pub fn new(config: MailServerConfig, email: String, auth: AuthCredential) -> Self {
Self {
session: None,
config,
email,
auth,
}
}
pub async fn connect(&mut self, password: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
pub async fn connect(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
tracing::debug!(
"ImapClient: Starting connection to {}:{}",
self.config.imap_host,
@@ -46,16 +70,20 @@ impl ImapClient {
let mut root_cert_store = rustls::RootCertStore::empty();
root_cert_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
tracing::debug!("ImapClient: Loading native certs...");
// Load native certs
match rustls_native_certs::load_native_certs() {
Ok(certs) => {
for cert in certs {
let _ = root_cert_store.add(cert);
// Skip rustls_native_certs on Android — the cert store path scan blocks
// or returns nothing useful; webpki_roots covers all major CAs.
#[cfg(not(target_os = "android"))]
{
tracing::debug!("ImapClient: Loading native certs...");
match rustls_native_certs::load_native_certs() {
Ok(certs) => {
for cert in certs {
let _ = root_cert_store.add(cert);
}
}
Err(e) => {
tracing::warn!("ImapClient: Failed to load native certs: {}", e);
}
}
Err(e) => {
tracing::warn!("ImapClient: Failed to load native certs: {}", e);
}
}
@@ -86,33 +114,100 @@ impl ImapClient {
.to_owned();
tracing::debug!("ImapClient: Performing TLS handshake for {}...", dns_name);
let tls_stream = connector.connect(server_name, stream).await?;
let tls_stream = tokio::time::timeout(
Duration::from_secs(10),
connector.connect(server_name, stream),
)
.await
.map_err(|_| format!("TLS handshake with {} timed out after 10 seconds", dns_name))?
.map_err(|e| format!("TLS handshake with {} failed: {}", dns_name, e))?;
// For OAuth2: fetch the token BEFORE the IMAP handshake so auth errors
// are clearly separated from connection errors.
let pre_fetched_token = match &self.auth {
AuthCredential::OAuth2 => {
tracing::debug!("ImapClient: Pre-fetching OAuth2 token for {}...", self.email);
let token = tokio::time::timeout(
Duration::from_secs(10),
crate::oauth::get_valid_token(&self.email),
)
.await
.map_err(|_| "OAuth2 token fetch timed out after 10s")?
.map_err(|e| e.to_string())?;
tracing::debug!("ImapClient: OAuth2 token obtained for {}", self.email);
Some(token)
}
_ => None,
};
// Create IMAP client
tracing::debug!("ImapClient: Creating async_imap client...");
let client = Client::new(tls_stream);
let mut client = Client::new(tls_stream);
// Login
// async_imap does NOT read the server greeting automatically. The greeting must be
// consumed before sending any command; otherwise authenticate()'s do_auth_handshake()
// reads it instead of the "+" continuation and hangs indefinitely.
tracing::debug!("ImapClient: Reading server greeting for {}...", self.config.imap_host);
match tokio::time::timeout(Duration::from_secs(10), client.read_response()).await {
Ok(Some(Ok(greeting))) => {
tracing::debug!("ImapClient: Server greeting: {:?}", greeting.parsed());
}
Ok(Some(Err(e))) => {
return Err(format!("Failed to read server greeting: {}", e).into());
}
Ok(None) => {
return Err("Server closed connection before greeting".into());
}
Err(_) => {
return Err("Timed out waiting for server greeting".into());
}
}
// Authenticate — explicit timeout so the caller gets a specific error message
// rather than waiting for the session_manager's global 20s wall-clock.
tracing::debug!("ImapClient: Attempting login for {}...", self.email);
let session = client
.login(&self.email, password)
let session = match &self.auth.clone() {
AuthCredential::Password(pw) => tokio::time::timeout(
Duration::from_secs(15),
client.login(&self.email, pw),
)
.await
.map_err(|_| "IMAP login timed out after 15s".to_string())?
.map_err(|(e, _client)| {
tracing::error!("ImapClient: Login failed: {}", e);
e.to_string()
})?;
})?,
AuthCredential::OAuth2 => {
let token = pre_fetched_token.unwrap();
let raw = format!(
"user={}\x01auth=Bearer {}\x01\x01",
self.email, token
);
tracing::debug!("ImapClient: Sending XOAUTH2 AUTHENTICATE for {}...", self.email);
tokio::time::timeout(
Duration::from_secs(15),
client.authenticate("XOAUTH2", XOAuth2 { response: raw.into_bytes() }),
)
.await
.map_err(|_| "XOAUTH2 authenticate timed out after 15s".to_string())?
.map_err(|(e, _client)| {
tracing::error!("ImapClient: XOAUTH2 auth failed: {}", e);
e.to_string()
})?
}
};
tracing::debug!("ImapClient: Successfully logged in as {}", self.email);
self.session = Some(session);
Ok(())
}
pub async fn reconnect(&mut self, password: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
pub async fn reconnect(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
// Force close existing session if any
if let Some(mut session) = self.session.take() {
let _ = session.logout().await;
}
self.connect(password).await
self.connect().await
}
pub async fn logout(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
@@ -139,13 +234,10 @@ impl ImapClient {
}
/// Use this before any operation to ensure we have a valid session.
pub async fn ensure_connected(
&mut self,
password: &str,
) -> Result<(), Box<dyn Error + Send + Sync>> {
pub async fn ensure_connected(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
if !self.keep_alive().await {
tracing::debug!("Reconnecting session for {}", self.email);
self.reconnect(password).await?;
self.reconnect().await?;
}
Ok(())
}