filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Imported Stalwart CLI crate.

Browse Source
This commit is contained in:
mdecimus
2023-07-03 18:48:43 +02:00
parent 0a8fdd9008
commit b30e022480
12 changed files with 2125 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
crates/cli/Cargo.toml Normal file
View File

@@ -0,0 +1,29 @@
[package]
name = "stalwart-cli"
description = "Stalwart Mail Server CLI"
authors = ["Stalwart Labs Ltd. <hello@stalw.art>"]
license = "AGPL-3.0-only"
repository = "https://github.com/stalwartlabs/cli"
homepage = "https://github.com/stalwartlabs/cli"
version = "0.3.0"
edition = "2021"
readme = "README.md"
resolver = "2"
[dependencies]
jmap-client = { git = "https://github.com/stalwartlabs/jmap-client", features = ["async"] }
mail-parser = { git = "https://github.com/stalwartlabs/mail-parser", features = ["full_encoding", "serde_support", "ludicrous_mode"] }
reqwest = { version = "0.11", default-features = false, features = ["rustls-tls-webpki-roots"]}
tokio = { version = "1.23", features = ["full"] }
num_cpus = "1.13.1"
clap = { version = "4.1.6", features = ["derive"] }
prettytable-rs = "0.10.0"
rpassword = "7.0"
indicatif = "0.17.0"
console = { version = "0.15", default-features = false, features = ["ansi-parsing"] }
serde = { version = "1.0", features = ["derive"]}
serde_json = "1.0"
csv = "1.1"
form_urlencoded = "1.1.0"
human-size = "0.4.2"
futures = "0.3.28"

135
crates/cli/src/main.rs Normal file
View File

@@ -0,0 +1,135 @@
/*
* Copyright (c) 2020-2023, Stalwart Labs Ltd.
*
* This file is part of the Stalwart Command Line Interface.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of
* the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
* in the LICENSE file at the top-level directory of this distribution.
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* You can be released from the requirements of the AGPLv3 license by
* purchasing a commercial license. Please contact licensing@stalw.art
* for more details.
*/
use std::{
collections::HashMap,
io::{BufRead, Write},
};
use clap::Parser;
use console::style;
use jmap_client::client::{Client, Credentials};
use modules::{
cli::{Cli, Commands},
get,
import::cmd_import,
post,
queue::cmd_queue,
report::cmd_report,
};
use crate::modules::OAuthResponse;
pub mod modules;
#[tokio::main]
async fn main() -> std::io::Result<()> {
let args = Cli::parse();
let is_jmap = args.command.is_jmap();
let credentials = if let Some(credentials) = args.credentials {
parse_credentials(&credentials)
} else if is_jmap {
let credentials = rpassword::prompt_password(
"\nEnter JMAP admin credentials or press [ENTER] to use OAuth: ",
)
.unwrap();
if !credentials.is_empty() {
parse_credentials(&credentials)
} else {
oauth(&args.url)
}
} else {
parse_credentials(&rpassword::prompt_password("\nEnter SMTP admin credentials: ").unwrap())
};
if is_jmap {
let client = Client::new()
.credentials(credentials)
.connect(&args.url)
.await
.unwrap_or_else(|err| {
eprintln!("Failed to connect to JMAP server {}: {}.", args.url, err);
std::process::exit(1);
});
match args.command {
Commands::Import(command) => cmd_import(client, command).await,
Commands::Queue(_) | Commands::Report(_) => unreachable!(),
}
} else {
match args.command {
Commands::Queue(command) => cmd_queue(&args.url, credentials, command),
Commands::Report(command) => cmd_report(&args.url, credentials, command),
_ => unreachable!(),
}
}
Ok(())
}
fn parse_credentials(credentials: &str) -> Credentials {
if let Some((account, secret)) = credentials.split_once(':') {
Credentials::basic(account, secret)
} else {
Credentials::basic("admin", credentials)
}
}
fn oauth(url: &str) -> Credentials {
let metadata = get(&format!("{}/.well-known/oauth-authorization-server", url));
let token_endpoint = metadata.property("token_endpoint");
let mut params = HashMap::from_iter([("client_id".to_string(), "Stalwart_CLI".to_string())]);
let response = post(metadata.property("device_authorization_endpoint"), &params);
params.insert(
"grant_type".to_string(),
"urn:ietf:params:oauth:grant-type:device_code".to_string(),
);
params.insert(
"device_code".to_string(),
response.property("device_code").to_string(),
);
print!(
"\nAuthenticate this request using code {} at {}. Please ENTER when done.",
style(response.property("user_code")).bold(),
style(response.property("verification_uri")).bold().dim()
);
std::io::stdout().flush().unwrap();
std::io::stdin().lock().lines().next();
let mut response = post(token_endpoint, &params);
if let Some(serde_json::Value::String(access_token)) = response.remove("access_token") {
Credentials::Bearer(access_token)
} else {
eprintln!(
"OAuth failed with code {}.",
response
.get("error")
.and_then(|s| s.as_str())
.unwrap_or("<unknown>")
);
std::process::exit(1);
}
}

213
crates/cli/src/modules/cli.rs Normal file
View File

