Registry testing - part 10

tests/src/telemetry/alerts.rs

@@ -0,0 +1,115 @@
+/*
+ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ */
+
+use crate::utils::server::TestServer;
+use common::BuildServer;
+use registry::{
+    schema::{
+        prelude::ObjectType,
+        structs::{
+            Alert, AlertEmail, AlertEmailProperties, AlertEvent, AlertEventProperties, Expression,
+        },
+    },
+    types::map::Map,
+};
+use trc::{ClusterEvent, Collector, EventType, MetricType};
+
+pub async fn test(test: &TestServer) {
+    println!("Running Alerts tests...");
+
+    // Create alerts
+    let admin = test.account("admin@example.org");
+    admin
+        .registry_create_object(Alert {
+            enable: true,
+            condition: Expression {
+                else_: "metric('domain.count') > 1 && metric('cluster.publisher-error') > 3".into(),
+                ..Default::default()
+            },
+            email_alert: AlertEmail::Enabled(AlertEmailProperties {
+                body: concat!(
+                    "Sorry for the bad news, but we found %{domain.count}% ",
+                    "domains and %{cluster.publisher-error}% cluster errors."
+                )
+                .to_string(),
+                from_address: "alert@example.com".to_string(),
+                from_name: "Alert Subsystem".to_string().into(),
+                subject: "Found %{cluster.publisher-error}% cluster errors".to_string(),
+                to: Map::new(vec!["jdoe@example.com".to_string()]),
+            }),
+            event_alert: AlertEvent::Enabled(AlertEventProperties {
+                event_message: "Yikes! Found %{cluster.publisher-error}% cluster errors!"
+                    .to_string()
+                    .into(),
+            }),
+        })
+        .await;
+    admin
+        .registry_create_object(Alert {
+            enable: true,
+            condition: Expression {
+                else_: "metric('domain.count') < 1 || metric('cluster.publisher-error') < 3".into(),
+                ..Default::default()
+            },
+            email_alert: AlertEmail::Disabled,
+            event_alert: AlertEvent::Enabled(AlertEventProperties {
+                event_message: "this should not have happened".to_string().into(),
+            }),
+        })
+        .await;
+    admin.reload_settings().await;
+
+    // Make sure the required metrics are set to 0
+    assert_eq!(
+        Collector::read_metric(MetricType::ClusterPublisherError),
+        0.0
+    );
+    assert_eq!(Collector::read_metric(MetricType::DomainCount), 1.0);
+    assert_eq!(Collector::read_metric(MetricType::TelemetryAlertEvent), 0.0);
+
+    // Increment metrics to trigger alerts
+    Collector::update_event_counter(EventType::Cluster(ClusterEvent::PublisherError), 5);
+    Collector::update_gauge(MetricType::DomainCount, 3);
+
+    // Make sure the values were set
+    assert_eq!(
+        Collector::read_metric(MetricType::ClusterPublisherError),
+        5.0
+    );
+    assert_eq!(Collector::read_metric(MetricType::DomainCount), 3.0);
+
+    // Process alerts
+    let message = test
+        .server
+        .inner
+        .build_server()
+        .process_alerts()
+        .await
+        .unwrap()
+        .pop()
+        .unwrap();
+    assert_eq!(message.from, "alert@example.com");
+    assert_eq!(message.to, vec!["jdoe@example.com".to_string()]);
+    let body = String::from_utf8(message.body).unwrap();
+    assert!(
+        body.contains("Sorry for the bad news, but we found 3 domains and 5 cluster errors."),
+        "{body:?}"
+    );
+    assert!(body.contains("Subject: Found 5 cluster errors"), "{body:?}");
+    assert!(
+        body.contains("From: \"Alert Subsystem\" <alert@example.com>"),
+        "{body:?}"
+    );
+    assert!(body.contains("To: <jdoe@example.com>"), "{body:?}");
+
+    // Make sure the event was triggered
+    assert_eq!(Collector::read_metric(MetricType::TelemetryAlertEvent), 1.0);
+
+    // Cleanup
+    admin.registry_destroy_all(ObjectType::Alert).await;
+    admin.reload_settings().await;
+    test.cleanup().await;
+}

tests/src/telemetry/metrics.rs

