From c51a1cf8884847691b44278b014c24364275aafb Mon Sep 17 00:00:00 2001 From: Mathew Beane Date: Fri, 2 Oct 2026 22:12:45 -0400 Subject: [PATCH] fix(sensing-server): apply --privacy-mode to REST, WebSocket and recordings --privacy-mode was declared in the MQTT arg group and read only by the MQTT publisher, so /api/v1/sensing/latest, /api/v1/pose/current, the /ws/sensing and /api/v1/stream/pose streams and recordings kept serving pose keypoints (and heart/breathing rate when vitals are published). - Move --privacy-mode (env RUVIEW_PRIVACY_MODE) out of MqttArgs to a top-level server flag. Spelling and env var are unchanged. The MQTT config takes its value from the server flag. - Add privacy_filter: one key filter for vital_signs, HR/BR leaves, enhanced_breathing, keypoints, pose_keypoints and joints_m. - REST: middleware strips those keys from every JSON response on the HTTP and WS ports. Unparseable JSON is withheld with a 500. - WebSocket: /ws/sensing frames and /api/v1/stream/pose messages are filtered; the pose physics assessment is withheld. - /api/v1/pose/current?view=both|refined returns 403 privacy_mode. - Recordings write filtered frames. - Default (flag off) behaviour is unchanged; the middleware is a no-op. Fixes #2094 Co-Authored-By: claude-flow --- docs/integrations/home-assistant.md | 3 +- docs/user-guide.md | 9 +- .../examples/mqtt_publisher.rs | 12 +- .../wifi-densepose-sensing-server/src/cli.rs | 22 +- .../wifi-densepose-sensing-server/src/lib.rs | 2 + .../wifi-densepose-sensing-server/src/main.rs | 259 +++++++++++++++++- .../src/mqtt/config.rs | 4 +- .../src/privacy_filter.rs | 259 ++++++++++++++++++ 8 files changed, 543 insertions(+), 27 deletions(-) create mode 100644 v2/crates/wifi-densepose-sensing-server/src/privacy_filter.rs diff --git a/docs/integrations/home-assistant.md b/docs/integrations/home-assistant.md index 8498a36f9..d678f42c3 100644 --- a/docs/integrations/home-assistant.md +++ b/docs/integrations/home-assistant.md @@ -128,7 +128,7 @@ Per ADR-115 §3.11.1, the Matter Bridge exposes a subset on standard clusters so | `--mqtt-rate-rssi ` | 0.1 | RSSI publish rate (Hz) | | `--mqtt-publish-pose` | off | Enable pose-keypoint publication | | `--mqtt-rate-pose ` | 1.0 | Pose publish rate when enabled | -| `--privacy-mode` | off | Strip HR/BR/pose from MQTT and Matter | +| `--privacy-mode` | off | Strip HR/BR/pose from MQTT, Matter, REST, WebSocket and recordings | | `--matter` | off | Enable the HA-FABRIC Matter Bridge | | `--matter-setup-file ` | — | Where to write the QR + manual code | | `--matter-reset` | off | Wipe fabric credentials and re-commission | @@ -172,6 +172,7 @@ When deploying in **healthcare**, **AAL (aging-in-place)**, or **commercial** se - **Strips** heart rate, breathing rate, and pose keypoints from every outbound MQTT publication. - **Suppresses discovery** for those entities entirely — HA never even sees they exist. +- **Applies to the rest of the server too.** The same fields are removed from REST responses, WebSocket frames and recordings, so another client on the HTTP or WebSocket port can't read what MQTT withholds. See [Privacy mode](../user-guide.md#privacy-mode-for-healthcare--aal) in the user guide. - **Keeps every semantic primitive enabled.** Sleeping / distress / room-active / etc are *inferred* states. The inference happens server-side and only the boolean or score crosses the wire. This is the architectural win that makes the platform deployable in regulated contexts. Always pair `--privacy-mode` with `--mqtt-tls` on non-localhost brokers. diff --git a/docs/user-guide.md b/docs/user-guide.md index 1d79d9416..a3d8a8d0b 100644 --- a/docs/user-guide.md +++ b/docs/user-guide.md @@ -843,7 +843,14 @@ Full design + operator guide: [`docs/integrations/home-assistant.md`](integratio sensing-server --mqtt --mqtt-host --mqtt-tls --privacy-mode ``` -`--privacy-mode` strips heart rate, breathing rate, and pose keypoints from MQTT **and** Matter — they never reach the wire. Semantic primitives stay published because they're inferred *states* server-side, not biometric *values*. This is the architectural win that makes ADR-115 healthcare- and enterprise-deployable. +`--privacy-mode` (env `RUVIEW_PRIVACY_MODE`) is a server-wide flag. It withholds heart rate, breathing rate, and pose keypoints from every output: + +- REST responses: the fields are removed from every JSON body (`/api/v1/sensing/latest`, `/api/v1/vital-signs`, `/api/v1/edge-vitals`, `/api/v1/pose/current`, ...). `/api/v1/pose/current?view=both|refined` returns `403` with `"code": "privacy_mode"`. +- WebSocket streams: `/ws/sensing` and `/api/v1/stream/pose` frames are filtered the same way. +- Recordings started with `POST /api/v1/recording/start` are written already filtered. +- MQTT and Matter: those entities are neither announced nor published. + +Presence, motion, person count, zones and the coarse posture label are still served, and semantic primitives stay published because they're inferred *states*, not biometric *values*. Some of those states (sleeping, possible distress) are derived from vitals, so they are still health-related information. Privacy mode reduces what leaves the server. It does not by itself make a deployment compliant with any regulation. ### Matter Bridge (Apple Home / Google Home / Alexa / SmartThings) diff --git a/v2/crates/wifi-densepose-sensing-server/examples/mqtt_publisher.rs b/v2/crates/wifi-densepose-sensing-server/examples/mqtt_publisher.rs index b1a9ee0dd..b317d0f47 100644 --- a/v2/crates/wifi-densepose-sensing-server/examples/mqtt_publisher.rs +++ b/v2/crates/wifi-densepose-sensing-server/examples/mqtt_publisher.rs @@ -61,14 +61,18 @@ use wifi_densepose_sensing_server::mqtt::{ async fn main() -> Result<(), Box> { tracing_subscriber::fmt::init(); - let args = { + let (args, privacy_mode) = { use clap::Parser; #[derive(Parser)] struct W { #[command(flatten)] m: MqttArgs, + /// Strip biometrics (HR/BR/pose) from MQTT + #[arg(long, env = "RUVIEW_PRIVACY_MODE")] + privacy_mode: bool, } - W::parse().m + let w = W::parse(); + (w.m, w.privacy_mode) }; if !args.mqtt { @@ -78,7 +82,9 @@ async fn main() -> Result<(), Box> { // 1. Build MqttConfig from CLI + run the security audit before any // network I/O. A failed audit short-circuits with a clear error. - let cfg = Arc::new(MqttConfig::from_args(&args)); + let mut cfg = MqttConfig::from_args(&args); + cfg.privacy_mode = privacy_mode; + let cfg = Arc::new(cfg); match audit(&cfg) { Ok(()) => {} Err(e) if !e.is_fatal() => { diff --git a/v2/crates/wifi-densepose-sensing-server/src/cli.rs b/v2/crates/wifi-densepose-sensing-server/src/cli.rs index 80e4356bf..0344c4586 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/cli.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/cli.rs @@ -3,10 +3,11 @@ use clap::Parser; use std::path::PathBuf; -/// MQTT publisher (HA auto-discovery) + privacy-mode flags, shared via -/// `#[command(flatten)]` by both `cli::Args` and the binary's `main::Args` -/// so the `--mqtt*` flags reach the actual `Args::parse()` the server uses -/// (the publisher in `mqtt::` is keyed off this group). ADR-115 §3.8/§3.10. +/// MQTT publisher (HA auto-discovery) flags, flattened into the binary's +/// `main::Args` so the `--mqtt*` flags reach the actual `Args::parse()` the +/// server uses (the publisher in `mqtt::` is keyed off this group). ADR-115 +/// §3.8. `--privacy-mode` is a top-level server flag (#2094), not part of +/// this group. #[derive(clap::Args, Debug, Clone)] pub struct MqttArgs { /// Enable MQTT publisher with HA auto-discovery @@ -80,10 +81,6 @@ pub struct MqttArgs { /// Pose publish rate (Hz) when --mqtt-publish-pose is set #[arg(long, default_value = "1.0")] pub mqtt_rate_pose: f64, - - /// Strip biometrics (HR/BR/pose) before any MQTT/Matter publish (ADR-115 §3.10). - #[arg(long, env = "RUVIEW_PRIVACY_MODE")] - pub privacy_mode: bool, } /// CLI arguments for the sensing server. @@ -264,11 +261,10 @@ pub struct Args { #[arg(long, default_value = "1.0")] pub mqtt_rate_pose: f64, - // ─── ADR-115 §3.10 — Privacy mode ────────────────────────────────────── - /// Strip biometrics (HR/BR/pose) before any MQTT or Matter publish. - /// Discovery for those entities is suppressed entirely — the controller - /// never sees them exist. Implements the ADR-106 primitive-isolation - /// contract at the integration boundary. + // ─── ADR-115 §3.10 / #2094 — Privacy mode ────────────────────────────── + /// Strip biometrics (heart rate, breathing rate, pose keypoints) from + /// every output: REST, WebSocket, recordings, MQTT and Matter. MQTT + /// discovery for those entities is suppressed entirely. #[arg(long, env = "RUVIEW_PRIVACY_MODE")] pub privacy_mode: bool, diff --git a/v2/crates/wifi-densepose-sensing-server/src/lib.rs b/v2/crates/wifi-densepose-sensing-server/src/lib.rs index 0cf0214f1..5d8d9e916 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/lib.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/lib.rs @@ -27,6 +27,8 @@ pub mod mqtt; pub mod path_safety; /// ADR-323: fail-closed pose physics integration and raw/refined view selection. pub mod pose_physics; +/// #2094: server-wide `--privacy-mode` filter for REST, WebSocket and recordings. +pub mod privacy_filter; /// ADR-295: canonical source-provenance state machine (synthetic can never /// present as live). pub mod provenance; diff --git a/v2/crates/wifi-densepose-sensing-server/src/main.rs b/v2/crates/wifi-densepose-sensing-server/src/main.rs index 180ae6f53..dafec8cdb 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/main.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/main.rs @@ -47,6 +47,7 @@ use wifi_densepose_sensing_server::bootstrap_baseline::{ }; // ADR-295 / ADR-297: canonical provenance state + per-node/room inference. use wifi_densepose_sensing_server::inference::{fuse_room, NodeInference, RoomInference}; +use wifi_densepose_sensing_server::privacy_filter; use wifi_densepose_sensing_server::provenance::SourceState; use ruvector_mincut::{DynamicMinCut, MinCutBuilder}; @@ -178,7 +179,14 @@ struct Args { #[arg(long)] disable_host_validation: bool, - /// MQTT publisher (HA auto-discovery) + privacy-mode flags (ADR-115). + /// Strip biometrics (heart rate, breathing rate, pose keypoints) from + /// every output: REST responses, WebSocket streams, recordings, MQTT and + /// Matter (#2094, ADR-115 §3.10). Presence, motion and person count are + /// still served. + #[arg(long, env = "RUVIEW_PRIVACY_MODE")] + privacy_mode: bool, + + /// MQTT publisher (HA auto-discovery) flags (ADR-115). /// Flattened so `--mqtt*` reach the binary's parser and the publisher /// in `mqtt::` is actually started (fixes #872). Uses the *lib* crate's /// `MqttArgs` type so it's compatible with `mqtt::config::from_args`. @@ -1941,6 +1949,9 @@ struct AppStateInner { recording_current_id: Option, /// Shutdown signal for the recording writer task. recording_stop_tx: Option>, + /// `--privacy-mode` (#2094): WebSocket frames and recordings drop + /// biometric fields; REST is filtered by middleware. + privacy_mode: bool, // ── Training fields (ADR-186 TRAIN-RECONNECT) ──────────────────────────── /// Live training state (shared status snapshot + cooperative cancel flag + /// background task handle) for the in-server trainer in `training_api`. @@ -2118,6 +2129,182 @@ mod adr323_pose_physics_http_tests { } } +/// #2094: `--privacy-mode` is a top-level flag and filters REST and WebSocket +/// output, not only MQTT. +#[cfg(test)] +mod privacy_mode_surface_tests { + use super::*; + use axum::{body::Body, http::Request}; + use tower::ServiceExt; + + fn update_with_biometrics() -> SensingUpdate { + serde_json::from_value(serde_json::json!({ + "type": "sensing_update", + "timestamp": 1.0, + "source": "simulated", + "tick": 7, + "nodes": [], + "features": { "mean_rssi": -50.0, "variance": 1.0, "motion_band_power": 0.5, + "breathing_band_power": 0.2, "dominant_freq_hz": 0.25, + "change_points": 0, "spectral_power": 1.0 }, + "classification": { "motion_level": "present_still", "presence": true, "confidence": 0.9 }, + "signal_field": { "grid_size": [1, 1, 1], "values": [0.0] }, + "vital_signs": { "breathing_rate_bpm": 14.0, "heart_rate_bpm": 62.0, + "breathing_confidence": 0.8, "heartbeat_confidence": 0.7, + "signal_quality": 0.9 }, + "pose_keypoints": [[0.1, 0.2, 0.0, 0.9]], + "persons": [{ "id": 1, "confidence": 0.9, "zone": "zone_1", + "bbox": { "x": 0.0, "y": 0.0, "width": 1.0, "height": 1.0 }, + "keypoints": [{ "name": "nose", "x": 1.0, "y": 2.0, "z": 0.0, "confidence": 0.9 }] }] + })) + .expect("test SensingUpdate") + } + + fn state(privacy_mode: bool) -> SharedState { + let mut inner = AppStateInner::minimal(); + inner.privacy_mode = privacy_mode; + inner.latest_update = Some(update_with_biometrics()); + Arc::new(RwLock::new(inner)) + } + + fn rest_app(privacy_mode: bool) -> Router { + Router::new() + .route("/api/v1/sensing/latest", get(latest)) + .route("/api/v1/vital-signs", get(vital_signs_endpoint)) + .route("/api/v1/pose/current", get(pose_current)) + .with_state(state(privacy_mode)) + .layer(axum::middleware::from_fn_with_state( + privacy_filter::PrivacyFilter::new(privacy_mode), + privacy_filter::redact_json_responses, + )) + } + + async fn get_json(app: Router, uri: &str) -> (StatusCode, serde_json::Value) { + let response = app + .oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap()) + .await + .unwrap(); + let status = response.status(); + let body = axum::body::to_bytes(response.into_body(), 1 << 20) + .await + .unwrap(); + (status, serde_json::from_slice(&body).unwrap()) + } + + #[test] + fn privacy_flag_is_top_level_and_defaults_off() { + assert!(!Args::parse_from(["sensing-server"]).privacy_mode); + let args = Args::parse_from(["sensing-server", "--privacy-mode"]); + assert!(args.privacy_mode); + assert!(!args.mqtt_opts.mqtt, "--privacy-mode must not imply --mqtt"); + } + + #[tokio::test] + async fn sensing_latest_keeps_biometrics_when_privacy_off() { + let (_, body) = get_json(rest_app(false), "/api/v1/sensing/latest").await; + assert!(body.get("vital_signs").is_some()); + assert!(body.get("pose_keypoints").is_some()); + assert!(body["persons"][0].get("keypoints").is_some()); + } + + #[tokio::test] + async fn sensing_latest_strips_biometrics_in_privacy_mode() { + let (status, body) = get_json(rest_app(true), "/api/v1/sensing/latest").await; + assert_eq!(status, StatusCode::OK); + assert!(body.get("vital_signs").is_none(), "{body}"); + assert!(body.get("pose_keypoints").is_none(), "{body}"); + assert!(body["persons"][0].get("keypoints").is_none(), "{body}"); + assert_eq!(body["classification"]["presence"], true); + assert_eq!(body["persons"][0]["zone"], "zone_1"); + } + + #[tokio::test] + async fn vital_signs_endpoint_strips_rates_in_privacy_mode() { + let (_, body) = get_json(rest_app(true), "/api/v1/vital-signs").await; + assert!(body.get("vital_signs").is_none(), "{body}"); + let (_, body) = get_json(rest_app(false), "/api/v1/vital-signs").await; + assert!(body["vital_signs"].get("heart_rate_bpm").is_some()); + } + + #[tokio::test] + async fn pose_current_strips_keypoints_and_refuses_skeletal_views() { + let (status, body) = get_json(rest_app(true), "/api/v1/pose/current").await; + assert_eq!(status, StatusCode::OK); + assert_eq!(body["total_persons"], 1); + assert!(body["persons"][0].get("keypoints").is_none(), "{body}"); + for view in ["both", "refined"] { + let (status, body) = + get_json(rest_app(true), &format!("/api/v1/pose/current?view={view}")).await; + assert_eq!(status, StatusCode::FORBIDDEN, "view={view}"); + assert_eq!(body["code"], "privacy_mode"); + } + } + + async fn first_ws_frame(app: Router, path: &str, tx: broadcast::Sender) -> String { + use futures_util::StreamExt; + use tokio_tungstenite::tungstenite::Message as TMsg; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + let _ = axum::serve(listener, app).await; + }); + let (mut ws, _) = tokio_tungstenite::connect_async(format!("ws://{addr}{path}")) + .await + .expect("WebSocket handshake"); + let frame = serde_json::to_string(&update_with_biometrics()).unwrap(); + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10); + // The handler subscribes after the upgrade, so keep sending until it hears one. + while tokio::time::Instant::now() < deadline { + let _ = tx.send(frame.clone()); + match tokio::time::timeout(std::time::Duration::from_millis(200), ws.next()).await { + Ok(Some(Ok(TMsg::Text(text)))) => { + let v: serde_json::Value = serde_json::from_str(&text).unwrap(); + if v["type"] == "sensing_update" || v["type"] == "pose_data" { + return text; + } + } + Ok(Some(Ok(_))) | Err(_) => {} + Ok(Some(Err(e))) => panic!("ws error: {e}"), + Ok(None) => panic!("ws closed"), + } + } + panic!("no frame on {path}"); + } + + async fn ws_frames(privacy_mode: bool) -> (String, String) { + let shared = state(privacy_mode); + let tx = shared.read().await.tx.clone(); + let app = Router::new() + .route("/ws/sensing", get(ws_sensing_handler)) + .route("/api/v1/stream/pose", get(ws_pose_handler)) + .with_state(shared); + let sensing = first_ws_frame(app.clone(), "/ws/sensing", tx.clone()).await; + let pose = first_ws_frame(app, "/api/v1/stream/pose", tx).await; + (sensing, pose) + } + + #[tokio::test] + async fn websocket_streams_strip_biometrics_in_privacy_mode() { + let (sensing, pose) = ws_frames(true).await; + for text in [&sensing, &pose] { + assert!(!text.contains("heart_rate_bpm"), "{text}"); + assert!(!text.contains("breathing_rate_bpm"), "{text}"); + assert!(!text.contains("keypoints"), "{text}"); + } + assert!(sensing.contains("\"presence\":true"), "{sensing}"); + assert!(pose.contains("\"persons\""), "{pose}"); + } + + #[tokio::test] + async fn websocket_streams_unchanged_when_privacy_off() { + let (sensing, pose) = ws_frames(false).await; + assert!(sensing.contains("heart_rate_bpm"), "{sensing}"); + assert!(sensing.contains("keypoints"), "{sensing}"); + assert!(pose.contains("keypoints"), "{pose}"); + } +} + /// If no ESP32 frame arrives within this duration, source reverts to offline. const ESP32_OFFLINE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); @@ -2609,6 +2796,7 @@ impl AppStateInner { recording_start_time: None, recording_current_id: None, recording_stop_tx: None, + privacy_mode: false, training_state: training_api::TrainingState::default(), training_progress_tx: broadcast::channel::(256).0, adaptive_model: None, @@ -4838,9 +5026,9 @@ async fn ws_sensing_handler( } async fn handle_ws_client(mut socket: WebSocket, state: SharedState) { - let mut rx = { + let (mut rx, privacy_mode) = { let s = state.read().await; - s.tx.subscribe() + (s.tx.subscribe(), s.privacy_mode) }; info!("WebSocket client connected (sensing)"); @@ -4854,6 +5042,14 @@ async fn handle_ws_client(mut socket: WebSocket, state: SharedState) { msg = rx.recv() => { match msg { Ok(json) => { + let json = if privacy_mode { + match privacy_filter::redact_json_str(&json) { + Some(json) => json, + None => continue, + } + } else { + json + }; if socket.send(Message::Text(json)).await.is_err() { break; } @@ -4947,9 +5143,9 @@ async fn ws_pose_handler( } async fn handle_ws_pose_client(mut socket: WebSocket, state: SharedState) { - let mut rx = { + let (mut rx, privacy_mode) = { let s = state.read().await; - s.tx.subscribe() + (s.tx.subscribe(), s.privacy_mode) }; info!("WebSocket client connected (pose)"); @@ -4974,7 +5170,10 @@ async fn handle_ws_pose_client(mut socket: WebSocket, state: SharedState) { // "signal_derived" — keypoints estimated from raw CSI features. let (model_loaded, physics_assessment) = { let s = state.read().await; - let physics = s.pose_physics.latest().and_then(|(raw, result)| { + // The physics assessment describes the skeleton, + // so privacy mode withholds it with the keypoints. + let latest = s.pose_physics.latest().filter(|_| !privacy_mode); + let physics = latest.and_then(|(raw, result)| { (raw.sequence == sensing.tick).then(|| { serde_json::json!({ "schema": "pose-refinement-v1", @@ -5032,7 +5231,7 @@ async fn handle_ws_pose_client(mut socket: WebSocket, state: SharedState) { sensing.persons.clone().unwrap_or_else(|| derive_pose_from_sensing(&sensing)) }; - let pose_msg = serde_json::json!({ + let mut pose_msg = serde_json::json!({ "type": "pose_data", "zone_id": "zone_1", "timestamp": sensing.timestamp, @@ -5057,6 +5256,9 @@ async fn handle_ws_pose_client(mut socket: WebSocket, state: SharedState) { } } }); + if privacy_mode { + privacy_filter::redact_value(&mut pose_msg); + } if socket.send(Message::Text(pose_msg.to_string())).await.is_err() { break; } @@ -6640,6 +6842,17 @@ async fn pose_current( "total_persons": persons.len(), "source": s.effective_source(), }); + // #2094: the raw/refined views are the skeleton itself. The legacy view is + // still served; the REST privacy middleware strips its keypoints. + if s.privacy_mode && !matches!(query.view, pose_physics::PoseView::Raw) { + return ( + StatusCode::FORBIDDEN, + Json(serde_json::json!({ + "code": "privacy_mode", + "detail": "pose views are disabled while --privacy-mode is set" + })), + ); + } match query.view { pose_physics::PoseView::Raw => (StatusCode::OK, Json(legacy)), pose_physics::PoseView::Both => { @@ -6985,6 +7198,8 @@ async fn start_recording( // Subscribe to the broadcast channel to capture CSI frames let mut rx = s.tx.subscribe(); + // #2094: biometric fields never reach disk in privacy mode. + let privacy_mode = s.privacy_mode; // Add initial recording entry s.recordings.push(serde_json::json!({ @@ -7008,6 +7223,14 @@ async fn start_recording( result = rx.recv() => { match result { Ok(frame_json) => { + let frame_json = if privacy_mode { + match privacy_filter::redact_json_str(&frame_json) { + Some(json) => json, + None => continue, + } + } else { + frame_json + }; if writeln!(writer, "{}", frame_json).is_err() { warn!("Recording {rec_id}: write error, stopping"); break; @@ -12042,6 +12265,13 @@ async fn main() { // clients drop oldest, identical backpressure shape. let (intro_tx, _) = broadcast::channel::(256); + if args.privacy_mode { + info!( + "Privacy mode ON: heart rate, breathing rate and pose keypoints are withheld from \ + REST, WebSocket, recordings and MQTT" + ); + } + // #872: actually start the MQTT publisher when `--mqtt` is set. The publisher // (mqtt::) consumes a typed VitalsSnapshot stream; we bridge the existing JSON // sensing broadcast into it with a defensive serde_json::Value mapping (absent @@ -12052,7 +12282,9 @@ async fn main() { #[cfg(feature = "mqtt")] { use wifi_densepose_sensing_server::mqtt; - let mcfg = std::sync::Arc::new(mqtt::config::MqttConfig::from_args(&args.mqtt_opts)); + let mut mcfg = mqtt::config::MqttConfig::from_args(&args.mqtt_opts); + mcfg.privacy_mode = args.privacy_mode; + let mcfg = std::sync::Arc::new(mcfg); match mcfg.validate() { Ok(()) => { let node_id = mcfg.client_id.clone(); @@ -12161,6 +12393,7 @@ async fn main() { recording_start_time: None, recording_current_id: None, recording_stop_tx: None, + privacy_mode: args.privacy_mode, // Training (ADR-186 TRAIN-RECONNECT) training_state: training_api::TrainingState::default(), training_progress_tx: broadcast::channel::(256).0, @@ -12426,6 +12659,10 @@ async fn main() { // so a client on :8765 can stream signed RuField FieldEvents alongside // `/ws/sensing`. Merged with its own FieldState (different state type). .merge(rufield_surface::router(field_surface.clone())) + .layer(axum::middleware::from_fn_with_state( + privacy_filter::PrivacyFilter::new(args.privacy_mode), + privacy_filter::redact_json_responses, + )) // ADR-272 FIX: this router had NO auth layer at all. `/ws/sensing` and // `/ws/field` on the dedicated WS port accepted unauthenticated // upgrades even with auth ON — and this is the port the UI actually @@ -12598,6 +12835,12 @@ async fn main() { // Merged AFTER `.with_state` (so http_app is already `Router<()>` and // can absorb the field router's own `FieldState`). .merge(rufield_surface::router(field_surface.clone())) + // #2094: strip biometrics from every JSON response in privacy mode + // (no-op otherwise). Applied after the merge so it covers every route. + .layer(axum::middleware::from_fn_with_state( + privacy_filter::PrivacyFilter::new(args.privacy_mode), + privacy_filter::redact_json_responses, + )) // Opt-in bearer auth (#443) + ADR-272 WebSocket gating. // // Applied AFTER the merge, and that ordering is load-bearing: axum diff --git a/v2/crates/wifi-densepose-sensing-server/src/mqtt/config.rs b/v2/crates/wifi-densepose-sensing-server/src/mqtt/config.rs index a430125b4..63887e209 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/mqtt/config.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/mqtt/config.rs @@ -92,7 +92,9 @@ impl MqttConfig { pose_hz: args.mqtt_rate_pose, }, publish_pose: args.mqtt_publish_pose, - privacy_mode: args.privacy_mode, + // `--privacy-mode` is a server-wide flag (#2094); the caller sets + // this from it after building the config. + privacy_mode: false, } } diff --git a/v2/crates/wifi-densepose-sensing-server/src/privacy_filter.rs b/v2/crates/wifi-densepose-sensing-server/src/privacy_filter.rs new file mode 100644 index 000000000..ad51370db --- /dev/null +++ b/v2/crates/wifi-densepose-sensing-server/src/privacy_filter.rs @@ -0,0 +1,259 @@ +//! Server-wide `--privacy-mode` output filter (#2094). +//! +//! `--privacy-mode` used to be read only by the MQTT publisher, so REST, the +//! WebSocket streams and recordings kept serving heart rate, breathing rate +//! and pose keypoints. This module is the one filter every other output +//! surface applies when the flag is set: +//! +//! - REST: [`redact_json_responses`] rewrites every `application/json` +//! response body. +//! - WebSocket and recordings: the server calls [`redact_json_str`] on each +//! broadcast frame before it is sent or written. +//! +//! The suppressed set matches the MQTT filter (`mqtt::privacy`): heart rate, +//! breathing rate and pose keypoints. Presence, motion, person count, zone +//! and the coarse posture label are kept, as they are on MQTT. Keys are +//! removed, not nulled, so a client can't tell a suppressed value from one +//! the server never had. +//! +//! The filter fails closed: a frame or JSON body that can't be parsed is +//! dropped (WebSocket, recording) or replaced with a 500 (REST) rather than +//! passed through unfiltered. + +use axum::{ + body::Body, + extract::{Request, State}, + http::{ + header::{CONTENT_LENGTH, CONTENT_TYPE}, + StatusCode, + }, + middleware::Next, + response::{IntoResponse, Response}, +}; + +/// JSON object keys removed, at any depth, from every output in privacy mode. +pub const BIOMETRIC_KEYS: &[&str] = &[ + // Vital signs: the `vital_signs` object carries HR/BR and their + // confidences; the leaf keys cover edge vitals, fused vitals and the + // mmWave block. + "vital_signs", + "breathing_rate_bpm", + "heart_rate_bpm", + "heartrate_bpm", + "hr_bpm", + "br_bpm", + "enhanced_breathing", + // Pose: per-person skeletons, model keypoints and refined joints. + "keypoints", + "pose_keypoints", + "joints_m", +]; + +/// Upper bound on a REST body the filter will buffer. +const MAX_FILTERED_BODY_BYTES: usize = 16 * 1024 * 1024; + +/// Whether privacy mode is on. Cheap to clone into router state. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct PrivacyFilter { + enabled: bool, +} + +impl PrivacyFilter { + pub fn new(enabled: bool) -> Self { + Self { enabled } + } + + pub fn is_enabled(&self) -> bool { + self.enabled + } +} + +/// Remove every [`BIOMETRIC_KEYS`] entry from `value`, recursively. +pub fn redact_value(value: &mut serde_json::Value) { + match value { + serde_json::Value::Object(map) => { + map.retain(|key, _| !BIOMETRIC_KEYS.contains(&key.as_str())); + for child in map.values_mut() { + redact_value(child); + } + } + serde_json::Value::Array(items) => { + for item in items { + redact_value(item); + } + } + _ => {} + } +} + +/// Redact one serialized JSON frame. `None` when the frame isn't valid JSON, +/// in which case the caller must drop it. +pub fn redact_json_str(json: &str) -> Option { + let mut value: serde_json::Value = serde_json::from_str(json).ok()?; + redact_value(&mut value); + serde_json::to_string(&value).ok() +} + +fn is_json(response: &Response) -> bool { + response + .headers() + .get(CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .is_some_and(|ct| ct.trim_start().starts_with("application/json")) +} + +/// Axum middleware: in privacy mode, strip [`BIOMETRIC_KEYS`] from every JSON +/// response body. A no-op when privacy mode is off. +pub async fn redact_json_responses( + State(filter): State, + request: Request, + next: Next, +) -> Response { + let response = next.run(request).await; + if !filter.is_enabled() || !is_json(&response) { + return response; + } + let (mut parts, body) = response.into_parts(); + let redacted = axum::body::to_bytes(body, MAX_FILTERED_BODY_BYTES) + .await + .ok() + .and_then(|bytes| { + let mut value: serde_json::Value = serde_json::from_slice(&bytes).ok()?; + redact_value(&mut value); + serde_json::to_vec(&value).ok() + }); + match redacted { + Some(bytes) => { + parts.headers.remove(CONTENT_LENGTH); + Response::from_parts(parts, Body::from(bytes)) + } + None => { + tracing::warn!("privacy mode: could not filter a JSON response; withholding it"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + axum::Json(serde_json::json!({ + "code": "privacy_filter_failed", + "detail": "response withheld because privacy mode could not filter it", + })), + ) + .into_response() + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::{routing::get, Router}; + use serde_json::json; + use tower::ServiceExt; + + fn sensing_update() -> serde_json::Value { + json!({ + "type": "sensing_update", + "classification": { "presence": true, "motion_level": "active", "confidence": 0.9 }, + "vital_signs": { "breathing_rate_bpm": 14.0, "heart_rate_bpm": 62.0, + "breathing_confidence": 0.8, "heartbeat_confidence": 0.7, + "signal_quality": 0.9 }, + "enhanced_breathing": { "rate_bpm": 14.1 }, + "pose_keypoints": [[0.1, 0.2, 0.0, 0.9]], + "posture": "standing", + "estimated_persons": 1, + "persons": [{ "id": 1, "confidence": 0.9, "zone": "zone_1", "pose": "standing", + "keypoints": [{ "name": "nose", "x": 1.0, "y": 2.0, "z": 0.0, "confidence": 0.9 }] }] + }) + } + + #[test] + fn redact_removes_vitals_and_pose_at_any_depth() { + let mut v = sensing_update(); + redact_value(&mut v); + assert!(v.get("vital_signs").is_none()); + assert!(v.get("enhanced_breathing").is_none()); + assert!(v.get("pose_keypoints").is_none()); + assert!(v["persons"][0].get("keypoints").is_none()); + // Non-biometric signals survive, as they do on MQTT. + assert_eq!(v["classification"]["presence"], true); + assert_eq!(v["estimated_persons"], 1); + assert_eq!(v["posture"], "standing"); + assert_eq!(v["persons"][0]["zone"], "zone_1"); + } + + #[test] + fn redact_strips_edge_and_fused_vitals_leaves() { + let mut v = json!({ + "type": "edge_fused_vitals", + "presence": true, + "breathing_rate_bpm": 15.0, + "heartrate_bpm": 72.0, + "mmwave": { "hr_bpm": 71.0, "br_bpm": 14.0, "present": true } + }); + redact_value(&mut v); + let text = v.to_string(); + for key in ["breathing_rate_bpm", "heartrate_bpm", "hr_bpm", "br_bpm"] { + assert!(!text.contains(key), "{key} leaked: {text}"); + } + assert_eq!(v["presence"], true); + assert_eq!(v["mmwave"]["present"], true); + } + + #[test] + fn redact_json_str_fails_closed_on_invalid_json() { + assert!(redact_json_str("not json").is_none()); + let out = redact_json_str(&sensing_update().to_string()).unwrap(); + assert!(!out.contains("heart_rate_bpm")); + assert!(!out.contains("keypoints")); + } + + fn app(enabled: bool) -> Router { + Router::new() + .route("/json", get(|| async { axum::Json(sensing_update()) })) + .route("/text", get(|| async { "heart_rate_bpm keypoints" })) + .route( + "/bad", + get(|| async { ([(CONTENT_TYPE, "application/json")], "{not json") }), + ) + .layer(axum::middleware::from_fn_with_state( + PrivacyFilter::new(enabled), + redact_json_responses, + )) + } + + async fn body(app: Router, path: &str) -> (StatusCode, String) { + let resp = app + .oneshot(Request::builder().uri(path).body(Body::empty()).unwrap()) + .await + .unwrap(); + let status = resp.status(); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + (status, String::from_utf8(bytes.to_vec()).unwrap()) + } + + #[tokio::test] + async fn middleware_strips_json_when_enabled() { + let (status, text) = body(app(true), "/json").await; + assert_eq!(status, StatusCode::OK); + assert!(!text.contains("heart_rate_bpm"), "{text}"); + assert!(!text.contains("keypoints"), "{text}"); + assert!(text.contains("\"presence\":true"), "{text}"); + } + + #[tokio::test] + async fn middleware_is_a_no_op_when_disabled() { + let (_, text) = body(app(false), "/json").await; + assert!(text.contains("heart_rate_bpm")); + assert!(text.contains("keypoints")); + } + + #[tokio::test] + async fn middleware_leaves_non_json_alone_and_withholds_unparseable_json() { + let (status, text) = body(app(true), "/text").await; + assert_eq!(status, StatusCode::OK); + assert_eq!(text, "heart_rate_bpm keypoints"); + let (status, text) = body(app(true), "/bad").await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR); + assert!(text.contains("privacy_filter_failed")); + } +}