@@ -0,0 +1,213 @@
/*
* Copyright (c) 2020-2023, Stalwart Labs Ltd.
*
* This file is part of the Stalwart Command Line Interface.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of
* the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
* in the LICENSE file at the top-level directory of this distribution.
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* You can be released from the requirements of the AGPLv3 license by
* purchasing a commercial license. Please contact licensing@stalw.art
* for more details.
*/
use clap::{Parser, Subcommand, ValueEnum};
use mail_parser::DateTime;
use serde::Deserialize;
#[derive(Parser)]
#[clap(version, about, long_about = None)]
#[clap(name = "stalwart-cli")]
pub struct Cli {
#[clap(subcommand)]
pub command: Commands,
/// JMAP or SMTP server base URL
#[clap(short, long)]
pub url: String,
/// Authentication credentials
#[clap(short, long)]
pub credentials: Option<String>,
}
#[derive(Subcommand)]
pub enum Commands {
/// Import accounts and domains
#[clap(subcommand)]
Import(ImportCommands),
/// Manage SMTP message queue
#[clap(subcommand)]
Queue(QueueCommands),
/// Manage SMTP DMARC/TLS report queue
#[clap(subcommand)]
Report(ReportCommands),
}
#[derive(Subcommand)]
pub enum ImportCommands {
/// Import messages and folders
Messages {
#[clap(value_enum)]
#[clap(short, long)]
format: MailboxFormat,
/// Number of threads to use for message import, defaults to the number of CPUs.
#[clap(short, long)]
num_threads: Option<usize>,
/// Account id to import messages into
account_id: String,
/// Path to the mailbox to import, or '-' for stdin (stdin only supported for mbox)
path: String,
},
}
#[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum)]
pub enum MailboxFormat {
/// Mbox format
Mbox,
/// Maildir and Maildir++ formats
Maildir,
/// Maildir with hierarchical folders (i.e. Dovecot)
MaildirNested,
}
#[derive(Subcommand)]
pub enum QueueCommands {
/// Shows messages queued for delivery
List {
/// Filter by sender address
#[clap(short, long)]
sender: Option<String>,
/// Filter by recipient
#[clap(short, long)]
rcpt: Option<String>,
/// Filter messages due for delivery before a certain datetime
#[clap(short, long)]
#[arg(value_parser = parse_datetime)]
before: Option<DateTime>,
/// Filter messages due for delivery after a certain datetime
#[clap(short, long)]
#[arg(value_parser = parse_datetime)]
after: Option<DateTime>,
/// Number of items to show per page
#[clap(short, long)]
page_size: Option<usize>,
},
/// Displays details about a queued message
Status {
#[clap(required = true)]
ids: Vec<String>,
},
/// Reschedule delivery
Retry {
/// Apply to messages matching a sender address
#[clap(short, long)]
sender: Option<String>,
/// Apply to a specific domain
#[clap(short, long)]
domain: Option<String>,
/// Apply to messages due before a certain datetime
#[clap(short, long)]
#[arg(value_parser = parse_datetime)]
before: Option<DateTime>,
/// Apply to messages due after a certain datetime
#[clap(short, long)]
#[arg(value_parser = parse_datetime)]
after: Option<DateTime>,
/// Schedule delivery at a specific time
#[clap(short, long)]
#[arg(value_parser = parse_datetime)]
time: Option<DateTime>,
// Reschedule one or multiple message ids
ids: Vec<String>,
},
/// Cancel delivery
Cancel {
/// Apply to messages matching a sender address
#[clap(short, long)]
sender: Option<String>,
/// Apply to specific recipients or domains
#[clap(short, long)]
rcpt: Option<String>,
/// Apply to messages due before a certain datetime
#[clap(short, long)]
#[arg(value_parser = parse_datetime)]
before: Option<DateTime>,
/// Apply to messages due after a certain datetime
#[clap(short, long)]
#[arg(value_parser = parse_datetime)]
after: Option<DateTime>,
// Cancel one or multiple message ids
ids: Vec<String>,
},
}
#[derive(Subcommand)]
pub enum ReportCommands {
/// Shows reports queued for delivery
List {
/// Filter by report domain
#[clap(short, long)]
domain: Option<String>,
/// Filter by report type
#[clap(short, long)]
#[clap(value_enum)]
format: Option<ReportFormat>,
/// Number of items to show per page
#[clap(short, long)]
page_size: Option<usize>,
},
/// Displays details about a queued report
Status {
#[clap(required = true)]
ids: Vec<String>,
},
/// Cancel report delivery
Cancel {
#[clap(required = true)]
ids: Vec<String>,
},
}
impl Commands {
pub fn is_jmap(&self) -> bool {
!matches!(self, Commands::Queue(_) | Commands::Report(_))
}
}
#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum, Deserialize)]
pub enum ReportFormat {
/// DMARC report
#[serde(rename = "dmarc")]
Dmarc,
/// TLS report
#[serde(rename = "tls")]
Tls,
}
fn parse_datetime(arg: &str) -> Result<DateTime, &'static str> {
if arg.contains('T') {
DateTime::parse_rfc3339(arg).ok_or("Failed to parse RFC3339 datetime")
} else {
DateTime::parse_rfc3339(&format!("{arg}T00:00:00Z"))
.ok_or("Failed to parse RFC3339 datetime")
}
}

444
crates/cli/src/modules/import.rs Normal file
View File