@@ -0,0 +1,164 @@
+/*
+ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ */
+
+use crate::utils::server::TestServer;
+use common::telemetry::metrics::store::{MetricsStore, SharedMetricHistory};
+use registry::{schema::prelude::ObjectType, types::datetime::UTCDateTime};
+use std::time::Duration;
+use store::{
+    rand::{self, Rng},
+    write::now,
+};
+use trc::*;
+use types::id::Id;
+
+pub async fn test(test: &TestServer) {
+    println!("Running Metrics tests...");
+
+    // Make sure there are no span entries in the db
+    let admin = test.account("admin@example.org");
+    assert_eq!(
+        admin
+            .registry_query(
+                ObjectType::Metric,
+                Vec::<(&str, &str)>::new(),
+                Vec::<&str>::new(),
+            )
+            .await
+            .object_ids()
+            .collect::<Vec<_>>(),
+        Vec::<Id>::new()
+    );
+
+    // Insert test metrics
+    insert_test_metrics(test).await;
+
+    // Fetch all metrics
+    let metric_ids = admin
+        .registry_query(
+            ObjectType::Metric,
+            Vec::<(&str, &str)>::new(),
+            Vec::<&str>::new(),
+        )
+        .await
+        .object_ids()
+        .collect::<Vec<_>>();
+    let response = admin
+        .registry_get_many(ObjectType::Metric, Vec::<&str>::new())
+        .await;
+    let metrics = response.list();
+    assert!(
+        metrics.len() > 2000,
+        "Found {} metrics, expected more than 2000",
+        metrics.len()
+    );
+    assert_eq!(metrics.len(), metric_ids.len());
+
+    // Fetch the last 48 hours of metrics
+    let metric_ids = admin
+        .registry_query(
+            ObjectType::Metric,
+            [(
+                "timestampIsGreaterThan",
+                UTCDateTime::from_timestamp((now() - (2 * 86400)) as i64).to_string(),
+            )],
+            Vec::<&str>::new(),
+        )
+        .await
+        .object_ids()
+        .collect::<Vec<_>>();
+    assert!(
+        metric_ids.len() > 20 && metric_ids.len() < 2000,
+        "Found {} metrics, expected more than 20 and less than 2000",
+        metric_ids.len()
+    );
+
+    // Purge metrics and make sure they are gone
+    test.server
+        .metrics_store()
+        .purge_metrics(Duration::from_secs(0))
+        .await
+        .unwrap();
+    assert_eq!(
+        admin
+            .registry_query(
+                ObjectType::Metric,
+                Vec::<(&str, &str)>::new(),
+                Vec::<&str>::new(),
+            )
+            .await
+            .object_ids()
+            .collect::<Vec<_>>(),
+        Vec::<Id>::new()
+    );
+}
+
+async fn insert_test_metrics(test: &TestServer) {
+    test.server
+        .metrics_store()
+        .purge_metrics(Duration::from_secs(0))
+        .await
+        .unwrap();
+    let mut start_time = now() - (90 * 24 * 60 * 60);
+    let timestamp = now();
+    let history = SharedMetricHistory::default();
+
+    while start_time < timestamp {
+        for event_type in [
+            EventType::Smtp(SmtpEvent::ConnectionStart),
+            EventType::Imap(ImapEvent::ConnectionStart),
+            EventType::Pop3(Pop3Event::ConnectionStart),
+            EventType::ManageSieve(ManageSieveEvent::ConnectionStart),
+            EventType::Http(HttpEvent::ConnectionStart),
+            EventType::Delivery(DeliveryEvent::AttemptStart),
+            EventType::Queue(QueueEvent::MessageQueued),
+            EventType::Queue(QueueEvent::AuthenticatedMessageQueued),
+            EventType::Queue(QueueEvent::DsnQueued),
+            EventType::Queue(QueueEvent::ReportQueued),
+            EventType::MessageIngest(MessageIngestEvent::Ham),
+            EventType::MessageIngest(MessageIngestEvent::Spam),
+            EventType::Auth(AuthEvent::Failed),
+            EventType::Security(SecurityEvent::AuthenticationBan),
+            EventType::Security(SecurityEvent::ScanBan),
+            EventType::Security(SecurityEvent::AbuseBan),
+            EventType::Security(SecurityEvent::LoiterBan),
+            EventType::Security(SecurityEvent::IpBlocked),
+            EventType::IncomingReport(IncomingReportEvent::DmarcReport),
+            EventType::IncomingReport(IncomingReportEvent::DmarcReportWithWarnings),
+            EventType::IncomingReport(IncomingReportEvent::TlsReport),
+            EventType::IncomingReport(IncomingReportEvent::TlsReportWithWarnings),
+        ] {
+            // Generate a random value between 0 and 100
+            Collector::update_event_counter(event_type, rand::rng().random_range(0..=100))
+        }
+
+        Collector::update_gauge(MetricType::QueueCount, rand::rng().random_range(0..=1000));
+        Collector::update_gauge(
+            MetricType::ServerMemory,
+            rand::rng().random_range(100 * 1024 * 1024..=300 * 1024 * 1024),
+        );
+
+        for metric_type in [
+            MetricType::MessageIngestTime,
+            MetricType::MessageIngestIndexTime,
+            MetricType::DeliveryTotalTime,
+            MetricType::DnsLookupTime,
+        ] {
+            Collector::update_histogram(metric_type, rand::rng().random_range(2..=1000))
+        }
+        Collector::update_histogram(
+            MetricType::DeliveryTotalTime,
+            rand::rng().random_range(1000..=5000),
+        );
+
+        test.server
+            .metrics_store()
+            .write_metrics(start_time.into(), history.clone())
+            .await
+            .unwrap();
+        start_time += 60 * 60 * 24;
+    }
+}

