From e774bc8c9aa5b479a93a9bf9c305daa9508ea41d Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Tue, 25 Aug 2026 12:59:41 +0300 Subject: [PATCH] WEB Debug + Trace Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com> --- src/api/mod.rs | 17 +- src/api/reload_tests.rs | 1 + src/api/web_status.rs | 540 ++++++++++++++++++ src/api/web_status/query.rs | 179 ++++++ src/api/web_status/tests.rs | 82 +++ src/config/hot_reload.rs | 2 +- src/config/hot_reload/fields.rs | 5 + src/config/hot_reload/reporting.rs | 9 + src/config/hot_reload/tests.rs | 36 ++ src/config/load/runtime_web.rs | 11 + src/config/load/strict_keys.rs | 17 +- src/config/load/strict_keys/check.rs | 8 + src/config/load/validate_web.rs | 41 +- src/config/load/validate_web/debug.rs | 31 + src/config/load/validate_web/memory.rs | 69 +++ .../tests/load_basic_tests/web_tests.rs | 47 ++ src/config/types.rs | 4 + src/config/types/web.rs | 17 + src/config/types/web_debug.rs | 100 ++++ src/maestro/listeners/control.rs | 17 +- src/maestro/orchestrator.rs | 11 +- src/maestro/reload_supervisor.rs | 5 + src/maestro/reload_supervisor_tests.rs | 9 + src/maestro/runtime_build.rs | 6 +- src/maestro/runtime_build_tests.rs | 21 + src/web/http.rs | 310 +++++----- src/web/http/activity.rs | 47 +- src/web/http/body.rs | 80 ++- src/web/http/decoy.rs | 1 + src/web/http/down.rs | 106 ++++ src/web/http/response.rs | 88 +++ src/web/http/tests.rs | 7 +- src/web/http/trace_tests.rs | 75 +++ src/web/manager.rs | 257 ++------- src/web/manager/credentials.rs | 280 +++++++++ src/web/manager/state.rs | 3 + src/web/mod.rs | 2 + src/web/session.rs | 32 ++ src/web/session/backend.rs | 74 ++- src/web/session/backend_tests.rs | 2 + src/web/session/downlink.rs | 2 + src/web/session/lanes/tests.rs | 2 + src/web/session/uplink.rs | 2 + src/web/trace/exchange.rs | 487 ++++++++++++++++ src/web/trace/mod.rs | 17 + src/web/trace/sanitize.rs | 297 ++++++++++ src/web/trace/store.rs | 442 ++++++++++++++ src/web/trace/types.rs | 281 +++++++++ 48 files changed, 3744 insertions(+), 435 deletions(-) create mode 100644 src/api/web_status.rs create mode 100644 src/api/web_status/query.rs create mode 100644 src/api/web_status/tests.rs create mode 100644 src/config/load/validate_web/debug.rs create mode 100644 src/config/load/validate_web/memory.rs create mode 100644 src/config/types/web_debug.rs create mode 100644 src/web/http/down.rs create mode 100644 src/web/http/response.rs create mode 100644 src/web/http/trace_tests.rs create mode 100644 src/web/manager/credentials.rs create mode 100644 src/web/trace/exchange.rs create mode 100644 src/web/trace/mod.rs create mode 100644 src/web/trace/sanitize.rs create mode 100644 src/web/trace/store.rs create mode 100644 src/web/trace/types.rs diff --git a/src/api/mod.rs b/src/api/mod.rs index 40bfdc7..cee3e82 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -30,6 +30,7 @@ use crate::startup::StartupTracker; use crate::stats::Stats; use crate::transport::UpstreamManager; use crate::transport::middle_proxy::MePool; +use crate::web::trace::WebTraceStore; mod config_edit; pub(crate) mod config_store; @@ -47,6 +48,7 @@ mod runtime_stats; mod runtime_watch; mod runtime_zero; mod users; +mod web_status; use config_store::{ current_revision, ensure_expected_revision, load_config_for_reload, load_config_from_disk, @@ -122,6 +124,7 @@ pub(super) struct ApiShared { pub(super) proxy_shared: Arc, pub(super) reload_control: ReloadControl, pub(super) active_runtime: Arc>, + pub(super) web_trace: Arc, } impl ApiShared { @@ -155,6 +158,7 @@ impl ApiShared { proxy_shared: runtime.proxy_shared.clone(), reload_control: self.reload_control.clone(), active_runtime: self.active_runtime.clone(), + web_trace: self.web_trace.clone(), } } } @@ -243,7 +247,8 @@ fn allowed_methods_for_path(path: &str) -> Option<&'static str> { | "/v1/runtime/tls-fingerprints" | "/v1/stats/users/active-ips" | "/v1/stats/users/quota" - | "/v1/stats/users" => Some(ALLOW_GET), + | "/v1/stats/users" + | "/web-status" => Some(ALLOW_GET), "/v1/system/reload" => Some(ALLOW_POST), "/v1/users" => Some(ALLOW_GET_POST), "/v1/config" => Some(ALLOW_GET_PATCH), @@ -279,6 +284,7 @@ pub async fn serve( reload_control: ReloadControl, mut active_runtime_rx: watch::Receiver>>>, mut runtime_watch_rx: watch::Receiver>, + web_trace: Arc, ) { let active_runtime = loop { if let Some(active_runtime) = active_runtime_rx.borrow().clone() { @@ -312,7 +318,7 @@ pub async fn serve( } }; - info!("API endpoint: http://{}/v1/*", listen); + info!("API endpoint: http://{}/v1/* and /web-status", listen); let runtime_state = Arc::new(ApiRuntimeState { process_started_at_epoch_secs, @@ -344,6 +350,7 @@ pub async fn serve( proxy_shared, reload_control, active_runtime, + web_trace, }); spawn_runtime_watchers( @@ -492,6 +499,12 @@ async fn handle( let result: Result>, ApiFailure> = async { match (method.as_str(), normalized_path) { + ("GET", "/web-status") => Ok(web_status::render( + query.as_deref(), + &shared.web_trace, + &cfg.web.debug, + ) + .await), ("GET", "/v1/health") => { let revision = current_revision(&shared.config_path).await?; let data = HealthData { diff --git a/src/api/reload_tests.rs b/src/api/reload_tests.rs index 0e62141..6d35607 100644 --- a/src/api/reload_tests.rs +++ b/src/api/reload_tests.rs @@ -112,6 +112,7 @@ fn reload_routes_expose_only_documented_methods_and_ids() { Some(ALLOW_GET) ); assert_eq!(reload_status_route_id("/v1/system/reload/42"), Some(42)); + assert_eq!(allowed_methods_for_path("/web-status"), Some(ALLOW_GET)); assert_eq!( reload_status_route_id("/v1/system/reload/not-a-number"), None diff --git a/src/api/web_status.rs b/src/api/web_status.rs new file mode 100644 index 0000000..bee469c --- /dev/null +++ b/src/api/web_status.rs @@ -0,0 +1,540 @@ +use std::collections::BTreeMap; +use std::sync::Arc; + +use base64::Engine as _; +use http_body_util::Full; +use hyper::body::Bytes; +use hyper::header::{self, HeaderValue}; +use hyper::{Response, StatusCode}; +use tokio::sync::OwnedSemaphorePermit; + +use crate::config::WebDebugConfig; +use crate::web::trace::{ + StoredTraceRecord, TraceRecord, TraceRecordKind, WebTraceStore, +}; + +const MAX_PAGE_BYTES: usize = 8 * 1024 * 1024; +const MAX_GROUPS: usize = 1024; + +// Query parsing and matching remain independent from bounded HTML rendering. +mod query; + +use query::{GroupBy, StatusQuery, client_ip, parse_query, record_matches}; + +struct GroupSummary { + count: usize, + latest_seq: u64, +} + +struct RenderedPage { + html: String, + _permit: OwnedSemaphorePermit, +} + +impl AsRef<[u8]> for RenderedPage { + fn as_ref(&self) -> &[u8] { + self.html.as_bytes() + } +} + +/// Renders the authenticated server-side WEB debugging table. +pub(super) async fn render( + raw_query: Option<&str>, + store: &Arc, + policy: &WebDebugConfig, +) -> Response> { + store.apply_policy(policy); + let query = match parse_query(raw_query, policy) { + Ok(query) => query, + Err(error) => return html_error(StatusCode::BAD_REQUEST, "Invalid query", &error), + }; + let Some(render_permit) = store.try_render_permit() else { + return html_error( + StatusCode::SERVICE_UNAVAILABLE, + "Renderer busy", + "Two WEB status pages are already rendering", + ); + }; + let now_millis = crate::web::trace::store_epoch_millis(); + let since_millis = query + .record + .is_none() + .then(|| now_millis.saturating_sub(query.window_secs.saturating_mul(1000))) + .unwrap_or(0); + let records = store.snapshot_matching(|record| record_matches(record, &query, since_millis)); + let status = store.status(); + let mut html = String::with_capacity(MAX_PAGE_BYTES); + push_page_start(&mut html); + html.push_str("

