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

Login screen and api endpoint reorganization

Browse Source
This commit is contained in:
Maurus Decimus
2026-04-05 19:57:23 +02:00
parent eecec0aa31
commit fde37b670f
29 changed files with 1005 additions and 1606 deletions
Download Patch File Download Diff File Expand all files Collapse all files

725
crates/http/src/api/diagnose.rs Normal file
View File

@@ -0,0 +1,725 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use common::{
Server,
config::smtp::{
queue::{HostOrIp, MxConfig},
resolver::{Policy, Tlsa},
},
};
use hyper::body::{Bytes, Frame};
use mail_auth::{IpLookupStrategy, mta_sts::TlsRpt};
use serde::{Deserialize, Serialize};
use smtp::outbound::{
client::{SmtpClient, StartTlsResult},
dane::{dnssec::TlsaLookup, verify::TlsaVerify},
error::ClientError,
lookup::{DnsLookup, ToNextHop},
mta_sts::{lookup::MtaStsLookup, verify::VerifyPolicy},
};
use std::{
net::{IpAddr, SocketAddr},
time::{Duration, Instant},
};
use tokio::{io::AsyncWriteExt, sync::mpsc};
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(tag = "type")]
pub(crate) enum DeliveryStage {
MxLookupStart {
domain: String,
},
MxLookupSuccess {
mxs: Vec<MX>,
elapsed: u64,
},
MxLookupError {
reason: String,
elapsed: u64,
},
MtaStsFetchStart,
MtaStsFetchSuccess {
policy: Policy,
elapsed: u64,
},
MtaStsFetchError {
reason: String,
elapsed: u64,
},
MtaStsNotFound {
elapsed: u64,
},
TlsRptLookupStart,
TlsRptLookupSuccess {
rua: Vec<ReportUri>,
elapsed: u64,
},
TlsRptLookupError {
reason: String,
elapsed: u64,
},
TlsRptNotFound {
elapsed: u64,
},
DeliveryAttemptStart {
hostname: String,
},
MtaStsVerifySuccess,
MtaStsVerifyError {
reason: String,
},
TlsaLookupStart,
TlsaLookupSuccess {
record: Tlsa,
elapsed: u64,
},
TlsaNotFound {
elapsed: u64,
reason: String,
},
TlsaLookupError {
elapsed: u64,
reason: String,
},
IpLookupStart,
IpLookupSuccess {
remote_ips: Vec<IpAddr>,
elapsed: u64,
},
IpLookupError {
reason: String,
elapsed: u64,
},
ConnectionStart {
remote_ip: IpAddr,
},
ConnectionSuccess {
elapsed: u64,
},
ConnectionError {
elapsed: u64,
reason: String,
},
ReadGreetingStart,
ReadGreetingSuccess {
elapsed: u64,
},
ReadGreetingError {
elapsed: u64,
reason: String,
},
EhloStart,
EhloSuccess {
elapsed: u64,
},
EhloError {
elapsed: u64,
reason: String,
},
StartTlsStart,
StartTlsSuccess {
elapsed: u64,
},
StartTlsError {
elapsed: u64,
reason: String,
},
DaneVerifySuccess,
DaneVerifyError {
reason: String,
},
MailFromStart,
MailFromSuccess {
elapsed: u64,
},
MailFromError {
reason: String,
elapsed: u64,
},
RcptToStart,
RcptToSuccess {
elapsed: u64,
},
RcptToError {
reason: String,
elapsed: u64,
},
QuitStart,
QuitCompleted {
elapsed: u64,
},
Completed,
}
#[derive(Debug, Serialize, Deserialize)]
pub(crate) struct MX {
pub exchanges: Vec<String>,
pub preference: u16,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(tag = "type")]
pub enum ReportUri {
Mail { email: String },
Http { url: String },
}
impl DeliveryStage {
pub fn to_frame(&self) -> Frame<Bytes> {
let payload = format!(
"event: event\ndata: [{}]\n\n",
serde_json::to_string(self).unwrap_or_default()
);
Frame::data(Bytes::from(payload))
}
}
trait ElapsedMs {
fn elapsed_ms(&self) -> u64;
}
impl ElapsedMs for Instant {
fn elapsed_ms(&self) -> u64 {
self.elapsed().as_millis() as u64
}
}
pub(crate) fn spawn_delivery_diagnose(
server: Server,
domain_or_email: String,
timeout: Duration,
) -> mpsc::Receiver<DeliveryStage> {
let (tx, rx) = mpsc::channel(10);
tokio::spawn(async move {
let _ = delivery_diagnose(tx, server, domain_or_email, timeout).await;
});
rx
}
async fn delivery_diagnose(
tx: mpsc::Sender<DeliveryStage>,
server: Server,
domain_or_email: String,
timeout: Duration,
) -> Result<(), mpsc::error::SendError<DeliveryStage>> {
let (domain, email) = if let Some((_, domain)) = domain_or_email.rsplit_once('@') {
(domain.to_string(), Some(domain_or_email))
} else {
(domain_or_email, None)
};
let local_host = &server.core.network.server_name;
tx.send(DeliveryStage::MxLookupStart {
domain: domain.to_string(),
})
.await?;
// Lookup MX
let now = Instant::now();
let mxs = match server
.core
.smtp
.resolvers
.dns
.mx_lookup(&domain, Some(&server.inner.cache.dns_mx))
.await
{
Ok(mxs) => mxs,
Err(err) => {
tx.send(DeliveryStage::MxLookupError {
reason: err.to_string(),
elapsed: now.elapsed_ms(),
})
.await?;
return Ok(());
}
};
// Obtain remote host list
let mx_config = MxConfig {
max_mx: mxs.len(),
max_multi_homed: 10,
ip_lookup_strategy: IpLookupStrategy::Ipv4thenIpv6,
};
let hosts = if let Some(hosts) = mxs.to_remote_hosts(&domain, &mx_config) {
tx.send(DeliveryStage::MxLookupSuccess {
mxs: mxs
.iter()
.map(|mx| MX {
exchanges: mx.exchanges.iter().map(|e| e.to_string()).collect(),
preference: mx.preference,
})
.collect(),
elapsed: now.elapsed_ms(),
})
.await?;
hosts
} else {
tx.send(DeliveryStage::MxLookupError {
reason: "Null MX record".to_string(),
elapsed: now.elapsed_ms(),
})
.await?;
return Ok(());
};
// Fetch MTA-STS policy
let now = Instant::now();
tx.send(DeliveryStage::MtaStsFetchStart).await?;
let mta_sts_policy = match server.lookup_mta_sts_policy(&domain, timeout).await {
Ok(policy) => {
tx.send(DeliveryStage::MtaStsFetchSuccess {
policy: policy.as_ref().clone(),
elapsed: now.elapsed_ms(),
})
.await?;
Some(policy)
}
Err(err) => {
if matches!(
&err,
smtp::outbound::mta_sts::Error::Dns(mail_auth::Error::DnsRecordNotFound(_))
) {
tx.send(DeliveryStage::MtaStsNotFound {
elapsed: now.elapsed_ms(),
})
.await?;
} else {
tx.send(DeliveryStage::MtaStsFetchError {
reason: err.to_string(),
elapsed: now.elapsed_ms(),
})
.await?;
}
None
}
};
// Fetch TLS-RPT settings
let now = Instant::now();
tx.send(DeliveryStage::TlsRptLookupStart).await?;
match server
.core
.smtp
.resolvers
.dns
.txt_lookup::<TlsRpt>(
format!("_smtp._tls.{domain}."),
Some(&server.inner.cache.dns_txt),
)
.await
{
Ok(record) => {
tx.send(DeliveryStage::TlsRptLookupSuccess {
rua: record
.rua
.iter()
.map(|r| match r {
mail_auth::mta_sts::ReportUri::Mail(email) => ReportUri::Mail {
email: email.clone(),
},
mail_auth::mta_sts::ReportUri::Http(url) => {
ReportUri::Http { url: url.clone() }
}
})
.collect(),
elapsed: now.elapsed_ms(),
})
.await?;
}
Err(err) => {
if matches!(&err, mail_auth::Error::DnsRecordNotFound(_)) {
tx.send(DeliveryStage::TlsRptNotFound {
elapsed: now.elapsed_ms(),
})
.await?;
} else {
tx.send(DeliveryStage::TlsRptLookupError {
reason: err.to_string(),
elapsed: now.elapsed_ms(),
})
.await?;
}
}
}
// Try with each host
'outer: for host in hosts {
let hostname = host.hostname();
tx.send(DeliveryStage::DeliveryAttemptStart {
hostname: hostname.to_string(),
})
.await?;
// Verify MTA-STS policy
if let Some(mta_sts_policy) = &mta_sts_policy {
if mta_sts_policy.verify(hostname) {
tx.send(DeliveryStage::MtaStsVerifySuccess).await?;
} else {
tx.send(DeliveryStage::MtaStsVerifyError {
reason: "Not authorized by policy".to_string(),
})
.await?;
continue;
}
}
// Fetch TLSA record
tx.send(DeliveryStage::TlsaLookupStart).await?;
let now = Instant::now();
let dane_policy = match server.tlsa_lookup(format!("_25._tcp.{hostname}.")).await {
Ok(Some(tlsa)) if tlsa.has_end_entities => {
tx.send(DeliveryStage::TlsaLookupSuccess {
record: tlsa.as_ref().clone(),
elapsed: now.elapsed_ms(),
})
.await?;
Some(tlsa)
}
Ok(Some(_)) => {
tx.send(DeliveryStage::TlsaLookupError {
elapsed: now.elapsed_ms(),
reason: "TLSA record does not have end entities".to_string(),
})
.await?;
None
}
Ok(None) => {
tx.send(DeliveryStage::TlsaNotFound {
elapsed: now.elapsed_ms(),
reason: "No TLSA DNSSEC records found".to_string(),
})
.await?;
None
}
Err(err) => {
if matches!(&err, mail_auth::Error::DnsRecordNotFound(_)) {
tx.send(DeliveryStage::TlsaNotFound {
elapsed: now.elapsed_ms(),
reason: "No TLSA records found for MX".to_string(),
})
.await?;
} else {
tx.send(DeliveryStage::TlsaLookupError {
elapsed: now.elapsed_ms(),
reason: err.to_string(),
})
.await?;
}
None
}
};
tx.send(DeliveryStage::IpLookupStart).await?;
let now = Instant::now();
let remote_ips = match host.fqdn_hostname() {
HostOrIp::Host(hostname) => {
match server
.ip_lookup(&hostname, IpLookupStrategy::Ipv4thenIpv6, usize::MAX)
.await
{
Ok(remote_ips) if !remote_ips.is_empty() => remote_ips,
Ok(_) => {
tx.send(DeliveryStage::IpLookupError {
reason: "No IP addresses found for host".to_string(),
elapsed: now.elapsed_ms(),
})
.await?;
continue;
}
Err(err) => {
tx.send(DeliveryStage::IpLookupError {
reason: err.to_string(),
elapsed: now.elapsed_ms(),
})
.await?;
continue;
}
}
}
HostOrIp::Ip(ip) => vec![ip],
};
tx.send(DeliveryStage::IpLookupSuccess {
remote_ips: remote_ips.clone(),
elapsed: now.elapsed_ms(),
})
.await?;
for remote_ip in remote_ips {
// Start connection
tx.send(DeliveryStage::ConnectionStart { remote_ip })
.await?;
let now = Instant::now();
match SmtpClient::connect(SocketAddr::new(remote_ip, 25), timeout, 0).await {
Ok(mut client) => {
tx.send(DeliveryStage::ConnectionSuccess {
elapsed: now.elapsed_ms(),
})
.await?;
// Read greeting
tx.send(DeliveryStage::ReadGreetingStart).await?;
let now = Instant::now();
if let Err(status) = client.read_greeting(hostname).await {
tx.send(DeliveryStage::ReadGreetingError {
elapsed: now.elapsed_ms(),
reason: status.to_string(),
})
.await?;
continue;
}
tx.send(DeliveryStage::ReadGreetingSuccess {
elapsed: now.elapsed_ms(),
})
.await?;
// Say EHLO
tx.send(DeliveryStage::EhloStart).await?;
let now = Instant::now();
let capabilities = match tokio::time::timeout(timeout, async {
client
.stream
.write_all(format!("EHLO {local_host}\r\n",).as_bytes())
.await?;
client.stream.flush().await?;
client.read_ehlo().await
})
.await
{
Ok(Ok(capabilities)) => {
tx.send(DeliveryStage::EhloSuccess {
elapsed: now.elapsed_ms(),
})
.await?;
capabilities
}
Ok(Err(err)) => {
tx.send(DeliveryStage::EhloError {
elapsed: now.elapsed_ms(),
reason: err.to_string(),
})
.await?;
continue;
}
Err(_) => {
tx.send(DeliveryStage::EhloError {
elapsed: now.elapsed_ms(),
reason: "Timed out reading response".to_string(),
})
.await?;
continue;
}
};
// Start TLS
tx.send(DeliveryStage::StartTlsStart).await?;
let now = Instant::now();
let mut client = match client
.try_start_tls(
&server.inner.data.smtp_connectors.pki_verify,
hostname,
&capabilities,
)
.await
{
StartTlsResult::Success { smtp_client } => {
tx.send(DeliveryStage::StartTlsSuccess {
elapsed: now.elapsed_ms(),
})
.await?;
smtp_client
}
StartTlsResult::Error { error } => {
tx.send(DeliveryStage::StartTlsError {
elapsed: now.elapsed_ms(),
reason: error.to_string(),
})
.await?;
continue;
}
StartTlsResult::Unavailable { response, .. } => {
tx.send(DeliveryStage::StartTlsError {
elapsed: now.elapsed_ms(),
reason: response.map(|r| r.to_string()).unwrap_or_else(|| {
"STARTTLS not advertised by host".to_string()
}),
})
.await?;
continue;
}
};
// Verify DANE policy
if let Some(dane_policy) = &dane_policy {
if let Err(err) = dane_policy.verify(
0,
hostname,
client.tls_connection().peer_certificates(),
) {
tx.send(DeliveryStage::DaneVerifyError {
reason: err.to_string(),
})
.await?;
} else {
tx.send(DeliveryStage::DaneVerifySuccess).await?;
}
}
// Say EHLO again (some SMTP servers require this)
tx.send(DeliveryStage::EhloStart).await?;
let now = Instant::now();
match tokio::time::timeout(timeout, async {
client
.stream
.write_all(format!("EHLO {local_host}\r\n",).as_bytes())
.await?;
client.stream.flush().await?;
client.read_ehlo().await
})
.await
{
Ok(Ok(_)) => {
tx.send(DeliveryStage::EhloSuccess {
elapsed: now.elapsed_ms(),
})
.await?;
}
Ok(Err(err)) => {
tx.send(DeliveryStage::EhloError {
elapsed: now.elapsed_ms(),
reason: err.to_string(),
})
.await?;
continue;
}
Err(_) => {
tx.send(DeliveryStage::EhloError {
elapsed: now.elapsed_ms(),
reason: "Timed out reading response".to_string(),
})
.await?;
continue;
}
}
// Verify recipient
let mut is_success = email.is_none();
if let Some(email) = &email {
// MAIL FROM
tx.send(DeliveryStage::MailFromStart).await?;
let now = Instant::now();
match client.cmd(b"MAIL FROM:<>\r\n").await.and_then(|r| {
if r.is_positive_completion() {
Ok(r)
} else {
Err(ClientError::UnexpectedReply(r))
}
}) {
Ok(_) => {
tx.send(DeliveryStage::MailFromSuccess {
elapsed: now.elapsed_ms(),
})
.await?;
// RCPT TO
tx.send(DeliveryStage::RcptToStart).await?;
let now = Instant::now();
match client
.cmd(format!("RCPT TO:<{email}>\r\n").as_bytes())
.await
.and_then(|r| {
if r.is_positive_completion() {
Ok(r)
} else {
Err(ClientError::UnexpectedReply(r))
}
}) {
Ok(_) => {
is_success = true;
tx.send(DeliveryStage::RcptToSuccess {
elapsed: now.elapsed_ms(),
})
.await?;
}
Err(err) => {
tx.send(DeliveryStage::RcptToError {
reason: err.to_string(),
elapsed: now.elapsed_ms(),
})
.await?;
}
}
}
Err(err) => {
tx.send(DeliveryStage::MailFromError {
reason: err.to_string(),
elapsed: now.elapsed_ms(),
})
.await?;
}
}
}
// QUIT
tx.send(DeliveryStage::QuitStart).await?;
let now = Instant::now();
client.quit().await;
tx.send(DeliveryStage::QuitCompleted {
elapsed: now.elapsed_ms(),
})
.await?;
if is_success {
break 'outer;
}
}
Err(err) => {
tx.send(DeliveryStage::ConnectionError {
elapsed: now.elapsed_ms(),
reason: err.to_string(),
})
.await?;
}
}
}
}
Ok(())
}