tests/src/telemetry/mod.rs

@@ -0,0 +1,65 @@
+/*
+ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ */
+
+pub mod alerts;
+pub mod metrics;
+pub mod tracing;
+pub mod webhooks;
+
+use crate::utils::server::TestServerBuilder;
+use registry::schema::structs::{Expression, Jmap, MetricsStore, MtaStageAuth, TracingStore};
+
+#[tokio::test(flavor = "multi_thread")]
+pub async fn telemetry_tests() {
+    let mut test = TestServerBuilder::new("telemetry_tests")
+        .await
+        .with_logging()
+        .with_default_listeners()
+        .await
+        .with_object(MetricsStore::Default)
+        .await
+        .with_object(TracingStore::Default)
+        .await
+        .with_object(Jmap {
+            get_max_results: 100_000,
+            query_max_results: 100_000,
+            ..Default::default()
+        })
+        .await
+        .with_object(MtaStageAuth {
+            require: Expression {
+                else_: "false".to_string(),
+                ..Default::default()
+            },
+            ..Default::default()
+        })
+        .await
+        .build()
+        .await;
+
+    // Create admin account
+    let admin = test
+        .create_user_account(
+            "admin",
+            "admin@example.org",
+            "these_pretzels_are_making_me_thirsty",
+            &[],
+        )
+        .await;
+    test.account("admin")
+        .assign_roles_to_account(admin.id(), &["user", "system"])
+        .await;
+    test.insert_account(admin);
+
+    alerts::test(&test).await;
+    metrics::test(&test).await;
+    tracing::test(&test).await;
+    webhooks::test(&test).await;
+
+    if test.is_reset() {
+        test.temp_dir.delete();
+    }
+}

tests/src/telemetry/tracing.rs