@@ -0,0 +1,444 @@
/*
* Copyright (c) 2020-2023, Stalwart Labs Ltd.
*
* This file is part of the Stalwart Command Line Interface.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of
* the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
* in the LICENSE file at the top-level directory of this distribution.
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* You can be released from the requirements of the AGPLv3 license by
* purchasing a commercial license. Please contact licensing@stalw.art
* for more details.
*/
use std::{
collections::HashMap,
io::{self, Cursor},
sync::{
atomic::{AtomicUsize, Ordering},
Arc, Mutex,
},
};
use console::style;
use futures::{stream::FuturesUnordered, StreamExt};
use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
use jmap_client::{
client::Client,
core::set::SetObject,
mailbox::{self, Role},
};
use mail_parser::mailbox::{
maildir,
mbox::{self, MessageIterator},
};
use crate::modules::UnwrapResult;
use super::{
cli::{ImportCommands, MailboxFormat},
read_file,
};
enum Mailbox {
Mbox(mbox::MessageIterator<Cursor<Vec<u8>>>),
Maildir(maildir::MessageIterator),
None,
}
#[derive(Debug)]
enum MailboxId<'x> {
ExistingId(&'x str),
CreateId(String),
None,
}
#[derive(Debug)]
struct Message {
identifier: String,
flags: Vec<maildir::Flag>,
internal_date: u64,
contents: Vec<u8>,
}
pub async fn cmd_import(client: Client, command: ImportCommands) {
match command {
ImportCommands::Messages {
num_threads,
format,
account_id,
path,
} => {
let account_id = Arc::new(account_id);
let mut create_mailboxes = Vec::new();
let mut create_mailbox_names = Vec::new();
let mut create_mailbox_ids = Vec::new();
eprintln!("{} Parsing mailbox...", style("[1/4]").bold().dim(),);
match format {
MailboxFormat::Mbox => {
create_mailbox_names.push(Vec::new());
create_mailboxes.push(Mailbox::Mbox(MessageIterator::new(Cursor::new(
read_file(&path),
))));
}
MailboxFormat::Maildir | MailboxFormat::MaildirNested => {
let (folder_sep, folder_split) = if format == MailboxFormat::Maildir {
(Some("."), ".")
} else {
(None, "/")
};
for folder in maildir::FolderIterator::new(path, folder_sep)
.unwrap_result("read Maildir folder")
{
let folder = folder.unwrap_result("read Maildir folder");
if let Some(folder_name) = folder.name() {
let mut folder_parts = Vec::new();
for folder_name in folder_name.split(folder_split) {
let mut folder_name = folder_name.trim();
if folder_name.is_empty() {
folder_name = ".";
}
folder_parts.push(folder_name.to_string());
if !create_mailbox_names.contains(&folder_parts) {
create_mailboxes.push(Mailbox::None);
create_mailbox_names.push(folder_parts.clone());
}
}
*create_mailboxes.last_mut().unwrap() = Mailbox::Maildir(folder);
} else {
create_mailboxes.push(Mailbox::Maildir(folder));
create_mailbox_names.push(Vec::new());
};
}
}
}
// Fetch all mailboxes for the account
eprintln!(
"{} Fetching existing mailboxes for account...",
style("[2/4]").bold().dim(),
);
let mut inbox_id = None;
let mut mailbox_ids = HashMap::new();
let mut children: HashMap<Option<&str>, Vec<&str>> =
HashMap::from_iter([(None, Vec::new())]);
let mut request = client.build();
request
.get_mailbox()
.account_id(account_id.as_ref())
.properties([
mailbox::Property::Name,
mailbox::Property::ParentId,
mailbox::Property::Role,
mailbox::Property::Id,
]);
let response = request
.send_get_mailbox()
.await
.unwrap_result("fetch mailboxes");
for mailbox in response.list() {
let mailbox_id = mailbox.id().unwrap();
if mailbox.role() == Role::Inbox {
inbox_id = mailbox_id.into();
}
children
.entry(mailbox.parent_id())
.or_insert_with(Vec::new)
.push(mailbox_id);
mailbox_ids.insert(mailbox_id, mailbox.name().unwrap_or("Untitled"));
}
let inbox_id =
inbox_id.unwrap_result("locate Inbox on account, please check the server logs.");
let mut it = children.get(&None).unwrap().iter();
let mut it_stack = Vec::new();
let mut name_stack = Vec::new();
let mut mailbox_names = HashMap::with_capacity(mailbox_ids.len());
// Build mailbox hierarchy on the server
eprintln!(
"{} Creating missing mailboxes...",
style("[3/4]").bold().dim(),
);
loop {
while let Some(mailbox_id) = it.next() {
let name = mailbox_ids[mailbox_id];
let mut mailbox_name = name_stack.clone();
mailbox_name.push(name.to_string());
mailbox_names.insert(mailbox_name, mailbox_id);
if let Some(next_it) = children.get(&Some(mailbox_id)).map(|c| c.iter()) {
name_stack.push(name.to_string());
it_stack.push(it);
it = next_it;
}
}
if let Some(prev_it) = it_stack.pop() {
name_stack.pop();
it = prev_it;
} else {
break;
}
}
// Check whether the mailboxes to be created already exist
let mut has_missing_mailboxes = false;
for mailbox_name in &create_mailbox_names {
create_mailbox_ids.push(if !mailbox_name.is_empty() {
if let Some(mailbox_id) = mailbox_names.get(mailbox_name) {
MailboxId::ExistingId(mailbox_id)
} else {
has_missing_mailboxes = true;
MailboxId::None
}
} else {
MailboxId::ExistingId(inbox_id)
});
}
// Create any missing mailboxes
if has_missing_mailboxes {
let mut request = client.build();
let set_request = request.set_mailbox().account_id(account_id.as_ref());
for pos in 0..create_mailbox_ids.len() {
if let MailboxId::None = create_mailbox_ids[pos] {
let mailbox_name = &create_mailbox_names[pos];
let create_request =
set_request.create().name(mailbox_name.last().unwrap());
if mailbox_name.len() > 1 {
let parent_mailbox_name = &mailbox_name[..mailbox_name.len() - 1];
let parent_mailbox_pos = create_mailbox_names
.iter()
.position(|n| n == parent_mailbox_name)
.unwrap();
match &create_mailbox_ids[parent_mailbox_pos] {
MailboxId::ExistingId(id) => {
create_request.parent_id((*id).into());
}
MailboxId::CreateId(id_ref) => {
create_request.parent_id_ref(id_ref);
}
MailboxId::None => unreachable!(),
}
} else {
create_request.parent_id(None::<String>);
}
create_mailbox_ids[pos] =
MailboxId::CreateId(create_request.create_id().unwrap());
}
}
// Create mailboxes
let mut response = request
.send_set_mailbox()
.await
.unwrap_result("create mailboxes");
for create_mailbox_id in create_mailbox_ids.iter_mut() {
if let MailboxId::CreateId(id) = create_mailbox_id {
*id = response
.created(id)
.unwrap_result("create mailbox")
.take_id();
}
}
}
// Import messages
eprintln!("{} Importing messages...", style("[4/4]").bold().dim(),);
let client = Arc::new(client);
let total_imported = Arc::new(AtomicUsize::from(0));
let m = MultiProgress::new();
let num_threads = num_threads.unwrap_or_else(|| num_cpus::get());
let spinner_style =
ProgressStyle::with_template("{prefix:.bold.dim} {spinner} {wide_msg}")
.unwrap()
.tick_chars("⠁⠂⠄⡀⢀⠠⠐⠈ ");
let pbs = Arc::new(Mutex::new((
(0..num_threads)
.map(|n| {
let pb = m.add(ProgressBar::new(40));
pb.set_style(spinner_style.clone());
pb.set_prefix(format!("[{}/?]", n + 1));
pb
})
.collect::<Vec<_>>(),
0usize,
)));
let failures = Arc::new(Mutex::new(Vec::new()));
let mut message_num = 0;
for ((mut mailbox, mailbox_id), mailbox_name) in create_mailboxes
.into_iter()
.zip(create_mailbox_ids)
.zip(create_mailbox_names)
{
let mut futures = FuturesUnordered::new();
let mut outputs = Vec::new();
let mailbox_id = Arc::new(match mailbox_id {
MailboxId::ExistingId(id) => id.to_string(),
MailboxId::CreateId(id) => id,
MailboxId::None => unreachable!(),
});
let mailbox_name = Arc::new(if !mailbox_name.is_empty() {
mailbox_name.join("/")
} else {
"Inbox".to_string()
});
while let Some(result) = mailbox.next() {
match result {
Ok(message) => {
message_num += 1;
let client = client.clone();
let account_id = account_id.clone();
let mailbox_id = mailbox_id.clone();
let mailbox_name = mailbox_name.clone();
let total_imported = total_imported.clone();
let pbs = pbs.clone();
let failures = failures.clone();
futures.push(async move {
// Update progress bar
{
let mut pbs = pbs.lock().unwrap();
let pb = &pbs.0[pbs.1 % pbs.0.len()];
pb.set_message(format!(
"Importing {}: {}/{}",
message_num, mailbox_name, message.identifier
));
pb.inc(1);
pbs.1 += 1;
}
if let Err(err) = client
.email_import_account(
&account_id,
message.contents,
[mailbox_id.as_ref()],
if !message.flags.is_empty() {
message
.flags
.into_iter()
.map(|f| match f {
maildir::Flag::Passed => "$passed",
maildir::Flag::Replied => "$answered",
maildir::Flag::Seen => "$seen",
maildir::Flag::Trashed => "$deleted",
maildir::Flag::Draft => "$draft",
maildir::Flag::Flagged => "$flagged",
})
.into()
} else {
None
},
if message.internal_date > 0 {
(message.internal_date as i64).into()
} else {
None
},
)
.await
{
failures.lock().unwrap().push(format!(
concat!(
"Failed to import message {} ",
"with identifier '{}': {}"
),
message_num, message.identifier, err
));
} else {
total_imported.fetch_add(1, Ordering::Relaxed);
}
});
if futures.len() == num_threads {
outputs.push(futures.next().await.unwrap());
}
}
Err(e) => {
failures
.lock()
.unwrap()
.push(format!("I/O error reading message: {}", e));
}
}
}
// Wait for remaining futures
while let Some(item) = futures.next().await {
outputs.push(item);
}
}
// Done
for pb in pbs.lock().unwrap().0.iter() {
pb.finish_with_message("Done");
}
let failures = failures.lock().unwrap();
eprintln!(
"\n\nSuccessfully imported {} messages.\n",
total_imported.load(Ordering::Relaxed)
);
if !failures.is_empty() {
eprintln!("There were {} failures:\n", failures.len());
for failure in failures.iter() {
eprintln!("{}", failure);
}
}
}
}
}
impl Iterator for Mailbox {
type Item = io::Result<Message>;
fn next(&mut self) -> Option<Self::Item> {
match self {
Mailbox::Mbox(it) => it.next().map(|r| {
r.map(|m| Message {
identifier: m.from().to_string(),
flags: Vec::new(),
internal_date: m.internal_date(),
contents: m.unwrap_contents(),
})
.map_err(|_| {
io::Error::new(io::ErrorKind::Other, "Failed to parse from mbox file.")
})
}),
Mailbox::Maildir(it) => it.next().map(|r| {
r.map(|m| Message {
identifier: m
.path()
.file_name()
.and_then(|f| f.to_str())
.unwrap_or("unknown")
.to_string(),
flags: m.flags().to_vec(),
internal_date: m.internal_date(),
contents: m.unwrap_contents(),
})
}),
Mailbox::None => None,
}
}
}