286
crates/http/src/api/mod.rs Normal file
View File

@@ -0,0 +1,286 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
// SPDX-SnippetBegin
// SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
// SPDX-License-Identifier: LicenseRef-SEL
#[cfg(feature = "enterprise")]
pub mod telemetry;
// SPDX-SnippetEnd
pub mod diagnose;
use crate::{
api::diagnose::{DeliveryStage, spawn_delivery_diagnose},
auth::{
authenticate::Authenticator, oauth::auth::OAuthApiHandler,
permissions::PermissionsApiHandler,
},
};
use common::{
Server,
auth::{AccessToken, oauth::GrantType},
};
use http_body_util::{StreamBody, combinators::BoxBody};
use http_proto::{
HttpRequest, HttpResponse, HttpSessionData, JsonResponse, ToHttpResponse,
request::{decode_path_element, fetch_body},
};
use hyper::{Method, StatusCode, header};
use jmap::api::{ToJmapHttpResponse, ToRequestError};
use jmap_proto::error::request::RequestError;
use registry::schema::enums::Permission;
use serde_json::json;
use std::time::Duration;
use utils::url_params::UrlParams;
pub trait ManagementApi: Sync + Send {
fn handle_api_request(
&self,
req: &mut HttpRequest,
session: &HttpSessionData,
) -> impl Future<Output = trc::Result<HttpResponse>> + Send;
fn management_access_token(
&self,
req: &HttpRequest,
session: &HttpSessionData,
) -> impl Future<Output = trc::Result<AccessToken>> + Send;
}
impl ManagementApi for Server {
#[allow(unused_variables)]
async fn handle_api_request(
&self,
req: &mut HttpRequest,
session: &HttpSessionData,
) -> trc::Result<HttpResponse> {
let is_post = req.method() == Method::POST;
let body = if is_post {
fetch_body(req, 1024 * 1024, session.session_id).await
} else {
None
};
let path = req.uri().path().split('/').skip(2).collect::<Vec<_>>();
match path.first().copied().unwrap_or_default() {
"auth" if is_post => {
self.is_http_anonymous_request_allowed(session.remote_ip)
.await?;
self.handle_login_request(
session,
body.ok_or_else(|| trc::LimitEvent::SizeRequest.into_err())?,
)
.await
}
"discover" => {
if let Some(email) = path.get(2).copied() {
self.is_http_anonymous_request_allowed(session.remote_ip)
.await?;
self.handle_discover_request(req, session, decode_path_element(email).as_ref())
.await
} else {
Err(trc::ResourceEvent::NotFound.into_err())
}
}
"permissions" => {
// Authenticate request
let (_in_flight, access_token) = self.authenticate_headers(req, session).await?;
self.handle_permissions_request(&access_token).await
}
"token" => {
let access_token = self.management_access_token(req, session).await?;
let account_id = access_token.account_id();
match path.get(1).copied() {
// SPDX-SnippetBegin
// SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
// SPDX-License-Identifier: LicenseRef-SEL
#[cfg(feature = "enterprise")]
Some("tracing") if self.core.is_enterprise_edition() => {
// Validate the access token
access_token.enforce_permission(Permission::LiveTracing)?;
// Issue a live telemetry token valid for 60 seconds
Ok(JsonResponse::new(json!({
"data": self.encode_access_token(GrantType::LiveTracing, account_id, "web", 60).await?,
}))
.into_http_response())
}
#[cfg(feature = "enterprise")]
Some("metrics") if self.core.is_enterprise_edition() => {
// Validate the access token
access_token.enforce_permission(Permission::LiveMetrics)?;
// Issue a live telemetry token valid for 60 seconds
Ok(JsonResponse::new(json!({
"data": self.encode_access_token(GrantType::LiveMetrics, account_id, "web", 60).await?,
}))
.into_http_response())
}
// SPDX-SnippetEnd
Some("delivery") => {
// Validate the access token
access_token.enforce_permission(Permission::LiveDeliveryTest)?;
// Issue a live telemetry token valid for 60 seconds
Ok(JsonResponse::new(json!({
"data": self.encode_access_token(GrantType::Diagnose, account_id, "web", 60).await?,
}))
.into_http_response())
}
Some("tracing") | Some("metrics") => {
Err(trc::ResourceEvent::NotFound
.ctx(trc::Key::Details, "Enterprise feature"))
}
_ => Err(trc::ResourceEvent::NotFound.into_err()),
}
}
"live" => {
let access_token = self.management_access_token(req, session).await?;
let params = UrlParams::new(req.uri().query());
let account_id = access_token.account_id();
match (
path.get(1).copied().unwrap_or_default(),
path.get(2).copied(),
req.method(),
) {
("delivery", Some(target), &Method::GET) => {
// Validate the access token
access_token.enforce_permission(Permission::LiveDeliveryTest)?;
let timeout = Duration::from_secs(
params
.parse::<u64>("timeout")
.filter(|interval| *interval >= 1)
.unwrap_or(30),
);
let mut rx = spawn_delivery_diagnose(
self.clone(),
decode_path_element(target).to_lowercase(),
timeout,
);
Ok(HttpResponse::new(StatusCode::OK)
.with_content_type("text/event-stream")
.with_cache_control("no-store")
.with_stream_body(BoxBody::new(StreamBody::new(
async_stream::stream! {
while let Some(stage) = rx.recv().await {
yield Ok(stage.to_frame());
}
yield Ok(DeliveryStage::Completed.to_frame());
},
))))
}
// SPDX-SnippetBegin
// SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
// SPDX-License-Identifier: LicenseRef-SEL
#[cfg(feature = "enterprise")]
("traces", _, &Method::GET) if self.core.is_enterprise_edition() => {
use crate::api::telemetry::TelemetryApi;
self.handle_telemetry_api_request(req, true, &access_token)
.await
}
#[cfg(feature = "enterprise")]
("metrics", _, &Method::GET) if self.core.is_enterprise_edition() => {
use crate::api::telemetry::TelemetryApi;
self.handle_telemetry_api_request(req, false, &access_token)
.await
}
// SPDX-SnippetEnd
("traces" | "metrics", _, &Method::GET) => {
Err(trc::ResourceEvent::NotFound
.ctx(trc::Key::Details, "Enterprise feature"))
}
_ => Err(trc::ResourceEvent::NotFound.into_err()),
}
}
_ => Err(trc::ResourceEvent::NotFound.into_err()),
}
}
async fn management_access_token(
&self,
req: &HttpRequest,
session: &HttpSessionData,
) -> trc::Result<AccessToken> {
let params = UrlParams::new(req.uri().query());
if let Some(token) = params.get("token") {
// SPDX-SnippetBegin
// SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
// SPDX-License-Identifier: LicenseRef-SEL
#[cfg(feature = "enterprise")]
if self.core.is_enterprise_edition() {
let path = req.uri().path();
let (grant_type, permissions) = if path.starts_with("/api/telemetry/traces") {
(GrantType::LiveTracing, Permission::LiveTracing)
} else if path.starts_with("/api/telemetry/metrics") {
(GrantType::LiveMetrics, Permission::LiveMetrics)
} else if path.starts_with("/api/diagnose") {
(GrantType::Diagnose, Permission::LiveDeliveryTest)
} else {
return Err(trc::ResourceEvent::NotFound.into_err());
};
self.validate_access_token(grant_type.into(), token)
.await
.map(|token_info| {
AccessToken::from_permissions(token_info.account_id, [permissions])
})
} else {
self.authenticate_headers(req, session)
.await
.map(|(_, token)| token)
}
// SPDX-SnippetEnd
#[cfg(not(feature = "enterprise"))]
{
self.authenticate_headers(req, session)
.await
.map(|(_, token)| token)
}
} else {
self.authenticate_headers(req, session)
.await
.map(|(_, token)| token)
}
}
}
pub trait ToManageHttpResponse {
fn into_http_response(self) -> HttpResponse;
}
impl ToManageHttpResponse for &trc::Error {
fn into_http_response(self) -> HttpResponse {
match self.as_ref() {
trc::EventType::Auth(
trc::AuthEvent::Failed | trc::AuthEvent::Error | trc::AuthEvent::TokenExpired,
) => HttpResponse::unauthorized(true),
_ => self.to_request_error().into_http_response(),
}
}
}
pub trait UnauthorizedResponse {
fn unauthorized(include_realms: bool) -> Self;
}
impl UnauthorizedResponse for HttpResponse {
fn unauthorized(include_realms: bool) -> Self {
(if include_realms {
HttpResponse::new(StatusCode::UNAUTHORIZED)
.with_header(header::WWW_AUTHENTICATE, "Bearer realm=\"Stalwart Server\"")
.with_header(header::WWW_AUTHENTICATE, "Basic realm=\"Stalwart Server\"")
} else {
HttpResponse::new(StatusCode::UNAUTHORIZED)
})
.with_content_type("application/problem+json")
.with_text_body(serde_json::to_string(&RequestError::unauthorized()).unwrap_or_default())
}
}