WEB status

"); + push_filter_form(&mut html, &query); + html.push_str("

Store

"); + summary_row(&mut html, "debug enabled", yes_no(status.policy.enabled)); + summary_row(&mut html, "body capture", body_mode(&status.policy)); + summary_row(&mut html, "window seconds", &query.window_secs.to_string()); + summary_row( + &mut html, + "records", + &format!("{} / {}", status.records, status.records_capacity), + ); + summary_row( + &mut html, + "bytes", + &format!("{} / {}", status.used_bytes, status.bytes_capacity), + ); + summary_row(&mut html, "matched", &records.len().to_string()); + summary_row( + &mut html, + "contention drops", + &status.contention_drops.to_string(), + ); + summary_row(&mut html, "evictions", &status.evictions.to_string()); + summary_row( + &mut html, + "byte truncations", + &status.byte_truncations.to_string(), + ); + summary_row( + &mut html, + "sequence range", + &format!( + "{} .. {}", + option_u64(status.earliest_seq), + option_u64(status.latest_seq) + ), + ); + html.push_str("
"); + if !query.group_by.is_empty() { + push_groups(&mut html, &records, &query.group_by); + } + push_records(&mut html, &records, &query); + html.push_str(""); + truncate_page(&mut html); + retained_html_response(StatusCode::OK, html, render_permit) +} + +fn push_page_start(html: &mut String) { + html.push_str("WEB status
"); +} + +fn push_filter_form(html: &mut String, query: &StatusQuery) { + html.push_str("

