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

JMAP Registry API implementation - part 10

Browse Source
This commit is contained in:
mdecimus
2026-03-06 19:50:10 +01:00
parent 9f9ddee965
commit d15efc6fb9
67 changed files with 2186 additions and 1879 deletions
Download Patch File Download Diff File Expand all files Collapse all files

502
crates/http/src/management/diagnose.rs
View File

@@ -4,129 +4,32 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::{
future::Future,
net::{IpAddr, SocketAddr},
time::{Duration, Instant},
};
use common::{
Server,
auth::{AccessToken, oauth::GrantType},
config::smtp::{
queue::MxConfig,
resolver::{Policy, Tlsa},
},
psl,
};
use http_body_util::{StreamBody, combinators::BoxBody};
use hyper::{
Method, StatusCode,
body::{Bytes, Frame},
};
use mail_auth::{
AuthenticatedMessage, DkimResult, DmarcResult, IpLookupStrategy, IprevOutput, IprevResult,
SpfOutput, SpfResult,
dmarc::{self, verify::DmarcParameters},
mta_sts::TlsRpt,
spf::verify::SpfParameters,
};
use hyper::body::{Bytes, Frame};
use mail_auth::{IpLookupStrategy, mta_sts::TlsRpt};
use serde::{Deserialize, Serialize};
use serde_json::json;
use smtp::outbound::{
client::{SmtpClient, StartTlsResult},
dane::{dnssec::TlsaLookup, verify::TlsaVerify},
lookup::{DnsLookup, ToNextHop},
mta_sts::{lookup::MtaStsLookup, verify::VerifyPolicy},
};
use std::{
net::{IpAddr, SocketAddr},
time::{Duration, Instant},
};
use tokio::{io::AsyncWriteExt, sync::mpsc};
use utils::url_params::UrlParams;
use http_proto::{request::decode_path_element, *};
pub trait TroubleshootApi: Sync + Send {
fn handle_diagnose_api_request(
&self,
req: &HttpRequest,
path: Vec<&str>,
access_token: &AccessToken,
body: Option<Vec<u8>>,
) -> impl Future<Output = trc::Result<HttpResponse>> + Send;
}
impl TroubleshootApi for Server {
async fn handle_diagnose_api_request(
&self,
req: &HttpRequest,
path: Vec<&str>,
access_token: &AccessToken,
body: Option<Vec<u8>>,
) -> trc::Result<HttpResponse> {
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(),
) {
("token", None, &Method::GET) => {
// 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())
}
("delivery", Some(target), &Method::GET) => {
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());
}))))
}
("dmarc", None, &Method::POST) => {
let request = serde_json::from_slice::<DmarcTroubleshootRequest>(
body.as_deref().unwrap_or_default(),
)
.map_err(|err| {
trc::EventType::Resource(trc::ResourceEvent::BadParameters).from_json_error(err)
})?;
let response = dmarc_diagnose(self, request).await.ok_or_else(|| {
trc::EventType::Resource(trc::ResourceEvent::BadParameters)
.reason("Failed to parse message body")
})?;
Ok(JsonResponse::new(json!({
"data": response,
}))
.into_http_response())
}
_ => Err(trc::ResourceEvent::NotFound.into_err()),
}
}
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(tag = "type")]
enum DeliveryStage {
pub(crate) enum DeliveryStage {
MxLookupStart {
domain: String,
},
@@ -253,7 +156,7 @@ enum DeliveryStage {
}
#[derive(Debug, Serialize, Deserialize)]
struct MX {
pub(crate) struct MX {
pub exchanges: Vec<String>,
pub preference: u16,
}
@@ -267,7 +170,7 @@ pub enum ReportUri {
}
impl DeliveryStage {
fn to_frame(&self) -> Frame<Bytes> {
pub fn to_frame(&self) -> Frame<Bytes> {
let payload = format!(
"event: event\ndata: [{}]\n\n",
serde_json::to_string(self).unwrap_or_default()
@@ -285,7 +188,7 @@ impl ElapsedMs for Instant {
self.elapsed().as_millis() as u64
}
}
fn spawn_delivery_diagnose(
pub(crate) fn spawn_delivery_diagnose(
server: Server,
domain_or_email: String,
timeout: Duration,
@@ -816,388 +719,3 @@ async fn delivery_diagnose(
Ok(())
}
#[derive(Debug, Serialize, Deserialize)]
struct DmarcTroubleshootRequest {
#[serde(rename = "remoteIp")]
remote_ip: IpAddr,
#[serde(rename = "ehloDomain")]
ehlo_domain: String,
#[serde(rename = "mailFrom")]
mail_from: String,
body: Option<String>,
}
#[derive(Debug, Serialize, Deserialize)]
struct DmarcTroubleshootResponse {
#[serde(rename = "spfEhloDomain")]
spf_ehlo_domain: String,
#[serde(rename = "spfEhloResult")]
spf_ehlo_result: AuthResult,
#[serde(rename = "spfMailFromDomain")]
spf_mail_from_domain: String,
#[serde(rename = "spfMailFromResult")]
spf_mail_from_result: AuthResult,
#[serde(rename = "ipRevResult")]
ip_rev_result: AuthResult,
#[serde(rename = "ipRevPtr")]
ip_rev_ptr: Vec<String>,
#[serde(rename = "dkimResults")]
dkim_results: Vec<AuthResult>,
#[serde(rename = "dkimPass")]
dkim_pass: bool,
#[serde(rename = "arcResult")]
arc_result: AuthResult,
#[serde(rename = "dmarcResult")]
dmarc_result: AuthResult,
#[serde(rename = "dmarcPass")]
dmarc_pass: bool,
#[serde(rename = "dmarcPolicy")]
dmarc_policy: DmarcPolicy,
elapsed: u64,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(tag = "type")]
pub enum AuthResult {
Pass,
Fail { details: Option<String> },
SoftFail { details: Option<String> },
TempError { details: Option<String> },
PermError { details: Option<String> },
Neutral { details: Option<String> },
None,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum DmarcPolicy {
None,
Quarantine,
Reject,
Unspecified,
}
async fn dmarc_diagnose(
server: &Server,
request: DmarcTroubleshootRequest,
) -> Option<DmarcTroubleshootResponse> {
let remote_ip = request.remote_ip;
let ehlo_domain = request.ehlo_domain.to_lowercase();
let mail_from = request.mail_from.to_lowercase();
let mail_from_domain = mail_from.rsplit_once('@').map(|(_, domain)| domain);
let local_host = &server.core.network.server_name;
let now = Instant::now();
let ehlo_spf_output = server
.core
.smtp
.resolvers
.dns
.verify_spf(
server
.inner
.cache
.build_auth_parameters(SpfParameters::verify_ehlo(
remote_ip,
&ehlo_domain,
local_host,
)),
)
.await;
let iprev = server
.core
.smtp
.resolvers
.dns
.verify_iprev(server.inner.cache.build_auth_parameters(remote_ip))
.await;
let mail_spf_output = if let Some(mail_from_domain) = mail_from_domain {
server
.core
.smtp
.resolvers
.dns
.check_host(server.inner.cache.build_auth_parameters(SpfParameters::new(
remote_ip,
mail_from_domain,
&ehlo_domain,
local_host,
&mail_from,
)))
.await
} else {
server
.core
.smtp
.resolvers
.dns
.check_host(server.inner.cache.build_auth_parameters(SpfParameters::new(
remote_ip,
&ehlo_domain,
&ehlo_domain,
local_host,
&format!("postmaster@{ehlo_domain}"),
)))
.await
};
let body = request
.body
.unwrap_or_else(|| format!("From: {mail_from}\r\nSubject: test\r\n\r\ntest"));
let auth_message = AuthenticatedMessage::parse_with_opts(body.as_bytes(), true)?;
let dkim_output = server
.core
.smtp
.resolvers
.dns
.verify_dkim(server.inner.cache.build_auth_parameters(&auth_message))
.await;
let dkim_pass = dkim_output
.iter()
.any(|d| matches!(d.result(), DkimResult::Pass));
let arc_output = server
.core
.smtp
.resolvers
.dns
.verify_arc(server.inner.cache.build_auth_parameters(&auth_message))
.await;
let dmarc_output = server
.core
.smtp
.resolvers
.dns
.verify_dmarc(server.inner.cache.build_auth_parameters(DmarcParameters {
message: &auth_message,
dkim_output: &dkim_output,
rfc5321_mail_from_domain: mail_from_domain.unwrap_or(ehlo_domain.as_str()),
spf_output: &mail_spf_output,
domain_suffix_fn: |domain| psl::domain_str(domain).unwrap_or(domain),
}))
.await;
let dmarc_pass = matches!(dmarc_output.spf_result(), DmarcResult::Pass)
|| matches!(dmarc_output.dkim_result(), DmarcResult::Pass);
let dmarc_result = if dmarc_pass {
DmarcResult::Pass
} else if dmarc_output.spf_result() != &DmarcResult::None {
dmarc_output.spf_result().clone()
} else if dmarc_output.dkim_result() != &DmarcResult::None {
dmarc_output.dkim_result().clone()
} else {
DmarcResult::None
};
Some(DmarcTroubleshootResponse {
spf_ehlo_domain: ehlo_spf_output.domain().to_string(),
spf_ehlo_result: (&ehlo_spf_output).into(),
spf_mail_from_domain: mail_spf_output.domain().to_string(),
spf_mail_from_result: (&mail_spf_output).into(),
ip_rev_ptr: iprev
.ptr
.as_ref()
.map(|ptr| ptr.iter().map(|s| s.to_string()).collect())
.unwrap_or_default(),
ip_rev_result: (&iprev).into(),
dkim_pass,
dkim_results: dkim_output
.iter()
.map(|result| result.result().into())
.collect(),
arc_result: arc_output.result().into(),
dmarc_result: (&dmarc_result).into(),
dmarc_policy: (&dmarc_output.policy()).into(),
dmarc_pass,
elapsed: now.elapsed_ms(),
})
}
impl From<&SpfOutput> for AuthResult {
fn from(value: &SpfOutput) -> Self {
match value.result() {
SpfResult::Pass => AuthResult::Pass,
SpfResult::Fail => AuthResult::Fail {
details: value.explanation().map(|e| e.to_string()),
},
SpfResult::SoftFail => AuthResult::SoftFail {
details: value.explanation().map(|e| e.to_string()),
},
SpfResult::Neutral => AuthResult::Neutral {
details: value.explanation().map(|e| e.to_string()),
},
SpfResult::TempError => AuthResult::TempError {
details: value.explanation().map(|e| e.to_string()),
},
SpfResult::PermError => AuthResult::PermError {
details: value.explanation().map(|e| e.to_string()),
},
SpfResult::None => AuthResult::None,
}
}
}
impl From<AuthResult> for SpfOutput {
fn from(value: AuthResult) -> Self {
match value {
AuthResult::Pass => SpfOutput::new(String::new()).with_result(SpfResult::Pass),
AuthResult::Fail { .. } => SpfOutput::new(String::new()).with_result(SpfResult::Fail),
AuthResult::SoftFail { .. } => {
SpfOutput::new(String::new()).with_result(SpfResult::SoftFail)
}
AuthResult::Neutral { .. } => {
SpfOutput::new(String::new()).with_result(SpfResult::Neutral)
}
AuthResult::TempError { .. } => {
SpfOutput::new(String::new()).with_result(SpfResult::TempError)
}
AuthResult::PermError { .. } => {
SpfOutput::new(String::new()).with_result(SpfResult::PermError)
}
AuthResult::None => SpfOutput::new(String::new()).with_result(SpfResult::None),
}
}
}
impl From<&IprevOutput> for AuthResult {
fn from(value: &IprevOutput) -> Self {
match &value.result {
IprevResult::Pass => AuthResult::Pass,
IprevResult::Fail(error) => AuthResult::Fail {
details: error.to_string().into(),
},
IprevResult::TempError(error) => AuthResult::TempError {
details: error.to_string().into(),
},
IprevResult::PermError(error) => AuthResult::PermError {
details: error.to_string().into(),
},
IprevResult::None => AuthResult::None,
}
}
}
impl From<AuthResult> for IprevResult {
fn from(value: AuthResult) -> Self {
match value {
AuthResult::Pass => IprevResult::Pass,
AuthResult::Fail { details } => {
IprevResult::Fail(mail_auth::Error::Io(details.unwrap_or_default()))
}
AuthResult::TempError { details } => {
IprevResult::TempError(mail_auth::Error::Io(details.unwrap_or_default()))
}
AuthResult::PermError { details } => {
IprevResult::PermError(mail_auth::Error::Io(details.unwrap_or_default()))
}
AuthResult::None => IprevResult::None,
_ => IprevResult::None,
}
}
}
impl From<&DkimResult> for AuthResult {
fn from(value: &DkimResult) -> Self {
match value {
DkimResult::Pass => AuthResult::Pass,
DkimResult::Neutral(error) => AuthResult::Neutral {
details: error.to_string().into(),
},
DkimResult::Fail(error) => AuthResult::Fail {
details: error.to_string().into(),
},
DkimResult::PermError(error) => AuthResult::PermError {
details: error.to_string().into(),
},
DkimResult::TempError(error) => AuthResult::TempError {
details: error.to_string().into(),
},
DkimResult::None => AuthResult::None,
}
}
}
impl From<AuthResult> for DkimResult {
fn from(value: AuthResult) -> Self {
match value {
AuthResult::Pass => DkimResult::Pass,
AuthResult::Neutral { details } => {
DkimResult::Neutral(mail_auth::Error::Io(details.unwrap_or_default()))
}
AuthResult::Fail { details } => {
DkimResult::Fail(mail_auth::Error::Io(details.unwrap_or_default()))
}
AuthResult::PermError { details } => {
DkimResult::PermError(mail_auth::Error::Io(details.unwrap_or_default()))
}
AuthResult::TempError { details } => {
DkimResult::TempError(mail_auth::Error::Io(details.unwrap_or_default()))
}
_ => DkimResult::None,
}
}
}
impl From<&DmarcResult> for AuthResult {
fn from(value: &DmarcResult) -> Self {
match value {
DmarcResult::Pass => AuthResult::Pass,
DmarcResult::Fail(error) => AuthResult::Fail {
details: error.to_string().into(),
},
DmarcResult::TempError(error) => AuthResult::TempError {
details: error.to_string().into(),
},
DmarcResult::PermError(error) => AuthResult::PermError {
details: error.to_string().into(),
},
DmarcResult::None => AuthResult::None,
}
}
}
impl From<AuthResult> for DmarcResult {
fn from(value: AuthResult) -> Self {
match value {
AuthResult::Pass => DmarcResult::Pass,
AuthResult::Fail { details } => {
DmarcResult::Fail(mail_auth::Error::Io(details.unwrap_or_default()))
}
AuthResult::TempError { details } => {
DmarcResult::TempError(mail_auth::Error::Io(details.unwrap_or_default()))
}
AuthResult::PermError { details } => {
DmarcResult::PermError(mail_auth::Error::Io(details.unwrap_or_default()))
}
AuthResult::None => DmarcResult::None,
_ => DmarcResult::None,
}
}
}
impl From<&dmarc::Policy> for DmarcPolicy {
fn from(value: &dmarc::Policy) -> Self {
match value {
dmarc::Policy::None => DmarcPolicy::None,
dmarc::Policy::Quarantine => DmarcPolicy::Quarantine,
dmarc::Policy::Reject => DmarcPolicy::Reject,
dmarc::Policy::Unspecified => DmarcPolicy::Unspecified,
}
}
}
impl From<DmarcPolicy> for dmarc::Policy {
fn from(value: DmarcPolicy) -> Self {
match value {
DmarcPolicy::None => dmarc::Policy::None,
DmarcPolicy::Quarantine => dmarc::Policy::Quarantine,
DmarcPolicy::Reject => dmarc::Policy::Reject,
DmarcPolicy::Unspecified => dmarc::Policy::Unspecified,
}
}
}

149
crates/http/src/management/mod.rs
View File

@@ -12,13 +12,23 @@ pub mod telemetry;
// SPDX-SnippetEnd
pub mod diagnose;
use crate::management::diagnose::TroubleshootApi;
use common::{Server, auth::AccessToken};
use http_proto::{HttpRequest, HttpResponse, HttpSessionData, request::fetch_body};
use hyper::{StatusCode, header};
use crate::management::diagnose::{DeliveryStage, spawn_delivery_diagnose};
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_manage_request(
@@ -41,37 +51,116 @@ impl ManagementApi for Server {
let path = req.uri().path().split('/').skip(2).collect::<Vec<_>>();
match path.first().copied().unwrap_or_default() {
"diagnose" => {
// Validate the access token
access_token.enforce_permission(Permission::Troubleshoot)?;
"token" => {
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::TracingLive)?;
self.handle_diagnose_api_request(req, path, access_token, body)
.await
}
// SPDX-SnippetBegin
// SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
// SPDX-License-Identifier: LicenseRef-SEL
#[cfg(feature = "enterprise")]
"telemetry" => {
// WARNING: TAMPERING WITH THIS FUNCTION IS STRICTLY PROHIBITED
// Any attempt to modify, bypass, or disable this license validation mechanism
// constitutes a severe violation of the Stalwart Enterprise License Agreement.
// Such actions may result in immediate termination of your license, legal action,
// and substantial financial penalties. Stalwart Labs LLC actively monitors for
// unauthorized modifications and will pursue all available legal remedies against
// violators to the fullest extent of the law, including but not limited to claims
// for copyright infringement, breach of contract, and fraud.
// 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::MetricsLive)?;
if self.core.is_enterprise_edition() {
use crate::management::telemetry::TelemetryApi;
// 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::Troubleshoot)?;
self.handle_telemetry_api_request(req, path, access_token)
.await
} else {
Err(trc::ResourceEvent::NotFound.ctx(trc::Key::Details, "Enterprise feature"))
// 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()),
}
}
// SPDX-SnippetEnd
"live" => {
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::Troubleshoot)?;
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::management::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::management::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()),
}
}

163
crates/http/src/management/telemetry.rs
View File

@@ -8,19 +8,15 @@
*
*/
use common::{
Server,
auth::{AccessToken, oauth::GrantType},
};
use common::{Server, auth::AccessToken};
use http_body_util::{StreamBody, combinators::BoxBody};
use http_proto::*;
use hyper::{
Method, StatusCode,
StatusCode,
body::{Bytes, Frame},
};
use mail_parser::DateTime;
use registry::schema::enums::Permission;
use serde_json::json;
use std::future::Future;
use std::{
fmt::Write,
@@ -38,7 +34,7 @@ pub trait TelemetryApi: Sync + Send {
fn handle_telemetry_api_request(
&self,
req: &HttpRequest,
path: Vec<&str>,
is_tracing: bool,
access_token: &AccessToken,
) -> impl Future<Output = trc::Result<HttpResponse>> + Send;
}
@@ -47,48 +43,40 @@ impl TelemetryApi for Server {
async fn handle_telemetry_api_request(
&self,
req: &HttpRequest,
path: Vec<&str>,
is_tracing: bool,
access_token: &AccessToken,
) -> trc::Result<HttpResponse> {
let params = UrlParams::new(req.uri().query());
let account_id = access_token.account_id();
let todo = "use same format as in JMAP API";
if is_tracing {
// Validate the access token
access_token.enforce_permission(Permission::TracingLive)?;
match (
path.get(1).copied().unwrap_or_default(),
path.get(2).copied(),
req.method(),
) {
("traces", Some("live"), &Method::GET) => {
// Validate the access token
access_token.enforce_permission(Permission::TracingLive)?;
let mut key_filters = AHashMap::new();
let mut filter = None;
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::try_parse(key.to_ascii_lowercase().as_str()) {
key_filters.insert(key, value.into_owned());
}
for (key, value) in params.into_inner() {
if key == "filter" {
filter = value.into_owned().into();
} else if let Some(key) = Key::try_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();
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)
Ok(HttpResponse::new(StatusCode::OK)
.with_content_type("text/event-stream")
.with_cache_control("no-store")
.with_stream_body(BoxBody::new(StreamBody::new(
@@ -185,68 +173,47 @@ impl TelemetryApi for Server {
}
},
))))
}
("live", Some("tracing-token"), &Method::GET) => {
// Validate the access token
access_token.enforce_permission(Permission::TracingLive)?;
} else {
// Validate the access token
access_token.enforce_permission(Permission::MetricsLive)?;
// 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())
}
("live", Some("metrics-token"), &Method::GET) => {
// Validate the access token
access_token.enforce_permission(Permission::MetricsLive)?;
// 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())
}
("metrics", Some("live"), &Method::GET) => {
// Validate the access token
access_token.enforce_permission(Permission::MetricsLive)?;
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);
}
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);
}
// 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)
Ok(HttpResponse::new(StatusCode::OK)
.with_content_type("text/event-stream")
.with_cache_control("no-store")
.with_stream_body(BoxBody::new(StreamBody::new(
@@ -309,8 +276,6 @@ impl TelemetryApi for Server {
}
},
))))
}
_ => Err(trc::ResourceEvent::NotFound.into_err()),
}
}
}