@@ -0,0 +1,170 @@
+/*
+ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ */
+
+use crate::utils::{server::TestServer, smtp::SmtpConnection};
+use common::telemetry::tracers::store::TracingStore;
+use registry::schema::{
+    prelude::{ObjectType, Property},
+    structs::Trace,
+};
+use std::time::Duration;
+use trc::{DeliveryEvent, EventType, SmtpEvent};
+use types::id::Id;
+
+pub async fn test(test: &TestServer) {
+    println!("Running Tracing tests...");
+
+    // Create test accounts
+    let admin = test.account("admin@example.org");
+    let account = test
+        .create_user_account(
+            "admin@example.org",
+            "jdoe@example.org",
+            "this is a very strong password",
+            &[],
+        )
+        .await;
+
+    // Make sure there are no span entries in the db
+    test.server
+        .tracing_store()
+        .purge_spans(Duration::from_secs(0), test.server.search_store().into())
+        .await
+        .unwrap();
+    assert_eq!(
+        admin
+            .registry_query(
+                ObjectType::Trace,
+                Vec::<(&str, &str)>::new(),
+                Vec::<&str>::new(),
+            )
+            .await
+            .object_ids()
+            .collect::<Vec<_>>(),
+        Vec::<Id>::new()
+    );
+
+    // Send an email
+    let mut lmtp = SmtpConnection::connect().await;
+    lmtp.ingest(
+        "bill@example.org",
+        &["jdoe@example.org"],
+        concat!(
+            "From: bill@example.org\r\n",
+            "To: jdoe@example.org\r\n",
+            "Subject: TPS Report\r\n",
+            "X-Spam-Status: No\r\n",
+            "\r\n",
+            "I'm going to need those TPS reports ASAP. ",
+            "So, if you could do that, that'd be great."
+        ),
+    )
+    .await;
+    lmtp.quit().await;
+    tokio::time::sleep(Duration::from_millis(300)).await;
+    test.server.notify_task_queue();
+    test.wait_for_tasks().await;
+
+    // There should be 2 spans
+    assert_eq!(
+        admin
+            .registry_query(
+                ObjectType::Trace,
+                Vec::<(&str, &str)>::new(),
+                Vec::<&str>::new(),
+            )
+            .await
+            .object_ids()
+            .count(),
+        2
+    );
+
+    // Purge should not delete anything at this point
+    test.server
+        .tracing_store()
+        .purge_spans(Duration::from_secs(2), test.server.search_store().into())
+        .await
+        .unwrap();
+
+    // There should be 2 spans
+    assert_eq!(
+        admin
+            .registry_query(
+                ObjectType::Trace,
+                Vec::<(&str, &str)>::new(),
+                Vec::<&str>::new(),
+            )
+            .await
+            .object_ids()
+            .count(),
+        2
+    );
+
+    // Search by spam type
+    for span_type in [
+        EventType::Delivery(DeliveryEvent::AttemptStart),
+        EventType::Smtp(SmtpEvent::ConnectionStart),
+    ] {
+        let span_ids = admin
+            .registry_query(
+                ObjectType::Trace,
+                [(Property::Event, span_type.as_str())],
+                Vec::<&str>::new(),
+            )
+            .await
+            .object_ids()
+            .collect::<Vec<_>>();
+
+        assert_eq!(span_ids.len(), 1, "{span_type:?}");
+        let trace = admin.registry_get::<Trace>(span_ids[0]).await;
+
+        assert_eq!(trace.events.iter().next().unwrap().event, span_type);
+    }
+
+    // Try searching
+    for keyword in ["bill@example.org", "jdoe@example.org", "example.org"] {
+        let span_ids = admin
+            .registry_query(
+                ObjectType::Trace,
+                [(Property::Text, keyword)],
+                Vec::<&str>::new(),
+            )
+            .await
+            .object_ids()
+            .collect::<Vec<_>>();
+
+        assert_eq!(span_ids.len(), 2, "keyword: {keyword}");
+
+        let trace_1 = admin.registry_get::<Trace>(span_ids[0]).await;
+        let trace_2 = admin.registry_get::<Trace>(span_ids[1]).await;
+
+        assert!(trace_1 != trace_2, "keyword: {keyword}");
+    }
+
+    // Purge should delete the span entries
+    tokio::time::sleep(Duration::from_millis(800)).await;
+    test.server
+        .tracing_store()
+        .purge_spans(Duration::from_secs(1), test.server.search_store().into())
+        .await
+        .unwrap();
+
+    assert_eq!(
+        admin
+            .registry_query(
+                ObjectType::Trace,
+                Vec::<(&str, &str)>::new(),
+                Vec::<&str>::new(),
+            )
+            .await
+            .object_ids()
+            .collect::<Vec<_>>(),
+        Vec::<Id>::new()
+    );
+
+    admin.destroy_account(account).await;
+    test.cleanup().await;
+}

tests/src/telemetry/webhooks.rs