Filters

"); + input(html, "window_secs", &query.window_secs.to_string()); + input(html, "ip", &query.ip.map(|value| value.to_string()).unwrap_or_default()); + input( + html, + "session", + &query.session.map(|value| value.to_string()).unwrap_or_default(), + ); + input( + html, + "user_agent", + query.user_agent.as_deref().unwrap_or_default(), + ); + input(html, "key", query.key.as_deref().unwrap_or_default()); + input(html, "limit", &query.limit.to_string()); + html.push_str("
"); +} + +fn input(html: &mut String, name: &str, value: &str) { + html.push_str(""); +} + +fn summary_row(html: &mut String, name: &str, value: &str) { + html.push_str(""); + escape(html, name); + html.push_str(""); + escape(html, value); + html.push_str(""); +} + +fn push_groups( + html: &mut String, + records: &[Arc], + groups: &[GroupBy], +) { + let mut summaries = BTreeMap::, GroupSummary>::new(); + let mut overflow = 0usize; + for stored in records { + let values = groups + .iter() + .map(|group| group_value(&stored.record, *group)) + .collect::>(); + if let Some(summary) = summaries.get_mut(&values) { + summary.count += 1; + summary.latest_seq = summary.latest_seq.max(stored.record.seq); + } else if summaries.len() < MAX_GROUPS { + summaries.insert( + values, + GroupSummary { + count: 1, + latest_seq: stored.record.seq, + }, + ); + } else { + overflow += 1; + } + } + let mut summaries = summaries.into_iter().collect::>(); + summaries.sort_by(|(left_values, left), (right_values, right)| { + right + .count + .cmp(&left.count) + .then_with(|| left_values.cmp(right_values)) + }); + html.push_str("

Groups