280
crates/http/src/api/telemetry.rs Normal file
View File

@@ -0,0 +1,280 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: LicenseRef-SEL
*
* This file is subject to the Stalwart Enterprise License Agreement (SEL) and
* is NOT open source software.
*
*/
use common::{Server, auth::AccessToken};
use http_body_util::{StreamBody, combinators::BoxBody};
use http_proto::*;
use hyper::{
StatusCode,
body::{Bytes, Frame},
};
use mail_parser::DateTime;
use registry::schema::enums::Permission;
use std::future::Future;
use std::{
fmt::Write,
time::{Duration, Instant},
};
use store::ahash::{AHashMap, AHashSet};
use trc::{
Collector, EventType, Key, MetricType, Value,
ipc::{bitset::Bitset, subscriber::SubscriberBuilder},
serializers::json::JsonEventSerializer,
};
use utils::url_params::UrlParams;
pub trait TelemetryApi: Sync + Send {
fn handle_telemetry_api_request(
&self,
req: &HttpRequest,
is_tracing: bool,
access_token: &AccessToken,
) -> impl Future<Output = trc::Result<HttpResponse>> + Send;
}
impl TelemetryApi for Server {
async fn handle_telemetry_api_request(
&self,
req: &HttpRequest,
is_tracing: bool,
access_token: &AccessToken,
) -> trc::Result<HttpResponse> {
let params = UrlParams::new(req.uri().query());
if is_tracing {
// Validate the access token
access_token.enforce_permission(Permission::LiveTracing)?;
let mut key_filters = AHashMap::new();
let mut filter = None;
for (key, value) in params.into_inner() {
if key == "filter" {
filter = value.into_owned().into();
} else if let Some(key) = Key::parse(key.to_ascii_lowercase().as_str()) {
key_filters.insert(key, value.into_owned());
}
}
let (_, mut rx) = SubscriberBuilder::new("live-tracer".to_string())
.with_interests(Box::new(Bitset::all()))
.with_lossy(false)
.register();
let throttle = Duration::from_secs(1);
let ping_interval = Duration::from_secs(30);
let ping_payload = Bytes::from(format!(
"event: ping\ndata: {{\"interval\": {}}}\n\n",
ping_interval.as_millis()
));
let mut last_ping = Instant::now();
let mut events = Vec::new();
let mut active_span_ids = AHashSet::new();
Ok(HttpResponse::new(StatusCode::OK)
.with_content_type("text/event-stream")
.with_cache_control("no-store")
.with_stream_body(BoxBody::new(StreamBody::new(
async_stream::stream! {
let mut last_message = Instant::now() - throttle;
let mut timeout = ping_interval;
loop {
match tokio::time::timeout(timeout, rx.recv()).await {
Ok(Some(event_batch)) => {
for event in event_batch {
if (filter.is_none() && key_filters.is_empty())
|| event
.span_id()
.is_some_and(|span_id| active_span_ids.contains(&span_id))
{
events.push(event);
} else {
let mut matched_keys = AHashSet::new();
for (key, value) in event
.keys
.iter()
.chain(event.inner.span.as_ref().map_or(([]).iter(), |s| s.keys.iter()))
{
if let Some(needle) = key_filters.get(key).or(filter.as_ref()) {
let matches = match value {
Value::String(haystack) => haystack.contains(needle),
Value::Timestamp(haystack) => {
DateTime::from_timestamp(*haystack as i64)
.to_rfc3339()
.contains(needle)
}
Value::Bool(true) => needle == "true",
Value::Bool(false) => needle == "false",
Value::Ipv4(haystack) => haystack.to_string().contains(needle),
Value::Ipv6(haystack) => haystack.to_string().contains(needle),
Value::Event(_) |
Value::Array(_) |
Value::UInt(_) |
Value::Int(_) |
Value::Float(_) |
Value::Duration(_) |
Value::Bytes(_) |
Value::None => false,
};
if matches {
matched_keys.insert(*key);
if filter.is_some() || matched_keys.len() == key_filters.len() {
if let Some(span_id) = event.span_id() {
active_span_ids.insert(span_id);
}
events.push(event);
break;
}
}
}
}
}
}
}
Ok(None) => {
break;
}
Err(_) => (),
}
timeout = if !events.is_empty() {
let elapsed = last_message.elapsed();
if elapsed >= throttle {
last_message = Instant::now();
yield Ok(Frame::data(Bytes::from(format!(
"event: trace\ndata: {}\n\n",
serde_json::to_string(
&JsonEventSerializer::new(std::mem::take(&mut events))
.with_description()).unwrap_or_default()
))));
ping_interval
} else {
throttle - elapsed
}
} else {
let elapsed = last_ping.elapsed();
if elapsed >= ping_interval {
last_ping = Instant::now();
yield Ok(Frame::data(ping_payload.clone()));
ping_interval
} else {
ping_interval - elapsed
}
};
}
},
))))
} else {
// Validate the access token
access_token.enforce_permission(Permission::LiveMetrics)?;
let interval = Duration::from_secs(
params
.parse::<u64>("interval")
.filter(|interval| *interval >= 1)
.unwrap_or(30),
);
let mut event_types = AHashSet::new();
let mut metric_types = AHashSet::new();
for metric_name in params.get("metrics").unwrap_or_default().split(',') {
let metric_name = metric_name.trim();
if !metric_name.is_empty() {
if let Some(event_type) = EventType::parse(metric_name) {
event_types.insert(event_type);
} else if let Some(metric_type) = MetricType::parse(metric_name) {
metric_types.insert(metric_type);
}
}
}
// Refresh expensive metrics
for metric_type in [
MetricType::QueueCount,
MetricType::UserCount,
MetricType::DomainCount,
] {
if metric_types.contains(&metric_type) {
let value = match metric_type {
MetricType::QueueCount => self.total_queued_messages().await?,
MetricType::UserCount => self.total_accounts().await? as u64,
MetricType::DomainCount => self.total_domains().await? as u64,
_ => unreachable!(),
};
Collector::update_gauge(metric_type, value);
}
}
Ok(HttpResponse::new(StatusCode::OK)
.with_content_type("text/event-stream")
.with_cache_control("no-store")
.with_stream_body(BoxBody::new(StreamBody::new(
async_stream::stream! {
loop {
let mut metrics = String::with_capacity(512);
metrics.push_str("event: metrics\ndata: [");
let mut is_first = true;
for counter in Collector::collect_counters(true) {
if event_types.is_empty() || event_types.contains(&counter.id()) {
if !is_first {
metrics.push(',');
} else {
is_first = false;
}
let _ = write!(
&mut metrics,
"{{\"id\":\"{}\",\"type\":\"counter\",\"value\":{}}}",
counter.id().as_str(),
counter.value()
);
}
}
for gauge in Collector::collect_gauges(true) {
if metric_types.is_empty() || metric_types.contains(&gauge.id()) {
if !is_first {
metrics.push(',');
} else {
is_first = false;
}
let _ = write!(
&mut metrics,
"{{\"id\":\"{}\",\"type\":\"gauge\",\"value\":{}}}",
gauge.id().as_str(),
gauge.get()
);
}
}
for histogram in Collector::collect_histograms(true) {
if metric_types.is_empty() || metric_types.contains(&histogram.id()) {
if !is_first {
metrics.push(',');
} else {
is_first = false;
}
let _ = write!(
&mut metrics,
"{{\"id\":\"{}\",\"type\":\"histogram\",\"count\":{},\"sum\":{}}}",
histogram.id().as_str(),
histogram.count(),
histogram.sum()
);
}
}
metrics.push_str("]\n\n");
yield Ok(Frame::data(Bytes::from(metrics)));
tokio::time::sleep(interval).await;
}
},
))))
}
}
}