@@ -0,0 +1,243 @@
+/*
+ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ */
+
+use crate::utils::server::TestServer;
+use crate::utils::smtp::SmtpConnection;
+use base64::{Engine, engine::general_purpose::STANDARD};
+use common::{manager::application::Resource, telemetry::tracers::store::TracingStore};
+use http_proto::{ToHttpResponse, request::fetch_body};
+use hyper::{body, server::conn::http1, service::service_fn};
+use hyper_util::rt::TokioIo;
+use jmap::api::ToJmapHttpResponse;
+use jmap_proto::error::request::RequestError;
+use registry::{
+    schema::{
+        enums::EventPolicy,
+        prelude::ObjectType,
+        structs::{SecretKeyOptional, SecretKeyValue, WebHook},
+    },
+    types::map::Map,
+};
+use ring::hmac;
+use std::{
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
+    time::Duration,
+};
+use store::parking_lot::Mutex;
+use tokio::{net::TcpListener, sync::watch};
+use trc::EventType;
+
+struct MockWebhookEndpoint {
+    pub _tx: watch::Sender<bool>,
+    pub events: Mutex<Vec<serde_json::Value>>,
+    pub reject: AtomicBool,
+}
+
+pub async fn test(test: &TestServer) {
+    println!("Running Webhooks tests...");
+
+    // Spawn mock webhook endpoint
+    let webhook = spawn_mock_webhook_endpoint();
+
+    // Add telemetry webhook
+    let admin = test.account("admin@example.org");
+    admin
+        .registry_create_object(WebHook {
+            enable: true,
+            url: "http://127.0.0.1:8821/hook".into(),
+            signature_key: SecretKeyOptional::Value(SecretKeyValue {
+                secret: "ovos-moles".into(),
+            }),
+            throttle: 100u64.into(),
+            allow_invalid_certs: true,
+            events: Map::new(
+                EventType::variants()
+                    .iter()
+                    .filter(|ev| {
+                        let ev = ev.as_str();
+                        ev.starts_with("smtp.connection-")
+                            || ev.starts_with("delivery.dsn")
+                            || ev.starts_with("message-ingest.")
+                    })
+                    .copied()
+                    .collect(),
+            ),
+            events_policy: EventPolicy::Include,
+            ..Default::default()
+        })
+        .await;
+    admin.reload_settings().await;
+
+    // Send test email
+    let john = test
+        .create_user_account(
+            "admin@example.org",
+            "jdoe@example.org",
+            "this is a very strong password",
+            &["john.doe@example.org"],
+        )
+        .await;
+    let mut lmtp = SmtpConnection::connect().await;
+    lmtp.ingest(
+        "bill@example.org",
+        &["jdoe@example.org"],
+        concat!(
+            "From: bill@example.org\r\n",
+            "To: jdoe@example.org\r\n",
+            "Subject: TPS Report\r\n",
+            "\r\n",
+            "I'm going to need those TPS reports ASAP. ",
+            "So, if you could do that, that'd be great."
+        ),
+    )
+    .await;
+    test.wait_for_tasks().await;
+
+    // Enable the webhook
+    webhook.assert_is_empty();
+    webhook.accept();
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // Check for events
+    webhook.assert_contains(&[
+        "smtp.connection-start",
+        "message-ingest.",
+        "delivery.dsn",
+        "\"from\": \"bill@example.org\"",
+        "\"jdoe@example.org\"",
+    ]);
+
+    // Cleanup
+    admin.registry_destroy_all(ObjectType::WebHook).await;
+    admin.reload_settings().await;
+    admin.destroy_account(john).await;
+    test.server
+        .tracing_store()
+        .purge_spans(Duration::from_secs(0), test.server.search_store().into())
+        .await
+        .unwrap();
+    test.cleanup().await;
+}
+
+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());
+    }
+}
+
+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 Webhooks 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, usize::MAX, 0).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
+                                        #[derive(serde::Deserialize)]
+                                        struct WebhookRequest {
+                                            events: Vec<serde_json::Value>,
+                                        }
+                                        let request = serde_json::from_slice::<WebhookRequest>(&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::new("application/json", "[]".to_string().into_bytes())
+                                                .into_http_response().build(),
+                                            )
+                                        } else {
+                                            //let c = print!("rejected webhook: {}", serde_json::to_string_pretty(&request).unwrap());
+
+                                            Ok::<_, hyper::Error>(
+                                                RequestError::not_found().into_http_response().build()
+                                            )
+                                        }
+
+                                    }
+                                }),
+                            )
+                            .await;
+                        }
+                        Err(err) => {
+                            panic!("Something went wrong: {err}" );
+                        }
+                    }
+                },
+                _ = rx_.changed() => {
+                    break;
+                }
+            };
+        }
+    });
+
+    endpoint_
+}