149
crates/cli/src/modules/mod.rs Normal file
View File

@@ -0,0 +1,149 @@
/*
* Copyright (c) 2020-2023, Stalwart Labs Ltd.
*
* This file is part of the Stalwart Command Line Interface.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of
* the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
* in the LICENSE file at the top-level directory of this distribution.
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* You can be released from the requirements of the AGPLv3 license by
* purchasing a commercial license. Please contact licensing@stalw.art
* for more details.
*/
use std::{collections::HashMap, fmt::Display, io::Read};
use jmap_client::principal::Property;
pub mod cli;
pub mod import;
pub mod queue;
pub mod report;
pub trait UnwrapResult<T> {
fn unwrap_result(self, action: &str) -> T;
}
impl<T> UnwrapResult<T> for Option<T> {
fn unwrap_result(self, message: &str) -> T {
match self {
Some(result) => result,
None => {
eprintln!("Failed to {}", message);
std::process::exit(1);
}
}
}
}
impl<T, E: Display> UnwrapResult<T> for Result<T, E> {
fn unwrap_result(self, message: &str) -> T {
match self {
Ok(result) => result,
Err(err) => {
eprintln!("Failed to {}: {}", message, err);
std::process::exit(1);
}
}
}
}
trait TableName {
fn table_name(&self) -> &'static str;
}
impl TableName for Property {
fn table_name(&self) -> &'static str {
match self {
Property::Id => "Id",
Property::Type => "Type",
Property::Name => "Name",
Property::Description => "Description",
Property::Email => "E-mail",
Property::Timezone => "Timezone",
Property::Capabilities => "Capabilities",
Property::Aliases => "Aliases",
Property::Secret => "Secret",
Property::DKIM => "DKIM",
Property::Quota => "Quota",
Property::Picture => "Picture",
Property::Members => "Members",
Property::ACL => "ACL",
}
}
}
pub fn read_file(path: &str) -> Vec<u8> {
if path == "-" {
let mut stdin = std::io::stdin().lock();
let mut raw_message = Vec::with_capacity(1024);
let mut buf = [0; 1024];
loop {
let n = stdin.read(&mut buf).unwrap();
if n == 0 {
break;
}
raw_message.extend_from_slice(&buf[..n]);
}
raw_message
} else {
std::fs::read(path).unwrap_or_else(|_| {
eprintln!("Failed to read file: {}", path);
std::process::exit(1);
})
}
}
pub fn get(url: &str) -> HashMap<String, serde_json::Value> {
serde_json::from_slice(
&reqwest::blocking::Client::builder()
.danger_accept_invalid_certs(true)
.build()
.unwrap_or_default()
.get(url)
.send()
.unwrap_result("send OAuth GET request")
.bytes()
.unwrap_result("fetch bytes"),
)
.unwrap_result("deserialize OAuth GET response")
}
pub fn post(url: &str, params: &HashMap<String, String>) -> HashMap<String, serde_json::Value> {
serde_json::from_slice(
&reqwest::blocking::Client::builder()
.danger_accept_invalid_certs(true)
.build()
.unwrap_or_default()
.post(url)
.form(params)
.send()
.unwrap_result("send OAuth POST request")
.bytes()
.unwrap_result("fetch bytes"),
)
.unwrap_result("deserialize OAuth POST response")
}
pub trait OAuthResponse {
fn property(&self, name: &str) -> &str;
}
impl OAuthResponse for HashMap<String, serde_json::Value> {
fn property(&self, name: &str) -> &str {
self.get(name)
.unwrap_result(&format!("find '{}' in OAuth response", name))
.as_str()
.unwrap_result(&format!("invalid '{}' value", name))
}
}