"); + for group in groups { + html.push_str(""); + } + html.push_str(""); + for (values, summary) in summaries { + html.push_str(""); + for value in values { + html.push_str(""); + } + html.push_str(""); + if html.len() >= MAX_PAGE_BYTES / 2 { + break; + } + } + if overflow != 0 { + html.push_str(""); + } + html.push_str("
"); + html.push_str(group.as_str()); + html.push_str("recordslatest seq
"); + escape(html, &value); + html.push_str(""); + html.push_str(&summary.count.to_string()); + html.push_str(""); + html.push_str(&summary.latest_seq.to_string()); + html.push_str("
Additional groups omitted: "); + html.push_str(&overflow.to_string()); + html.push_str("
"); +} + +fn group_value(record: &TraceRecord, group: GroupBy) -> String { + match group { + GroupBy::Ip => client_ip(record).map(|value| value.to_string()), + GroupBy::Session => record.identity.session_id.map(|value| value.to_string()), + GroupBy::UserAgent => record.user_agent.clone(), + GroupBy::Key => record.identity.key_fingerprint.clone(), + } + .unwrap_or_else(|| "-".to_string()) +} + +fn push_records(html: &mut String, records: &[Arc], query: &StatusQuery) { + html.push_str("

Records

"); + let mut shown = 0usize; + for stored in records.iter().take(query.limit) { + if html.len() >= MAX_PAGE_BYTES.saturating_sub(64 * 1024) { + break; + } + push_record(html, &stored.record); + shown += 1; + } + if shown == 0 { + html.push_str(""); + } + html.push_str("
seqtimekindroute/eventmethodstatusIPsessionuser / keyUser-Agentdetails
No matching records
"); + if records.len() > shown && shown != 0 { + let before = records[shown - 1].record.seq; + html.push_str("

Next page

"); + } + html.push_str("
"); +} + +fn push_record(html: &mut String, record: &TraceRecord) { + html.push_str(""); + html.push_str(&record.seq.to_string()); + html.push_str(""); + escape(html, &format_time(record.epoch_millis)); + let (kind, route, method, status) = match &record.kind { + TraceRecordKind::Http(http) => ( + "http", + http.route.as_str(), + http.method.as_str(), + http.status.map(|value| value.to_string()).unwrap_or_else(|| "-".to_string()), + ), + TraceRecordKind::Lifecycle(event) => ( + "lifecycle", + event.event.as_str(), + "-", + event.reason.unwrap_or("-").to_string(), + ), + }; + for value in [kind, route, method, status.as_str()] { + html.push_str(""); + escape(html, value); + } + html.push_str(""); + escape( + html, + &client_ip(record) + .map(|value| value.to_string()) + .unwrap_or_else(|| "-".to_string()), + ); + html.push_str(""); + escape(html, &option_u64(record.identity.session_id)); + html.push_str(""); + escape(html, record.identity.user.as_deref().unwrap_or("-")); + html.push_str(" / "); + escape( + html, + record.identity.key_fingerprint.as_deref().unwrap_or("-"), + ); + html.push_str(""); + escape(html, record.user_agent.as_deref().unwrap_or("-")); + html.push_str("
request → response"); + match &record.kind { + TraceRecordKind::Http(http) => { + html.push_str("

"); + escape(html, &http.method); + html.push(' '); + escape(html, &http.path); + html.push_str("

"); + push_headers(html, "request headers", &http.request_headers); + push_body(html, "request body", http.request_body.as_ref()); + push_headers(html, "response headers", &http.response_headers); + push_body(html, "response body", http.response_body.as_ref()); + if let Some(timings) = &http.timings { + html.push_str("

timings

