From 97183cf867ee450abc7a026f282043ff53d26c0b Mon Sep 17 00:00:00 2001 From: joshrmcdaniel <80354972+joshrmcdaniel@users.noreply.github.com> Date: Sun, 4 Oct 2026 13:19:58 -0500 Subject: [PATCH 01/10] stats --- .gitignore | 5 +- CHANGELOG.md | 17 + Cargo.lock | 11 + README.md | 81 +++ privaxy/Cargo.toml | 1 + privaxy/src/server/configuration/mod.rs | 5 + .../src/server/configuration/statistics.rs | 62 ++ privaxy/src/server/lib.rs | 32 +- privaxy/src/server/proxy/html_rewriter.rs | 54 ++ privaxy/src/server/proxy/serve.rs | 29 +- privaxy/src/server/statistics.rs | 141 +++- privaxy/src/server/statistics/history.rs | 682 ++++++++++++++++++ .../src/server/statistics/history/tests.rs | 450 ++++++++++++ privaxy/src/server/web_gui/history.rs | 122 ++++ privaxy/src/server/web_gui/mod.rs | 17 + web_frontend/Cargo.toml | 1 + web_frontend/src/dashboard.rs | 3 +- web_frontend/src/history.rs | 281 ++++++++ web_frontend/src/main.rs | 9 + web_frontend/src/settings.rs | 8 + web_frontend/src/statistics_settings.rs | 319 ++++++++ web_frontend/tests/ui-regressions.cjs | 89 ++- 22 files changed, 2366 insertions(+), 53 deletions(-) create mode 100644 privaxy/src/server/configuration/statistics.rs create mode 100644 privaxy/src/server/statistics/history.rs create mode 100644 privaxy/src/server/statistics/history/tests.rs create mode 100644 privaxy/src/server/web_gui/history.rs create mode 100644 web_frontend/src/history.rs create mode 100644 web_frontend/src/statistics_settings.rs diff --git a/.gitignore b/.gitignore index 8ca8437..1da5405 100644 --- a/.gitignore +++ b/.gitignore @@ -7,4 +7,7 @@ filters/__pycache__ node_modules .vscode **/target/* -testing/ \ No newline at end of file +testing/ + +# Local opt-in request history can contain browsing metadata. +statistics.redb diff --git a/CHANGELOG.md b/CHANGELOG.md index c09ce61..5740921 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,23 @@ ## Unreleased +- Added opt-in persistent statistics and individual request history using redb. + History includes time ranges, trends, top domains/clients, and pagination, with + privacy levels that remove identifiers before collection. URL paths, query + strings, headers, and bodies are never saved in statistics. +- Added Statistics settings for retention, a request-count limit, clearing saved + history, and excluding domain patterns from both statistics and the live + Requests log. Exclusions do not change filtering behavior and work even when + persistence is disabled. +- Bounded statistics buffering, batched transactional writes, visible storage + errors, and shutdown flushing keep storage failures off the proxy request path. + Existing installations retain memory-only statistics until explicitly enabled. +- Dashboard top blocked entries now use domains and select the top 50 after + sorting; client tracking is bounded. Dashboard counters remain per-process; + the new History page shows statistics retained across restarts. +- Count each modified HTML response once when both head scripts and cosmetic + styles are injected, instead of counting the two injection stages separately. + ## v0.8.0 - 2026-10-02 Release highlights and upgrade notes are in [RELEASE.md](RELEASE.md). diff --git a/Cargo.lock b/Cargo.lock index 0ffb544..c06596d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2192,6 +2192,7 @@ dependencies = [ "once_cell", "openssl", "portable-atomic", + "redb", "regex", "reqwest", "rustls", @@ -2393,6 +2394,15 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "redb" +version = "4.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb338a6c67830a61bed824b78c2bb034ab1ff401a20616bd7957524f4fdc2f22" +dependencies = [ + "libc", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -3657,6 +3667,7 @@ dependencies = [ "gloo-net 0.6.0", "gloo-timers 0.4.0", "gloo-utils 0.3.0", + "js-sys", "log", "num-format", "readonly", diff --git a/README.md b/README.md index 8b1315b..9f8c277 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,7 @@ features, dependency updates, an improved UI, and server-friendly configuration - [Cert-pinned hosts (exclusions)](#4-cert-pinned-hosts-exclusions) - [Filter only selected hosts (inclusions)](#5-filter-only-selected-hosts-inclusions) - [HTTP CONNECT and proxy chaining](#6-http-connect-and-proxy-chaining) + - [Persistent statistics and privacy](#7-persistent-statistics-and-privacy) - [Screenshots](#screenshots) - [Acknowledgements](#acknowledgements) @@ -73,6 +74,7 @@ blocker more than it strictly replaces one. - Suppport for [Adblock Plus filters](https://adblockplus.org/filter-cheatsheet), such as [easylist](https://easylist.to/). - Web graphical user interface with a statistics display as well as a live request explorer. +- Opt-in persistent statistics and request history, with privacy levels, retention limits, and domain exclusions. - Support for uBlock origin's `js` syntax. - Support for uBlock origin's `redirect` syntax. - Support for uBlock origin's scriptlets. @@ -319,6 +321,85 @@ for those steps. > `password_hash` value from the config file and restart. The web UI will > force the setup flow again. +### 7. Persistent statistics and privacy + +Enable collection under **Settings → Statistics**, then open **History** for +saved totals, hourly/daily trends, top domains/clients, and a paginated request +log. The Dashboard continues to show the current process's live counters. +Persistence is **off by default**; enabling it starts collecting new activity, +without importing the live Requests feed. + +History records timestamps, destination hostnames, client IP addresses, and +whether a request was blocked or successfully forwarded. It never stores URL +paths, query strings, headers, cookies, or request/response bodies. Raw CONNECT +tunnels and requests that fail before being forwarded are not counted. Clients +behind another proxy share that proxy's client address. + +The privacy levels follow [Pi-hole's approach](https://docs.pi-hole.net/ftldns/privacylevels/): + +| Level | Saved identifiers | +| --- | --- | +| Show domains and clients | Hostname and client IP | +| Hide domains | Client IP only | +| Hide domains and clients (default) | Neither; request times and outcomes remain | +| Anonymous | No new history; session counters only | + +Hidden identifiers are removed **before queueing or storage**, without retaining +a reversible mapping or a hash. While collection is enabled, raising privacy +also redacts identifiers in existing history, including when entering anonymous +mode. Lowering privacy cannot restore them. These controls apply to statistics; +the separate live Requests feed and diagnostic logs can still contain full URLs. + +**Exclude domains from statistics and request logs** skips matching new traffic +in session counters, modified-response counts, saved history, and the live +Requests feed, even with persistence disabled. Filtering and interception still +apply. Use one case-insensitive hostname or wildcard pattern per line: +`example.com` matches exactly; `*.example.com` matches its subdomains. Do not +include schemes, ports, or paths. Existing history and diagnostic logs are +unaffected. + +Defaults are **7 days** and **100,000 saved requests**. The oldest records expire +when either limit is reached; valid limits are 1–365 days and 100–1,000,000 +requests. History totals and top lists describe retained requests, so they can +decrease as older entries expire. Modified-response totals are separate hourly +aggregates retained by age. Time windows include the current hour and preceding +hours. **Clear saved history** removes the database and starts fresh if enabled. +Turning collection off stops recording and closes the database; it retains the +existing file until cleared or collection is re-enabled and retention runs. + +The equivalent configuration section is: + +```toml +[statistics] +enabled = false +privacy = "hide_domains_and_clients" # full, hide_domains, hide_domains_and_clients, anonymous +retention_days = 7 +max_entries = 100000 +excluded_domains = ["example.com", "*.example.com"] +``` + +UI changes apply immediately; file edits also apply on SIGHUP. Data stays local +in `statistics.redb` under `PRIVAXY_BASE_PATH` (`/etc/privaxy` by default, `/conf` +in Docker). Keep `/conf` on a persistent Docker volume for history to survive +container replacement. No new ports or external database service are required. +On Unix, the database is restricted to its owner (`0600`). Storage is not +encrypted, and redaction/deletion is not secure erasure of filesystem snapshots, +backups, or recoverable disk pages. + +[redb](https://github.com/cberner/redb) transactions run on a dedicated worker, +with a bounded queue and batches flushed every five seconds or 512 events. +SIGTERM/SIGINT also flush accepted events; an abrupt kill can lose the pending +batch. Expired rows are pruned every minute and free space is compacted hourly; +the row limit bounds retained data, not an exact file size. Disk/write errors +pause recording and appear in Statistics settings without stopping the proxy. +Free space and save settings again to retry. Queue overflow drops statistics +events and reports their count instead of delaying proxy requests. + +The authenticated APIs are `GET/PUT /api/settings/statistics` and +`GET/DELETE /api/history`. History accepts `hours`, `limit` (1–200), `before` +(the previous response's `next_before` cursor), and `blocked=true|false`. +Pagination and the blocked filter affect request rows, not the window's totals. + ## Screenshots
diff --git a/privaxy/Cargo.toml b/privaxy/Cargo.toml index 4fae88e..1899117 100644 --- a/privaxy/Cargo.toml +++ b/privaxy/Cargo.toml @@ -137,6 +137,7 @@ tera = { version = "1", default-features = false } argon2 = { version = "0.5", features = ["std"] } hmac = "0.12" filterlists-api = { path = "../filterlists-api", features = ["reqwest"] } +redb = "4.3.0" # getrandom 0.3 (pulled in via warp -> tokio-tungstenite -> rand) selects a # backend that calls the libc `getrandom()` wrapper on mips{,el}-unknown-linux-gnu, diff --git a/privaxy/src/server/configuration/mod.rs b/privaxy/src/server/configuration/mod.rs index 1cac41a..faeb422 100644 --- a/privaxy/src/server/configuration/mod.rs +++ b/privaxy/src/server/configuration/mod.rs @@ -7,6 +7,7 @@ mod ca; mod filter; mod filter_failures; mod network; +mod statistics; mod updater; mod userscript; pub use auth::*; @@ -15,6 +16,7 @@ pub use filter::*; pub use filter_failures::*; use futures::future::join_all; pub use network::*; +pub use statistics::*; use std::env; use std::path::{Path, PathBuf}; pub use updater::*; @@ -85,6 +87,8 @@ pub struct Configuration { pub auth: Auth, #[serde(default)] pub debug: DebugConfig, + #[serde(default)] + pub statistics: StatisticsConfig, /// Userscripts and the master switch governing them. Absent from /// configuration files written before the feature existed, hence the serde /// default. @@ -588,6 +592,7 @@ impl Configuration { custom_filters: Vec::new(), auth: Auth::new_initialized(), debug: DebugConfig::default(), + statistics: StatisticsConfig::default(), userscripts: UserScriptsConfig::default(), }) } diff --git a/privaxy/src/server/configuration/statistics.rs b/privaxy/src/server/configuration/statistics.rs new file mode 100644 index 0000000..9f93c39 --- /dev/null +++ b/privaxy/src/server/configuration/statistics.rs @@ -0,0 +1,62 @@ +use serde::{Deserialize, Serialize}; + +/// These levels describe collected data, not just what the UI displays. +#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize, PartialEq, Eq, PartialOrd, Ord)] +#[serde(rename_all = "snake_case")] +pub enum StatisticsPrivacy { + Full, + HideDomains, + #[default] + HideDomainsAndClients, + Anonymous, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +#[serde(default, deny_unknown_fields)] +pub struct StatisticsConfig { + pub enabled: bool, + pub privacy: StatisticsPrivacy, + pub retention_days: u16, + pub max_entries: u32, + pub excluded_domains: Vec, +} + +impl Default for StatisticsConfig { + fn default() -> Self { + Self { + enabled: false, + privacy: StatisticsPrivacy::default(), + retention_days: 7, + max_entries: 100_000, + excluded_domains: Vec::new(), + } + } +} + +impl StatisticsConfig { + pub fn validate(&self) -> Result<(), String> { + if self.excluded_domains.len() > 1000 + || self.excluded_domains.iter().any(|pattern| { + let pattern = pattern.trim(); + pattern.is_empty() + || pattern.len() > 253 + || !pattern + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b".-*?".contains(&b)) + }) + { + return Err("Use at most 1,000 domain patterns, without schemes, ports, or paths (for example example.com or *.example.com)".into()); + } + if !(1..=365).contains(&self.retention_days) { + return Err("Statistics retention must be between 1 and 365 days".into()); + } + if !(100..=1_000_000).contains(&self.max_entries) { + return Err("Request history limit must be between 100 and 1,000,000 entries".into()); + } + Ok(()) + } + + pub fn persists(&self) -> bool { + self.enabled && self.privacy != StatisticsPrivacy::Anonymous + } +} diff --git a/privaxy/src/server/lib.rs b/privaxy/src/server/lib.rs index 7e32955..f56006b 100644 --- a/privaxy/src/server/lib.rs +++ b/privaxy/src/server/lib.rs @@ -104,7 +104,7 @@ pub(crate) fn parse_ip_address(ip_str: &str) -> IpAddr { IpAddr::from_str(ip_str).unwrap() } -async fn handle_signals() -> (Arc, Arc) { +async fn handle_signals(statistics: statistics::Statistics) -> (Arc, Arc) { let notify_shutdown = Arc::new(Notify::new()); let notify_reload = Arc::new(Notify::new()); let notify_shutdown_clone = notify_shutdown.clone(); @@ -115,6 +115,8 @@ async fn handle_signals() -> (Arc, Arc) { signal(SignalKind::hangup()).expect("failed to set up SIGHUP signal handler"); let mut term_signal = signal(SignalKind::terminate()).expect("failed to set up SIGTERM signal handler"); + let mut int_signal = + signal(SignalKind::interrupt()).expect("failed to set up SIGINT signal handler"); loop { tokio::select! { @@ -122,9 +124,16 @@ async fn handle_signals() -> (Arc, Arc) { log::info!("Received SIGHUP signal, restarting child processes..."); notify_reload_clone.notify_waiters(); } - _ = term_signal.recv() => { - log::info!("Received SIGTERM signal, shutting down gracefully..."); + _ = async { tokio::select! { _ = term_signal.recv() => {}, _ = int_signal.recv() => {} } } => { + log::info!("Shutting down; flushing persistent statistics..."); notify_shutdown_clone.notify_waiters(); + if let Some(history) = &statistics.history { + match tokio::time::timeout(Duration::from_secs(8), history.shutdown()).await { + Ok(Ok(())) => {}, + Ok(Err(error)) => log::error!("Could not flush statistics: {error}"), + Err(_) => log::error!("Timed out flushing statistics"), + } + } std::process::exit(0); } } @@ -220,7 +229,7 @@ pub async fn start_privaxy() -> PrivaxyServer { } }; - let statistics = statistics::Statistics::new(); + let statistics = statistics::Statistics::with_history(configuration.statistics.clone()).await; let statistics_clone = statistics.clone(); let tls_failure_store = TlsFailureStore::new(configuration.ignored_tls_failures.clone()); @@ -272,7 +281,7 @@ pub async fn start_privaxy() -> PrivaxyServer { let configuration_save_lock = Arc::new(tokio::sync::Mutex::new(())); - let (_notify_shutdown, notify_reload) = handle_signals().await; + let (_notify_shutdown, notify_reload) = handle_signals(statistics.clone()).await; let block_disable_ref = blocking_disabled_store.clone(); let local_exclusion_store_ref = local_exclusion_store.clone(); @@ -361,6 +370,19 @@ pub async fn start_privaxy() -> PrivaxyServer { // The UI updates this store live; SIGHUP must also pick up file // edits to the mode and both inclusion/exclusion lists. local_exclusion_store.replace_configuration(&cfg); + { + // Serialize this re-read/apply with the settings API. A reload + // must not restore a stale privacy policy over a concurrent save. + let _guard = cfg_lock_backend.lock().await; + match configuration::Configuration::read_from_home().await { + Ok(current) => { + if let Err(error) = statistics.configure_history(current.statistics).await { + log::error!("Could not apply statistics settings: {error}"); + } + } + Err(error) => log::error!("Could not reload statistics settings: {error}"), + } + } // Same for the TLS-failure ignore set: pick up hand-edited // `ignored_tls_failures` entries on SIGHUP reload. tls_failure_store.set_ignored(cfg.ignored_tls_failures.clone()); diff --git a/privaxy/src/server/proxy/html_rewriter.rs b/privaxy/src/server/proxy/html_rewriter.rs index d3cfd65..7d3ae4c 100644 --- a/privaxy/src/server/proxy/html_rewriter.rs +++ b/privaxy/src/server/proxy/html_rewriter.rs @@ -427,6 +427,10 @@ impl Rewriter { endpoint_tokens, } = self; + // Head scriptlets and the trailing cosmetic stylesheet belong to one + // response. Their separate tasks share a single counting guard. + let statistics = statistics.for_response(); + let CosmeticBlockerResult { hidden_selectors, style_selectors, @@ -676,6 +680,56 @@ impl Rewriter { #[cfg(test)] mod tests { use super::*; + + #[tokio::test] + async fn head_and_body_injections_count_each_response_only_once() { + use crate::blocker::BlockingDisabledStore; + use http_body_util::BodyExt; + use std::sync::RwLock; + + let statistics = Statistics::new(); + for expected in 1..=2 { + let requester = + AdblockRequester::new(BlockingDisabledStore(Arc::new(RwLock::new(false)))); + let mut cosmetics = requester + .get_cosmetic_response("https://example.test/".into()) + .await; + cosmetics.hidden_selectors.push(".ad-test".into()); + let (input, receiver) = mpsc::channel(2); + let (sender, mut body) = body_channel(); + let rewriter = Rewriter::new( + requester, + receiver, + sender, + statistics.clone(), + "fixture".into(), + cosmetics, + false, + Vec::new(), + GmStorageStore::in_memory(), + HashMap::new(), + ); + input + .send(Ok(Bytes::from_static( + b"
ad
", + ))) + .await + .unwrap(); + drop(input); + let task = tokio::task::spawn_blocking(move || rewriter.rewrite()); + let mut output = Vec::new(); + while let Some(frame) = body.frame().await { + if let Ok(bytes) = frame.unwrap().into_data() { + output.extend_from_slice(&bytes); + } + } + task.await.unwrap(); + let html = String::from_utf8(output).unwrap(); + assert!(html.contains(" Result, hyper::Error> { - let scheme_string = scheme.to_string(); - let uri = match http::uri::Builder::new() .scheme(scheme) .authority(authority) @@ -349,14 +347,13 @@ pub(crate) async fn serve( let req = Request::from_parts(parts, body); log::debug!("{} {}", req.method(), req.uri()); - - statistics.increment_top_clients(client_ip_address); + let statistics = statistics.for_domain(uri.host()); if !intercept { let mut req = req; req.headers_mut().remove(http::header::PROXY_AUTHORIZATION); req.headers_mut().remove("proxy-connection"); - if broadcast_sender.receiver_count() > 0 { + if statistics.should_collect() && broadcast_sender.receiver_count() > 0 { let _ = broadcast_sender.send(Event { now: chrono::Utc::now(), method: req.method().to_string(), @@ -367,7 +364,7 @@ pub(crate) async fn serve( // Forward with hyper so bodies, compression and CSP remain untouched. // This path skips adblock, DoH policy and all HTML/userscript injection. let response = if req.headers().contains_key(http::header::UPGRADE) { - perform_two_ends_upgrade(req, uri, hyper_client).await + perform_two_ends_upgrade(req, uri.clone(), hyper_client).await } else { let (mut parts, body) = req.into_parts(); parts.headers.remove(http::header::HOST); @@ -385,7 +382,7 @@ pub(crate) async fn serve( } } }; - statistics.increment_proxied_requests(); + statistics.record_request(uri.host(), client_ip_address, false); return Ok(response); } @@ -403,7 +400,7 @@ pub(crate) async fn serve( log::debug!("Refusing DoH request: {}", uri); // Events only feed the live requests view; when nobody is subscribed, // building one (two Strings and a timestamp) is wasted work. - if broadcast_sender.receiver_count() > 0 { + if statistics.should_collect() && broadcast_sender.receiver_count() > 0 { let _result = broadcast_sender.send(Event { now: chrono::Utc::now(), method: req.method().to_string(), @@ -411,7 +408,7 @@ pub(crate) async fn serve( is_request_blocked: true, }); } - statistics.increment_blocked_requests(); + statistics.record_request(uri.host(), client_ip_address, true); return Ok(get_empty_response(StatusCode::BAD_GATEWAY)); } @@ -426,7 +423,7 @@ pub(crate) async fn serve( request_type, ); - if broadcast_sender.receiver_count() > 0 { + if statistics.should_collect() && broadcast_sender.receiver_count() > 0 { let _result = broadcast_sender.send(Event { now: chrono::Utc::now(), method: req.method().to_string(), @@ -436,13 +433,7 @@ pub(crate) async fn serve( } if is_request_blocked { - statistics.increment_blocked_requests(); - statistics.increment_top_blocked_paths(format!( - "{}://{}{}", - scheme_string, - uri.host().unwrap(), - uri.path() - )); + statistics.record_request(uri.host(), client_ip_address, true); // adblock-rust fuses many same-option patterns into one filter and // reports the union as the matched filter, which can be enormous; cap @@ -476,7 +467,7 @@ pub(crate) async fn serve( return Ok(get_empty_response(StatusCode::BAD_GATEWAY)); }; let response = perform_two_ends_upgrade(req, outbound_uri, hyper_client).await; - statistics.increment_proxied_requests(); + statistics.record_request(uri.host(), client_ip_address, false); return Ok(response); } @@ -532,7 +523,7 @@ pub(crate) async fn serve( } }; - statistics.increment_proxied_requests(); + statistics.record_request(uri.host(), client_ip_address, false); *new_response.headers_mut() = response.headers().clone(); diff --git a/privaxy/src/server/statistics.rs b/privaxy/src/server/statistics.rs index 133984b..ec0c7c0 100644 --- a/privaxy/src/server/statistics.rs +++ b/privaxy/src/server/statistics.rs @@ -1,13 +1,16 @@ +use crate::configuration::{StatisticsConfig, StatisticsPrivacy}; use portable_atomic::{AtomicU64, Ordering}; use serde::Serialize; use std::{ collections::HashMap, net::IpAddr, - sync::{Arc, Mutex}, + sync::{atomic::AtomicBool, Arc, Mutex}, }; use uluru::LRUCache; -const ENTRIES_PER_STATISTICS_TABLE: u8 = 50; +pub mod history; + +const ENTRIES_PER_STATISTICS_TABLE: usize = 50; #[derive(Debug, Serialize)] pub struct SerializableStatistics { @@ -27,6 +30,9 @@ pub struct Statistics { pub modified_responses: Arc, pub top_blocked_paths: Arc>>, pub top_clients: Arc>>, + pub history: Option, + collect: bool, + response_modified: Option>, } impl Default for Statistics { @@ -43,6 +49,79 @@ impl Statistics { modified_responses: Arc::new(AtomicU64::new(0)), top_blocked_paths: Arc::new(Mutex::new(LRUCache::default())), top_clients: Arc::new(Mutex::new(HashMap::new())), + history: None, + collect: true, + response_modified: None, + } + } + + pub async fn with_history(config: StatisticsConfig) -> Self { + let path = crate::configuration::get_config_file().with_file_name("statistics.redb"); + Self { + history: Some(history::HistoryStore::start(path, config).await), + ..Self::new() + } + } + + pub async fn configure_history(&self, config: StatisticsConfig) -> Result<(), String> { + let Some(history) = &self.history else { + return Ok(()); + }; + if config == history.config() && history.status().error.is_none() { + return Ok(()); + } + let result = history.configure(config).await; + // Discard session identifiers too. Lowering privacy later must not + // resurrect identifiers from before the change. + self.top_blocked_paths.lock().unwrap().clear(); + self.top_clients.lock().unwrap().clear(); + result + } + + /// Carry one domain's collection decision through response rewriting too. + /// Statistics exclusions never change interception or blocking decisions. + pub fn for_domain(&self, domain: Option<&str>) -> Self { + Self { + collect: self.collect && !self.history.as_ref().is_some_and(|h| h.is_excluded(domain)), + ..self.clone() + } + } + + pub fn should_collect(&self) -> bool { + self.collect + } + + pub(crate) fn for_response(&self) -> Self { + Self { + response_modified: Some(Arc::new(AtomicBool::new(false))), + ..self.clone() + } + } + + pub fn record_request(&self, domain: Option<&str>, client: IpAddr, blocked: bool) { + if !self.collect || self.history.as_ref().is_some_and(|h| h.is_excluded(domain)) { + return; + } + let privacy = self + .history + .as_ref() + .map(|h| h.session_privacy()) + .unwrap_or(StatisticsPrivacy::Full); + if privacy < StatisticsPrivacy::HideDomainsAndClients { + self.increment_top_clients(client); + } + if blocked { + self.increment_blocked_requests(); + if privacy == StatisticsPrivacy::Full { + if let Some(domain) = domain { + self.increment_top_blocked_paths(domain.to_ascii_lowercase()); + } + } + } else { + self.increment_proxied_requests(); + } + if let Some(history) = &self.history { + history.request(domain, client, blocked); } } @@ -60,7 +139,10 @@ impl Statistics { } pub fn increment_top_clients(&self, client: IpAddr) { - *self.top_clients.lock().unwrap().entry(client).or_insert(0) += 1; + let mut clients = self.top_clients.lock().unwrap(); + if clients.len() < 1000 || clients.contains_key(&client) { + *clients.entry(client).or_insert(0) += 1; + } } pub fn increment_proxied_requests(&self) -> u64 { @@ -72,46 +154,55 @@ impl Statistics { } pub fn increment_modified_responses(&self) -> u64 { + if !self.collect + || self + .response_modified + .as_ref() + .is_some_and(|modified| modified.swap(true, Ordering::Relaxed)) + { + return self.modified_responses.load(Ordering::Relaxed); + } + if let Some(history) = &self.history { + history.modified(); + } self.modified_responses.fetch_add(1, Ordering::Relaxed) + 1 } pub fn get_serialized(&self) -> SerializableStatistics { - SerializableStatistics { + let mut serialized = SerializableStatistics { proxied_requests: self.proxied_requests.load(Ordering::Relaxed), blocked_requests: self.blocked_requests.load(Ordering::Relaxed), modified_responses: self.modified_responses.load(Ordering::Relaxed), top_blocked_paths: { let top_blocked_paths = self.top_blocked_paths.lock().unwrap(); - let mut top_blocked_paths_iterator = top_blocked_paths.iter(); - - let mut top_blocked_paths = (0..=ENTRIES_PER_STATISTICS_TABLE) - .filter_map(|_| { - let (path, count) = top_blocked_paths_iterator.next()?; - - Some((path.clone(), *count)) - }) - .collect::>(); - - top_blocked_paths.sort_by_key(|entry| std::cmp::Reverse(entry.1)); + let mut top_blocked_paths = top_blocked_paths.iter().cloned().collect::>(); + top_blocked_paths + .sort_unstable_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0))); + top_blocked_paths.truncate(ENTRIES_PER_STATISTICS_TABLE); top_blocked_paths }, top_clients: { let top_clients = self.top_clients.lock().unwrap(); - let mut top_clients_iter = top_clients.iter(); - - let mut top_clients = (0..=ENTRIES_PER_STATISTICS_TABLE) - .filter_map(|_| { - let (ipv4, count) = top_clients_iter.next()?; - - Some((ipv4.to_string(), *count)) - }) + let mut top_clients = top_clients + .iter() + .map(|(ip, count)| (ip.to_string(), *count)) .collect::>(); - - top_clients.sort_by_key(|entry| std::cmp::Reverse(entry.1)); + top_clients.sort_unstable_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0))); + top_clients.truncate(ENTRIES_PER_STATISTICS_TABLE); top_clients }, + }; + if let Some(history) = &self.history { + let privacy = history.session_privacy(); + if privacy >= StatisticsPrivacy::HideDomains { + serialized.top_blocked_paths.clear(); + } + if privacy >= StatisticsPrivacy::HideDomainsAndClients { + serialized.top_clients.clear(); + } } + serialized } } diff --git a/privaxy/src/server/statistics/history.rs b/privaxy/src/server/statistics/history.rs new file mode 100644 index 0000000..7c776a7 --- /dev/null +++ b/privaxy/src/server/statistics/history.rs @@ -0,0 +1,682 @@ +//! Opt-in, bounded request history. Only the worker thread touches redb. +//! Request tasks redact identifiers before trying a non-blocking channel send. +use crate::configuration::{StatisticsConfig, StatisticsPrivacy}; +use portable_atomic::{AtomicU64, Ordering}; +use redb::{Database, ReadableDatabase, ReadableTable, ReadableTableMetadata, TableDefinition}; +use serde::{Deserialize, Serialize}; +use std::{ + collections::{BTreeMap, HashMap}, + fs::OpenOptions, + net::IpAddr, + path::PathBuf, + sync::{mpsc, Arc, Mutex, RwLock}, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, +}; +use tokio::sync::oneshot; + +const REQUESTS: TableDefinition = TableDefinition::new("requests_v1"); +const MODIFIED: TableDefinition = TableDefinition::new("modified_hours_v1"); +const META: TableDefinition<&str, u64> = TableDefinition::new("metadata"); +const FLUSH_INTERVAL: Duration = Duration::from_secs(5); +const QUEUE_CAPACITY: usize = 4096; +const BATCH_SIZE: usize = 512; +const HOUR_MS: u64 = 3_600_000; +const DAY_MS: u64 = 24 * HOUR_MS; +type StoreResult = Result>; + +#[derive(Debug, Clone, Default, Serialize)] +pub struct HistoryStatus { + pub active: bool, + pub error: Option, + pub dropped_events: u64, + pub stored_requests: u64, + pub database_bytes: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RequestRecord { + pub id: u64, + pub timestamp: u64, + pub domain: Option, + pub client: Option, + pub blocked: bool, +} + +impl RequestRecord { + fn redact(&mut self, privacy: StatisticsPrivacy) { + if privacy >= StatisticsPrivacy::HideDomains { + self.domain = None; + } + if privacy >= StatisticsPrivacy::HideDomainsAndClients { + self.client = None; + } + } +} + +#[derive(Debug, Clone, Copy, Default, Serialize)] +pub struct Counts { + pub proxied_requests: u64, + pub blocked_requests: u64, + pub modified_responses: u64, +} + +#[derive(Debug, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct HistoryQuery { + pub hours: Option, + pub before: Option, + pub blocked: Option, + pub limit: Option, +} + +impl HistoryQuery { + pub fn validate(&self) -> Result<(), String> { + if self.hours.is_some_and(|v| v == 0 || v > 365 * 24) { + return Err("History window must be between 1 and 8760 hours".into()); + } + if self.limit.is_some_and(|v| v == 0 || v > 200) { + return Err("History page size must be between 1 and 200".into()); + } + Ok(()) + } +} + +#[derive(Debug, Default, Serialize)] +pub struct HistorySnapshot { + pub status: HistoryStatus, + pub totals: Counts, + pub hourly: Vec<(u64, Counts)>, + pub top_domains: Vec<(String, u64)>, + pub top_blocked_domains: Vec<(String, u64)>, + pub top_clients: Vec<(String, u64)>, + pub requests: Vec, + pub next_before: Option, +} + +#[derive(Debug, Default)] +struct Policy { + config: StatisticsConfig, + generation: u64, + stopped: bool, + exclusions: Vec, +} + +enum Event { + Request(RequestRecord), + Modified(u64), +} + +enum Command { + Event(u64, Event), + Configure(StatisticsConfig, u64, oneshot::Sender>), + Query( + HistoryQuery, + oneshot::Sender>, + ), + Clear(u64, oneshot::Sender>), + Shutdown(oneshot::Sender>), +} + +#[derive(Debug, Clone)] +pub struct HistoryStore { + sender: mpsc::SyncSender, + policy: Arc>, + status: Arc>, + dropped: Arc, + operations: Arc>, +} + +impl HistoryStore { + pub async fn start(path: PathBuf, config: StatisticsConfig) -> Self { + let (sender, receiver) = mpsc::sync_channel(QUEUE_CAPACITY); + let status = Arc::new(Mutex::new(HistoryStatus::default())); + let dropped = Arc::new(AtomicU64::new(0)); + let store = Self { + sender, + policy: Arc::new(RwLock::new(Policy::default())), + status: status.clone(), + dropped: dropped.clone(), + operations: Arc::new(tokio::sync::Mutex::new(())), + }; + let worker = Worker { + path, + database: None, + config: StatisticsConfig::default(), + generation: 0, + pending: Vec::new(), + status, + dropped, + last_id: 0, + last_flush: Instant::now(), + last_compaction: Instant::now(), + last_prune: Instant::now() - Duration::from_secs(60), + }; + match std::thread::Builder::new() + .name("statistics".into()) + .spawn(move || worker.run(receiver)) + { + Ok(_) => { + if let Err(error) = store.configure(config).await { + log::error!("Persistent statistics unavailable: {error}"); + store.status.lock().unwrap().error = Some(error); + } + } + Err(error) => store.status.lock().unwrap().error = Some(error.to_string()), + } + store + } + + pub fn config(&self) -> StatisticsConfig { + self.policy.read().unwrap().config.clone() + } + + pub fn session_privacy(&self) -> StatisticsPrivacy { + let policy = self.policy.read().unwrap(); + if policy.config.enabled { + policy.config.privacy + } else { + StatisticsPrivacy::Full + } + } + + pub fn status(&self) -> HistoryStatus { + let mut status = self.status.lock().unwrap().clone(); + status.dropped_events = self.dropped.load(Ordering::Relaxed); + status + } + + pub fn is_excluded(&self, domain: Option<&str>) -> bool { + let policy = self.policy.read().unwrap(); + domain.is_some_and(|domain| matches_exclusion(&policy, domain)) + } + + pub fn request(&self, domain: Option<&str>, client: IpAddr, blocked: bool) { + let policy = self.policy.read().unwrap(); + if policy.stopped + || !policy.config.persists() + || domain.is_some_and(|domain| matches_exclusion(&policy, domain)) + { + return; + } + // No URL, path, query string, headers, request body, or client/domain hash + // is ever queued. Hidden values cannot be restored by changing settings. + let record = RequestRecord { + id: 0, + timestamp: now_ms(), + domain: if policy.config.privacy == StatisticsPrivacy::Full { + domain + .filter(|v| v.len() <= 253) + .map(|v| v.trim_end_matches('.').to_ascii_lowercase()) + } else { + None + }, + client: (policy.config.privacy < StatisticsPrivacy::HideDomainsAndClients) + .then(|| client.to_string()), + blocked, + }; + self.enqueue(&policy, Event::Request(record)); + } + + pub fn modified(&self) { + let policy = self.policy.read().unwrap(); + if !policy.stopped && policy.config.persists() { + self.enqueue(&policy, Event::Modified(now_ms())); + } + } + + fn enqueue(&self, policy: &Policy, event: Event) { + if self + .sender + .try_send(Command::Event(policy.generation, event)) + .is_err() + { + self.dropped.fetch_add(1, Ordering::Relaxed); + } + } + + async fn send( + &self, + command: Command, + reply: oneshot::Receiver>, + ) -> Result { + let sender = self.sender.clone(); + tokio::task::spawn_blocking(move || sender.send(command)) + .await + .map_err(|e| e.to_string())? + .map_err(|_| "Statistics worker has stopped".to_string())?; + reply + .await + .map_err(|_| "Statistics worker has stopped".to_string())? + } + + pub async fn configure(&self, config: StatisticsConfig) -> Result<(), String> { + config.validate()?; + let _operation = self.operations.lock().await; + // Pause collection across the command boundary. An old queued event + // cannot reintroduce identifiers after a privacy change or clear. + let generation = { + let mut policy = self.policy.write().unwrap(); + policy.stopped = true; + policy.generation += 1; + policy.generation + }; + let (tx, rx) = oneshot::channel(); + let result = self + .send(Command::Configure(config.clone(), generation, tx), rx) + .await; + let mut policy = self.policy.write().unwrap(); + policy.exclusions = config + .excluded_domains + .iter() + .map(|pattern| { + wildmatch::WildMatch::new( + &pattern.trim().trim_end_matches('.').to_ascii_lowercase(), + ) + }) + .collect(); + policy.config = config; + policy.stopped = result.is_err(); + result + } + + pub async fn query(&self, query: HistoryQuery) -> Result { + query.validate()?; + let (tx, rx) = oneshot::channel(); + self.send(Command::Query(query, tx), rx).await + } + + pub async fn clear(&self) -> Result<(), String> { + let _operation = self.operations.lock().await; + let generation = { + let mut policy = self.policy.write().unwrap(); + policy.stopped = true; + policy.generation += 1; + policy.generation + }; + let (tx, rx) = oneshot::channel(); + let result = self.send(Command::Clear(generation, tx), rx).await; + self.policy.write().unwrap().stopped = result.is_err(); + result + } + + pub async fn shutdown(&self) -> Result<(), String> { + let _operation = self.operations.lock().await; + self.policy.write().unwrap().stopped = true; + let (tx, rx) = oneshot::channel(); + self.send(Command::Shutdown(tx), rx).await + } +} + +struct Worker { + path: PathBuf, + database: Option, + config: StatisticsConfig, + generation: u64, + pending: Vec, + status: Arc>, + dropped: Arc, + last_id: u64, + last_flush: Instant, + last_compaction: Instant, + last_prune: Instant, +} + +impl Worker { + fn run(mut self, receiver: mpsc::Receiver) { + loop { + match receiver.recv_timeout(Duration::from_secs(1)) { + Ok(Command::Event(generation, event)) => { + if generation == self.generation + && self.database.is_some() + && self.status.lock().unwrap().error.is_none() + { + self.pending.push(event); + } else { + self.dropped.fetch_add(1, Ordering::Relaxed); + } + } + Ok(Command::Configure(config, generation, reply)) => { + let result = self + .configure(config, generation) + .map_err(|e| e.to_string()); + self.report(&result); + let _ = reply.send(result); + } + Ok(Command::Query(query, reply)) => { + let result = self + .flush() + .and_then(|()| self.query(query)) + .map_err(|e| e.to_string()); + if let Err(ref error) = result { + self.fail(error.clone()); + } + let _ = reply.send(result); + } + Ok(Command::Clear(generation, reply)) => { + self.generation = generation; + self.pending.clear(); + self.database = None; + let result = self.clear().map_err(|e| e.to_string()); + self.report(&result); + let _ = reply.send(result); + } + Ok(Command::Shutdown(reply)) => { + let result = self.flush().map_err(|e| e.to_string()); + self.database = None; + let _ = reply.send(result); + break; + } + Err(mpsc::RecvTimeoutError::Disconnected) => { + if let Err(error) = self.flush() { + self.fail(error.to_string()); + } + break; + } + Err(mpsc::RecvTimeoutError::Timeout) => {} + } + if self.pending.len() >= BATCH_SIZE || self.last_flush.elapsed() >= FLUSH_INTERVAL { + if let Err(error) = self.flush() { + self.fail(error.to_string()); + } + self.last_flush = Instant::now(); + } + } + } + + fn fail(&mut self, error: String) { + let mut status = self.status.lock().unwrap(); + if status.error.as_ref() != Some(&error) { + log::error!("Persistent statistics paused: {error}"); + } + status.active = false; + status.error = Some(error); + self.dropped + .fetch_add(self.pending.len() as u64, Ordering::Relaxed); + self.pending.clear(); + } + + fn report(&mut self, result: &Result<(), String>) { + if let Err(error) = result { + self.fail(error.clone()); + } + } + + fn configure(&mut self, config: StatisticsConfig, generation: u64) -> StoreResult<()> { + // Previously accepted events still have the previous privacy policy. + // Flush them before redacting the database with the new policy. + let flush_result = self.flush(); + if let Err(ref error) = flush_result { + self.fail(error.to_string()); + } + self.pending.clear(); + self.database = None; + self.config = config; + self.generation = generation; + *self.status.lock().unwrap() = HistoryStatus::default(); + if !self.config.persists() { + // Disabled stores are never opened. Anonymous mode may redact a + // database from an earlier opt-in, but never creates one or records + // new events. Switching back cannot recover old identifiers. + if self.config.enabled && self.path.exists() { + self.open()?; + self.database = None; + self.status.lock().unwrap().active = false; + } + return flush_result; + } + self.open()?; + self.last_prune = Instant::now() - Duration::from_secs(60); + self.flush()?; + Ok(()) + } + + fn open(&mut self) -> StoreResult<()> { + let mut options = OpenOptions::new(); + options.read(true).write(true).create(true).truncate(false); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let file = options.open(&self.path)?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + file.set_permissions(std::fs::Permissions::from_mode(0o600))?; + } + let mut database = Database::builder() + .set_cache_size(8 * 1024 * 1024) + .create_file(file)?; + let tx = database.begin_write()?; + let redact; + { + let mut meta = tx.open_table(META)?; + if meta.get("version")?.is_some_and(|v| v.value() != 1) { + return Err("Unsupported statistics database version".into()); + } + meta.insert("version", 1)?; + let previous = meta.get("privacy")?.map(|v| v.value()).unwrap_or(0); + redact = (self.config.privacy as u64) > previous; + meta.insert("privacy", self.config.privacy as u64)?; + } + { + let mut table = tx.open_table(REQUESTS)?; + self.last_id = table.last()?.map(|(id, _)| id.value()).unwrap_or(0); + if redact { + let ids = table + .iter()? + .map(|r| r.map(|(k, _)| k.value())) + .collect::, _>>()?; + for id in ids { + let mut record: RequestRecord = + serde_json::from_slice(table.get(id)?.unwrap().value())?; + record.redact(self.config.privacy); + table.insert(id, serde_json::to_vec(&record)?.as_slice())?; + } + } + } + tx.open_table(MODIFIED)?; + tx.commit()?; + if redact { + database.compact()?; + } + self.database = Some(database); + self.status.lock().unwrap().active = true; + Ok(()) + } + + fn clear(&mut self) -> StoreResult<()> { + match std::fs::remove_file(&self.path) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error.into()), + } + self.last_id = 0; + self.dropped.store(0, Ordering::Relaxed); + *self.status.lock().unwrap() = HistoryStatus::default(); + if self.config.persists() { + self.open()?; + } + Ok(()) + } + + fn flush(&mut self) -> StoreResult<()> { + if self.status.lock().unwrap().error.is_some() { + return Ok(()); + } + let Some(database) = self.database.as_ref() else { + return Ok(()); + }; + let prune = self.last_prune.elapsed() >= Duration::from_secs(60); + if self.pending.is_empty() && !prune { + return Ok(()); + } + let now = now_ms(); + let cutoff = now.saturating_sub(u64::from(self.config.retention_days) * DAY_MS); + let tx = database.begin_write()?; + let stored; + { + let mut requests = tx.open_table(REQUESTS)?; + let mut modified = tx.open_table(MODIFIED)?; + for event in &self.pending { + match event { + Event::Request(record) => { + let mut record = record.clone(); + // Microsecond-sized IDs remain ordered and fit exactly in + // JavaScript numbers for centuries. The clock may step back. + self.last_id = (record.timestamp * 1000).max(self.last_id + 1); + record.id = self.last_id; + requests.insert(record.id, serde_json::to_vec(&record)?.as_slice())?; + } + Event::Modified(timestamp) => { + let hour = timestamp / HOUR_MS * HOUR_MS; + let count = modified.get(hour)?.map(|v| v.value()).unwrap_or(0) + 1; + modified.insert(hour, count)?; + } + } + } + // Also inspect stored timestamps: an NTP clock adjustment must not + // make old rows immortal merely because their IDs are monotonic. + if prune { + let mut invalid_row = false; + requests.retain(|_, bytes| { + match serde_json::from_slice::(bytes) { + Ok(record) => record.timestamp >= cutoff, + Err(_) => { + invalid_row = true; + true + } + } + })?; + if invalid_row { + return Err("Invalid record in statistics database; recording paused".into()); + } + } + while requests.len()? > u64::from(self.config.max_entries) { + let first = requests.first()?.unwrap().0.value(); + requests.remove(first)?; + } + modified.retain(|hour, _| hour + HOUR_MS > cutoff)?; + stored = requests.len()?; + } + tx.commit()?; + self.pending.clear(); + if prune { + self.last_prune = Instant::now(); + } + if self.last_compaction.elapsed() >= Duration::from_secs(3600) { + self.database.as_mut().unwrap().compact()?; + self.last_compaction = Instant::now(); + } + let mut status = self.status.lock().unwrap(); + status.stored_requests = stored; + status.database_bytes = std::fs::metadata(&self.path)?.len(); + Ok(()) + } + + fn query(&self, query: HistoryQuery) -> StoreResult { + let mut snapshot = HistorySnapshot { + status: self.status.lock().unwrap().clone(), + ..Default::default() + }; + snapshot.status.dropped_events = self.dropped.load(Ordering::Relaxed); + let Some(database) = &self.database else { + return Ok(snapshot); + }; + // A failed privacy transition must never reveal previously collected data. + if snapshot.status.error.is_some() { + return Ok(snapshot); + } + let hours = query + .hours + .unwrap_or(24) + .min(self.config.retention_days * 24); + let cutoff = (now_ms() / HOUR_MS * HOUR_MS).saturating_sub(u64::from(hours - 1) * HOUR_MS); + let limit = query.limit.unwrap_or(50); + let tx = database.begin_read()?; + let table = tx.open_table(REQUESTS)?; + let mut hourly: BTreeMap = BTreeMap::new(); + let mut domains = HashMap::new(); + let mut blocked_domains = HashMap::new(); + let mut clients = HashMap::new(); + // IDs are at least timestamp * 1000, even after a clock correction, + // so this range skips older pages without losing in-window records. + for entry in table.range(cutoff.saturating_mul(1000)..)?.rev() { + let (_, bytes) = entry?; + let record: RequestRecord = serde_json::from_slice(bytes.value())?; + if record.timestamp < cutoff { + continue; + } + let counts = hourly + .entry(record.timestamp / HOUR_MS * HOUR_MS) + .or_default(); + if record.blocked { + snapshot.totals.blocked_requests += 1; + counts.blocked_requests += 1; + } else { + snapshot.totals.proxied_requests += 1; + counts.proxied_requests += 1; + } + if let Some(ref domain) = record.domain { + *domains.entry(domain.clone()).or_insert(0) += 1; + if record.blocked { + *blocked_domains.entry(domain.clone()).or_insert(0) += 1; + } + } + if let Some(ref client) = record.client { + *clients.entry(client.clone()).or_insert(0) += 1; + } + if query.before.is_none_or(|id| record.id < id) + && query.blocked.is_none_or(|v| record.blocked == v) + { + if snapshot.requests.len() < limit { + snapshot.requests.push(record); + } else if snapshot.next_before.is_none() { + snapshot.next_before = snapshot.requests.last().map(|r| r.id); + } + } + } + for entry in tx + .open_table(MODIFIED)? + .range(cutoff / HOUR_MS * HOUR_MS..)? + { + let (hour, value) = entry?; + let count = value.value(); + snapshot.totals.modified_responses += count; + hourly.entry(hour.value()).or_default().modified_responses = count; + } + snapshot.hourly = hourly.into_iter().collect(); + snapshot.top_domains = top_entries(domains); + snapshot.top_blocked_domains = top_entries(blocked_domains); + snapshot.top_clients = top_entries(clients); + Ok(snapshot) + } +} + +fn top_entries(entries: HashMap) -> Vec<(String, u64)> { + let mut entries: Vec<_> = entries.into_iter().collect(); + entries.sort_unstable_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0))); + entries.truncate(50); + entries +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64 +} + +fn matches_exclusion(policy: &Policy, domain: &str) -> bool { + if policy.exclusions.is_empty() { + return false; + } + let domain = domain.trim_end_matches('.').to_ascii_lowercase(); + policy + .exclusions + .iter() + .any(|pattern| pattern.matches(&domain)) +} + +#[cfg(test)] +mod tests; diff --git a/privaxy/src/server/statistics/history/tests.rs b/privaxy/src/server/statistics/history/tests.rs new file mode 100644 index 0000000..d8d7fed --- /dev/null +++ b/privaxy/src/server/statistics/history/tests.rs @@ -0,0 +1,450 @@ +use super::*; +use crate::statistics::Statistics; + +struct Fixture(PathBuf); +impl Fixture { + fn new() -> Self { + let path = std::env::temp_dir().join(format!( + "privaxy-stats-test-{}", + crate::configuration::generate_random_hex(12) + )); + std::fs::create_dir(&path).unwrap(); + Self(path) + } + fn database(&self) -> PathBuf { + self.0.join("statistics.redb") + } +} +impl Drop for Fixture { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +fn enabled(privacy: StatisticsPrivacy) -> StatisticsConfig { + StatisticsConfig { + enabled: true, + privacy, + ..Default::default() + } +} +fn client() -> IpAddr { + "192.0.2.10".parse().unwrap() +} +async fn snapshot(store: &HistoryStore) -> HistorySnapshot { + store.query(HistoryQuery::default()).await.unwrap() +} + +#[tokio::test] +async fn disabled_and_anonymous_do_not_create_databases_or_retroactively_collect() { + for config in [ + StatisticsConfig::default(), + enabled(StatisticsPrivacy::Anonymous), + ] { + let fixture = Fixture::new(); + let store = HistoryStore::start(fixture.database(), config).await; + store.request(Some("private.example.test"), client(), true); + store.modified(); + assert!(snapshot(&store).await.requests.is_empty()); + assert!(!fixture.database().exists()); + store + .configure(enabled(StatisticsPrivacy::Full)) + .await + .unwrap(); + assert!(snapshot(&store).await.requests.is_empty()); + store.shutdown().await.unwrap(); + } +} + +#[tokio::test] +async fn requests_and_modified_counts_survive_restart_without_url_details() { + let fixture = Fixture::new(); + let config = enabled(StatisticsPrivacy::Full); + let store = HistoryStore::start(fixture.database(), config.clone()).await; + let uri: http::Uri = "https://EXAMPLE.test:443/private-account?secret=do-not-store" + .parse() + .unwrap(); + store.request(uri.host(), client(), true); + store.request(Some("allowed.example.test"), client(), false); + store.modified(); + store.shutdown().await.unwrap(); + let store = HistoryStore::start(fixture.database(), config).await; + let data = snapshot(&store).await; + assert_eq!(data.totals.blocked_requests, 1); + assert_eq!(data.totals.proxied_requests, 1); + assert_eq!(data.totals.modified_responses, 1); + assert_eq!(data.requests[1].domain.as_deref(), Some("example.test")); + assert_eq!(data.requests[1].client.as_deref(), Some("192.0.2.10")); + assert_eq!(data.top_blocked_domains, vec![("example.test".into(), 1)]); + store.shutdown().await.unwrap(); + let bytes = std::fs::read(fixture.database()).unwrap(); + let text = String::from_utf8_lossy(&bytes); + assert!(!text.contains("private-account")); + assert!(!text.contains("do-not-store")); + assert!(!text.contains(":443")); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + assert_eq!( + std::fs::metadata(fixture.database()) + .unwrap() + .permissions() + .mode() + & 0o777, + 0o600 + ); + } +} + +#[tokio::test] +async fn identifiers_are_redacted_before_queueing_and_cannot_be_recovered() { + for privacy in [ + StatisticsPrivacy::HideDomains, + StatisticsPrivacy::HideDomainsAndClients, + ] { + let fixture = Fixture::new(); + let store = HistoryStore::start(fixture.database(), enabled(privacy)).await; + store.request(Some("private.example.test"), client(), true); + store + .configure(enabled(StatisticsPrivacy::Full)) + .await + .unwrap(); + let data = snapshot(&store).await; + assert_eq!(data.requests.len(), 1); + assert!(data.requests[0].domain.is_none()); + assert_eq!( + data.requests[0].client.is_none(), + privacy == StatisticsPrivacy::HideDomainsAndClients + ); + store.shutdown().await.unwrap(); + let bytes = std::fs::read(fixture.database()).unwrap(); + let text = String::from_utf8_lossy(&bytes); + assert!(!text.contains("private.example.test")); + if privacy == StatisticsPrivacy::HideDomainsAndClients { + assert!(!text.contains("192.0.2.10")); + } + } +} + +#[tokio::test] +async fn increasing_privacy_redacts_old_rows_including_before_anonymous_mode() { + for privacy in [ + StatisticsPrivacy::HideDomains, + StatisticsPrivacy::HideDomainsAndClients, + StatisticsPrivacy::Anonymous, + ] { + let fixture = Fixture::new(); + let store = HistoryStore::start(fixture.database(), enabled(StatisticsPrivacy::Full)).await; + store.request(Some("private.example.test"), client(), true); + // The pending record must also be covered by this transition. + store.configure(enabled(privacy)).await.unwrap(); + store + .configure(enabled(StatisticsPrivacy::Full)) + .await + .unwrap(); + let data = snapshot(&store).await; + assert_eq!(data.requests.len(), 1); + assert!(data.requests[0].domain.is_none()); + assert_eq!( + data.requests[0].client.is_none(), + privacy >= StatisticsPrivacy::HideDomainsAndClients + ); + store.shutdown().await.unwrap(); + } +} + +#[tokio::test] +async fn retention_entry_cap_and_cursor_pagination_are_enforced() { + let fixture = Fixture::new(); + let mut config = enabled(StatisticsPrivacy::Full); + config.max_entries = 100; + let store = HistoryStore::start(fixture.database(), config.clone()).await; + for i in 0..150 { + store.request( + Some(&format!("host-{i}.example.test")), + client(), + i % 2 == 0, + ); + } + let query = || HistoryQuery { + limit: Some(10), + blocked: Some(true), + ..Default::default() + }; + let first = store.query(query()).await.unwrap(); + assert_eq!(first.status.stored_requests, 100); + assert_eq!(first.totals.blocked_requests, 50); + assert_eq!(first.totals.proxied_requests, 50); + assert!(first.requests.iter().all(|r| r.blocked)); + assert_eq!(first.requests.len(), 10); + let second = store + .query(HistoryQuery { + before: first.next_before, + ..query() + }) + .await + .unwrap(); + assert_eq!(second.requests.len(), 10); + assert!(second + .requests + .iter() + .all(|r| r.id < first.requests.last().unwrap().id)); + assert_eq!(second.totals.blocked_requests, 50); + let generation = store.policy.read().unwrap().generation; + store + .sender + .send(Command::Event( + generation, + Event::Request(RequestRecord { + id: 0, + timestamp: now_ms() - 2 * DAY_MS, + domain: Some("old.example.test".into()), + client: None, + blocked: true, + }), + )) + .unwrap(); + config.retention_days = 1; + store.configure(config).await.unwrap(); + let data = snapshot(&store).await; + // Old timestamp with a newer ID is pruned too (clock corrections). + assert_eq!(data.status.stored_requests, 99); + store.shutdown().await.unwrap(); +} + +#[tokio::test] +async fn clear_and_disable_do_not_replay_queued_records() { + let fixture = Fixture::new(); + let config = enabled(StatisticsPrivacy::Full); + let store = HistoryStore::start(fixture.database(), config.clone()).await; + for _ in 0..100 { + store.request(Some("old.example.test"), client(), true); + } + store.clear().await.unwrap(); + assert!(snapshot(&store).await.requests.is_empty()); + store.configure(StatisticsConfig::default()).await.unwrap(); + store.request(Some("disabled.example.test"), client(), false); + store.clear().await.unwrap(); + assert!(!fixture.database().exists()); + store.configure(config).await.unwrap(); + assert!(snapshot(&store).await.requests.is_empty()); + store.shutdown().await.unwrap(); +} + +#[tokio::test] +async fn corrupt_storage_is_reported_without_truncation_or_breaking_session_counts() { + let fixture = Fixture::new(); + std::fs::write(fixture.database(), b"corrupt-fixture-do-not-truncate").unwrap(); + let history = HistoryStore::start(fixture.database(), enabled(StatisticsPrivacy::Full)).await; + assert!(history.status().error.is_some()); + assert_eq!( + std::fs::read(fixture.database()).unwrap(), + b"corrupt-fixture-do-not-truncate" + ); + let stats = Statistics { + history: Some(history.clone()), + ..Statistics::new() + }; + stats.record_request(Some("example.test"), client(), true); + assert_eq!(stats.get_serialized().blocked_requests, 1); + history.clear().await.unwrap(); + assert!(history.status().error.is_none()); + stats.record_request(Some("example.test"), client(), true); + assert_eq!(snapshot(&history).await.totals.blocked_requests, 1); + history.shutdown().await.unwrap(); +} + +#[tokio::test] +async fn excluded_domains_skip_counters_history_and_the_live_log_even_when_disabled() { + for persistent in [false, true] { + let fixture = Fixture::new(); + let config = StatisticsConfig { + enabled: persistent, + privacy: StatisticsPrivacy::Full, + excluded_domains: vec!["EXAMPLE.test".into(), "*.private.test".into()], + ..Default::default() + }; + let history = HistoryStore::start(fixture.database(), config).await; + let stats = Statistics { + history: Some(history.clone()), + ..Statistics::new() + }; + for authority in ["example.test:443", "Sub.Private.Test.:8443"] { + let uri: http::Uri = format!("https://{authority}/secret").parse().unwrap(); + let request_stats = stats.for_domain(uri.host()); + assert!(!request_stats.should_collect()); + request_stats.record_request(uri.host(), client(), true); + request_stats.increment_modified_responses(); + // The history collector independently enforces the same policy. + history.request(uri.host(), client(), true); + } + let data = stats.get_serialized(); + assert_eq!(data.blocked_requests, 0); + assert_eq!(data.modified_responses, 0); + assert!(data.top_clients.is_empty()); + assert!(snapshot(&history).await.requests.is_empty()); + for domain in ["private.test", "sub.example.test", "example.test.evil.test"] { + assert!(stats.for_domain(Some(domain)).should_collect()); + } + history.shutdown().await.unwrap(); + } +} + +#[test] +fn rejects_unbounded_settings_and_urls_in_domain_exclusions() { + for pattern in [ + "https://example.test", + "example.test:443", + "example.test/path", + " ", + ] { + let config = StatisticsConfig { + excluded_domains: vec![pattern.into()], + ..Default::default() + }; + assert!(config.validate().is_err()); + } + for days in [0, 366] { + assert!(StatisticsConfig { + retention_days: days, + ..Default::default() + } + .validate() + .is_err()); + } + for limit in [0, 99, 1_000_001] { + assert!(StatisticsConfig { + max_entries: limit, + ..Default::default() + } + .validate() + .is_err()); + } + assert!(HistoryQuery { + limit: Some(201), + ..Default::default() + } + .validate() + .is_err()); +} + +#[test] +fn a_full_queue_drops_events_without_blocking_requests() { + let (sender, receiver) = mpsc::sync_channel(1); + let store = HistoryStore { + sender, + policy: Arc::new(RwLock::new(Policy { + config: enabled(StatisticsPrivacy::HideDomainsAndClients), + ..Default::default() + })), + status: Arc::new(Mutex::new(HistoryStatus::default())), + dropped: Arc::new(AtomicU64::new(0)), + operations: Arc::new(tokio::sync::Mutex::new(())), + }; + store.request(Some("private.example.test"), client(), true); + store.request(Some("private.example.test"), client(), true); + assert_eq!(store.status().dropped_events, 1); + let Command::Event(_, Event::Request(record)) = receiver.try_recv().unwrap() else { + panic!("expected a request"); + }; + assert!(record.domain.is_none()); + assert!(record.client.is_none()); +} + +#[derive(Debug)] +struct FailingDisk { + inner: redb::backends::FileBackend, + full: Arc, +} +impl redb::StorageBackend for FailingDisk { + fn len(&self) -> std::io::Result { + self.inner.len() + } + fn read(&self, offset: u64, out: &mut [u8]) -> std::io::Result<()> { + self.inner.read(offset, out) + } + fn set_len(&self, len: u64) -> std::io::Result<()> { + if self.full.load(Ordering::Relaxed) { + return Err(std::io::ErrorKind::StorageFull.into()); + } + self.inner.set_len(len) + } + fn sync_data(&self) -> std::io::Result<()> { + if self.full.load(Ordering::Relaxed) { + return Err(std::io::ErrorKind::StorageFull.into()); + } + self.inner.sync_data() + } + fn write(&self, offset: u64, bytes: &[u8]) -> std::io::Result<()> { + if self.full.load(Ordering::Relaxed) { + return Err(std::io::ErrorKind::StorageFull.into()); + } + self.inner.write(offset, bytes) + } + fn close(&self) -> std::io::Result<()> { + self.inner.close() + } +} + +#[test] +fn disk_full_pauses_recording_and_preserves_committed_history() { + let fixture = Fixture::new(); + let mut worker = Worker { + path: fixture.database(), + database: None, + config: enabled(StatisticsPrivacy::Full), + generation: 0, + pending: Vec::new(), + status: Arc::new(Mutex::new(HistoryStatus::default())), + dropped: Arc::new(AtomicU64::new(0)), + last_id: 0, + last_flush: Instant::now(), + last_compaction: Instant::now(), + last_prune: Instant::now(), + }; + worker.open().unwrap(); + let event = || { + Event::Request(RequestRecord { + id: 0, + timestamp: now_ms(), + domain: Some("example.test".into()), + client: None, + blocked: true, + }) + }; + worker.pending.push(event()); + worker.flush().unwrap(); + worker.database = None; + let full = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let file = OpenOptions::new() + .read(true) + .write(true) + .open(fixture.database()) + .unwrap(); + let backend = FailingDisk { + inner: redb::backends::FileBackend::new(file).unwrap(), + full: full.clone(), + }; + worker.database = Some(Database::builder().create_with_backend(backend).unwrap()); + full.store(true, Ordering::Relaxed); + worker.pending.push(event()); + let error = worker.flush().unwrap_err(); + worker.fail(error.to_string()); + assert!(!worker.status.lock().unwrap().active); + assert!(worker.status.lock().unwrap().error.is_some()); + assert!(worker.pending.is_empty()); + assert_eq!(worker.dropped.load(Ordering::Relaxed), 1); + worker.database = None; + full.store(false, Ordering::Relaxed); + worker + .configure(enabled(StatisticsPrivacy::Full), 1) + .unwrap(); + assert_eq!( + worker + .query(HistoryQuery::default()) + .unwrap() + .totals + .blocked_requests, + 1 + ); +} diff --git a/privaxy/src/server/web_gui/history.rs b/privaxy/src/server/web_gui/history.rs new file mode 100644 index 0000000..8f9d5e1 --- /dev/null +++ b/privaxy/src/server/web_gui/history.rs @@ -0,0 +1,122 @@ +use super::{json_status, with_arc, with_configuration_save_lock}; +use crate::{ + configuration::{Configuration, StatisticsConfig}, + statistics::{history::HistoryQuery, Statistics}, +}; +use http::StatusCode; +use serde::Serialize; +use std::{convert::Infallible, sync::Arc}; +use warp::{filters::BoxedFilter, Filter, Reply}; + +#[derive(Serialize)] +struct SettingsResponse { + config: StatisticsConfig, + status: crate::statistics::history::HistoryStatus, +} + +fn error(message: &str) -> Box { + json_status(StatusCode::INTERNAL_SERVER_ERROR, message) +} + +async fn get_settings(statistics: Statistics) -> Result, Infallible> { + let Some(history) = statistics.history else { + return Ok(error("Statistics worker unavailable")); + }; + Ok(match Configuration::read_from_home().await { + Ok(configuration) => Box::new(warp::reply::json(&SettingsResponse { + config: configuration.statistics, + status: history.status(), + })), + Err(err) => error(&err.to_string()), + }) +} + +async fn put_settings( + config: StatisticsConfig, + statistics: Statistics, + lock: Arc>, +) -> Result, Infallible> { + if let Err(message) = config.validate() { + return Ok(json_status(StatusCode::UNPROCESSABLE_ENTITY, &message)); + } + let _guard = lock.lock().await; + let mut configuration = match Configuration::read_from_home().await { + Ok(configuration) => configuration, + Err(err) => return Ok(error(&err.to_string())), + }; + configuration.statistics = config; + if let Err(err) = configuration.save().await { + return Ok(error(&err.to_string())); + } + // A saved opt-in and a working database are distinct; report disk errors + // alongside the saved config. Saving again retries without proxy reloads. + if let Err(err) = statistics.configure_history(configuration.statistics).await { + log::error!("Could not apply statistics settings: {err}"); + } + get_settings(statistics).await +} + +async fn get_history( + query: HistoryQuery, + statistics: Statistics, +) -> Result, Infallible> { + if let Err(message) = query.validate() { + return Ok(json_status(StatusCode::UNPROCESSABLE_ENTITY, &message)); + } + let Some(history) = statistics.history else { + return Ok(error("Statistics worker unavailable")); + }; + Ok(match history.query(query).await { + Ok(snapshot) => Box::new(warp::reply::json(&snapshot)), + Err(err) => error(&err), + }) +} + +async fn clear_history( + statistics: Statistics, + lock: Arc>, +) -> Result, Infallible> { + let _guard = lock.lock().await; + let Some(history) = statistics.history else { + return Ok(error("Statistics worker unavailable")); + }; + Ok(match history.clear().await { + Ok(()) => Box::new(warp::reply::with_status("", StatusCode::NO_CONTENT)), + Err(err) => error(&err), + }) +} + +pub(super) fn settings_routes( + statistics: Statistics, + lock: Arc>, +) -> BoxedFilter<(impl Reply,)> { + let get = warp::get() + .and(warp::path::end()) + .and(with_arc(statistics.clone())) + .and_then(get_settings); + let put = warp::put() + .and(warp::path::end()) + .and(warp::body::content_length_limit(512 * 1024)) + .and(warp::body::json()) + .and(with_arc(statistics)) + .and(with_configuration_save_lock(lock)) + .and_then(put_settings); + get.or(put).boxed() +} + +pub(super) fn routes( + statistics: Statistics, + lock: Arc>, +) -> BoxedFilter<(impl Reply,)> { + let get = warp::get() + .and(warp::path::end()) + .and(warp::query::()) + .and(with_arc(statistics.clone())) + .and_then(get_history); + let delete = warp::delete() + .and(warp::path::end()) + .and(with_arc(statistics)) + .and(with_configuration_save_lock(lock)) + .and_then(clear_history); + get.or(delete).boxed() +} diff --git a/privaxy/src/server/web_gui/mod.rs b/privaxy/src/server/web_gui/mod.rs index 54209ce..2486462 100644 --- a/privaxy/src/server/web_gui/mod.rs +++ b/privaxy/src/server/web_gui/mod.rs @@ -23,6 +23,7 @@ pub(crate) mod events; pub(crate) mod exclusions; mod filterlists; pub(crate) mod filters; +mod history; mod inclusions; pub(crate) mod logs; mod pac; @@ -155,7 +156,21 @@ fn create_api_routes( ws.on_upgrade(move |websocket| events::events(websocket, events_sender)) }); + let history_route = warp::path("history") + .and(require_auth.clone()) + .and(history::routes( + statistics.clone(), + configuration_save_lock.clone(), + )); + let statistics_settings_route = warp::path!("settings" / "statistics" / ..) + .and(require_auth.clone()) + .and(history::settings_routes( + statistics.clone(), + configuration_save_lock.clone(), + )); + let statistics_route = warp::path("statistics") + .and(warp::path::end()) .and(require_auth.clone()) .and(warp::ws()) .map(move |ws: warp::ws::Ws| { @@ -257,6 +272,8 @@ fn create_api_routes( let api_inner = auth_routes .or(events_route) .or(statistics_route) + .or(history_route) + .or(statistics_settings_route) .or(logs_route) .or(filters_route) .or(custom_filters_route) diff --git a/web_frontend/Cargo.toml b/web_frontend/Cargo.toml index 38caf90..2ba4290 100644 --- a/web_frontend/Cargo.toml +++ b/web_frontend/Cargo.toml @@ -12,6 +12,7 @@ num-format = "0.4.0" serde = { version = "1.0.137", features = ["derive"] } gloo-utils = "0.3" wasm-bindgen = "0.2.125" +js-sys = "0.3" wasm-bindgen-futures = "0.4" futures = "0.3.21" log = "0.4.17" diff --git a/web_frontend/src/dashboard.rs b/web_frontend/src/dashboard.rs index 9972a02..83015a6 100644 --- a/web_frontend/src/dashboard.rs +++ b/web_frontend/src/dashboard.rs @@ -112,6 +112,7 @@ impl Component for Dashboard { if !self.connected {

{"Connecting to statistics… Retrying automatically. Displayed counts may be out of date."}

} +

{"Live statistics since this process started. "}{"View saved history and trends"}

@@ -150,7 +151,7 @@ impl Component for Dashboard {
-

{"Top blocked paths"}

+

{"Top blocked domains"}

    diff --git a/web_frontend/src/history.rs b/web_frontend/src/history.rs new file mode 100644 index 0000000..4c82555 --- /dev/null +++ b/web_frontend/src/history.rs @@ -0,0 +1,281 @@ +use crate::button::{get_css, ButtonColor}; +use crate::{api, statistics_settings::HistoryStatus}; +use gloo_timers::callback::Interval; +use num_format::{Locale, ToFormattedString}; +use serde::Deserialize; +use std::collections::BTreeMap; +use wasm_bindgen_futures::spawn_local; +use web_sys::HtmlSelectElement; +use yew::prelude::*; + +#[derive(Clone, Debug, Default, Deserialize)] +pub struct Counts { + proxied_requests: u64, + blocked_requests: u64, + modified_responses: u64, +} + +#[derive(Clone, Debug, Deserialize)] +struct Record { + id: u64, + timestamp: u64, + domain: Option, + client: Option, + blocked: bool, +} + +#[derive(Clone, Debug, Deserialize)] +pub struct Snapshot { + status: HistoryStatus, + totals: Counts, + hourly: Vec<(u64, Counts)>, + top_domains: Vec<(String, u64)>, + top_blocked_domains: Vec<(String, u64)>, + top_clients: Vec<(String, u64)>, + requests: Vec, + next_before: Option, +} + +pub enum Message { + Load, + Poll, + Loaded(u64, Result), + Hours(String), + Filter(String), + Older, + Newest, +} + +pub struct HistoryPage { + snapshot: Option, + hours: u16, + filter: String, + before: Option, + loading: bool, + error: Option, + generation: u64, + _interval: Interval, +} + +impl Component for HistoryPage { + type Message = Message; + type Properties = (); + fn create(ctx: &Context) -> Self { + ctx.link().send_message(Message::Load); + let link = ctx.link().clone(); + Self { + snapshot: None, + hours: 24, + filter: String::new(), + before: None, + loading: false, + error: None, + generation: 0, + _interval: Interval::new(30_000, move || link.send_message(Message::Poll)), + } + } + + fn update(&mut self, ctx: &Context, message: Message) -> bool { + match message { + Message::Poll if self.loading || self.before.is_some() => return false, + Message::Load | Message::Poll => { + self.load(ctx); + } + Message::Loaded(generation, result) => { + if generation != self.generation { + return false; + } + self.loading = false; + match result { + Ok(snapshot) => { + self.error = snapshot.status.error.clone(); + self.snapshot = Some(snapshot); + } + Err(error) => self.error = Some(error), + } + } + Message::Hours(hours) => { + self.hours = hours.parse().unwrap_or(24); + self.before = None; + self.snapshot = None; + self.load(ctx); + } + Message::Filter(filter) => { + self.filter = filter; + self.before = None; + self.load(ctx); + } + Message::Older => { + if self.loading { + return false; + } + self.before = self.snapshot.as_ref().and_then(|s| s.next_before); + self.load(ctx); + } + Message::Newest => { + self.before = None; + self.load(ctx); + } + } + true + } + + fn view(&self, ctx: &Context) -> Html { + let on_hours = ctx.link().callback(|e: Event| { + Message::Hours(e.target_unchecked_into::().value()) + }); + let on_filter = ctx.link().callback(|e: Event| { + Message::Filter(e.target_unchecked_into::().value()) + }); + html! { <> +
    +

    {"History"}

    +
    + + + + {"Statistics settings"} +
    +
    +

    {"Saved statistics survive restarts. Totals and top lists cover retained requests in this window; the request filter and pagination only affect the log. Times are UTC. The latest page refreshes every 30 seconds."}

    + if let Some(error) = &self.error { } + if self.loading {

    {"Loading history…"}

    } + if let Some(snapshot) = &self.snapshot { + if !snapshot.status.active && snapshot.status.error.is_none() { +

    {"Persistent statistics are off or anonymous mode is selected. Enable history in Statistics settings to start collecting it."}

    + } else if snapshot.status.active { + if snapshot.status.dropped_events > 0 { } +
    + {for [("Proxied requests", snapshot.totals.proxied_requests), ("Blocked requests", snapshot.totals.blocked_requests), ("Modified responses", snapshot.totals.modified_responses)].into_iter().map(|(label, count)| html! { +
    {label}
    {count.to_formatted_string(&Locale::en)}
    + })} +
    +
    +

    {"Requests over time"}

    + {self.chart(snapshot)} +
    +
    + {top_list("Top domains", &snapshot.top_domains)} + {top_list("Top blocked domains", &snapshot.top_blocked_domains)} + {top_list("Top clients", &snapshot.top_clients)} +
    +
    +
    +

    {"Request history"}

    + +
    + if snapshot.requests.is_empty() {

    {"No saved requests in this window."}

    } +
    + + + {for snapshot.requests.iter().map(|record| html! { + + + + + + + })} +
    {"Time (UTC)"}{"Domain"}{"Client"}{"Result"}
    {timestamp(record.timestamp)}{record.domain.as_deref().unwrap_or("Hidden")}{record.client.as_deref().unwrap_or("Hidden")}{if record.blocked {"Blocked"} else {"Proxied"}}
    +
    +
    + + +
    +
    + } + } + } + } +} + +impl HistoryPage { + fn load(&mut self, ctx: &Context) { + self.loading = true; + self.error = None; + self.generation += 1; + let generation = self.generation; + let mut url = format!("/api/history?hours={}&limit=50", self.hours); + if let Some(before) = self.before { + url.push_str(&format!("&before={before}")); + } + if !self.filter.is_empty() { + url.push_str(&format!("&blocked={}", self.filter)); + } + let link = ctx.link().clone(); + spawn_local(async move { + link.send_message(Message::Loaded(generation, api::get_json(&url).await)); + }); + } + + fn chart(&self, snapshot: &Snapshot) -> Html { + let bucket_ms = if self.hours <= 24 { + 3_600_000 + } else { + 86_400_000 + }; + let mut buckets = BTreeMap::::new(); + for (time, counts) in &snapshot.hourly { + let bucket = buckets.entry(time / bucket_ms * bucket_ms).or_default(); + bucket.0 += counts.proxied_requests; + bucket.1 += counts.blocked_requests; + } + if buckets.is_empty() { + return html! {

    {"No saved activity yet."}

    }; + } + let now = js_sys::Date::now() as u64; + let last = now / bucket_ms * bucket_ms; + let first = if self.hours == 8760 { + (*buckets.first_key_value().unwrap().0).min(last) + } else { + (now / 3_600_000 * 3_600_000).saturating_sub(u64::from(self.hours - 1) * 3_600_000) + / bucket_ms + * bucket_ms + }; + let max = buckets + .values() + .map(|(p, b)| p + b) + .max() + .unwrap_or(1) + .max(1) as f64; + html! { <> +

    {format!("Peak: {} requests per {}", max as u64, if self.hours <= 24 {"hour"} else {"day"})}

    + +
    {timestamp(first)}{timestamp(last)}
    +

    {if self.hours <= 24 {"Hourly totals · Blue: proxied · Red: blocked"} else {"Daily totals · Blue: proxied · Red: blocked"}}

    + } + } +} + +fn top_list(title: &str, entries: &[(String, u64)]) -> Html { + html! {
    +

    {title}

    + if entries.is_empty() {

    {"No visible data at this privacy level or in this window."}

    } +
      {for entries.iter().take(10).map(|(label, count)| html! { +
    1. {label}{count.to_formatted_string(&Locale::en)}
      2. + })}
    +
    } +} + +fn timestamp(value: u64) -> String { + let iso = js_sys::Date::new(&wasm_bindgen::JsValue::from_f64(value as f64)) + .to_iso_string() + .as_string() + .unwrap_or_default(); + iso.get(..19).unwrap_or(&iso).replace('T', " ") +} diff --git a/web_frontend/src/main.rs b/web_frontend/src/main.rs index 72909a1..e354a71 100644 --- a/web_frontend/src/main.rs +++ b/web_frontend/src/main.rs @@ -19,6 +19,7 @@ mod filter_failures; mod filterlists; mod filters; mod general; +mod history; mod inclusions; mod live_stream; mod logs; @@ -27,6 +28,7 @@ mod requests; mod save_button; mod settings; mod settings_textarea; +mod statistics_settings; mod submit_banner; mod tls_failures; mod userscript_edit; @@ -43,6 +45,8 @@ enum Route { Dashboard, #[at("/requests")] Requests, + #[at("/history")] + History, #[at("/settings/:s")] Settings, #[at("/exclude")] @@ -109,6 +113,7 @@ fn switch(route: Route) -> Html {
    classes={ get_classes(route, Route::Dashboard) } to={Route::Dashboard}>{ "Dashboard" }> classes={ get_classes(route, Route::Requests) } to={Route::Requests}>{ "Requests" }> + classes={ get_classes(route, Route::History) } to={Route::History}>{ "History" }> classes={ get_classes(route, Route::Settings) } to={settings::SettingsRoute::Filters}>{ "Settings" }>
    @@ -129,6 +134,10 @@ fn switch(route: Route) -> Html { Route::Settings => { html! {<>{navigation}
    render={settings::switch_settings} />
    } } + Route::History => { + set_title("History"); + html! { <>{navigation}
    } + } Route::Exclude => { set_title("Exclude host"); html! { <>{navigation}
    } diff --git a/web_frontend/src/settings.rs b/web_frontend/src/settings.rs index dc9d653..0484d53 100644 --- a/web_frontend/src/settings.rs +++ b/web_frontend/src/settings.rs @@ -6,6 +6,7 @@ use crate::general::GeneralSettings; use crate::pac::PacSettingsPage; use crate::set_title; use crate::settings_textarea::SettingsTextarea; +use crate::statistics_settings::StatisticsSettingsPage; use crate::userscripts::UserScriptsPage; use yew::prelude::*; use yew::{html, Html}; @@ -29,6 +30,8 @@ pub enum SettingsRoute { Account, #[at("/settings/debug")] Debug, + #[at("/settings/statistics")] + Statistics, } pub fn switch_settings(route: SettingsRoute) -> Html { @@ -112,6 +115,10 @@ pub fn switch_settings(route: SettingsRoute) -> Html { html! { } } + SettingsRoute::Statistics => { + set_title("Settings - Statistics"); + html! { } + } }; html! {
    @@ -124,6 +131,7 @@ pub fn switch_settings(route: SettingsRoute) -> Html { classes={get_classes(route, SettingsRoute::Pac)} to={SettingsRoute::Pac}> { "PAC" }> classes={get_classes(route, SettingsRoute::Account)} to={SettingsRoute::Account}> { "Account" }> classes={get_classes(route, SettingsRoute::Debug)} to={SettingsRoute::Debug}> { "Debug" }> + classes={get_classes(route, SettingsRoute::Statistics)} to={SettingsRoute::Statistics}> { "Statistics" }>
    { content }
    diff --git a/web_frontend/src/statistics_settings.rs b/web_frontend/src/statistics_settings.rs new file mode 100644 index 0000000..7e36fd2 --- /dev/null +++ b/web_frontend/src/statistics_settings.rs @@ -0,0 +1,319 @@ +use crate::api; +use gloo_net::http::Request; +use serde::{Deserialize, Serialize}; +use wasm_bindgen_futures::spawn_local; +use web_sys::{HtmlInputElement, HtmlSelectElement, HtmlTextAreaElement}; +use yew::prelude::*; + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] +pub struct StatisticsConfig { + pub enabled: bool, + pub privacy: String, + pub retention_days: u16, + pub max_entries: u32, + #[serde(default)] + pub excluded_domains: Vec, +} + +#[derive(Clone, Debug, Default, Deserialize, PartialEq)] +pub struct HistoryStatus { + pub active: bool, + pub error: Option, + pub dropped_events: u64, + pub stored_requests: u64, + pub database_bytes: u64, +} + +#[derive(Clone, Deserialize)] +pub struct StatisticsSettings { + pub config: StatisticsConfig, + pub status: HistoryStatus, +} + +pub enum Message { + Load, + Loaded(Result), + Enabled(bool), + Privacy(String), + Days(String), + Limit(String), + Exclusions(String), + Save, + Saved(Result), + ConfirmClear(bool), + Clear, + Cleared(Result<(), String>), +} + +pub struct StatisticsSettingsPage { + settings: Option, + days: String, + limit: String, + exclusions: String, + busy: bool, + confirm_clear: bool, + error: Option, + notice: Option, +} + +impl Component for StatisticsSettingsPage { + type Message = Message; + type Properties = (); + + fn create(ctx: &Context) -> Self { + ctx.link().send_message(Message::Load); + Self { + settings: None, + days: String::new(), + limit: String::new(), + exclusions: String::new(), + busy: false, + confirm_clear: false, + error: None, + notice: None, + } + } + + fn update(&mut self, ctx: &Context, message: Message) -> bool { + match message { + Message::Load => { + self.busy = true; + self.error = None; + let link = ctx.link().clone(); + spawn_local(async move { + link.send_message(Message::Loaded( + api::get_json("/api/settings/statistics").await, + )); + }); + } + Message::Loaded(result) | Message::Saved(result) => { + self.busy = false; + match result { + Ok(settings) => { + self.days = settings.config.retention_days.to_string(); + self.limit = settings.config.max_entries.to_string(); + self.exclusions = settings.config.excluded_domains.join("\n"); + self.error = settings.status.error.clone(); + self.settings = Some(settings); + } + Err(error) => { + self.error = Some(error); + self.notice = None; + } + } + } + Message::Enabled(value) => { + if self.busy { + return false; + } + if let Some(settings) = &mut self.settings { + settings.config.enabled = value; + } + self.notice = None; + } + Message::Privacy(value) => { + if self.busy { + return false; + } + if let Some(settings) = &mut self.settings { + settings.config.privacy = value; + } + self.notice = None; + } + Message::Days(value) => { + if self.busy { + return false; + } + self.days = value; + self.notice = None; + } + Message::Limit(value) => { + if self.busy { + return false; + } + self.limit = value; + self.notice = None; + } + Message::Exclusions(value) => { + if self.busy { + return false; + } + self.exclusions = value; + self.notice = None; + } + Message::Save => { + if self.busy { + return false; + } + let Some(settings) = &self.settings else { + return false; + }; + let (Ok(days @ 1..=365), Ok(limit @ 100..=1_000_000)) = + (self.days.parse::(), self.limit.parse::()) + else { + self.error = Some("Choose 1–365 days and 100–1,000,000 requests.".into()); + return true; + }; + let mut config = settings.config.clone(); + config.retention_days = days; + config.max_entries = limit; + config.excluded_domains = self + .exclusions + .lines() + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(|s| s.trim_end_matches('.').to_ascii_lowercase()) + .collect(); + self.busy = true; + self.error = None; + self.notice = None; + let link = ctx.link().clone(); + spawn_local(async move { + let result = async { + let response = Request::put("/api/settings/statistics") + .json(&config) + .map_err(|e| e.to_string())? + .send() + .await + .map_err(|e| e.to_string())?; + if !response.ok() { + return Err(api::response_error(response).await); + } + response + .json::() + .await + .map_err(|e| e.to_string()) + } + .await; + link.send_message(Message::Saved(result)); + }); + self.notice = Some("Settings saved".into()); + } + Message::ConfirmClear(value) => { + if self.busy { + return false; + } + self.confirm_clear = value; + } + Message::Clear => { + if self.busy || !self.confirm_clear { + return false; + } + self.busy = true; + self.error = None; + self.notice = None; + let link = ctx.link().clone(); + spawn_local(async move { + let result = async { + let response = Request::delete("/api/history") + .send() + .await + .map_err(|e| e.to_string())?; + if response.ok() { + Ok(()) + } else { + Err(api::response_error(response).await) + } + } + .await; + link.send_message(Message::Cleared(result)); + }); + } + Message::Cleared(result) => { + self.busy = false; + self.confirm_clear = false; + match result { + Ok(()) => { + self.notice = Some("Saved history deleted".into()); + if let Some(settings) = &mut self.settings { + settings.status.stored_requests = 0; + settings.status.database_bytes = 0; + settings.status.dropped_events = 0; + settings.status.error = None; + } + } + Err(error) => self.error = Some(error), + } + } + } + true + } + + fn view(&self, ctx: &Context) -> Html { + let Some(settings) = &self.settings else { + return html! { <> +

    {"Statistics"}

    + if let Some(error) = &self.error { + + + } else {

    {"Loading…"}

    } + }; + }; + let config = &settings.config; + let on_enabled = ctx.link().callback(|e: Event| { + Message::Enabled(e.target_unchecked_into::().checked()) + }); + let on_privacy = ctx.link().callback(|e: Event| { + Message::Privacy(e.target_unchecked_into::().value()) + }); + let on_days = ctx.link().callback(|e: InputEvent| { + Message::Days(e.target_unchecked_into::().value()) + }); + let on_limit = ctx.link().callback(|e: InputEvent| { + Message::Limit(e.target_unchecked_into::().value()) + }); + let on_exclusions = ctx.link().callback(|e: InputEvent| { + Message::Exclusions(e.target_unchecked_into::().value()) + }); + html! { <> +

    {"Statistics"}

    +

    {"History is off by default. When enabled, it stays on this server and survives restarts. It records request times, hostnames, client IPs, and blocking decisions according to your privacy level. Full URLs, query strings, headers, and bodies are never saved in statistics."}

    + if let Some(error) = &self.error { } +
    +
    +
    + +
    +
    +

    {"Disabling stops recording. Previously saved history remains until you delete it or re-enable recording and retention removes it."}

    +
    +
    +
    + + +
    +

    {"Hidden identifiers are removed before collection. While enabled, increasing privacy also removes those identifiers from saved history. They cannot be recovered by lowering privacy. Anonymous mode keeps only session counters and does not record history."}

    +
    +
    + + +
    +
    + + +

    {"The oldest requests are removed when either limit is reached. Reducing a limit deletes older history when saved."}

    +
    + +