548
crates/cli/src/modules/queue.rs Normal file
View File

@@ -0,0 +1,548 @@
/*
* Copyright (c) 2020-2023, Stalwart Labs Ltd.
*
* This file is part of the Stalwart Command Line Interface.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of
* the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
* in the LICENSE file at the top-level directory of this distribution.
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* You can be released from the requirements of the AGPLv3 license by
* purchasing a commercial license. Please contact licensing@stalw.art
* for more details.
*/
use super::{cli::QueueCommands, UnwrapResult};
use console::Term;
use human_size::{Byte, SpecificSize};
use jmap_client::client::Credentials;
use mail_parser::DateTime;
use prettytable::{format::Alignment, Attr, Cell, Row, Table};
use reqwest::header::AUTHORIZATION;
use serde::{de::DeserializeOwned, Deserialize, Deserializer};
#[derive(Debug, Deserialize, PartialEq, Eq)]
pub struct Message {
pub return_path: String,
pub domains: Vec<Domain>,
#[serde(deserialize_with = "deserialize_datetime")]
pub created: DateTime,
pub size: usize,
#[serde(default)]
pub priority: i16,
pub env_id: Option<String>,
}
#[derive(Debug, Deserialize, PartialEq, Eq)]
pub struct Domain {
pub name: String,
pub status: Status,
pub recipients: Vec<Recipient>,
pub retry_num: u32,
#[serde(deserialize_with = "deserialize_maybe_datetime")]
pub next_retry: Option<DateTime>,
#[serde(deserialize_with = "deserialize_maybe_datetime")]
pub next_notify: Option<DateTime>,
#[serde(deserialize_with = "deserialize_datetime")]
pub expires: DateTime,
}
#[derive(Debug, Deserialize, PartialEq, Eq)]
pub struct Recipient {
pub address: String,
pub status: Status,
pub orcpt: Option<String>,
}
#[derive(Debug, PartialEq, Eq, Deserialize)]
pub enum Status {
#[serde(rename = "scheduled")]
Scheduled,
#[serde(rename = "completed")]
Completed(String),
#[serde(rename = "temp_fail")]
TemporaryFailure(String),
#[serde(rename = "perm_fail")]
PermanentFailure(String),
}
pub fn cmd_queue(url: &str, credentials: Credentials, command: QueueCommands) {
match command {
QueueCommands::List {
sender,
rcpt,
before,
after,
page_size,
} => {
let stdout = Term::buffered_stdout();
let ids = query_messages(url, &credentials, &sender, &rcpt, &before, &after);
let ids_len = ids.len();
let page_size = page_size.map(|p| std::cmp::max(p, 1)).unwrap_or(20);
let pages_total = (ids_len as f64 / page_size as f64).ceil() as usize;
for (page_num, chunk) in ids.chunks(page_size).enumerate() {
// Build table
let mut table = Table::new();
table.add_row(Row::new(
["ID", "Delivery Due", "Sender", "Recipients", "Size"]
.iter()
.map(|p| Cell::new(p).with_style(Attr::Bold))
.collect(),
));
for (message, id) in smtp_manage_request::<Vec<Option<Message>>>(
&build_query(url, "/queue/status?ids=", chunk),
&credentials,
)
.into_iter()
.zip(chunk)
{
if let Some(message) = message {
let mut rcpts = String::new();
let mut deliver_at = i64::MAX;
let mut deliver_pos = 0;
for (pos, domain) in message.domains.iter().enumerate() {
if let Some(next_retry) = &domain.next_retry {
let ts = next_retry.to_timestamp();
if ts < deliver_at {
deliver_at = ts;
deliver_pos = pos;
}
}
for rcpt in &domain.recipients {
if !rcpts.is_empty() {
rcpts.push('\n');
}
rcpts.push_str(&rcpt.address);
rcpts.push_str(" (");
rcpts.push_str(rcpt.status.status_short());
rcpts.push(')');
}
}
let mut cells = Vec::new();
cells.push(Cell::new(&format!("{id:X}")));
cells.push(if deliver_at != i64::MAX {
Cell::new(
&message.domains[deliver_pos]
.next_retry
.as_ref()
.unwrap()
.to_rfc822(),
)
} else {
Cell::new("None")
});
cells.push(Cell::new(if !message.return_path.is_empty() {
&message.return_path
} else {
"<>"
}));
cells.push(Cell::new(&rcpts));
cells.push(Cell::new(
&SpecificSize::new(message.size as u32, Byte)
.unwrap()
.to_string(),
));
table.add_row(Row::new(cells));
}
}
eprintln!();
table.printstd();
eprintln!();
if page_num + 1 != pages_total {
eprintln!("\n--- Press any key to continue or 'q' to exit ---");
if let Ok('q' | 'Q') = stdout.read_char() {
break;
}
}
}
eprintln!("\n{ids_len} queued message(s) found.")
}
QueueCommands::Status { ids } => {
for (message, id) in smtp_manage_request::<Vec<Option<Message>>>(
&build_query(url, "/queue/status?ids=", &parse_ids(&ids)),
&credentials,
)
.into_iter()
.zip(&ids)
{
let mut table = Table::new();
table.add_row(Row::new(vec![
Cell::new("ID").with_style(Attr::Bold),
Cell::new(id),
]));
if let Some(message) = message {
table.add_row(Row::new(vec![
Cell::new("Sender").with_style(Attr::Bold),
Cell::new(if !message.return_path.is_empty() {
&message.return_path
} else {
"<>"
}),
]));
table.add_row(Row::new(vec![
Cell::new("Created").with_style(Attr::Bold),
Cell::new(&message.created.to_rfc822()),
]));
table.add_row(Row::new(vec![
Cell::new("Size").with_style(Attr::Bold),
Cell::new(
&SpecificSize::new(message.size as u32, Byte)
.unwrap()
.to_string(),
),
]));
if let Some(env_id) = &message.env_id {
table.add_row(Row::new(vec![
Cell::new("Env-Id").with_style(Attr::Bold),
Cell::new(env_id),
]));
}
if message.priority != 0 {
table.add_row(Row::new(vec![
Cell::new("Priority").with_style(Attr::Bold),
Cell::new(&message.priority.to_string()),
]));
}
for domain in &message.domains {
table.add_row(Row::new(vec![Cell::new_align(
&domain.name,
Alignment::RIGHT,
)
.with_style(Attr::Bold)
.with_style(Attr::Italic(true))
.with_hspan(2)]));
table.add_row(Row::new(vec![
Cell::new("Status").with_style(Attr::Bold),
Cell::new(domain.status.status()),
]));
table.add_row(Row::new(vec![
Cell::new("Details").with_style(Attr::Bold),
Cell::new(domain.status.details()),
]));
table.add_row(Row::new(vec![
Cell::new("Retry #").with_style(Attr::Bold),
Cell::new(&domain.retry_num.to_string()),
]));
if let Some(dt) = &domain.next_retry {
table.add_row(Row::new(vec![
Cell::new("Delivery Due").with_style(Attr::Bold),
Cell::new(&dt.to_rfc822()),
]));
}
if let Some(dt) = &domain.next_notify {
table.add_row(Row::new(vec![
Cell::new("Notify at").with_style(Attr::Bold),
Cell::new(&dt.to_rfc822()),
]));
}
table.add_row(Row::new(vec![
Cell::new("Expires").with_style(Attr::Bold),
Cell::new(&domain.expires.to_rfc822()),
]));
let mut rcpts = Table::new();
rcpts.add_row(Row::new(vec![
Cell::new("Address").with_style(Attr::Bold),
Cell::new("Status").with_style(Attr::Bold),
Cell::new("Details").with_style(Attr::Bold),
]));
for rcpt in &domain.recipients {
rcpts.add_row(Row::new(vec![
Cell::new(&rcpt.address),
Cell::new(rcpt.status.status()),
Cell::new(rcpt.status.details()),
]));
}
table.add_row(Row::new(vec![
Cell::new("Recipients").with_style(Attr::Bold),
Cell::from(&rcpts),
]));
}
} else {
table.add_row(Row::new(vec![Cell::new_align(
"-- Not found --",
Alignment::CENTER,
)
.with_hspan(2)]));
}
eprintln!();
table.printstd();
eprintln!();
}
}
QueueCommands::Retry {
sender,
domain,
before,
after,
time,
ids,
} => {
let (parsed_ids, ids) = if ids.is_empty() {
if sender.is_some() || domain.is_some() || before.is_some() || after.is_some() {
let parsed_ids =
query_messages(url, &credentials, &sender, &domain, &before, &after);
let ids = parsed_ids.iter().map(|id| format!("{id:X}")).collect();
(parsed_ids, ids)
} else {
(vec![], vec![])
}
} else {
(parse_ids(&ids), ids)
};
if ids.is_empty() {
eprintln!("No messages were found.");
std::process::exit(1);
}
let mut query = form_urlencoded::Serializer::new(format!("{url}/queue/retry?"));
if let Some(filter) = &domain {
query.append_pair("filter", filter);
}
if let Some(at) = time {
query.append_pair("at", &at.to_rfc3339());
}
query.append_pair("ids", &append_ids(String::new(), &parsed_ids));
let mut success_count = 0;
let mut failed_list = vec![];
for (success, id) in smtp_manage_request::<Vec<bool>>(&query.finish(), &credentials)
.into_iter()
.zip(ids)
{
if success {
success_count += 1;
} else {
failed_list.push(id);
}
}
eprint!("\nSuccessfully rescheduled {success_count} message(s).");
if !failed_list.is_empty() {
eprint!(" Unable to reschedule id(s): {}.", failed_list.join(", "));
}
eprintln!();
}
QueueCommands::Cancel {
sender,
rcpt,
before,
after,
ids,
} => {
let (parsed_ids, ids) = if ids.is_empty() {
if sender.is_some() || rcpt.is_some() || before.is_some() || after.is_some() {
let parsed_ids =
query_messages(url, &credentials, &sender, &rcpt, &before, &after);
let ids = parsed_ids.iter().map(|id| format!("{id:X}")).collect();
(parsed_ids, ids)
} else {
(vec![], vec![])
}
} else {
(parse_ids(&ids), ids)
};
if ids.is_empty() {
eprintln!("No messages were found.");
std::process::exit(1);
}
let mut query = form_urlencoded::Serializer::new(format!("{url}/queue/cancel?"));
if let Some(filter) = &rcpt {
query.append_pair("filter", filter);
}
query.append_pair("ids", &append_ids(String::new(), &parsed_ids));
let mut success_count = 0;
let mut failed_list = vec![];
for (success, id) in smtp_manage_request::<Vec<bool>>(&query.finish(), &credentials)
.into_iter()
.zip(ids)
{
if success {
success_count += 1;
} else {
failed_list.push(id);
}
}
eprint!("\nCancelled delivery of {success_count} message(s).");
if !failed_list.is_empty() {
eprint!(
" Unable to cancel delivery for id(s): {}.",
failed_list.join(", ")
);
}
eprintln!();
}
}
}
#[derive(Deserialize)]
#[serde(untagged)]
pub enum Response<T> {
Data { data: T },
Error { error: String, details: String },
}
pub fn smtp_manage_request<T: DeserializeOwned>(url: &str, credentials: &Credentials) -> T {
match serde_json::from_slice::<Response<T>>(
&reqwest::blocking::Client::builder()
.danger_accept_invalid_certs(url.starts_with("https://127.0.0.1"))
.build()
.unwrap_or_default()
.get(url)
.header(
AUTHORIZATION,
match credentials {
Credentials::Basic(s) => format!("Basic {s}"),
Credentials::Bearer(s) => format!("Bearer {s}"),
},
)
.send()
.unwrap_result("send GET request")
.bytes()
.unwrap_result("fetch bytes"),
)
.unwrap_result("deserialize response")
{
Response::Data { data } => data,
Response::Error { error, details } => {
eprintln!("Request failed: {details} ({error:?})");
std::process::exit(1);
}
}
}
fn query_messages(
url: &str,
credentials: &Credentials,
from: &Option<String>,
rcpt: &Option<String>,
before: &Option<DateTime>,
after: &Option<DateTime>,
) -> Vec<u64> {
let mut query = form_urlencoded::Serializer::new(format!("{url}/queue/list?"));
if let Some(sender) = from {
query.append_pair("from", sender);
}
if let Some(rcpt) = rcpt {
query.append_pair("to", rcpt);
}
if let Some(before) = before {
query.append_pair("before", &before.to_rfc3339());
}
if let Some(after) = after {
query.append_pair("after", &after.to_rfc3339());
}
smtp_manage_request::<Vec<u64>>(&query.finish(), credentials)
}
fn deserialize_maybe_datetime<'de, D>(deserializer: D) -> Result<Option<DateTime>, D::Error>
where
D: Deserializer<'de>,
{
if let Some(value) = Option::<&str>::deserialize(deserializer)? {
if let Some(value) = DateTime::parse_rfc3339(value) {
Ok(Some(value))
} else {
Err(serde::de::Error::custom(
"Failed to parse RFC3339 timestamp",
))
}
} else {
Ok(None)
}
}
pub fn deserialize_datetime<'de, D>(deserializer: D) -> Result<DateTime, D::Error>
where
D: Deserializer<'de>,
{
if let Some(value) = DateTime::parse_rfc3339(<&str>::deserialize(deserializer)?) {
Ok(value)
} else {
Err(serde::de::Error::custom(
"Failed to parse RFC3339 timestamp",
))
}
}
fn parse_ids(ids: &[String]) -> Vec<u64> {
let mut result = Vec::with_capacity(ids.len());
for id in ids {
match u64::from_str_radix(id, 16) {
Ok(id) => {
result.push(id);
}
Err(_) => {
eprintln!("Failed to parse id {id:?}.");
std::process::exit(1);
}
}
}
result
}
fn build_query(url: &str, path: &str, ids: &[u64]) -> String {
let mut query = String::with_capacity(url.len() + path.len() + (ids.len() * 10));
query.push_str(url);
query.push_str(path);
append_ids(query, ids)
}
fn append_ids(mut query: String, ids: &[u64]) -> String {
for (pos, id) in ids.iter().enumerate() {
if pos != 0 {
query.push(',');
}
query.push_str(&id.to_string());
}
query
}
impl Status {
fn status_short(&self) -> &str {
match self {
Status::Scheduled => "scheduled",
Status::Completed(_) => "delivered",
Status::TemporaryFailure(_) => "tempfail",
Status::PermanentFailure(_) => "permfail",
}
}
fn status(&self) -> &str {
match self {
Status::Scheduled => "Scheduled",
Status::Completed(_) => "Delivered",
Status::TemporaryFailure(_) => "Temporary Failure",
Status::PermanentFailure(_) => "Permanent Failure",
}
}
fn details(&self) -> &str {
match self {
Status::Scheduled => "",
Status::Completed(status) => status,
Status::TemporaryFailure(status) => status,
Status::PermanentFailure(status) => status,
}
}
}

