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

Webhooks implementation (closes #480 closes #233)

Browse Source
This commit is contained in:
mdecimus
2024-06-20 19:11:15 +02:00
parent 5ff6bc895c
commit 68a189ed9f
72 changed files with 2368 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
tests/src/imap/mod.rs
View File

@@ -45,7 +45,8 @@ use std::{
use ::managesieve::core::ManageSieveSessionManager;
use common::{
config::server::{ServerProtocol, Servers},
Core,
webhooks::manager::spawn_webhook_manager,
Core, Ipc, IPC_CHANNEL_BUFFER,
};
use ::store::Stores;
@@ -53,7 +54,7 @@ use ahash::AHashSet;
use directory::backend::internal::manage::ManageDirectory;
use imap::core::{ImapSessionManager, Inner, IMAP};
use imap_proto::ResponseType;
use jmap::{api::JmapSessionManager, services::IPC_CHANNEL_BUFFER, JMAP};
use jmap::{api::JmapSessionManager, JMAP};
use pop3::Pop3SessionManager;
use smtp::core::{SmtpSessionManager, SMTP};
use tokio::{
@@ -321,9 +322,18 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest {
// Parse acceptors
servers.parse_tcp_acceptors(&mut config, shared_core.clone());
// Init servers
// Spawn webhook manager
let webhook_tx = spawn_webhook_manager(shared_core.clone());
// Setup IPC channels
let (delivery_tx, delivery_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
let smtp = SMTP::init(&mut config, shared_core.clone(), delivery_tx).await;
let ipc = Ipc {
delivery_tx,
webhook_tx,
};
// Init servers
let smtp = SMTP::init(&mut config, shared_core.clone(), ipc).await;
let jmap = JMAP::init(
&mut config,
delivery_rx,
@@ -369,6 +379,7 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest {
),
};
});
// Create tables and test accounts
let lookup = DirectoryStore {
store: shared_core

28
tests/src/jmap/auth_limits.rs
View File

@@ -71,6 +71,16 @@ pub async fn test(params: &mut JMAPTest) {
// Reset rate limiters
server.inner.concurrency_limiter.clear();
params.webhook.clear();
// Incorrect passwords should be rejected with a 401 error
assert!(matches!(
Client::new()
.credentials(Credentials::basic("jdoe@example.com", "abcde"))
.accept_invalid_certs(true)
.connect("https://127.0.0.1:8899")
.await,
Err(jmap_client::Error::Problem(err)) if err.status() == Some(401)));
// Wait until the beginning of the 5 seconds bucket
const LIMIT: u64 = 5;
@@ -79,15 +89,6 @@ pub async fn test(params: &mut JMAPTest) {
let range_end = (range_start * LIMIT) + LIMIT;
tokio::time::sleep(Duration::from_secs(range_end - now)).await;
// Incorrect passwords should be rejected with a 401 error
assert!(matches!(
Client::new()
.credentials(Credentials::basic("jdoe@example.com", "abcde"))
.accept_invalid_certs(true)
.connect("https://127.0.0.1:8899")
.await,
Err(jmap_client::Error::Problem(err)) if err.status() == Some(401)));
// Invalid authentication requests should be rate limited
let mut n_401 = 0;
let mut n_429 = 0;
@@ -280,4 +281,13 @@ pub async fn test(params: &mut JMAPTest) {
params.client.set_default_account_id(&account_id);
destroy_all_mailboxes(params).await;
assert_is_empty(server).await;
// Check webhook events
params.webhook.assert_contains(&[
"auth.failure",
"auth.success",
"auth.banned",
"\"login\": \"jdoe@example.com\"",
"\"accountType\": \"individual\"",
]);
}

12
tests/src/jmap/delivery.rs
View File

@@ -104,6 +104,7 @@ pub async fn test(params: &mut JMAPTest) {
// Delivering to individuals
let mut lmtp = SmtpConnection::connect().await;
params.webhook.clear();
lmtp.ingest(
"bill@example.com",
@@ -320,6 +321,17 @@ pub async fn test(params: &mut JMAPTest) {
destroy_all_mailboxes(params).await;
}
assert_is_empty(server).await;
// Check webhook events
params.webhook.assert_contains(&[
"message.accepted",
"message.appended",
"dsn",
"\"returnPath\": \"bill@example.com\"",
"\"sender\": \"bill@example.com\"",
"\"address\": \"john.doe@example.com\"",
"\"type\": \"success\"",
]);
}
pub struct SmtpConnection {

36
tests/src/jmap/mod.rs
View File

@@ -30,15 +30,12 @@ use base64::{
use common::{
config::server::{ServerProtocol, Servers},
manager::config::{ConfigManager, Patterns},
Core,
webhooks::manager::spawn_webhook_manager,
Core, Ipc, IPC_CHANNEL_BUFFER,
};
use hyper::{header::AUTHORIZATION, Method};
use imap::core::{ImapSessionManager, IMAP};
use jmap::{
api::JmapSessionManager,
services::{housekeeper::Event, IPC_CHANNEL_BUFFER},
JMAP,
};
use jmap::{api::JmapSessionManager, services::housekeeper::Event, JMAP};
use jmap_client::client::{Client, Credentials};
use jmap_proto::{error::request::RequestError, types::id::Id};
use managesieve::core::ManageSieveSessionManager;
@@ -54,6 +51,7 @@ use store::{
};
use tokio::sync::{mpsc, watch};
use utils::config::Config;
use webhooks::{spawn_mock_webhook_endpoint, MockWebhookEndpoint};
use crate::{add_test_certs, directory::DirectoryStore, store::TempDir, AssertConfig};
@@ -82,6 +80,7 @@ pub mod stress_test;
pub mod thread_get;
pub mod thread_merge;
pub mod vacation_response;
pub mod webhooks;
pub mod websocket;
const SERVER: &str = r#"
@@ -287,6 +286,15 @@ refresh-token-renew = "2s"
expn = true
vrfy = true
[webhook."test"]
url = "http://127.0.0.1:8821/hook"
events = ["auth.success", "auth.failure", "auth.banned", "auth.error",
"message.accepted", "message.rejected", "message.appended",
"account.over-quota", "dsn", "double-bounce", "report.incoming.dmarc",
"report.incoming.tls", "report.incoming.arf", "report.outgoing"]
signature-key = "ovos-moles"
throttle = "100ms"
"#;
#[tokio::test(flavor = "multi_thread")]
@@ -314,6 +322,7 @@ pub async fn jmap_tests() {
)
.await;
webhooks::test(&mut params).await;
email_query::test(&mut params, delete).await;
email_get::test(&mut params).await;
email_set::test(&mut params).await;
@@ -379,6 +388,7 @@ pub struct JMAPTest {
client: Client,
directory: DirectoryStore,
temp_dir: TempDir,
webhook: Arc<MockWebhookEndpoint>,
shutdown_tx: watch::Sender<bool>,
}
@@ -485,9 +495,18 @@ async fn init_jmap_tests(store_id: &str, delete_if_exists: bool) -> JMAPTest {
// Parse acceptors
servers.parse_tcp_acceptors(&mut config, shared_core.clone());
// Init servers
// Spawn webhook manager
let webhook_tx = spawn_webhook_manager(shared_core.clone());
// Setup IPC channels
let (delivery_tx, delivery_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
let smtp = SMTP::init(&mut config, shared_core.clone(), delivery_tx).await;
let ipc = Ipc {
delivery_tx,
webhook_tx,
};
// Init servers
let smtp = SMTP::init(&mut config, shared_core.clone(), ipc).await;
let jmap = JMAP::init(
&mut config,
delivery_rx,
@@ -569,6 +588,7 @@ async fn init_jmap_tests(store_id: &str, delete_if_exists: bool) -> JMAPTest {
client,
directory,
shutdown_tx,
webhook: spawn_mock_webhook_endpoint(),
}
}

7
tests/src/jmap/thread_merge.rs
View File

@@ -27,7 +27,10 @@ use crate::{
jmap::{assert_is_empty, mailbox::destroy_all_mailboxes},
store::deflate_test_resource,
};
use jmap::{email::ingest::IngestEmail, IngestError};
use jmap::{
email::ingest::{IngestEmail, IngestSource},
IngestError,
};
use jmap_client::{email, mailbox::Role};
use jmap_proto::types::{collection::Collection, id::Id};
use mail_parser::{mailbox::mbox::MessageIterator, MessageParser};
@@ -264,7 +267,7 @@ async fn test_multi_thread(params: &mut JMAPTest) {
mailbox_ids: vec![mailbox_id],
keywords: vec![],
received_at: None,
skip_duplicates: true,
source: IngestSource::Smtp,
encrypt: false,
})
.await

183
tests/src/jmap/webhooks.rs Normal file
View File

@@ -0,0 +1,183 @@
/*
* Copyright (c) 2023 Stalwart Labs Ltd.
*
* This file is part of Stalwart Mail Server.
*
* 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::{
sync::{
atomic::{AtomicBool, Ordering},
Arc,
},
time::Duration,
};
use base64::{engine::general_purpose::STANDARD, Engine};
use common::{
manager::webadmin::Resource,
webhooks::{WebhookEvent, WebhookEvents},
};
use hyper::{body, server::conn::http1, service::service_fn};
use hyper_util::rt::TokioIo;
use jmap::api::http::{fetch_body, ToHttpResponse};
use jmap_proto::error::request::RequestError;
use ring::hmac;
use store::parking_lot::Mutex;
use tokio::{net::TcpListener, sync::watch};
use super::JMAPTest;
pub struct MockWebhookEndpoint {
pub tx: watch::Sender<bool>,
pub events: Mutex<Vec<WebhookEvent>>,
pub reject: AtomicBool,
}
pub async fn test(params: &mut JMAPTest) {
println!("Running Webhook tests...");
// Webhooks endpoint starts disabled by default, make sure there are no events.
tokio::time::sleep(Duration::from_millis(200)).await;
params.webhook.assert_is_empty();
// Enable the endpoint
params.webhook.accept();
tokio::time::sleep(Duration::from_millis(1000)).await;
// Check for events
params.webhook.assert_contains(&["auth.success"]);
}
impl MockWebhookEndpoint {
pub fn assert_contains(&self, expected: &[&str]) {
let events =
serde_json::to_string_pretty(&self.events.lock().drain(..).collect::<Vec<_>>())
.unwrap();
for string in expected {
if !events.contains(string) {
panic!(
"Expected events to contain '{}', but it did not. Events: {}",
string, events
);
}
}
}
pub fn accept(&self) {
self.reject.store(false, Ordering::Relaxed);
}
pub fn reject(&self) {
self.reject.store(true, Ordering::Relaxed);
}
pub fn clear(&self) {
self.events.lock().clear();
}
pub fn assert_is_empty(&self) {
assert!(self.events.lock().is_empty());
}
}
pub fn spawn_mock_webhook_endpoint() -> Arc<MockWebhookEndpoint> {
let (tx, rx) = watch::channel(true);
let endpoint_ = Arc::new(MockWebhookEndpoint {
tx,
events: Mutex::new(vec![]),
reject: true.into(),
});
let endpoint = endpoint_.clone();
tokio::spawn(async move {
let listener = TcpListener::bind("127.0.0.1:8821")
.await
.unwrap_or_else(|e| {
panic!("Failed to bind mock Milter server to 127.0.0.1:8821: {e}");
});
let mut rx_ = rx.clone();
loop {
tokio::select! {
stream = listener.accept() => {
match stream {
Ok((stream, _)) => {
let _ = http1::Builder::new()
.keep_alive(false)
.serve_connection(
TokioIo::new(stream),
service_fn(|mut req: hyper::Request<body::Incoming>| {
let endpoint = endpoint.clone();
async move {
// Verify HMAC signature
let key = hmac::Key::new(hmac::HMAC_SHA256, "ovos-moles".as_bytes());
let body = fetch_body(&mut req, 1024 * 1024).await.unwrap();
let tag = STANDARD.decode(req.headers().get("X-Signature").unwrap().to_str().unwrap()).unwrap();
hmac::verify(&key, &body, &tag).expect("Invalid signature");
// Deserialize JSON
let request = serde_json::from_slice::<WebhookEvents>(&body)
.expect("Failed to parse JSON");
if !endpoint.reject.load(Ordering::Relaxed) {
//let c = print!("received webhook: {}", serde_json::to_string_pretty(&request).unwrap());
// Add events
endpoint.events.lock().extend(request.events);
Ok::<_, hyper::Error>(
Resource {
content_type: "application/json",
contents: "[]".to_string().into_bytes(),
}
.into_http_response(),
)
} else {
//let c = print!("rejected webhook: {}", serde_json::to_string_pretty(&request).unwrap());
Ok::<_, hyper::Error>(
RequestError::not_found().into_http_response()
)
}
}
}),
)
.await;
}
Err(err) => {
panic!("Something went wrong: {err}" );
}
}
},
_ = rx_.changed() => {
//println!("Mock jMilter server stopping");
break;
}
};
}
});
endpoint_
}