service/head accepted: 0 us\nrequest body: ");
+                html.push_str(&option_u64(timings.request_body_us));
+                html.push_str(" us\nresponse ready: ");
+                html.push_str(&option_u64(timings.response_ready_us));
+                html.push_str(" us\nresponse body consumed/polled: ");
+                html.push_str(&option_u64(timings.response_body_us));
+                html.push_str(" us\n(kernel flush and TCP ACK are not observed)
"); + } + if !http.frames.is_empty() { + html.push_str("

frames

"); + for frame in &http.frames { + html.push_str(""); + for value in [ + frame.direction.as_str().to_string(), + frame.frame_type.unwrap_or("-").to_string(), + frame.stream_id.map(|v| v.to_string()).unwrap_or_else(|| "-".to_string()), + frame.payload_len.map(|v| v.to_string()).unwrap_or_else(|| "-".to_string()), + frame.window_delta.map(|v| v.to_string()).unwrap_or_else(|| "-".to_string()), + frame.parse_error.unwrap_or("-").to_string(), + ] { + html.push_str(""); + } + html.push_str(""); + } + html.push_str("
dirtypestream/lanepayloadWINDOWerror
"); + escape(html, &value); + html.push_str("
"); + } + } + TraceRecordKind::Lifecycle(event) => { + html.push_str("
event: ");
+            html.push_str(event.event.as_str());
+            html.push_str("\nstream: ");
+            html.push_str(&event.stream_id.map(|v| v.to_string()).unwrap_or_else(|| "-".to_string()));
+            html.push_str("\nreason: ");
+            html.push_str(event.reason.unwrap_or("-"));
+            html.push_str("
"); + } + } + html.push_str("
"); +} + +fn push_headers(html: &mut String, title: &str, headers: &[crate::web::trace::TraceHeader]) { + html.push_str("

"); + escape(html, title); + html.push_str("

");
+    for header in headers {
+        escape(html, &header.name);
+        html.push_str(": ");
+        escape(html, header.value.as_deref().unwrap_or("[value omitted]"));
+        html.push('\n');
+    }
+    html.push_str("
"); +} + +fn push_body( + html: &mut String, + title: &str, + body: Option<&crate::web::trace::TraceBodySnapshot>, +) { + html.push_str("

"); + escape(html, title); + html.push_str("

"); + let Some(body) = body else { + html.push_str("

capture off

"); + return; + }; + html.push_str("

observed="); + html.push_str(&body.observed_bytes.to_string()); + html.push_str(" captured="); + html.push_str(&body.captured.len().to_string()); + html.push_str(" state="); + html.push_str(body.state.as_str()); + html.push_str(" truncated="); + html.push_str(yes_no(body.truncated)); + html.push_str("

");
+    let available = MAX_PAGE_BYTES.saturating_sub(html.len()).saturating_sub(4096);
+    let raw_limit = available.saturating_mul(3) / 4;
+    let shown = body.captured.len().min(raw_limit);
+    base64::engine::general_purpose::STANDARD
+        .encode_string(&body.captured[..shown], html);
+    if shown < body.captured.len() {
+        html.push_str("\n[page output truncated]");
+    }
+    html.push_str("
"); +} + +fn pagination_url(query: &StatusQuery, before_seq: u64) -> String { + let mut serializer = url::form_urlencoded::Serializer::new(String::from("/web-status?")); + serializer.append_pair("window_secs", &query.window_secs.to_string()); + if let Some(ip) = query.ip { + serializer.append_pair("ip", &ip.to_string()); + } + if let Some(session) = query.session { + serializer.append_pair("session", &session.to_string()); + } + if let Some(user_agent) = &query.user_agent { + serializer.append_pair("user_agent", user_agent); + } + if let Some(key) = &query.key { + serializer.append_pair("key", key); + } + for group in &query.group_by { + serializer.append_pair("group_by", group.as_str()); + } + serializer.append_pair("limit", &query.limit.to_string()); + serializer.append_pair("before_seq", &before_seq.to_string()); + serializer.finish() +} + +fn format_time(epoch_millis: u64) -> String { + chrono::DateTime::from_timestamp_millis(epoch_millis as i64) + .map(|value| value.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)) + .unwrap_or_else(|| epoch_millis.to_string()) +} + +fn option_u64(value: Option) -> String { + value.map(|value| value.to_string()).unwrap_or_else(|| "-".to_string()) +} + +fn body_mode(policy: &WebDebugConfig) -> &'static str { + match policy.body_capture { + crate::config::WebDebugBodyCapture::Off => "off", + crate::config::WebDebugBodyCapture::Metadata => "metadata", + crate::config::WebDebugBodyCapture::Prefix => "prefix", + crate::config::WebDebugBodyCapture::Full => "full", + } +} + +fn yes_no(value: bool) -> &'static str { + if value { "yes" } else { "no" } +} + +fn escape(output: &mut String, value: &str) { + for character in value.chars() { + match character { + '&' => output.push_str("&"), + '<' => output.push_str("<"), + '>' => output.push_str(">"), + '"' => output.push_str("""), + '\'' => output.push_str("'"), + _ => output.push(character), + } + } +} + +fn truncate_page(html: &mut String) { + const SUFFIX: &str = "[page output truncated]"; + if html.len() <= MAX_PAGE_BYTES { + return; + } + let mut end = MAX_PAGE_BYTES.saturating_sub(SUFFIX.len()); + while !html.is_char_boundary(end) { + end -= 1; + } + html.truncate(end); + html.push_str(SUFFIX); +} + +fn html_error(status: StatusCode, title: &str, message: &str) -> Response> { + let mut html = String::new(); + push_page_start(&mut html); + html.push_str("

"); + escape(&mut html, title); + html.push_str("

"); + escape(&mut html, message); + html.push_str("

"); + html_response(status, html) +} + +fn html_response(status: StatusCode, html: String) -> Response> { + html_bytes_response(status, Bytes::from(html)) +} + +fn retained_html_response( + status: StatusCode, + html: String, + permit: OwnedSemaphorePermit, +) -> Response> { + html_bytes_response( + status, + Bytes::from_owner(RenderedPage { + html, + _permit: permit, + }), + ) +} + +fn html_bytes_response(status: StatusCode, html: Bytes) -> Response> { + let mut response = Response::new(Full::new(html)); + *response.status_mut() = status; + response.headers_mut().insert( + header::CONTENT_TYPE, + HeaderValue::from_static("text/html; charset=utf-8"), + ); + response + .headers_mut() + .insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store")); + response.headers_mut().insert( + header::CONTENT_SECURITY_POLICY, + HeaderValue::from_static("default-src 'none'; style-src 'unsafe-inline'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'"), + ); + response.headers_mut().insert( + header::REFERRER_POLICY, + HeaderValue::from_static("no-referrer"), + ); + response.headers_mut().insert( + header::X_CONTENT_TYPE_OPTIONS, + HeaderValue::from_static("nosniff"), + ); + response + .headers_mut() + .insert(header::X_FRAME_OPTIONS, HeaderValue::from_static("DENY")); + response +} + +#[cfg(test)] +#[path = "web_status/tests.rs"] +mod tests; diff --git a/src/api/web_status/query.rs b/src/api/web_status/query.rs new file mode 100644 index 0000000..106c2ad --- /dev/null +++ b/src/api/web_status/query.rs @@ -0,0 +1,179 @@ +use std::collections::BTreeSet; +use std::net::IpAddr; + +use crate::config::WebDebugConfig; +use crate::web::trace::TraceRecord; + +const DEFAULT_LIMIT: usize = 200; +const MAX_LIMIT: usize = 1000; + +/// Supported status-page grouping dimensions. +#[derive(Clone, Copy, PartialEq, Eq)] +pub(super) enum GroupBy { + Ip, + Session, + UserAgent, + Key, +} + +impl GroupBy { + fn parse(value: &str) -> Option { + match value { + "ip" => Some(Self::Ip), + "session" => Some(Self::Session), + "user_agent" => Some(Self::UserAgent), + "key" => Some(Self::Key), + _ => None, + } + } + + /// Returns the canonical query and table label. + pub(super) const fn as_str(self) -> &'static str { + match self { + Self::Ip => "ip", + Self::Session => "session", + Self::UserAgent => "user_agent", + Self::Key => "key", + } + } +} + +/// Validated bounded status-page filter and pagination state. +pub(super) struct StatusQuery { + pub(super) window_secs: u64, + pub(super) ip: Option, + pub(super) session: Option, + pub(super) user_agent: Option, + pub(super) key: Option, + pub(super) group_by: Vec, + pub(super) limit: usize, + pub(super) before_seq: Option, + pub(super) record: Option, +} + +/// Parses a strict query without accepting unknown or ambiguous fields. +pub(super) fn parse_query( + raw: Option<&str>, + policy: &WebDebugConfig, +) -> Result { + let mut query = StatusQuery { + window_secs: policy.default_window_secs, + ip: None, + session: None, + user_agent: None, + key: None, + group_by: Vec::new(), + limit: DEFAULT_LIMIT, + before_seq: None, + record: None, + }; + let mut seen = BTreeSet::new(); + for (name, value) in url::form_urlencoded::parse(raw.unwrap_or_default().as_bytes()) { + let name = name.as_ref(); + let value = value.as_ref(); + if name != "group_by" && !seen.insert(name.to_string()) { + return Err(format!("{name} must not repeat")); + } + match name { + "window_secs" => { + query.window_secs = parse_positive_u64(value, "window_secs")?; + } + "ip" => { + let parsed = value + .parse::() + .map_err(|_| "ip must be a canonical IP address".to_string())?; + if parsed.to_string() != value { + return Err("ip must use canonical formatting".to_string()); + } + query.ip = Some(parsed); + } + "session" => query.session = Some(parse_positive_u64(value, "session")?), + "user_agent" => { + if value.is_empty() || value.len() > 512 { + return Err("user_agent must contain 1..512 bytes".to_string()); + } + query.user_agent = Some(value.to_string()); + } + "key" => { + if value.is_empty() || value.len() > 64 { + return Err("key must contain 1..64 bytes".to_string()); + } + query.key = Some(value.to_string()); + } + "group_by" => { + let group = GroupBy::parse(value) + .ok_or_else(|| "group_by must be ip, session, user_agent, or key".to_string())?; + if query.group_by.contains(&group) { + return Err("group_by values must not repeat".to_string()); + } + query.group_by.push(group); + } + "limit" => { + query.limit = value + .parse::() + .ok() + .filter(|value| (1..=MAX_LIMIT).contains(value)) + .ok_or_else(|| "limit must be within 1..1000".to_string())?; + } + "before_seq" => { + query.before_seq = Some(parse_positive_u64(value, "before_seq")?); + } + "record" => query.record = Some(parse_positive_u64(value, "record")?), + _ => return Err(format!("unknown query field `{name}`")), + } + } + if query.window_secs > policy.max_window_secs { + return Err(format!( + "window_secs must not exceed {}", + policy.max_window_secs + )); + } + Ok(query) +} + +fn parse_positive_u64(value: &str, field: &str) -> Result { + value + .parse::() + .ok() + .filter(|value| *value != 0) + .ok_or_else(|| format!("{field} must be a positive integer")) +} + +/// Applies the complete filter predicate to one immutable record. +pub(super) fn record_matches( + record: &TraceRecord, + query: &StatusQuery, + since_millis: u64, +) -> bool { + !(record.epoch_millis < since_millis + || query.before_seq.is_some_and(|before| record.seq >= before) + || query.record.is_some_and(|seq| record.seq != seq) + || query.ip.is_some_and(|ip| client_ip(record) != Some(ip)) + || query + .session + .is_some_and(|session| record.identity.session_id != Some(session)) + || query.user_agent.as_ref().is_some_and(|needle| { + record + .user_agent + .as_deref() + .is_none_or(|value| !contains_ascii_case_insensitive(value, needle)) + }) + || query.key.as_ref().is_some_and(|key| { + record.identity.user.as_deref() != Some(key) + && record.identity.key_fingerprint.as_deref() != Some(key) + })) +} + +fn contains_ascii_case_insensitive(value: &str, needle: &str) -> bool { + let needle = needle.as_bytes(); + needle.is_empty() + || value + .as_bytes() + .windows(needle.len()) + .any(|window| window.eq_ignore_ascii_case(needle)) +} + +/// Returns the trusted effective address or direct peer fallback. +pub(super) fn client_ip(record: &TraceRecord) -> Option { + record.effective_ip.or(record.peer_ip) +} diff --git a/src/api/web_status/tests.rs b/src/api/web_status/tests.rs new file mode 100644 index 0000000..5139ec7 --- /dev/null +++ b/src/api/web_status/tests.rs @@ -0,0 +1,82 @@ +use http_body_util::BodyExt as _; + +use super::*; +use crate::web::trace::{TraceIdentity, TraceLifecycleEvent}; + +#[test] +fn query_rejects_noncanonical_ip_and_excessive_window() { + let policy = WebDebugConfig::default(); + assert!(parse_query(Some("ip=2001%3A0db8%3A%3A1"), &policy).is_err()); + assert!(parse_query(Some("window_secs=3601"), &policy).is_err()); + assert!(parse_query(Some("session=1&session=2"), &policy).is_err()); +} + +#[test] +fn html_escaping_covers_active_markup_characters() { + let mut output = String::new(); + escape(&mut output, "