203
crates/cli/src/modules/report.rs Normal file
View File

@@ -0,0 +1,203 @@
/*
* Copyright (c) 2020-2023, Stalwart Labs Ltd.
*
* This file is part of the Stalwart Command Line Interface.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of
* the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
* in the LICENSE file at the top-level directory of this distribution.
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* You can be released from the requirements of the AGPLv3 license by
* purchasing a commercial license. Please contact licensing@stalw.art
* for more details.
*/
use super::cli::{ReportCommands, ReportFormat};
use crate::modules::queue::{deserialize_datetime, smtp_manage_request};
use console::Term;
use human_size::{Byte, SpecificSize};
use jmap_client::client::Credentials;
use mail_parser::DateTime;
use prettytable::{format::Alignment, Attr, Cell, Row, Table};
use serde::Deserialize;
#[derive(Debug, Deserialize)]
pub struct Report {
pub domain: String,
#[serde(rename = "type")]
pub type_: ReportFormat,
#[serde(deserialize_with = "deserialize_datetime")]
pub range_from: DateTime,
#[serde(deserialize_with = "deserialize_datetime")]
pub range_to: DateTime,
pub size: usize,
}
pub fn cmd_report(url: &str, credentials: Credentials, command: ReportCommands) {
match command {
ReportCommands::List {
domain,
format,
page_size,
} => {
let stdout = Term::buffered_stdout();
let mut query = form_urlencoded::Serializer::new(format!("{url}/report/list?"));
if let Some(domain) = &domain {
query.append_pair("domain", domain);
}
if let Some(format) = &format {
query.append_pair("type", format.id());
}
let ids = smtp_manage_request::<Vec<String>>(&query.finish(), &credentials);
let ids_len = ids.len();
let page_size = page_size.map(|p| std::cmp::max(p, 1)).unwrap_or(20);
let pages_total = (ids_len as f64 / page_size as f64).ceil() as usize;
for (page_num, chunk) in ids.chunks(page_size).enumerate() {
// Build table
let mut table = Table::new();
table.add_row(Row::new(
["ID", "Domain", "Type", "From Date", "To Date", "Size"]
.iter()
.map(|p| Cell::new(p).with_style(Attr::Bold))
.collect(),
));
for (report, id) in smtp_manage_request::<Vec<Option<Report>>>(
&format!("{url}/report/status?ids={}", chunk.join(",")),
&credentials,
)
.into_iter()
.zip(chunk)
{
if let Some(report) = report {
table.add_row(Row::new(vec![
Cell::new(id),
Cell::new(&report.domain),
Cell::new(report.type_.name()),
Cell::new(&report.range_from.to_rfc822()),
Cell::new(&report.range_to.to_rfc822()),
Cell::new(
&SpecificSize::new(report.size as u32, Byte)
.unwrap()
.to_string(),
),
]));
}
}
eprintln!();
table.printstd();
eprintln!();
if page_num + 1 != pages_total {
eprintln!("\n--- Press any key to continue or 'q' to exit ---");
if let Ok('q' | 'Q') = stdout.read_char() {
break;
}
}
}
eprintln!("\n{ids_len} queued message(s) found.")
}
ReportCommands::Status { ids } => {
for (report, id) in smtp_manage_request::<Vec<Option<Report>>>(
&format!("{url}/report/status?ids={}", ids.join(",")),
&credentials,
)
.into_iter()
.zip(&ids)
{
let mut table = Table::new();
table.add_row(Row::new(vec![
Cell::new("ID").with_style(Attr::Bold),
Cell::new(id),
]));
if let Some(report) = report {
table.add_row(Row::new(vec![
Cell::new("Domain Name").with_style(Attr::Bold),
Cell::new(&report.domain),
]));
table.add_row(Row::new(vec![
Cell::new("Type").with_style(Attr::Bold),
Cell::new(report.type_.name()),
]));
table.add_row(Row::new(vec![
Cell::new("From Date").with_style(Attr::Bold),
Cell::new(&report.range_from.to_rfc822()),
]));
table.add_row(Row::new(vec![
Cell::new("To Date").with_style(Attr::Bold),
Cell::new(&report.range_to.to_rfc822()),
]));
table.add_row(Row::new(vec![
Cell::new("Size").with_style(Attr::Bold),
Cell::new(
&SpecificSize::new(report.size as u32, Byte)
.unwrap()
.to_string(),
),
]));
} else {
table.add_row(Row::new(vec![Cell::new_align(
"-- Not found --",
Alignment::CENTER,
)
.with_hspan(2)]));
}
eprintln!();
table.printstd();
eprintln!();
}
}
ReportCommands::Cancel { ids } => {
let mut success_count = 0;
let mut failed_list = vec![];
for (success, id) in smtp_manage_request::<Vec<bool>>(
&format!("{url}/report/cancel?ids={}", ids.join(",")),
&credentials,
)
.into_iter()
.zip(ids)
{
if success {
success_count += 1;
} else {
failed_list.push(id);
}
}
eprint!("\nRemoved {success_count} report(s).");
if !failed_list.is_empty() {
eprint!(
" Unable to remove report id(s): {}.",
failed_list.join(", ")
);
}
eprintln!();
}
}
}
impl ReportFormat {
fn id(&self) -> &'static str {
match self {
ReportFormat::Dmarc => "dmarc",
ReportFormat::Tls => "tls",
}
}
fn name(&self) -> &'static str {
match self {
ReportFormat::Dmarc => "DMARC",
ReportFormat::Tls => "TLS",
}
}
}